ecms-engine.git

package com.zhitan.engine.service.impl;
 
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.zhitan.engine.config.InfluxDBConfig;
import com.zhitan.engine.influxdb.InfluxdbRepository;
import com.zhitan.engine.entity.*;
import com.zhitan.engine.repository.*;
import com.zhitan.engine.service.DataCleaningService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
 
import java.math.BigDecimal;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
 
import static java.util.List.*;
 
/**
 * 数据清洗服务实现类
 */
@Slf4j
@Service
public class DataCleaningServiceImpl implements DataCleaningService {
    private final String TAG = "tag";
    private final String FIELD_VALUE = "value";
    @Autowired
    private InfluxDBConfig influxDBConfig;
    @Autowired
    private InfluxdbRepository influxdbRepository;
 
    @Autowired
    private IndexStorageRepository indexStorageRepository;
 
    @Autowired
    private DataItemRepository dataItemRepository;
 
    @Autowired
    private ElectricityDataItemRepository electricityDataItemRepository;
 
    @Autowired
    private ElectricityPriceRepository electricityPriceRepository;
 
    @Autowired
    private ElectricityPriceDateRepository electricityPriceDateRepository;
 
    private static final Pattern CALC_TEXT_PATTERN = Pattern.compile("accumulate\\('([^']+)'\\)");
 
    @Override
    public void calculateHourlyElectricity(LocalDateTime dateTime) {
        calculateElectricityByTimeType("HOUR", dateTime);
    }
 
    @Override
    public void calculateDailyElectricity(LocalDateTime dateTime) {
        calculateElectricityByTimeType("DAY", dateTime);
    }
 
    @Override
    public void calculateMonthlyElectricity(LocalDateTime dateTime) {
        calculateElectricityByTimeType("MONTH", dateTime);
    }
 
    @Override
    public void calculateYearlyElectricity(LocalDateTime dateTime) {
        calculateElectricityByTimeType("YEAR", dateTime);
    }
 
    @Override
    @Transactional
    public void calculateElectricityByTimeType(String timeType, LocalDateTime dateTime) {
        log.info("开始{}统计，时间点：{}", timeType, dateTime);
        
        // 获取指定时间类型的统计规则
        List<IndexStorage> indexStorages = indexStorageRepository.findByTimeType(timeType);
        
        for (IndexStorage indexStorage : indexStorages) {
            try {
                // 解析计算表达式
                if ("CALC".equals(indexStorage.getCalcType()) && StringUtils.hasText(indexStorage.getCalcText())) {
                    processCalcRule(indexStorage, dateTime);
                }
                
                // 如果是按小时统计且需要计算尖峰平谷
                if ("HOUR".equals(timeType) && indexStorage.getIsPvCalc() != null && indexStorage.getIsPvCalc() == 1) {
                    calculatePeakValleyElectricity(indexStorage.getIndexId(), dateTime);
                }
            } catch (Exception e) {
                log.error("处理索引{}的{}统计失败：{}", indexStorage.getIndexId(), timeType, e.getMessage(), e);
            }
        }
        
        log.info("完成{}统计，时间点：{}", timeType, dateTime);
    }
 
    /**
     * 处理计算规则
     *
     * @param indexStorage 索引存储配置
     * @param dateTime     统计时间点
     */
    private void processCalcRule(IndexStorage indexStorage, LocalDateTime dateTime) {
        String calcText = indexStorage.getCalcText();
        String timeType = indexStorage.getTimeType();
        
        // 解析计算表达式：accumulate('00601925032100026643_5_ActiveZT')
        Matcher matcher = CALC_TEXT_PATTERN.matcher(calcText);
        if (matcher.find()) {
            String tag = matcher.group(1);
            
            // 计算时间范围
            LocalDateTime[] timeRange = calculateTimeRange(timeType, dateTime);
            LocalDateTime startTime = timeRange[0];
            LocalDateTime endTime = timeRange[1];
            
            // 生成时间编码
            String timeCode = generateTimeCode(timeType, dateTime);
            
            // 从InfluxDB查询数据并计算累计值
            Double accumulatedValue = influxdbRepository.calculateAccumulatedValue(tag, startTime, endTime);
            
            double finalValue = accumulatedValue;
            
            // 保存到数据库
            saveDataItem(indexStorage.getIndexId(), timeCode, startTime, endTime, finalValue, timeType);
        } else {
            log.warn("无法解析计算表达式：{}", calcText);
        }
    }
 
    /**
     * 保存数据项
     *
     * @param indexId    索引ID
     * @param timeCode   时间编码
     * @param beginTime  开始时间
     * @param endTime    结束时间
     * @param value      值
     * @param timeType   时间类型
     */
    private void saveDataItem(String indexId, String timeCode, LocalDateTime beginTime, LocalDateTime endTime,
                             double value, String timeType) {
        // 查询是否已存在
        DataItem existingItem = dataItemRepository.findByIndexIdAndTimeCode(indexId, timeCode);
        
        if (existingItem == null) {
            existingItem = new DataItem();
            existingItem.setIndexId(indexId);
            existingItem.setTimeCode(timeCode);
            existingItem.setCreateTime(LocalDateTime.now());
        }
        
        existingItem.setBeginTime(beginTime);
        existingItem.setEndTime(endTime);
        existingItem.setValue(value);
        existingItem.setQuality("GOOD");
        existingItem.setUpdateTime(LocalDateTime.now());
        existingItem.setTimeType(timeType);
        existingItem.setDataTime(beginTime);
        
        dataItemRepository.save(existingItem);
        log.info("保存数据项成功：indexId={}, timeCode={}, value={}", indexId, timeCode, value);
    }
 
    @Override
    public void calculateTotalElectricity(LocalDateTime dateTime) {
        log.info("开始执行全厂总电量统计，时间点：{}", dateTime);
        
        // 定义需要统计的电表网关列表
        List<String> oldGatewayList = List.of("00601925032100028432", "00601925032100026840");
        List<String> sevenGatewayList = List.of("00601925032100028685", "00601925032100027392", "00601925032100028050", "00601925032100028751","00601925032100026275","00601925032100028117","00601925032100028814","00601925032100028990");
        List<String> outGatewayList = List.of("00601925032100031307");
 
 
        // 定义需要统计的参数类型及其统计方式
        Map<String, String> parameterTypes = new java.util.HashMap<>(Map.of(
                "ActiveZT", "SUM" // 正向总有功电能 - 求和
                , "ActiveZN", "SUM" // 正向总无功电能 - 求和
                , "ActiveElectricity", "SUM" // 有功电能 - 求和
                , "NoActiveElectricity", "SUM" // 无功功率 - 求和
                , "ExpZN", "SUM" // 反向总无功电能 - 求和
                , "Exp", "SUM" // 反向有功电能 - 求和
                , "ActivePow", "SUM" // 正向有功功率 - 求和
                , "VoltageUAB", "AVG" // 总电压 - 平均值
                , "VoltageUBC", "AVG" // 总电压 - 平均值
                , "VoltageUCA", "AVG" // 总电压 - 平均值
 
        ));
        parameterTypes.put("CurrentA", "SUM");
        parameterTypes.put("CurrentB", "SUM");
        parameterTypes.put("CurrentC", "SUM");
        parameterTypes.put("Io", "SUM");
        parameterTypes.put("VoltageA", "AVG");
        parameterTypes.put("VoltageB", "AVG");
        parameterTypes.put("VoltageC", "AVG");
 
        // 保存总表数据
        String old = "old_";
        String seven = "seven_";
        String out = "out_";
        String total = "total_";
        // 遍历所有参数类型进行统计
        parameterTypes.forEach((paramType, calcType) -> {
            TotalCount totalCount = new TotalCount();
            statistics(paramType, calcType, oldGatewayList, old, totalCount);
            statistics(paramType, calcType, sevenGatewayList, seven, totalCount);
            statistics(paramType, calcType, outGatewayList, out, totalCount);
            clacAndSave(paramType, calcType,total,  totalCount.totalValue, totalCount.totalCount);
 
        });
        
        log.info("全厂总电量统计完成，时间点：{}", dateTime);
    }
 
    private static final int METER_COUNT_PER_GATEWAY = 8;
 
    private static class TotalCount {
        public int totalCount;
        public double totalValue;
        public TotalCount() {
            totalCount = 0;
            totalValue = 0;
        }
    }
 
private void statistics(String paramType, String calcType, List<String> gatewayCodes, String area, TotalCount totalCount) {
    try {
        double sumValue = 0;
        int meterCount = 0;
 
        // 遍历老楼车间所有网关下的电表
        for (String gatewayCode : gatewayCodes) {
            // 遍历每个网关下的8个电表
            for (int i = 1; i <= METER_COUNT_PER_GATEWAY; i++) {
                // 构建tag名称格式：网关编号_电表序号_参数类型
                // 示例：00601925032100028432_1_ActiveZT
                String tag = String.format("%s_%d_%s", gatewayCode, i, paramType);
                // 查询最后一次保存数据
                Double value = influxdbRepository.getLastPoint(influxDBConfig.getMeasurement(),TAG, tag);
                if (value != null) {
                    sumValue += value;
                    meterCount++;
                }
            }
        }
 
        totalCount.totalCount += meterCount;
        totalCount.totalValue += sumValue;
        clacAndSave(paramType, calcType, area, sumValue, meterCount);
 
    } catch (Exception e) {
        log.error("统计参数{}失败：{}", paramType, e.getMessage(), e);
    }
}
 
 
    private void clacAndSave(String paramType, String calcType, String area, double sumValue, int meterCount) {
        // 计算最终值
        double finalValue = "SUM".equals(calcType) ? sumValue : (meterCount > 0 ? sumValue / meterCount : 0);
        String sumTag = area + paramType;
        Point point = Point
                .measurement(influxDBConfig.getMeasurement())
                .addTag(TAG, sumTag)
                .addField(FIELD_VALUE, finalValue)
                .time(Instant.now(), WritePrecision.S);
 
        influxdbRepository.writePoint(point);
        log.info("统计完成：{}={} ({}个电表)", sumTag, finalValue, meterCount);
    }
 
    @Override
    @Transactional
    public void calculatePeakValleyElectricity(String indexId, LocalDateTime dateTime) {
        log.info("开始计算尖峰平谷用电量，索引ID：{}，时间点：{}", indexId, dateTime);
        
        try {
            // 查询当前生效的电价配置
            List<ElectricityPriceDate> priceDates = electricityPriceDateRepository.findEffectiveByDate(dateTime.toLocalDate());
            if (priceDates.isEmpty()) {
                log.warn("未找到{}生效的电价配置", dateTime.toLocalDate());
                return;
            }
            
            // 获取电价配置
            ElectricityPriceDate priceDate = priceDates.get(0);
            List<ElectricityPrice> prices = electricityPriceRepository.findByParentId(priceDate.getId());
            if (prices.isEmpty()) {
                log.warn("未找到电价配置，parentId={}", priceDate.getId());
                return;
            }
            
            // 按电价类型分组，如果存在重复的电价类型，保留最后一个配置
            Map<String, ElectricityPrice> priceMap = prices.stream()
                    .collect(Collectors.toMap(
                        ElectricityPrice::getType,
                        price -> price,
                        (existing, replacement) -> replacement
                    ));
            
            // 生成时间编码
            String timeCode = generateTimeCode("HOUR", dateTime);
            
            // 查询该小时的用电量
            DataItem dataItem = dataItemRepository.findByIndexIdAndTimeCode(indexId, timeCode);
            if (dataItem == null || dataItem.getValue() == null) {
                log.warn("未找到用电量数据，indexId={}，timeCode={}", indexId, timeCode);
                return;
            }
            
            // 确定当前小时属于哪个电价类型
            LocalTime currentTime = dateTime.toLocalTime();
            String electricityType = determineElectricityType(currentTime, prices);
            if (electricityType == null) {
                log.warn("无法确定电价类型，时间：{}", currentTime);
                return;
            }
            
            // 获取电价
            ElectricityPrice price = priceMap.get(electricityType);
            // 注意：effecticityPrice是实体类中的属性名，虽然拼写看起来不太规范
            BigDecimal priceValue = null;
            try {
                priceValue = price.getEffecticityPrice();
                if (priceValue == null) {
                    priceValue = BigDecimal.ZERO;
                    log.warn("电价值为空，使用默认值0，电价类型：{}", electricityType);
                }
            } catch (Exception e) {
                priceValue = BigDecimal.ZERO;
                log.error("获取电价失败，使用默认值0，电价类型：{}，错误：{}", electricityType, e.getMessage(), e);
            }
            
            // 计算电费
            BigDecimal electricity = BigDecimal.valueOf(dataItem.getValue());
            BigDecimal cost = electricity.multiply(priceValue);
            
            // 保存尖峰平谷数据
            saveElectricityDataItem(indexId, timeCode, electricityType, dataItem.getDataTime(), 
                    dataItem.getBeginTime(), dataItem.getEndTime(), electricity, cost, priceValue);
            
            // 同时更新日、月、年的尖峰平谷统计数据
          //  updatePeakValleyStatistics(indexId, dateTime, electricityType, electricity, cost);
            
            log.info("计算尖峰平谷用电量完成，索引ID：{}，时间点：{}，类型：{}，用电量：{}，电费：{}", 
                    indexId, dateTime, electricityType, electricity, cost);
        } catch (Exception e) {
            log.error("计算尖峰平谷用电量失败：{}", e.getMessage(), e);
        }
    }
 
    /**
     * 确定电价类型
     *
     * @param time     当前时间
     * @param prices 电价配置映射
     * @return 电价类型
     */
    private String determineElectricityType(LocalTime time, List<ElectricityPrice> prices) {
        for (ElectricityPrice price : prices) {
            if (isTimeInRange(time, price.getStartTime(), price.getStopTime())) {
                return price.getType();
            }
        }
        return null;
    }
 
    /**
     * 判断时间是否在范围内
     *
     * @param time      当前时间
     * @param startTime 开始时间
     * @param stopTime  结束时间
     * @return 是否在范围内
     */
    private boolean isTimeInRange(LocalTime time, LocalTime startTime, LocalTime stopTime) {
        // 处理跨天的情况
        if (stopTime.isBefore(startTime)) {
            return !time.isBefore(startTime) || !time.isAfter(stopTime);
        } else {
            return !time.isBefore(startTime) && !time.isAfter(stopTime);
        }
    }
 
    /**
     * 保存尖峰平谷数据
     *
     * @param indexId         索引ID
     * @param timeCode        时间编码
     * @param electricityType 用电类型
     * @param dataTime        数据时间
     * @param beginTime       开始时间
     * @param endTime         结束时间
     * @param electricity     用电量
     * @param cost            电费
     * @param price           电价
     */
    @Autowired
    private EnergyIndexRepository energyIndexRepository;
 
    private void saveElectricityDataItem(String indexId, String timeCode, String electricityType, 
                                        LocalDateTime dataTime, LocalDateTime beginTime, LocalDateTime endTime, 
                                        BigDecimal electricity, BigDecimal cost, BigDecimal price) {
        // 查询是否已存在
        ElectricityDataItem existingItem = electricityDataItemRepository
                .findByIndexIdAndTimeCodeAndElectricityType(indexId, timeCode, electricityType);
        
        // 查询指标code
        String indexCode = null;
        try {
            EnergyIndex energyIndex = energyIndexRepository.findByIndexId(indexId);
            if (energyIndex != null) {
                indexCode = energyIndex.getCode();
            }
        } catch (Exception e) {
            log.error("查询指标code失败：{}", e.getMessage(), e);
        }
        
        if (existingItem == null) {
            existingItem = new ElectricityDataItem();
            existingItem.setIndexId(indexId);
            existingItem.setTimeCode(timeCode);
            existingItem.setElectricityType(electricityType);
            existingItem.setCreateTime(LocalDateTime.now());
        }
        
        // 设置指标code
        existingItem.setIndexCode(indexCode);
        
        existingItem.setDataTime(dataTime);
        existingItem.setBeginTime(beginTime);
        existingItem.setEndTime(endTime);
        existingItem.setElectricity(electricity);
        existingItem.setCost(cost);
        existingItem.setPrice(price);
        existingItem.setTimeType("HOUR");
        existingItem.setUpdateTime(LocalDateTime.now());
        
        electricityDataItemRepository.save(existingItem);
    }
 
    /**
     * 更新日、月、年的尖峰平谷统计数据
     *
     * @param indexId         索引ID
     * @param dateTime        数据时间
     * @param electricityType 用电类型
     * @param electricity     用电量
     * @param cost            电费
     */
    private void updatePeakValleyStatistics(String indexId, LocalDateTime dateTime, 
                                          String electricityType, BigDecimal electricity, BigDecimal cost) {
        try {
            // 更新日统计
            updateDailyPeakValleyStatistics(indexId, dateTime, electricityType, electricity, cost);
            
            // 更新月统计
            updateMonthlyPeakValleyStatistics(indexId, dateTime, electricityType, electricity, cost);
            
            // 更新年统计
            updateYearlyPeakValleyStatistics(indexId, dateTime, electricityType, electricity, cost);
        } catch (Exception e) {
            log.error("更新尖峰平谷统计数据失败：{}", e.getMessage(), e);
        }
    }
    
    /**
     * 更新日尖峰平谷统计数据
     */
    private void updateDailyPeakValleyStatistics(String indexId, LocalDateTime dateTime, 
                                              String electricityType, BigDecimal electricity, BigDecimal cost) {
        // 生成日时间编码
        String timeCode = generateTimeCode("DAY", dateTime);
        
        // 计算时间范围
        LocalDateTime[] timeRange = calculateTimeRange("DAY", dateTime);
        
        // 更新或创建统计数据
        updateElectricityDataItem(indexId, timeCode, electricityType, dateTime, 
                                timeRange[0], timeRange[1], electricity, cost, "DAY");
    }
    
    /**
     * 更新月尖峰平谷统计数据
     */
    private void updateMonthlyPeakValleyStatistics(String indexId, LocalDateTime dateTime, 
                                                String electricityType, BigDecimal electricity, BigDecimal cost) {
        // 生成月时间编码
        String timeCode = generateTimeCode("MONTH", dateTime);
        
        // 计算时间范围
        LocalDateTime[] timeRange = calculateTimeRange("MONTH", dateTime);
        
        // 更新或创建统计数据
        updateElectricityDataItem(indexId, timeCode, electricityType, dateTime, 
                                timeRange[0], timeRange[1], electricity, cost, "MONTH");
    }
    
    /**
     * 更新年尖峰平谷统计数据
     */
    private void updateYearlyPeakValleyStatistics(String indexId, LocalDateTime dateTime, 
                                               String electricityType, BigDecimal electricity, BigDecimal cost) {
        // 生成年时间编码
        String timeCode = generateTimeCode("YEAR", dateTime);
        
        // 计算时间范围
        LocalDateTime[] timeRange = calculateTimeRange("YEAR", dateTime);
        
        // 更新或创建统计数据
        updateElectricityDataItem(indexId, timeCode, electricityType, dateTime, 
                                timeRange[0], timeRange[1], electricity, cost, "YEAR");
    }
    
    /**
     * 更新或创建尖峰平谷统计数据
     */
    private void updateElectricityDataItem(String indexId, String timeCode, String electricityType, 
                                         LocalDateTime dataTime, LocalDateTime beginTime, LocalDateTime endTime, 
                                         BigDecimal electricity, BigDecimal cost, String timeType) {
        // 查询是否已存在
        ElectricityDataItem existingItem = electricityDataItemRepository
                .findByIndexIdAndTimeCodeAndElectricityType(indexId, timeCode, electricityType);
        
        if (existingItem == null) {
            // 创建新记录
            existingItem = new ElectricityDataItem();
            existingItem.setIndexId(indexId);
            existingItem.setTimeCode(timeCode);
            existingItem.setElectricityType(electricityType);
            existingItem.setCreateTime(LocalDateTime.now());
            existingItem.setElectricity(electricity);
            existingItem.setCost(cost);
            // 避免除零异常
            if (electricity.compareTo(BigDecimal.ZERO) > 0) {
                existingItem.setPrice(cost.divide(electricity, 2, BigDecimal.ROUND_HALF_UP));
            } else {
                existingItem.setPrice(BigDecimal.ZERO);
                log.warn("用电量为零，无法计算电价，设置为默认值0，索引ID={}，时间编码={}，类型={}", indexId, timeCode, electricityType);
            }
        } else {
            // 更新已有记录，累加电量和电费
            existingItem.setElectricity(existingItem.getElectricity().add(electricity));
            existingItem.setCost(existingItem.getCost().add(cost));
            // 重新计算平均电价
            if (existingItem.getElectricity().compareTo(BigDecimal.ZERO) > 0) {
                existingItem.setPrice(existingItem.getCost().divide(existingItem.getElectricity(), 2, BigDecimal.ROUND_HALF_UP));
            }
        }
        
        existingItem.setDataTime(dataTime);
        existingItem.setBeginTime(beginTime);
        existingItem.setEndTime(endTime);
        existingItem.setTimeType(timeType);
        existingItem.setUpdateTime(LocalDateTime.now());
        
        electricityDataItemRepository.save(existingItem);
        log.info("更新{}尖峰平谷统计数据成功：indexId={}, timeCode={}, type={}, electricity={}, cost={}", 
                timeType, indexId, timeCode, electricityType, electricity, cost);
    }
 
    /**
     * 计算时间范围
     *
     * @param timeType 时间类型
     * @param dateTime 统计时间点
     * @return 时间范围[开始时间, 结束时间]
     */
    private LocalDateTime[] calculateTimeRange(String timeType, LocalDateTime dateTime) {
        LocalDateTime startTime;
        LocalDateTime endTime;
        
        switch (timeType) {
            case "HOUR":
                startTime = dateTime.withMinute(0).withSecond(0).withNano(0);
                endTime = startTime.plusHours(1).minusNanos(1);
                break;
            case "DAY":
                startTime = dateTime.withHour(0).withMinute(0).withSecond(0).withNano(0);
                endTime = startTime.plusDays(1).minusNanos(1);
                break;
            case "MONTH":
                startTime = dateTime.withDayOfMonth(1).withHour(0).withMinute(0).withSecond(0).withNano(0);
                endTime = startTime.plusMonths(1).minusNanos(1);
                break;
            case "YEAR":
                startTime = dateTime.withDayOfYear(1).withHour(0).withMinute(0).withSecond(0).withNano(0);
                endTime = startTime.plusYears(1).minusNanos(1);
                break;
            default:
                throw new IllegalArgumentException("不支持的时间类型：" + timeType);
        }
        
        return new LocalDateTime[]{startTime, endTime};
    }
 
    /**
     * 生成时间编码
     *
     * @param timeType 时间类型
     * @param dateTime 统计时间点
     * @return 时间编码
     */
    private String generateTimeCode(String timeType, LocalDateTime dateTime) {
        DateTimeFormatter formatter;
        
        switch (timeType) {
            case "HOUR":
                formatter = DateTimeFormatter.ofPattern("yyyyMMddHH");
                return "H" + dateTime.format(formatter);
            case "DAY":
                formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
                return "D" + dateTime.format(formatter);
            case "MONTH":
                formatter = DateTimeFormatter.ofPattern("yyyyMM");
                return "M" + dateTime.format(formatter);
            case "YEAR":
                formatter = DateTimeFormatter.ofPattern("yyyy");
                return "Y" + dateTime.format(formatter);
            default:
                throw new IllegalArgumentException("不支持的时间类型：" + timeType);
        }
    }
}