refactor(engine): 重构全厂总电量统计逻辑

baoshiwei

2025-06-21 d9714be7130e14e063e6499637e1cc5241ff9dd3

refactor(engine): 重构全厂总电量统计逻辑

- 将固定的网关代码列表改为配置化的参数
- 优化了数据统计的逻辑，按区域分别统计后再汇总
- 新增 TotalCount 类用于统计总数和总值
- 重构了代码结构，提高了可读性和可维护性

 src/main/java/com/zhitan/engine/scheduler/DataCleaningScheduler.java

@@ -98,6 +98,10 @@
            log.error("手动触发{}统计任务执行失败：{}", timeType, e.getMessage(), e);
        }
    }

    /**
     * 汇总上级电表数据
     */
    @Scheduled(cron = "0 * * * * ?")
    public void totalElectricityTask() {
        try {

 src/main/java/com/zhitan/engine/service/impl/DataCleaningServiceImpl.java

@@ -176,12 +176,9 @@
        log.info("开始执行全厂总电量统计，时间点：{}", dateTime);
        
        // 定义需要统计的电表网关列表
        List<String> gatewayCodes = new java.util.ArrayList<>(List.of("00601925032100028432", "00601925032100026840", "00601925032100028685", "00601925032100027392", "00601925032100028050", "00601925032100028751"));
        gatewayCodes.add("00601925032100026275");
        gatewayCodes.add("00601925032100028117");
        gatewayCodes.add("00601925032100028814");
        gatewayCodes.add("00601925032100028990");
        gatewayCodes.add("00601925032100031307");
        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");


        // 定义需要统计的参数类型及其统计方式
@@ -196,13 +193,7 @@
                , "VoltageUAB", "AVG" // 总电压 - 平均值
                , "VoltageUBC", "AVG" // 总电压 - 平均值
                , "VoltageUCA", "AVG" // 总电压 - 平均值
//            ,"CurrentA", "SUM" // 总电流 - 求和
//            ,"CurrentB", "SUM" // 总电流 - 求和
//            ,"CurrentC", "SUM" // 总电流 - 求和
//            ,"Io", "SUM" // 总电流 - 求和
//            ,"VoltageA", "AVG" // 总电压 - 平均值
//            ,"VoltageB", "AVG" // 总电压 - 平均值
//            ,"VoltageC", "AVG"

        ));
        parameterTypes.put("CurrentA", "SUM");
        parameterTypes.put("CurrentB", "SUM");
@@ -211,57 +202,81 @@
        parameterTypes.put("VoltageA", "AVG");
        parameterTypes.put("VoltageB", "AVG");
        parameterTypes.put("VoltageC", "AVG");
        
        // 计算时间范围（当前分钟）
        LocalDateTime startTime = dateTime.withSecond(0).withNano(0);
        LocalDateTime endTime = startTime.plusMinutes(1);
        

        // 保存总表数据
        String old = "old_";
        String seven = "seven_";
        String out = "out_";
        String total = "total_";
        // 遍历所有参数类型进行统计
        parameterTypes.forEach((paramType, calcType) -> {
            try {
                double totalValue = 0;
                int meterCount = 0;
                
                // 遍历所有网关下的电表
                for (String gatewayCode : gatewayCodes) {
                    // 遍历每个网关下的8个电表
                    for (int i = 1; i <= 8; i++) {
                        // 构建tag名称格式：网关编号_电表序号_参数类型
                        // 示例：00601925032100028432_1_ActiveZT
                        String tag = String.format("%s_%d_%s", gatewayCode, i, paramType);
                        // 查询最后一次保存数据
            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);

                        Double value = influxdbRepository.getLastPoint(influxDBConfig.getMeasurement(),TAG,tag);
                        if (value != null) {
                            totalValue +=  value ;
                            meterCount++;
                        }
                    }
                }
                
                // 计算最终值
                double finalValue = "SUM".equals(calcType) ? totalValue : (meterCount > 0 ? totalValue / meterCount : 0);
                
                // 保存总表数据
                String totalTag = "total_" + paramType;

                Point point = Point
                        .measurement(influxDBConfig.getMeasurement())
                        .addTag(TAG, totalTag)
                        .addField(FIELD_VALUE, finalValue)
                        .time(Instant.now(), WritePrecision.S);

                influxdbRepository.writePoint(point);
                
                log.info("统计完成：{}={} ({}个电表)", totalTag, finalValue, meterCount);
            } catch (Exception e) {
                log.error("统计参数{}失败：{}", paramType, e.getMessage(), e);
            }
        });
        
        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) {