Merge remote-tracking branch 'origin/master'

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/ExecutionCenter.java

@@ -123,9 +123,9 @@ public class ExecutionCenter extends BaseController {
                     startDate = DateUtil.offsetDay(startDate, 1);
                 }
                 calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
-               // CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
-                calculatedParam.setType(0);
-                flowService.execute2Resp(methodDescribe, calculatedParam);
+                CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
+                repairParam.setType(0);
+                flowService.execute2Resp(methodDescribe, repairParam);
             }
         } else {
             //非补招

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IDataLimitRateAsync.java

@@ -4,26 +4,29 @@ import com.njcn.device.biz.pojo.po.Overlimit;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 public interface IDataLimitRateAsync {
 
     /**
      * limit_rate多线程算法
+     *
      * @param dataDate
      * @param list
      * @param phase
      * @param overLimitMap
      * @param size
      * @param i
-     * @param type 系统类型
+     * @param type         系统类型
+     * @return
      * @Author: wr
      * @Date: 2025/12/17 12:16
      */
-    void lineDataRate(String dataDate,
-                      List<String> list,
-                      List<String> phase,
-                      Map<String, Overlimit> overLimitMap,
-                      int size,
-                      int i,
-                      int type);
+    CompletableFuture<Void> lineDataRate(String dataDate,
+                                         List<String> list,
+                                         List<String> phase,
+                                         Map<String, Overlimit> overLimitMap,
+                                         int size,
+                                         int i,
+                                         int type);
 }

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataComAssServiceImpl.java

@@ -58,7 +58,7 @@ public class DataComAssServiceImpl implements IDataComAssService {
         lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
         lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
         List<String> lineIdList = calculatedParam.getIdList();
-        getAbnormalData(lineParam);
+        //getAbnormalData(lineParam);
         for (String lineId : lineIdList) {
             DataComassesDPO rStatComassesDpo = new DataComassesDPO();
             rStatComassesDpo.setTime(calculatedParam.getDataDate());

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DayDataServiceImpl.java

@@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 /**
@@ -64,9 +65,13 @@ public class DayDataServiceImpl implements IDayDataService {
     private DataInharmVFeignClient dataInharmVFeignClient;
     @Resource
     private DataPltFeignClient dataPltFeignClient;
-
     @Resource
     private PqDataVerifyFeignClient pqDataVerifyFeignClient;
+    /**
+     * 查询配置 辽宁版本比较特殊，没有CP95值，采用平均值进行判断
+     */
+    @Value("${version.used:master}")
+    private String versionUsed;
 
     @Override
     public void dataVHandler(CalculatedParam calculatedParam) {
@@ -102,7 +107,8 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataVHandler(item3,valueTypes,dto,true);
+                            //默认CP95值取 CP95的最大值；如果有特殊要求，可以根据平均值的数据取CP95
+                            channelDataVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -150,7 +156,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataIHandler(item3,valueTypes,dto,true);
+                            channelDataIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -198,7 +204,7 @@ public class DayDataServiceImpl implements IDayDataService {
                                 dto.setPhasicType(item2.getPhasicType());
                                 dto.setValueType(item4.toUpperCase());
                                 dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                                channelDataFlickerHandler(item3,dto,item4,true);
+                                channelDataFlickerHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
                                 result.add(dto);
                             });
                         });
@@ -247,7 +253,7 @@ public class DayDataServiceImpl implements IDayDataService {
                                 dto.setPhasicType(item2.getPhasicType());
                                 dto.setValueType(item4.toUpperCase());
                                 dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                                channelDataFlucHandler(item3,dto,item4,true);
+                                channelDataFlucHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
                                 result.add(dto);
                             });
                         });
@@ -296,7 +302,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmPhasicIHandler(item3,valueTypes,dto,true);
+                            channelDataHarmPhasicIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -344,7 +350,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmPhasicVHandler(item3,valueTypes,dto,true);
+                            channelDataHarmPhasicVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -392,7 +398,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmPowerPHandler(item3,valueTypes,dto,true);
+                            channelDataHarmPowerPHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -440,7 +446,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmPowerQHandler(item3,valueTypes,dto,true);
+                            channelDataHarmPowerQHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -488,7 +494,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmPowerSHandler(item3,valueTypes,dto,true);
+                            channelDataHarmPowerSHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -536,7 +542,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmRateIHandler(item3,valueTypes,dto,true);
+                            channelDataHarmRateIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -584,7 +590,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataHarmRateVHandler(item3,valueTypes,dto,true);
+                            channelDataHarmRateVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -632,7 +638,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataInHarmIHandler(item3,valueTypes,dto,true);
+                            channelDataInHarmIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -680,7 +686,7 @@ public class DayDataServiceImpl implements IDayDataService {
                             dto.setPhasicType(item2.getPhasicType());
                             dto.setValueType(item3.getValueType().toUpperCase());
                             dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                            channelDataInHarmVHandler(item3,valueTypes,dto,true);
+                            channelDataInHarmVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
                             result.add(dto);
                         });
                     });
@@ -728,7 +734,7 @@ public class DayDataServiceImpl implements IDayDataService {
                                 dto.setPhasicType(item2.getPhasicType());
                                 dto.setValueType(item4.toUpperCase());
                                 dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
-                                channelDataPltHandler(item3,dto,item4,true);
+                                channelDataPltHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
                                 result.add(dto);
                             });
                         });
@@ -1549,23 +1555,20 @@ public class DayDataServiceImpl implements IDayDataService {
         valueType = valueType.toUpperCase();
         if (scheme) {
             switch (valueType) {
-                case InfluxDbSqlConstant.MAX:
-                case InfluxDbSqlConstant.CP95:
+                case "MAX":
+                case "CP95":
                     Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
-//                    result = max.orElse(null);
                     result = max.orElse(0.0);
                     break;
-                case InfluxDbSqlConstant.MIN:
+                case "MIN":
                     Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
-//                    result = min.orElse(null);
                     result = min.orElse(0.0);
                     break;
-                case InfluxDbSqlConstant.AVG_WEB:
+                case "AVG":
                     OptionalDouble average = list.stream()
                             .filter(Objects::nonNull)
                             .mapToDouble(Double::doubleValue)
                             .average();
-//                    result = average.isPresent() ? average.getAsDouble() : null;
                     result = average.isPresent() ? average.getAsDouble() : 0.0;
                     break;
                 default:
@@ -1573,29 +1576,42 @@ public class DayDataServiceImpl implements IDayDataService {
             }
         } else {
             switch (valueType) {
-                case InfluxDbSqlConstant.MAX:
+                case "MAX":
                     Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
-//                    result = max.orElse(null);
                     result = max.orElse(0.0);
                     break;
-                case InfluxDbSqlConstant.MIN:
+                case "MIN":
                     Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
-//                    result = min.orElse(null);
                     result = min.orElse(0.0);
                     break;
-                case InfluxDbSqlConstant.AVG_WEB:
+                case "AVG":
                     OptionalDouble average = list.stream()
                             .filter(Objects::nonNull)
                             .mapToDouble(Double::doubleValue)
                             .average();
-//                    result = average.isPresent() ? average.getAsDouble() : null;
                     result = average.isPresent() ? average.getAsDouble() : 0.0;
                     break;
-                case InfluxDbSqlConstant.CP95:
-                    list.sort(Collections.reverseOrder());
-                    int discardCount = (int) Math.ceil(list.size() * 0.05);
-                    List<Double> remainingList = list.subList(discardCount, list.size());
-//                    result = remainingList.isEmpty() ? null : remainingList.get(0);
+                case "CP95":
+                    if (list == null || list.isEmpty()) {
+                        result = 0.0;
+                        break;
+                    }
+                    // 过滤掉 null 元素
+                    List<Double> filteredList = list.stream()
+                            .filter(Objects::nonNull)
+                            .collect(Collectors.toList());
+
+                    if (filteredList.isEmpty()) {
+                        result = 0.0;
+                        break;
+                    }
+                    filteredList.sort(Collections.reverseOrder());
+                    int discardCount = (int) Math.ceil(filteredList.size() * 0.05);
+                    if (discardCount >= filteredList.size()) {
+                        result = 0.0;
+                        break;
+                    }
+                    List<Double> remainingList = filteredList.subList(discardCount, filteredList.size());
                     result = remainingList.isEmpty() ? 0.0 : remainingList.get(0);
                     break;
                 default:
@@ -1610,4 +1626,28 @@ public class DayDataServiceImpl implements IDayDataService {
         return result;
     }
 
+    /**
+     * 将平均值赋值给 CP95
+     */
+    public void putCP95ByAvg(List<CommonMinuteDto> list) {
+        list.forEach(po1->{
+            List<CommonMinuteDto.PhasicType> phasicTypeList = po1.getPhasicTypeList();
+            phasicTypeList.forEach(po2 -> {
+                List<CommonMinuteDto.ValueType> valueTypeList = po2.getValueTypeList();
+                AtomicReference<List<List<Double>>> valueList = new AtomicReference<>(new ArrayList<>());
+                valueTypeList.forEach(po3 -> {
+                    if (po3.getValueType().equals(InfluxDbSqlConstant.AVG_WEB)) {
+                        valueList.set(po3.getValueList());
+                    }
+                    if (po3.getValueType().equals(InfluxDbSqlConstant.CP95)) {
+                        po3.setValueList((List<List<Double>>) valueList);
+                    }
+                });
+            });
+        });
+    }
+
+
+
+
 }

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataCrossingServiceImpl.java

@@ -46,6 +46,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -108,11 +109,26 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
         List<List<String>> pendingIds = ListUtils.partition(lineIds, 1);
         ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
         MemorySizeUtil.getNowMemory();
+
+        List<CompletableFuture<Void>> futures = new ArrayList<>();
         for (int i = 0; i < pendingIds.size(); i++) {
             logger.info(calculatedParam.getDataDate()+" 总分区数据：" + pendingIds.size() + "=====》当前第"+(i + 1)+"小分区");
             List<String> list = pendingIds.get(i);
-            dataLimitRateAsync.lineDataRate(calculatedParam.getDataDate(), list, phase, overLimitMap, pendingIds.size(), (i + 1), lineParam.getType());
+            // 获取Future
+            CompletableFuture<Void> future = dataLimitRateAsync.lineDataRate(
+                    calculatedParam.getDataDate(),
+                    list,
+                    phase,
+                    overLimitMap,
+                    pendingIds.size(),
+                    (i + 1),
+                    lineParam.getType()
+            );
+
+            futures.add(future);
         }
+        // 等待所有任务完成
+        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
         System.gc();
     }
 

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataLimitRateAsyncImpl.java

@@ -18,6 +18,7 @@ import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
@@ -25,6 +26,7 @@ import org.springframework.util.CollectionUtils;
 import javax.annotation.Resource;
 import java.lang.reflect.Method;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 /**
@@ -52,17 +54,22 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
     private DataLimitRateFeignClient dataLimitRateFeignClient;
     @Resource
     private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient;
+    /**
+     * 查询配置 辽宁版本比较特殊，没有CP95值，采用平均值进行判断
+     */
+    @Value("${version.used:master}")
+    private String versionUsed;
 
 
     @Override
     @Async("asyncExecutor")
-    public void lineDataRate(String dataDate,
-                             List<String> list,
-                             List<String> phase,
-                             Map<String, Overlimit> overLimitMap,
-                             int size,
-                             int i,
-                             int type) {
+    public CompletableFuture<Void> lineDataRate(String dataDate,
+                                                List<String> list,
+                                                List<String> phase,
+                                                Map<String, Overlimit> overLimitMap,
+                                                int size,
+                                                int i,
+                                                int type) {
         List<DataLimitDetailDto> result = new ArrayList<>();
         LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
         lineParam.setStartTime(TimeUtils.getBeginOfDay(dataDate));
@@ -79,6 +86,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
         List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(lineParam).getData();
         //电流数据
         List<DataIDto> dataIList = dataIFeignClient.getRawData(lineParam).getData();
+
         /**
          * 功能描述:获取influxDB -> data_v ->
          *              总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
@@ -121,76 +129,140 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
          *功能描述:获取influxDB -> data_harmrate_v ->
          *              2-25次谐波电压含有率 -> A相||B相||C相的日95%概率值
          */
-        Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream()
-                .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        Map<String, List<DataHarmDto>> harmRateV;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            harmRateV = dataVHarmList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        } else {
+            harmRateV = dataVHarmList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_i -> 2-25次谐波电流 -> 日95%概率值
          */
-        Map<String, List<DataIDto>> dataI = dataIList.stream()
-                .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataIDto::getLineId));
-
+        Map<String, List<DataIDto>> dataI;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            dataI = dataIList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataIDto::getLineId));
+        } else {
+            dataI = dataIList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataIDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_inharm_v -> 0.5-15.5次间谐波电压含有率 -> 日95%概率值
          */
-        Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream()
-                .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        Map<String, List<DataHarmDto>> inHarmV;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            inHarmV = dataVInHarmList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        } else {
+            inHarmV = dataVInHarmList.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataHarmDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_v -> 电压总谐波畸变率 -> 日95%概率值
          */
-        Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream()
-                .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataVDto::getLineId));
+        Map<String, List<DataVDto>> dataVThd;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            dataVThd = dataVAllTime.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataVDto::getLineId));
+        } else {
+            dataVThd = dataVAllTime.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataVDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值
          */
-        Map<String, List<DataVDto>> dataVUnbalance = dataVAllTime.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataVDto::getLineId));
+        Map<String, List<DataVDto>> dataVUnbalance;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            dataVUnbalance = dataVAllTime.stream()
+                    .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()) ||
+                            InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataVDto::getLineId));
+        } else {
+            dataVUnbalance = dataVAllTime.stream()
+                    .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
+                            InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataVDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_i -> 负序电流 -> 最大值 && 日95%概率值
          */
-        Map<String, List<DataIDto>> dataINeg = dataIList.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
-                .collect(Collectors.groupingBy(DataIDto::getLineId));
+
+        Map<String, List<DataIDto>> dataINeg;
+        if (Objects.equals(versionUsed, "liaoning")) {
+            dataINeg = dataIList.stream()
+                    .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()) ||
+                            InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
+                    .peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
+                    .collect(Collectors.groupingBy(DataIDto::getLineId));
+        } else {
+            dataINeg = dataIList.stream()
+                    .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
+                            InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataIDto::getLineId));
+        }
 
         /**
          * 功能描述:获取influxDB -> data_v -> 频率偏差 -> 最大值 && 最小值
          */
         Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream()
-                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.MIN.equals(x.getValueType()) ||
-                        InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
+                .filter(x -> InfluxDBTableConstant.MIN.equalsIgnoreCase(x.getValueType()) ||
+                        InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
         /**
          * 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值
          */
         Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream()
                 .filter(x -> phase.contains(x.getPhasicType()))
-                .filter(x -> InfluxDBTableConstant.MAX.equals(x.getValueType()))
+                .filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
                 .collect(Collectors.groupingBy(DataVDto::getLineId));
 
         /**
          * 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
          */
-        Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
-                .filter(x -> phase.contains(x.getPhasicType()))
-                .collect(Collectors.groupingBy(DataPltDto::getLineId));
+        Map<String, List<DataPltDto>> dataPlt;
+        if (type == 2) {
+            dataPlt = dataPltAllTime.stream()
+                    .filter(x -> phase.contains(x.getPhasicType()))
+                    .filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
+                    .collect(Collectors.groupingBy(DataPltDto::getLineId));
+        } else {
+            dataPlt = dataPltAllTime.stream()
+                    .filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
+                    .collect(Collectors.groupingBy(DataPltDto::getLineId));
+        }
 
         for (String item : list) {
             if (ObjectUtil.isNotNull(overLimitMap.get(item))) {
@@ -229,6 +301,8 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
             MemorySizeUtil.getNowMemory();
         }
         System.gc();
+
+        return CompletableFuture.completedFuture(null);
     }
 
     /**
@@ -296,9 +370,6 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
         result.add(c);
         result.add(t);
 
-
-
-
         return result;
     }
 

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionCalcImpl.java

@@ -80,19 +80,21 @@ public class PollutionCalcImpl implements IPollutionCalc {
                     LineDevGetDTO line = lineDetailMap.get(id);
                     if (limitMap.containsKey(id)) {
                         Overlimit overlimit = limitMap.get(id);
-                        dataPollutionD = new DataPollutionD();
-                        dataPollutionD.setLineId(id);
-                        dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));
-                        dataPollutionD.setPollutionType(vHarmonicLimit);
                         lineParam.setValueType(Arrays.asList(line.getTimeInterval() + ""));
                         lineParam.setLineId(Arrays.asList(id));
 
-                        List<DataVDto> dataVDtoList = dataVFeignClient.getGroupByTimeDataV(lineParam).getData();
-                        List<DataHarmDto> dataHarmDtoList = dataHarmRateVFeignClient.getGroupByTimeHarmRateV(lineParam).getData();
-                        //计算谐波电压污染值
-                        dataPollutionD.setValue(PubUtils.doubleRound(2, calcVAllPollutionValue(dataVDtoList, dataHarmDtoList, overlimit) * line.getTimeInterval()));
+                        if(!dataVFeignClient.excludeZeroData(lineParam).getData()){
+                            dataPollutionD = new DataPollutionD();
+                            dataPollutionD.setLineId(id);
+                            dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));
+                            dataPollutionD.setPollutionType(vHarmonicLimit);
+                            List<DataVDto> dataVDtoList = dataVFeignClient.getGroupByTimeDataV(lineParam).getData();
+                            List<DataHarmDto> dataHarmDtoList = dataHarmRateVFeignClient.getGroupByTimeHarmRateV(lineParam).getData();
+                            //计算谐波电压污染值
+                            dataPollutionD.setValue(PubUtils.doubleRound(2, calcVAllPollutionValue(dataVDtoList, dataHarmDtoList, overlimit) * line.getTimeInterval()));
+                            list.add(dataPollutionD);
+                        }
 
-                        list.add(dataPollutionD);
                         dataPollutionD = new DataPollutionD();
                         dataPollutionD.setLineId(id);
                         dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));

algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/SpecialAnalysisServiceImpl.java

@@ -27,7 +27,7 @@ import com.njcn.event.api.TransientFeignClient;
 import com.njcn.event.file.pojo.dto.WaveDataDTO;
 import com.njcn.event.pojo.param.EventCountParam;
 import com.njcn.event.pojo.po.RmpEventDetailPO;
-import com.njcn.supervision.api.UserLedgerFeignClient;
+import com.njcn.supervision.api.UserLedgerOldFeignClient;
 import com.njcn.supervision.pojo.vo.user.NewUserReportVO;
 import com.njcn.system.api.DicDataFeignClient;
 import com.njcn.system.api.DictTreeFeignClient;
@@ -84,7 +84,7 @@ public class SpecialAnalysisServiceImpl implements ISpecialAnalysisService {
     @Resource
     private CommLineClient commLineClient;
     @Resource
-    private UserLedgerFeignClient userLedgerFeignClient;
+    private UserLedgerOldFeignClient userLedgerFeignClient;
     @Resource
     private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient;
 

algorithm/algorithm-boot/src/main/resources/bootstrap-jb.yml

@@ -0,0 +1,76 @@
+#当前服务的基本信息
+microservice:
+  ename: @artifactId@
+  name: '@name@'
+  version: @version@
+  sentinel:
+    url: @sentinel.url@
+  gateway:
+    url: @gateway.url@
+server:
+  port: 10401
+#feign接口开启服务熔断降级处理
+feign:
+  sentinel:
+    enabled: true
+  compression:
+    request:
+      enabled: true
+      # 配置压缩支持的MIME TYPE
+      mime-types: text/xml,application/xml,application/json
+      # 配置压缩数据大小的下限
+      min-request-size: 1024
+    response:
+      # 配置响应GZIP压缩
+      enabled: true
+spring:
+  application:
+    name: @artifactId@
+  cloud:
+    nacos:
+      discovery:
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+      config:
+        username: @nacos.username@
+        password: @nacos.password@
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        file-extension: yaml
+        shared-configs:
+          - data-id: share-config.yaml
+            refresh: true
+          - data-id: share-config-datasource-db.yaml
+            refresh: true
+  main:
+    allow-bean-definition-overriding: true
+
+liteflow:
+  rule-source-ext-data-map:
+    serverAddr: @nacos.url@
+    dataId: prepare_liteflow
+    group: DEFAULT_GROUP
+    namespace: @nacos.namespace@
+    username: @nacos.username@
+    password: @nacos.password@
+  when-max-wait-time: 3600000
+  print-banner: false
+
+
+#项目日志的配置
+logging:
+  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  level:
+    root: info
+    com.alibaba.nacos.client: INFO
+
+
+#mybatis配置信息
+mybatis-plus:
+  #别名扫描
+  type-aliases-package: com.njcn.harmonic.pojo
+line:
+  num: 20
+
+mqtt:
+  client-id: @artifactId@${random.value}

algorithm/algorithm-boot/src/main/resources/bootstrap-sjzx.yml

@@ -0,0 +1,79 @@
+#当前服务的基本信息
+microservice:
+  ename: @artifactId@
+  name: '@name@'
+  version: @version@
+  sentinel:
+    url: @sentinel.url@
+  gateway:
+    url: @gateway.url@
+server:
+  port: 10401
+#feign接口开启服务熔断降级处理
+feign:
+  sentinel:
+    enabled: true
+  compression:
+    request:
+      enabled: true
+      # 配置压缩支持的MIME TYPE
+      mime-types: text/xml,application/xml,application/json
+      # 配置压缩数据大小的下限
+      min-request-size: 1024
+    response:
+      # 配置响应GZIP压缩
+      enabled: true
+spring:
+  application:
+    name: @artifactId@
+  cloud:
+    nacos:
+      discovery:
+        ip: @service.server.url@
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
+      config:
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
+        file-extension: yaml
+        shared-configs:
+          - data-id: share-config.yaml
+            refresh: true
+          - data-Id: share-config-datasource-db.yaml
+            refresh: true
+  main:
+    allow-bean-definition-overriding: true
+
+liteflow:
+  rule-source-ext-data-map:
+    serverAddr: @nacos.url@
+    dataId: prepare_liteflow
+    group: DEFAULT_GROUP
+    namespace: @nacos.namespace@
+    username: @nacos.username@
+    password: @nacos.password@
+  when-max-wait-time: 3600000
+  print-banner: false
+
+
+#项目日志的配置
+logging:
+  #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  level:
+    root: info
+    com.alibaba.nacos.client: INFO
+
+
+#mybatis配置信息
+mybatis-plus:
+  #别名扫描
+  type-aliases-package: com.njcn.harmonic.pojo
+line:
+  num: 20
+
+#mqtt:
+#  client-id: @artifactId@${random.value}

algorithm/algorithm-boot/src/main/resources/bootstrap.yml

@@ -1,72 +1,3 @@
-#当前服务的基本信息
-microservice:
-  ename: @artifactId@
-  name: '@name@'
-  version: @version@
-  sentinel:
-    url: @sentinel.url@
-  gateway:
-    url: @gateway.url@
-server:
-  port: 10401
-#feign接口开启服务熔断降级处理
-feign:
-  sentinel:
-    enabled: true
-  compression:
-    request:
-      enabled: true
-      # 配置压缩支持的MIME TYPE
-      mime-types: text/xml,application/xml,application/json
-      # 配置压缩数据大小的下限
-      min-request-size: 1024
-    response:
-      # 配置响应GZIP压缩
-      enabled: true
 spring:
-  application:
-    name: @artifactId@
-  cloud:
-    nacos:
-      discovery:
-        ip: @service.server.url@
-        server-addr: @nacos.url@
-        namespace: @nacos.namespace@
-      config:
-        server-addr: @nacos.url@
-        namespace: @nacos.namespace@
-        file-extension: yaml
-        shared-configs:
-          - data-id: share-config.yaml
-            refresh: true
-          - data-Id: share-config-datasource-db.yaml
-            refresh: true
-  main:
-    allow-bean-definition-overriding: true
-
-liteflow:
-  rule-source-ext-data-map:
-    serverAddr: @nacos.url@
-    dataId: prepare_liteflow
-    group: DEFAULT_GROUP
-    namespace: @nacos.namespace@
-  when-max-wait-time: 600000
-  print-banner: false
-
-
-#项目日志的配置
-logging:
-  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
-  level:
-    root: error
-
-
-#mybatis配置信息
-mybatis-plus:
-  #别名扫描
-  type-aliases-package: com.njcn.harmonic.pojo
-line:
-  num: 10
-
-mqtt:
-  client-id: @artifactId@${random.value}
+  profiles:
+    active: @spring.profiles.active@

analysis/stat/stat-boot/src/main/resources/bootstrap.yml

@@ -22,16 +22,20 @@ spring:
         ip: @service.server.url@
         server-addr: @nacos.url@
         namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
       config:
         server-addr: @nacos.url@
         namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
         file-extension: yaml
         shared-configs:
           - data-id: share-config.yaml
             refresh: true
-          - data-Id: share-config-datasource-db.yaml
+          - data-id: share-config-datasource-db.yaml
             refresh: true
-          - data-Id: data-platform.yaml
+          - data-id: data-platform.yaml
             refresh: true
   main:
     allow-bean-definition-overriding: true
@@ -39,7 +43,7 @@ spring:
 
 #项目日志的配置
 logging:
-  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info
 
@@ -50,4 +54,4 @@ mybatis-plus:
   type-aliases-package: com.njcn.harmonic.pojo
 
 mqtt:
-  client-id: @artifactId@${random.value}
+  client-id: @artifactId@${random.value}

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataVFeignClient.java

@@ -68,4 +68,8 @@ public interface DataVFeignClient {
     //按时间分组获取原始数据
     @PostMapping("/getGroupByTimeDataV")
     HttpResult<List<DataVDto>> getGroupByTimeDataV(@RequestBody LineCountEvaluateParam lineParam);
+
+
+    @PostMapping("/excludeZeroData")
+    HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam);
 }

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataVFeignClientFallbackFactory.java

@@ -116,6 +116,12 @@ public class DataVFeignClientFallbackFactory implements FallbackFactory<DataVFei
                 log.error("{}异常，降级处理，异常为：{}","DataV按时间分组获取原始数据",cause.toString());
                 throw new BusinessException(finalExceptionEnum);
             }
+
+            @Override
+            public HttpResult<Boolean> excludeZeroData(LineCountEvaluateParam lineParam) {
+                log.error("{}异常，降级处理，异常为：{}","判断是否存在是零飘数据",cause.toString());
+                throw new BusinessException(finalExceptionEnum);
+            }
         };
     }
 }

data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/dto/PqsCommunicateDto.java

@@ -18,6 +18,8 @@ public class PqsCommunicateDto {
     private String description;
 
     private Integer type;
+    //是否更新updateTime标志；数据上送更新1，状态翻转不更新0
+    private Integer flag=0;
 
 
 }

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataVController.java

@@ -166,6 +166,15 @@ public class DataVController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, dataV, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/excludeZeroData")
+    @ApiOperation("判断是否存在是零飘数据")
+    public HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam) {
+        String methodDescribe = getMethodDescribe("excludeZeroData");
+        Boolean b = dataVQuery.excludeZeroData(lineParam);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, b, methodDescribe);
+    }
+
 
 
 

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataV.java

@@ -81,4 +81,17 @@ public interface IDataV extends IMppService<RStatDataVD> {
     List<DataVDto> getDataV(LineCountEvaluateParam lineParam);
 
     List<DataVDto> getGroupByTimeDataV(LineCountEvaluateParam lineParam);
+
+
+    /**
+     * 查看监测点的电压，判断是否存在是零飘数据
+     * 如果零飘数据占比一半多，则抛弃该监测点数据
+     * @param lineParam
+     * @return: java.lang.Boolean
+     * @Author: wr
+     * @Date: 2026/1/15 9:26
+     */
+    Boolean excludeZeroData(LineCountEvaluateParam lineParam);
+
+
 }

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/LnDataDealServiceImpl.java

@@ -97,7 +97,7 @@ public class LnDataDealServiceImpl implements LnDataDealService {
                 pqsCommunicateDto.setTime(LocalDateTimeUtil.format(dataVDTO.getTimeid(), DatePattern.NORM_DATETIME_PATTERN));
                 pqsCommunicateDto.setDevId(temp);
                 pqsCommunicateDto.setType(1);
-
+                pqsCommunicateDto.setFlag(1);
 
                 iPqsCommunicate.insertion(pqsCommunicateDto);
             });

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataHarmpowerPImpl.java

@@ -128,7 +128,7 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
                         CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
                         value.setValueType(valueType);
                         List<List<Double>> lists;
-                        if (Objects.equals(phasicType, "T")) {
+                        if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
                             lists = extractDataLists(valueTypeList, "Tot");
                         } else {
                             lists = extractDataLists(valueTypeList, "");

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataHarmpowerQImpl.java

@@ -127,7 +127,7 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
                         value.setValueType(valueType);
 
                         List<List<Double>> lists;
-                        if (Objects.equals(phasicType, "T")) {
+                        if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
                             lists = extractDataLists(valueTypeList, "Tot");
                         } else {
                             lists = extractDataLists(valueTypeList, "");

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataHarmpowerSImpl.java

@@ -126,7 +126,7 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
                         CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
                         value.setValueType(valueType);
                         List<List<Double>> lists;
-                        if (Objects.equals(phasicType, "T")) {
+                        if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
                             lists = extractDataLists(valueTypeList, "Tot");
                         } else {
                             lists = extractDataLists(valueTypeList, "");

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataVImpl.java

@@ -208,7 +208,7 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
         List<DataV> list = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
         if (CollUtil.isNotEmpty(list)) {
             Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime();
-            if(CollUtil.isNotEmpty(abnormalTime)){
+            if (CollUtil.isNotEmpty(abnormalTime)) {
                 if (abnormalTime.containsKey(lineParam.getLineId().get(0))) {
                     List<String> timeList = abnormalTime.get(lineParam.getLineId().get(0));
                     //有异常数据,当前监测点自身的异常数据
@@ -216,7 +216,7 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
                         result.addAll(list.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()));
                     }
                 }
-            }else{
+            } else {
                 result.addAll(list);
             }
         }
@@ -390,8 +390,8 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
                 .groupBy(DataV::getLineId)
                 .between(DataV::getTime, startTime, endTime);
         List<MeasurementCountDTO> measurementCountDTOList = dataVMapper.getMeasurementCount(influxQueryWrapper);
-        if(CollUtil.isNotEmpty(measurementCountDTOList)){
-            for(MeasurementCountDTO m : measurementCountDTOList){
+        if (CollUtil.isNotEmpty(measurementCountDTOList)) {
+            for (MeasurementCountDTO m : measurementCountDTOList) {
                 MeasurementCountDTO.MeasurementCountResultDTO p = new MeasurementCountDTO.MeasurementCountResultDTO();
                 p.setFreq(m.getFreq());
                 p.setLineId(m.getLineId());
@@ -425,6 +425,27 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
         return result;
     }
 
+
+    @Override
+    public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
+        InfluxQueryWrapper dataVQueryWrapper = new InfluxQueryWrapper(DataV.class);
+        dataVQueryWrapper.eq(DataV::getLineId, lineParam.getLineId().get(0))
+                .eq(DataV::getValueType, InfluxDbSqlConstant.AVG_WEB)
+                .ne(DataV::getPhasicType, InfluxDBTableConstant.PHASE_TYPE_T)
+                .select(DataV::getRms)
+                .between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime());
+        List<DataV> rmsResult = dataVMapper.selectByQueryWrapper(dataVQueryWrapper);
+        if (CollUtil.isNotEmpty(rmsResult)) {
+            List<DataV> exceptionData = rmsResult.stream().filter(dataV -> dataV.getRms() < 1.0).collect(Collectors.toList());
+            if (CollUtil.isNotEmpty(exceptionData)) {
+                if (exceptionData.size() * 2 >= rmsResult.size()) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     /**
      * 按监测点集合、时间条件获取dataV分钟数据
      * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理
@@ -483,53 +504,53 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
                 CsLinePO po = lineMap.get(lineId);
                 InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class);
                 influxQueryWrapper.eq(DataV::getLineId, lineId)
-                        .eq(DataV::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
+                        .eq(DataV::getProcess, Integer.toString(devsMap.get(devId).getProcess()))
                         .select(DataV::getLineId)
                         .select(DataV::getPhasicType)
                         .select(DataV::getValueType)
                         //频率
-                        .select("Pq_Freq","freq")
+                        .select("Pq_Freq", "freq")
                         //频率偏差
-                        .select("Pq_FreqDev","freq_dev")
+                        .select("Pq_FreqDev", "freq_dev")
                         //相电压有效值
-                        .select("Pq_RmsU","rms")
+                        .select("Pq_RmsU", "rms")
                         //线电压有效值
-                        .select("Pq_RmsLU","rms_lvr")
+                        .select("Pq_RmsLU", "rms_lvr")
                         //电压负序
-                        .select("Pq_SeqNegU","v_neg")
+                        .select("Pq_SeqNegU", "v_neg")
                         //电压正序
-                        .select("Pq_SeqPosU","v_pos")
+                        .select("Pq_SeqPosU", "v_pos")
                         //电压零序
-                        .select("Pq_SeqZeroU","v_zero")
+                        .select("Pq_SeqZeroU", "v_zero")
                         //电压负序不平衡度
-                        .select("Pq_UnbalNegU","v_unbalance")
+                        .select("Pq_UnbalNegU", "v_unbalance")
                         .between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime())
                         .eq(DataV::getQualityFlag, "0");
                 if (CollUtil.isNotEmpty(lineParam.getPhasicType())) {
                     influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType());
                 }
                 if (Objects.isNull(po.getLineNo())) {
-                    influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getClDid()));
+                    influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getClDid()));
                 } else {
-                    influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getLineNo()));
+                    influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getLineNo()));
                 }
                 //判断接线方式 (0-星型 1-角型 2-V型) 星型是相电压  角型或者v型是线电压
-                if (Objects.equals(po.getConType(),0)) {
+                if (Objects.equals(po.getConType(), 0)) {
                     //相电压偏差
-                    influxQueryWrapper.select("Pq_UDev","vu_dev");
+                    influxQueryWrapper.select("Pq_UDev", "vu_dev");
                     //相电压谐波总畸变率
-                    influxQueryWrapper.select("Pq_ThdU","v_thd");
+                    influxQueryWrapper.select("Pq_ThdU", "v_thd");
                     //相电压基波有效值
-                    influxQueryWrapper.select("Pq_RmsFundU","v_1");
+                    influxQueryWrapper.select("Pq_RmsFundU", "v_1");
                     //2-50次 相电压谐波有效值
                     influxQueryWrapper.samePrefixAndSuffix("Pq_HarmUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
                 } else {
                     //线电压偏差
-                    influxQueryWrapper.select("Pq_LUDev","vu_dev");
+                    influxQueryWrapper.select("Pq_LUDev", "vu_dev");
                     //线电压谐波总畸变率
-                    influxQueryWrapper.select("Pq_ThdLU","v_thd");
+                    influxQueryWrapper.select("Pq_ThdLU", "v_thd");
                     //线电压基波有效值
-                    influxQueryWrapper.select("Pq_RmsFundLU","v_1");
+                    influxQueryWrapper.select("Pq_RmsFundLU", "v_1");
                     //2-50次 线电压谐波有效值
                     influxQueryWrapper.samePrefixAndSuffix("Pq_HarmLUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
                 }

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbPqsCommunicateImpl.java

@@ -129,7 +129,9 @@ public class InfluxdbPqsCommunicateImpl implements IPqsCommunicate {
         //更新mysql数据
         DevComFlagDTO devComFlagDTO = new DevComFlagDTO();
         devComFlagDTO.setId(pqsCommunicateDto.getDevId());
-        devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER));
+        if(Objects.equals(pqsCommunicateDto.getFlag(),1)){
+            devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER));
+        }
         devComFlagDTO.setStatus(pqsCommunicateDto.getType());
 
         deviceFeignClient.updateDevComFlag(devComFlagDTO);

data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java

@@ -167,6 +167,11 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
         return Collections.emptyList();
     }
 
+    @Override
+    public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
+        return null;
+    }
+
     private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) {
         List<DataVDto> result = new ArrayList<>();
         Map<String, List<DataVDto>> lineMap = list.stream().collect(Collectors.groupingBy(DataVDto::getLineId));

data-processing/data-processing-boot/src/main/resources/bootstrap-jb.yml

@@ -0,0 +1,78 @@
+#当前服务的基本信息
+microservice:
+  ename: @artifactId@
+  name: '@name@'
+  version: @version@
+  sentinel:
+    url: @sentinel.url@
+  gateway:
+    url: @gateway.url@
+server:
+  port: 10405
+  compression:
+    enabled: true
+    mime-types: application/json,application/xml,text/html,text/xml,text/plain
+    min-response-size: 1024
+#feign接口开启服务熔断降级处理
+feign:
+  sentinel:
+    enabled: true
+spring:
+  application:
+    name: @artifactId@
+  #nacos注册中心以及配置中心的指定
+  cloud:
+    nacos:
+      discovery:
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+      config:
+        username: @nacos.username@
+        password: @nacos.password@
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        file-extension: yaml
+        shared-configs:
+          - data-id: share-config.yaml
+            refresh: true
+          - data-id: share-config-datasource-db.yaml
+            refresh: true
+  main:
+    allow-bean-definition-overriding: true
+  servlet:
+    multipart:
+      max-file-size: 100MB
+      max-request-size: 100MB
+  jackson:
+    time-zone: GMT+8
+
+
+#项目日志的配置
+logging:
+  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  level:
+    root: error
+
+
+mqtt:
+  client-id: @artifactId@${random.value}
+
+data:
+  source:
+    query: Influxdb
+#    insert: Influxdb
+    insert: Relation
+#mybatis配置信息
+mybatis-plus:
+  configuration:
+    #配置sql日志输出
+    log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
+
+
+#  type-aliases-package: com.njcn.harmonic.pojo
+#  type-handlers-package: com.njcn.db.handler
+#  global-config:
+#    db-config:
+#      date-format: yyyy-MM-dd HH:mm:ss
+#  global-config:
+#    enable-sql-runner: true

data-processing/data-processing-boot/src/main/resources/bootstrap-sjzx.yml

@@ -0,0 +1,81 @@
+#当前服务的基本信息
+microservice:
+  ename: @artifactId@
+  name: '@name@'
+  version: @version@
+  sentinel:
+    url: @sentinel.url@
+  gateway:
+    url: @gateway.url@
+server:
+  port: 10405
+  compression:
+    enabled: true
+    mime-types: application/json,application/xml,text/html,text/xml,text/plain
+    min-response-size: 1024
+#feign接口开启服务熔断降级处理
+feign:
+  sentinel:
+    enabled: true
+spring:
+  application:
+    name: @artifactId@
+  #nacos注册中心以及配置中心的指定
+  cloud:
+    nacos:
+      discovery:
+        ip: @service.server.url@
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
+      config:
+        server-addr: @nacos.url@
+        namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
+        file-extension: yaml
+        shared-configs:
+          - data-id: share-config.yaml
+            refresh: true
+          - data-id: share-config-datasource-db.yaml
+            refresh: true
+  main:
+    allow-bean-definition-overriding: true
+  servlet:
+    multipart:
+      max-file-size: 100MB
+      max-request-size: 100MB
+  jackson:
+    time-zone: GMT+8
+
+
+#项目日志的配置
+logging:
+  #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  level:
+    root: error
+
+
+#mqtt:
+#  client-id: @artifactId@${random.value}
+
+data:
+  source:
+    query: Influxdb
+#    insert: Influxdb
+    insert: Relation
+#mybatis配置信息
+mybatis-plus:
+  configuration:
+    #配置sql日志输出
+    log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
+
+
+#  type-aliases-package: com.njcn.harmonic.pojo
+#  type-handlers-package: com.njcn.db.handler
+#  global-config:
+#    db-config:
+#      date-format: yyyy-MM-dd HH:mm:ss
+#  global-config:
+#    enable-sql-runner: true

data-processing/data-processing-boot/src/main/resources/bootstrap.yml

@@ -1,77 +1,3 @@
-#当前服务的基本信息
-microservice:
-  ename: @artifactId@
-  name: '@name@'
-  version: @version@
-  sentinel:
-    url: @sentinel.url@
-  gateway:
-    url: @gateway.url@
-server:
-  port: 10405
-  compression:
-    enabled: true
-    mime-types: application/json,application/xml,text/html,text/xml,text/plain
-    min-response-size: 1024
-#feign接口开启服务熔断降级处理
-feign:
-  sentinel:
-    enabled: true
 spring:
-  application:
-    name: @artifactId@
-  #nacos注册中心以及配置中心的指定
-  cloud:
-    nacos:
-      discovery:
-        ip: @service.server.url@
-        server-addr: @nacos.url@
-        namespace: @nacos.namespace@
-      config:
-        server-addr: @nacos.url@
-        namespace: @nacos.namespace@
-        file-extension: yaml
-        shared-configs:
-          - data-id: share-config.yaml
-            refresh: true
-          - data-Id: share-config-datasource-db.yaml
-            refresh: true
-  main:
-    allow-bean-definition-overriding: true
-  servlet:
-    multipart:
-      max-file-size: 100MB
-      max-request-size: 100MB
-  jackson:
-    time-zone: GMT+8
-
-
-#项目日志的配置
-logging:
-  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
-  level:
-    root: error
-
-
-mqtt:
-  client-id: @artifactId@${random.value}
-
-data:
-  source:
-    query: Influxdb
-#    insert: Influxdb
-    insert: Relation
-#mybatis配置信息
-mybatis-plus:
-  configuration:
-    #配置sql日志输出
-    log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
-
-
-#  type-aliases-package: com.njcn.harmonic.pojo
-#  type-handlers-package: com.njcn.db.handler
-#  global-config:
-#    db-config:
-#      date-format: yyyy-MM-dd HH:mm:ss
-#  global-config:
-#    enable-sql-runner: true
+  profiles:
+    active: @spring.profiles.active@

message/message-boot/src/main/resources/bootstrap.yml

@@ -22,16 +22,20 @@ spring:
         ip: @service.server.url@
         server-addr: @nacos.url@
         namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
       config:
         server-addr: @nacos.url@
         namespace: @nacos.namespace@
+        username: @nacos.username@
+        password: @nacos.password@
         file-extension: yaml
         shared-configs:
           - data-id: share-config.yaml
             refresh: true
-          - data-Id: share-config-datasource-db.yaml
+          - data-id: share-config-datasource-db.yaml
             refresh: true
-          - data-Id: data-platform.yaml
+          - data-id: data-platform.yaml
             refresh: true
   main:
     allow-bean-definition-overriding: true
@@ -39,7 +43,7 @@ spring:
 
 #项目日志的配置
 logging:
-  config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
+  #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
   level:
     root: info
 
@@ -50,4 +54,4 @@ mybatis-plus:
   type-aliases-package: com.njcn.harmonic.pojo
 
 mqtt:
-  client-id: @artifactId@${random.value}
+  client-id: @artifactId@${random.value}

pom.xml

@@ -15,7 +15,7 @@
     <module>algorithm</module>
     <module>log</module>
     <module>data-processing</module>
-    <module>migration-influxdb</module>
+    <!--<module>migration-influxdb</module>-->
   </modules>
 
   <distributionManagement>
@@ -32,47 +32,13 @@
   </distributionManagement>
 
   <properties>
+    <spring.profiles.active>sjzx</spring.profiles.active>
     <!--内网-->
-<!--        <middle.server.url>192.168.1.22</middle.server.url>-->
-<!--        <service.server.url>192.168.1.121</service.server.url>-->
-<!--        <docker.server.url>192.168.1.22</docker.server.url>-->
-<!--        <nacos.url>${middle.server.url}:18848</nacos.url>-->
-<!--        <nacos.namespace>07ec981d-1377-4920-8760-b5a98ff7e668</nacos.namespace>-->
-
-<!--          <middle.server.url>192.168.21.129</middle.server.url>-->
-<!--          <service.server.url>192.168.21.1</service.server.url>-->
-<!--          <docker.server.url>192.168.1.22</docker.server.url>-->
-<!--          <nacos.url>${middle.server.url}:18848</nacos.url>-->
-<!--          <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
-
-<!--    <middle.server.url>127.0.0.1</middle.server.url>-->
-<!--    &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
-<!--    <service.server.url>127.0.0.1</service.server.url>-->
-<!--    &lt;!&ndash;docker仓库地址&ndash;&gt;-->
-<!--    <docker.server.url>127.0.0.1</docker.server.url>-->
-<!--    &lt;!&ndash;nacos的ip:port&ndash;&gt;-->
-<!--    <nacos.url>${middle.server.url}:8848</nacos.url>-->
-<!--    <nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>-->
-<!--              <middle.server.url>24.43.102.139</middle.server.url>-->
-<!--              <service.server.url>24.43.102.139</service.server.url>-->
-<!--              <docker.server.url>24.43.102.139</docker.server.url>-->
-<!--              <nacos.url>${middle.server.url}:18848</nacos.url>-->
-<!--              <nacos.namespace></nacos.namespace>-->
-
-            <middle.server.url>127.0.0.1</middle.server.url>
-            <service.server.url>127.0.0.1</service.server.url>
-            <docker.server.url>127.0.0.1</docker.server.url>
-            <nacos.url>${middle.server.url}:8848</nacos.url>
-            <nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>
-
-<!--            <middle.server.url>192.168.21.129</middle.server.url>-->
-<!--            &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
-<!--            <service.server.url>192.168.21.1</service.server.url>-->
-<!--            &lt;!&ndash;docker仓库地址&ndash;&gt;-->
-<!--            <docker.server.url>192.168.1.22</docker.server.url>-->
-<!--          <nacos.url>${middle.server.url}:18848</nacos.url>-->
-<!--          <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
-
+    <middle.server.url>192.168.1.65</middle.server.url>
+    <service.server.url>192.168.1.65</service.server.url>
+    <docker.server.url>192.168.1.22</docker.server.url>
+    <nacos.url>${middle.server.url}:18848</nacos.url>
+    <nacos.namespace></nacos.namespace>
     <!--sentinel:port-->
     <sentinel.url>${middle.server.url}:8080</sentinel.url>
     <!--网关地址，主要用于配置swagger中认证token-->
@@ -87,6 +53,12 @@
     <micro.sdk>common-microservice</micro.sdk>
     <!--        <micro.sdk>common-huawei</micro.sdk>-->
     <docker.operate>site</docker.operate>
+
+    <!--nacos开启鉴权后的用户名-->
+    <nacos.username>nacos</nacos.username>
+    <!--nacos的ip:port-->
+    <nacos.password>njcnpqs</nacos.password>
+
     <java.version>1.8</java.version>
     <maven.compiler.source>8</maven.compiler.source>
     <maven.compiler.target>8</maven.compiler.target>