Middleware/cn-data-platform
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

污区数据部分代码提交

Browse Source
This commit is contained in:
chendaofei
2025-03-19 13:51:56 +08:00
parent b9005d7fc8
commit a72c956921
15 changed files with 652 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files

977
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java
View File

@@ -1,496 +1,481 @@
//package com.njcn.algorithm.serviceimpl.line;
//
//import cn.hutool.core.date.DatePattern;
//import cn.hutool.core.date.LocalDateTimeUtil;
//import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
//import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
//import com.njcn.algorithm.pojo.bo.CalculatedParam;
//import com.njcn.algorithm.service.line.IPollutionService;
//import com.njcn.dataProcess.api.DataHarmRateVFeignClient;
//import com.njcn.dataProcess.api.DataVFeignClient;
//import com.njcn.dataProcess.dto.DataVDTO;
//import com.njcn.dataProcess.dto.PollutionDTO;
//import com.njcn.dataProcess.param.LineCountEvaluateParam;
//import com.njcn.dataProcess.pojo.dto.DataHarmDto;
//import com.njcn.dataProcess.pojo.dto.DataVDto;
//import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
//import com.njcn.device.biz.pojo.po.Overlimit;
//import com.njcn.device.pq.pojo.dto.PublicDTO;
//import com.njcn.system.api.DicDataFeignClient;
//import com.njcn.system.enums.DicDataTypeEnum;
//import com.njcn.system.pojo.po.DictData;
//import lombok.RequiredArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.stereotype.Service;
//
//import java.math.BigDecimal;
//import java.math.RoundingMode;
//import java.time.LocalDate;
//import java.util.*;
//import java.util.function.BinaryOperator;
//import java.util.function.Function;
//import java.util.stream.Collectors;
//import java.util.stream.Stream;
//
///**
// * @Author: cdf
// * @CreateTime: 2025-03-12
// * @Description: 测点污区数据
// */
//@Service
//@Slf4j
//@RequiredArgsConstructor
//public class PollutionServiceImpl implements IPollutionService {
//
// private final DicDataFeignClient dicDataFeignClient;
//
// private final CommTerminalGeneralClient commTerminalGeneralClient;
//
// private final DataVFeignClient dataVFeignClient;
//
// private final DataHarmRateVFeignClient dataHarmRateVFeignClient;
//
// @Override
// public void handleDay(CalculatedParam<String> calculatedParam) {
// List<PollutionDTO> pollutionList ;
//
// List<DictData> dictData = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.POLLUTION_STATIS.getCode()).getData();
// LocalDate local = LocalDateTimeUtil.parseDate(calculatedParam.getDataDate());
// String beginDay =LocalDateTimeUtil.format(LocalDateTimeUtil.beginOfDay(LocalDateTimeUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN)),DatePattern.NORM_DATETIME_PATTERN);
// String endDay = LocalDateTimeUtil.format(LocalDateTimeUtil.endOfDay(LocalDateTimeUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN)),DatePattern.NORM_DATETIME_PATTERN);
// List<String> idList = calculatedParam.getIdList();
// List<Overlimit> overlimitList = commTerminalGeneralClient.getOverLimitDataByIds(idList).getData();
// Map<String,Overlimit> limitMap = overlimitList.stream().collect(Collectors.toMap(Overlimit::getId,Function.identity()));
//
//
//
// //指标数据获取
// List<PublicDTO> harmonicVoltageList = getHarmonicVoltage(idList,limitMap,beginDay,endDay);
// List<PublicDTO> harmonicCurrentList = getHarmonicCurrent(idList,limitMap,beginDay,endDay);
// List<PublicDTO> frequencyDeviationList = getFrequencyDeviation(idList,limitMap,beginDay,endDay);
// List<PublicDTO> voltageDeviationList = getVoltageDeviation(idList,limitMap,beginDay,endDay);
// List<PublicDTO> threePhaseVoltageList = getThreePhaseVoltageUnbalance(idList,limitMap,beginDay,endDay);
// List<PublicDTO> negativeSequenceList = getNegativeSequenceCurrent(idList,limitMap,beginDay,endDay);
// List<PublicDTO> interHarmonicVoltageList = getInterharmonicVoltage(idList,limitMap,beginDay,endDay);
// List<PublicDTO> voltageFlickerList = getVoltageFlicker(idList,limitMap,beginDay,endDay);
//
// List<PollutionDTO> lineIdList = new ArrayList<>();
// idList.forEach(item->{
// PollutionDTO pollutionDTO = new PollutionDTO();
// pollutionDTO.setLineId(item);
// lineIdList.add(pollutionDTO);
// });
// pollutionList = processPollutionList(lineIdList,harmonicVoltageList,harmonicCurrentList,frequencyDeviationList,voltageDeviationList,threePhaseVoltageList,negativeSequenceList,interHarmonicVoltageList,voltageFlickerList);
// insertPollutionDayMySql(pollutionList, dictData, local);
// }
//
//
// /**
// * 谐波电压：取监测点最新的A、B、C三相数据，再取电压总谐波畸变率、各次谐波电压含有率（2~25次）中的最大值，作为结果
// */
// private List<PublicDTO> getHarmonicVoltage(List<String> ids,Map<String,Overlimit> limitMap,String beginTime,String endTime){
// List<PublicDTO> list = new ArrayList<>();
//
// Map<String,PublicDTO> threePhase = getThreePhaseData(ids,limitMap,beginTime,endTime);
// Map<String,PublicDTO> distortionRate = getDistortionRateData(ids,limitMap,beginTime,endTime);
// for (String key : threePhase.keySet()) {
// list.add(threePhase.get(key));
// }
// for (String key : distortionRate.keySet()) {
// list.add(distortionRate.get(key));
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = list.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 出参处理
// */
// private List<PublicDTO> process(Map<String, Optional<PublicDTO>> outMap){
// List<PublicDTO> outList = new ArrayList<>();
// for (String key : outMap.keySet()) {
// outList.add(outMap.get(key).get());
// }
// return outList;
// }
//
// private Map<String, PublicDTO> getThreePhaseData(List<String> ids,Map<String,Overlimit> limitMap,String beginTime,String endTime){
// LineCountEvaluateParam lineCountEvaluateParam = new LineCountEvaluateParam();
// lineCountEvaluateParam.setLineId(ids);
// lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
// lineCountEvaluateParam.setPhasicType(Stream.of("A","B","C").collect(Collectors.toList()));
// lineCountEvaluateParam.setStartTime(beginTime);
// lineCountEvaluateParam.setEndTime(endTime);
// List<DataVDto> threePhaseList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData();
// Map<String,List<DataVDto>> dataVMap = threePhaseList.stream().collect(Collectors.groupingBy(DataVDto::getLineId));
//
// Map<String,PublicDTO> map = new HashMap<>();
// dataVMap.forEach((lineId,vList)->{
// PublicDTO publicDTO = new PublicDTO();
// if(limitMap.containsKey(lineId)){
// Overlimit overlimit = limitMap.get(lineId);
// double val = vList.stream().mapToDouble(DataVDto::getVThd).max().getAsDouble();
// double vUnbalance = val/overlimit.getUaberrance();
// publicDTO.setId(lineId);
// publicDTO.setData(vUnbalance);
// map.put(lineId,publicDTO);
// }
// });
// return map;
// }
//
// private Map<String,PublicDTO> getDistortionRateData(List<String> ids,Map<String,Overlimit> limitMap,String beginTime,String endTime){
// Map<String,PublicDTO> mapResult = new HashMap<>();
//
// LineCountEvaluateParam lineCountEvaluateParam = new LineCountEvaluateParam();
// lineCountEvaluateParam.setLineId(ids);
// lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
// lineCountEvaluateParam.setPhasicType(Stream.of("A","B","C").collect(Collectors.toList()));
// lineCountEvaluateParam.setStartTime(beginTime);
// lineCountEvaluateParam.setEndTime(endTime);
// List<DataHarmDto> distortionRateList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData();
//
//
//
// Map<String,List<DataHarmDto>> harmDataMap = distortionRateList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId));
// harmDataMap.forEach((lineId,vList)->{
// Overlimit overlimit = limitMap.get(lineId);
// List<PublicDTO> temList = new ArrayList<>();
// for(DataHarmDto dayHarmrateV : vList){
// PublicDTO publicDTO = new PublicDTO();
// double v2 = dayHarmrateV.getV2()/overlimit.getUharm2();
// double v3 = dayHarmrateV.getV3()/overlimit.getUharm3();
// double v4 = dayHarmrateV.getV4()/overlimit.getUharm4();
// double v5 = dayHarmrateV.getV5()/overlimit.getUharm5();
// double v6 = dayHarmrateV.getV6()/overlimit.getUharm6();
// double v7 = dayHarmrateV.getV7()/overlimit.getUharm7();
// double v8 = dayHarmrateV.getV8()/overlimit.getUharm8();
// double v9 = dayHarmrateV.getV9()/overlimit.getUharm9();
// double v10 = dayHarmrateV.getV10()/overlimit.getUharm10();
// double v11 = dayHarmrateV.getV11()/overlimit.getUharm11();
// double v12 = dayHarmrateV.getV12()/overlimit.getUharm12();
// double v13 = dayHarmrateV.getV13()/overlimit.getUharm13();
// double v14 = dayHarmrateV.getV14()/overlimit.getUharm14();
// double v15 = dayHarmrateV.getV15()/overlimit.getUharm15();
// double v16 = dayHarmrateV.getV16()/overlimit.getUharm16();
// double v17 = dayHarmrateV.getV17()/overlimit.getUharm17();
// double v18 = dayHarmrateV.getV18()/overlimit.getUharm18();
// double v19 = dayHarmrateV.getV19()/overlimit.getUharm19();
// double v20 = dayHarmrateV.getV20()/overlimit.getUharm20();
// double v21 = dayHarmrateV.getV21()/overlimit.getUharm21();
// double v22 = dayHarmrateV.getV22()/overlimit.getUharm22();
// double v23 = dayHarmrateV.getV23()/overlimit.getUharm23();
// double v24 = dayHarmrateV.getV24()/overlimit.getUharm24();
// double v25 = dayHarmrateV.getV25()/overlimit.getUharm25();
// List<Double> data = Stream.of(v2,v3,v4,v5,v6,v7,v8,v9,v10,v11,v12,v13,v14,v15,v16,v17,v18,v19,v20,v21,v22,v23,v24,v25).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO.setId(lineId);
// publicDTO.setData(result);
// temList.add(publicDTO);
// }
// PublicDTO publicDTO = new PublicDTO();
// publicDTO.setId(lineId);
// publicDTO.setData(temList.stream().mapToDouble(PublicDTO::getData).max().getAsDouble());
// mapResult.put(lineId,publicDTO);
// });
//
// return mapResult;
// }
//
// /**
// * 谐波电流：各次谐波电流（2~25次），取各监测点最新的A、B、C三相数据。
// */
// private List<PublicDTO> getHarmonicCurrent(List<String> ids,Map<String,Overlimit> limitMap,String beginTime,String endTime){
// List<RStatDataIDPO> list = dataIDService.list(new LambdaQueryWrapper<RStatDataIDPO>()
// .eq(RStatDataIDPO::getValueType, "CP95")
// .in(RStatDataIDPO::getPhaseType, Arrays.asList("A", "B", "C"))
// .ge(RStatDataIDPO::getTime, beginTime)
// .le(RStatDataIDPO::getTime, endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
// for (RStatDataIDPO dayI : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayI.getLineId(),overlimit.getId())){
// double v2 = dayI.getI2()/overlimit.getIharm2();
// double v3 = dayI.getI3()/overlimit.getIharm3();
// double v4 = dayI.getI4()/overlimit.getIharm4();
// double v5 = dayI.getI5()/overlimit.getIharm5();
// double v6 = dayI.getI6()/overlimit.getIharm6();
// double v7 = dayI.getI7()/overlimit.getIharm7();
// double v8 = dayI.getI8()/overlimit.getIharm8();
// double v9 = dayI.getI9()/overlimit.getIharm9();
// double v10 = dayI.getI10()/overlimit.getIharm10();
// double v11 = dayI.getI11()/overlimit.getIharm11();
// double v12 = dayI.getI12()/overlimit.getIharm12();
// double v13 = dayI.getI13()/overlimit.getIharm13();
// double v14 = dayI.getI14()/overlimit.getIharm14();
// double v15 = dayI.getI15()/overlimit.getIharm15();
// double v16 = dayI.getI16()/overlimit.getIharm16();
// double v17 = dayI.getI17()/overlimit.getIharm17();
// double v18 = dayI.getI18()/overlimit.getIharm18();
// double v19 = dayI.getI19()/overlimit.getIharm19();
// double v20 = dayI.getI20()/overlimit.getIharm20();
// double v21 = dayI.getI21()/overlimit.getIharm21();
// double v22 = dayI.getI22()/overlimit.getIharm22();
// double v23 = dayI.getI23()/overlimit.getIharm23();
// double v24 = dayI.getI24()/overlimit.getIharm24();
// double v25 = dayI.getI25()/overlimit.getIharm25();
// data = Stream.of(v2,v3,v4,v5,v6,v7,v8,v9,v10,v11,v12,v13,v14,v15,v16,v17,v18,v19,v20,v21,v22,v23,v24,v25).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayI.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 频率偏差：各监测点最新的T相数据，取频率上下偏差的绝对值中的最大值。
// */
// private List<PublicDTO> getFrequencyDeviation(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataVDPO> threePhaseList = dataVDService.list(new QueryWrapper<RStatDataVDPO>()
// .select("line_id","abs(freq_dev) as freq_dev ")
// .in("value_type", Arrays.asList("MIN","MAX"))
// .in("phasic_type", Arrays.asList("T"))
// .ge("time", beginTime)
// .le("time",endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//// String sql = "SELECT line_id,abs(freq_dev) AS freq_dev FROM day_v where phasic_type = 'T' and (value_type = 'MIN' or value_type = 'MAX') "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +"group by line_id order by time desc limit 2 tz('Asia/Shanghai')";
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// List<DayVPO> list = resultMapper.toPOJO(sqlResult, DayVPO.class);
// for (RStatDataVDPO dayV : threePhaseList) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayV.getLineId(),overlimit.getId())){
// double freqDev = dayV.getFreqDev()/overlimit.getFreqDev();
// data = Stream.of(freqDev).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayV.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 电压偏差：各监测点最新的A、B、C三相数据，取电压上下偏差的绝对值中的最大值。
// */
// private List<PublicDTO> getVoltageDeviation(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataVDPO> list = dataVDService.list(new QueryWrapper<RStatDataVDPO>()
// .select("line_id","vu_dev","vl_dev","value_type")
// .in("value_type", Arrays.asList("MIN","MAX"))
// .in("phasic_type", Arrays.asList("A","B","C"))
// .ge("time", beginTime)
// .le("time", endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//// String sql = "SELECT line_id,vu_dev,vl_dev,value_type FROM day_v where (phasic_type = 'A' or phasic_type = 'B' or phasic_type = 'C') and (value_type = 'MIN' or value_type = 'MAX') "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +" group by line_id order by time desc limit 6 tz('Asia/Shanghai')";
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// List<DayVPO> list = resultMapper.toPOJO(sqlResult, DayVPO.class);
// for (RStatDataVDPO dayV : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayV.getLineId(),overlimit.getId())){
// double vlDev = Math.abs(dayV.getVlDev()/overlimit.getUvoltageDev());
// double vuDev = Math.abs(dayV.getVuDev()/overlimit.getVoltageDev());
// data = Stream.of(vuDev,vlDev).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayV.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 三相电压不平衡度：各监测点最新的T相数据。
// */
// private List<PublicDTO> getThreePhaseVoltageUnbalance(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataVDPO> list = dataVDService.list(new QueryWrapper<RStatDataVDPO>()
// .select("line_id","v_unbalance","value_type")
// .in("value_type", Arrays.asList("CP95","MAX"))
// .in("phasic_type", Arrays.asList("T"))
// .ge("time", beginTime)
// .le("time", endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//// String sql = "SELECT line_id,v_unbalance,value_type FROM day_v where phasic_type = 'T' and (value_type = 'CP95' or value_type = 'MAX') "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +" group by line_id order by time desc limit 2 tz('Asia/Shanghai')";
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// List<DayVPO> list = resultMapper.toPOJO(sqlResult, DayVPO.class);
// for (RStatDataVDPO dayV : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayV.getLineId(),overlimit.getId())){
// double vUnbalance = Math.abs(dayV.getVUnbalance()/overlimit.getUbalance());
// data = Stream.of(vUnbalance).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayV.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 负序电流：各监测点最新的T相数据。
// */
// private List<PublicDTO> getNegativeSequenceCurrent(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataIDPO> list = dataIDService.list(new QueryWrapper<RStatDataIDPO>()
// .select("line_id","i_neg","value_type")
// .in("value_type", Arrays.asList("CP95","MAX"))
// .in("phasic_type", Arrays.asList("T"))
// .ge("time", beginTime)
// .le("time", endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//// String sql = "SELECT line_id,i_neg,value_type FROM day_i where phasic_type = 'T' and (value_type = 'CP95' or value_type = 'MAX') "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +" group by line_id order by time desc limit 2 tz('Asia/Shanghai')";
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// List<DayIPO> list = resultMapper.toPOJO(sqlResult, DayIPO.class);
// for (RStatDataIDPO dayI : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayI.getLineId(),overlimit.getId()) && overlimit.getINeg() != 0){
// BigDecimal dayiNeg = BigDecimal.valueOf(dayI.getINeg());
// BigDecimal overNeg = BigDecimal.valueOf(overlimit.getINeg());
// double outNeg = dayiNeg.divide(overNeg,9, RoundingMode.HALF_UP).doubleValue();
// double iNeg = Math.abs(outNeg);
// data = Stream.of(iNeg).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayI.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 间谐波电压含有率：各监测点最新的A、B、C三相数据。
// */
// private List<PublicDTO> getInterharmonicVoltage(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataInharmVDPO> list = inharmVDService.list(new QueryWrapper<RStatDataInharmVDPO>()
// .in("value_type", Arrays.asList("CP95"))
// .in("phasic_type", Arrays.asList("A","B","C"))
// .ge("time", beginTime)
// .le("time", endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//// String sql = "SELECT * FROM day_inharm_v where (phasic_type = 'A' or phasic_type = 'B' or phasic_type = 'C') and value_type = 'CP95' "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +" group by line_id order by time desc limit 3 tz('Asia/Shanghai')";
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// List<DayInharmVPO> list = resultMapper.toPOJO(sqlResult, DayInharmVPO.class);
// for (RStatDataInharmVDPO dayInharmV : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayInharmV.getLineId(),overlimit.getId())){
// double v1 = Math.abs(dayInharmV.getV1()/overlimit.getInuharm1());
// double v2 = Math.abs(dayInharmV.getV2()/overlimit.getInuharm2());
// double v3 = Math.abs(dayInharmV.getV3()/overlimit.getInuharm3());
// double v4 = Math.abs(dayInharmV.getV4()/overlimit.getInuharm4());
// double v5 = Math.abs(dayInharmV.getV5()/overlimit.getInuharm5());
// double v6 = Math.abs(dayInharmV.getV6()/overlimit.getInuharm6());
// double v7 = Math.abs(dayInharmV.getV7()/overlimit.getInuharm7());
// double v8 = Math.abs(dayInharmV.getV8()/overlimit.getInuharm8());
// double v9 = Math.abs(dayInharmV.getV9()/overlimit.getInuharm9());
// double v10 = Math.abs(dayInharmV.getV10()/overlimit.getInuharm10());
// double v11 = Math.abs(dayInharmV.getV11()/overlimit.getInuharm11());
// double v12 = Math.abs(dayInharmV.getV12()/overlimit.getInuharm12());
// double v13 = Math.abs(dayInharmV.getV13()/overlimit.getInuharm13());
// double v14 = Math.abs(dayInharmV.getV14()/overlimit.getInuharm14());
// double v15 = Math.abs(dayInharmV.getV15()/overlimit.getInuharm15());
// double v16 = Math.abs(dayInharmV.getV16()/overlimit.getInuharm16());
// data = Stream.of(v1,v2,v3,v4,v5,v6,v7,v8,v9,v10,v11,v12,v13,v14,v15,v16).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayInharmV.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
// /**
// * 长时电压闪变：各监测点最新的A、B、C三相数据。
// */
// private List<PublicDTO> getVoltageFlicker(List<Overlimit> overLimitList,String beginTime,String endTime){
// List<RStatDataPltDPO> list = pltDService.list(new QueryWrapper<RStatDataPltDPO>()
// .in("value_type", Arrays.asList("CP95"))
// .in("phasic_type", Arrays.asList("A","B","C"))
// .ge("time", beginTime)
// .le("time", endTime)
// );
// List<Double> data;
// PublicDTO publicDTO;
// List<PublicDTO> lineData = new ArrayList<>();
//
//// String sql = "SELECT * FROM day_plt where (phasic_type = 'A' or phasic_type = 'B' or phasic_type = 'C') and value_type = 'CP95' "+ processDate(dataDate,Integer.valueOf(BizParamConstant.STAT_BIZ_DAY)) +" group by line_id order by time desc limit 3 tz('Asia/Shanghai')";
//// InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
////
//// QueryResult sqlResult = influxDbUtils.query(sql);
//// List<DayPltPO> list = resultMapper.toPOJO(sqlResult, DayPltPO.class);
// for (RStatDataPltDPO dayPlt : list) {
// for (Overlimit overlimit : overLimitList) {
// if (Objects.equals(dayPlt.getLineId(),overlimit.getId())){
// double plt = Math.abs(dayPlt.getPlt()/overlimit.getFlicker());
// data = Stream.of(plt).collect(Collectors.toList());
// double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
// publicDTO = new PublicDTO();
// publicDTO.setId(dayPlt.getLineId());
// publicDTO.setData(result);
// lineData.add(publicDTO);
// }
// }
// }
// Comparator<PublicDTO> comparator = Comparator.comparing(PublicDTO::getData);
// Map<String, Optional<PublicDTO>> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
// return process(outMap);
// }
//
//
//}
package com.njcn.algorithm.serviceimpl.line;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.api.R;
import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.service.line.IPollutionService;
import com.njcn.dataProcess.api.DataHarmRateVFeignClient;
import com.njcn.dataProcess.api.DataIFeignClient;
import com.njcn.dataProcess.api.DataPltFeignClient;
import com.njcn.dataProcess.api.DataVFeignClient;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.PollutionDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.po.relation.RMpPollutionDPO;
import com.njcn.dataProcess.pojo.dto.DataHarmDto;
import com.njcn.dataProcess.pojo.dto.DataIDto;
import com.njcn.dataProcess.pojo.dto.DataPltDto;
import com.njcn.dataProcess.pojo.dto.DataVDto;
import com.njcn.dataProcess.pojo.po.RStatDataVD;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.po.Overlimit;
import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.enums.DicDataTypeEnum;
import com.njcn.system.pojo.po.DictData;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.util.*;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* @Author: cdf
* @CreateTime: 2025-03-12
* @Description: 测点污区数据
*/
@Service
@Slf4j
@RequiredArgsConstructor
public class PollutionServiceImpl implements IPollutionService {
private final DicDataFeignClient dicDataFeignClient;
private final CommTerminalGeneralClient commTerminalGeneralClient;
private final DataVFeignClient dataVFeignClient;
private final DataHarmRateVFeignClient dataHarmRateVFeignClient;
private final DataPltFeignClient dataPltFeignClient;
private final DataIFeignClient dataIFeignClient;
@Override
public void handleDay(CalculatedParam<String> calculatedParam) {
List<RMpPollutionDPO> pollutionList;
List<DictData> dictDataList = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.POLLUTION_STATIS.getCode()).getData();
Map<String, DictData> dictData = dictDataList.stream().collect(Collectors.toMap(DictData::getId, Function.identity()));
LocalDate local = LocalDateTimeUtil.parseDate(calculatedParam.getDataDate());
String beginDay = LocalDateTimeUtil.format(LocalDateTimeUtil.beginOfDay(LocalDateTimeUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN)), DatePattern.NORM_DATETIME_PATTERN);
String endDay = LocalDateTimeUtil.format(LocalDateTimeUtil.endOfDay(LocalDateTimeUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN)), DatePattern.NORM_DATETIME_PATTERN);
List<String> idList = calculatedParam.getIdList();
List<Overlimit> overlimitList = commTerminalGeneralClient.getOverLimitDataByIds(idList).getData();
Map<String, Overlimit> limitMap = overlimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
LineCountEvaluateParam lineCountEvaluateParam = new LineCountEvaluateParam();
lineCountEvaluateParam.setLineId(idList);
lineCountEvaluateParam.setStartTime(beginDay);
lineCountEvaluateParam.setEndTime(endDay);
//指标数据获取
Map<String, RMpPollutionDPO> harmonicVoltageList = getHarmonicVoltage(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> harmonicCurrentList = getHarmonicCurrent(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> frequencyDeviationList = getFrequencyDeviation(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> voltageDeviationList = getVoltageDeviation(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> threePhaseVoltageList = getThreePhaseVoltageUnbalance(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> negativeSequenceList = getNegativeSequenceCurrent(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> interHarmonicVoltageList = getInterharmonicVoltage(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> voltageFlickerList = getVoltageFlicker(lineCountEvaluateParam, limitMap);
pollutionList = processPollutionList(local,idList,dictData, harmonicVoltageList, harmonicCurrentList, frequencyDeviationList, voltageDeviationList,
threePhaseVoltageList, negativeSequenceList, interHarmonicVoltageList, voltageFlickerList);
//TODO 插入数据库
}
private void assPollution(String lineId, LocalDate local, DictData dic, Map<String, RMpPollutionDPO> map,List<RMpPollutionDPO> result) {
if(map.containsKey(lineId)){
RMpPollutionDPO tem = map.get(lineId);
tem.setPollutionType(dic.getId());
tem.setDataDate(local);
result.add(tem);
}else{
RMpPollutionDPO tem = new RMpPollutionDPO();
tem.setLineId(lineId);
tem.setPollutionType(dic.getId());
tem.setDataDate(local);
tem.setValue(0.0);
result.add(tem);
}
}
/**
* 参数拼装处理
*/
private List<RMpPollutionDPO> processPollutionList(LocalDate local, List<String> lineIds, Map<String, DictData> dictData,
Map<String, RMpPollutionDPO> harmonicVoltageMap, Map<String, RMpPollutionDPO> harmonicCurrentMap,
Map<String, RMpPollutionDPO> frequencyDeviationMap, Map<String, RMpPollutionDPO> voltageDeviationMap,
Map<String, RMpPollutionDPO> threePhaseVoltageMap, Map<String, RMpPollutionDPO> negativeSequenceMap,
Map<String, RMpPollutionDPO> interharmonicVoltageMap, Map<String,RMpPollutionDPO> voltageFlickerMap) {
List<RMpPollutionDPO> result = new ArrayList<>();
DictData dicVHarmonic = dictData.get("V_Harmonic");
DictData dicIHarmonic = dictData.get("I_All");
DictData I_Neg = dictData.get("I_Neg");
DictData V_Inharm = dictData.get("V_Inharm");
DictData V_Dev = dictData.get("V_Dev");
DictData V_Unbalance = dictData.get("V_Unbalance");
DictData Plt = dictData.get("Plt");
DictData Freq_Dev = dictData.get("Freq_Dev");
for (String lineId : lineIds) {
assPollution(lineId,local,dicVHarmonic,harmonicVoltageMap,result);
assPollution(lineId,local,dicIHarmonic,harmonicCurrentMap,result);
assPollution(lineId,local,I_Neg,negativeSequenceMap,result);
assPollution(lineId,local,Plt,voltageFlickerMap,result);
assPollution(lineId,local,V_Dev,voltageDeviationMap,result);
assPollution(lineId,local,V_Unbalance,threePhaseVoltageMap,result);
assPollution(lineId,local,Freq_Dev,frequencyDeviationMap,result);
assPollution(lineId,local,V_Inharm,interharmonicVoltageMap,result);
}
return result;
}
/**
* 谐波电压：取监测点最新的A、B、C三相数据，再取电压总谐波畸变率、各次谐波电压含有率（2~25次）中的最大值，作为结果
*/
private Map<String, RMpPollutionDPO> getHarmonicVoltage(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
List<RMpPollutionDPO> list = new ArrayList<>();
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
Map<String, RMpPollutionDPO> threePhase = getThreePhaseData(lineCountEvaluateParam, limitMap);
Map<String, RMpPollutionDPO> distortionRate = getDistortionRateData(lineCountEvaluateParam, limitMap);
for (String key : threePhase.keySet()) {
list.add(threePhase.get(key));
}
for (String key : distortionRate.keySet()) {
list.add(distortionRate.get(key));
}
Comparator<RMpPollutionDPO> comparator = Comparator.comparing(RMpPollutionDPO::getValue);
Map<String, Optional<RMpPollutionDPO>> outMap = list.stream().collect(Collectors.groupingBy(RMpPollutionDPO::getLineId, Collectors.reducing(BinaryOperator.maxBy(comparator))));
return process(outMap);
}
/**
* 出参处理
*/
private Map<String, RMpPollutionDPO> process(Map<String, Optional<RMpPollutionDPO>> outMap) {
Map<String, RMpPollutionDPO> map = new HashMap<>();
for (String key : outMap.keySet()) {
map.put(key, outMap.get(key).get());
}
return map;
}
private Map<String, RMpPollutionDPO> getThreePhaseData(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
List<DataVDto> threePhaseList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData();
Map<String, List<DataVDto>> dataVMap = threePhaseList.stream().collect(Collectors.groupingBy(DataVDto::getLineId));
Map<String, RMpPollutionDPO> map = new HashMap<>();
dataVMap.forEach((lineId, vList) -> {
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double val = vList.stream().mapToDouble(DataVDto::getVThd).max().getAsDouble();
double vUnbalance = val / overlimit.getUaberrance();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(vUnbalance);
map.put(lineId, RMpPollutionDPO);
}
});
return map;
}
private Map<String, RMpPollutionDPO> getDistortionRateData(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> mapResult = new HashMap<>();
List<DataHarmDto> distortionRateList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData();
Map<String, List<DataHarmDto>> harmDataMap = distortionRateList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId));
harmDataMap.forEach((lineId, vList) -> {
Overlimit overlimit = limitMap.get(lineId);
List<RMpPollutionDPO> temList = new ArrayList<>();
for (DataHarmDto dayHarmrateV : vList) {
RMpPollutionDPO tem = new RMpPollutionDPO();
double v2 = dayHarmrateV.getV2() / overlimit.getUharm2();
double v3 = dayHarmrateV.getV3() / overlimit.getUharm3();
double v4 = dayHarmrateV.getV4() / overlimit.getUharm4();
double v5 = dayHarmrateV.getV5() / overlimit.getUharm5();
double v6 = dayHarmrateV.getV6() / overlimit.getUharm6();
double v7 = dayHarmrateV.getV7() / overlimit.getUharm7();
double v8 = dayHarmrateV.getV8() / overlimit.getUharm8();
double v9 = dayHarmrateV.getV9() / overlimit.getUharm9();
double v10 = dayHarmrateV.getV10() / overlimit.getUharm10();
double v11 = dayHarmrateV.getV11() / overlimit.getUharm11();
double v12 = dayHarmrateV.getV12() / overlimit.getUharm12();
double v13 = dayHarmrateV.getV13() / overlimit.getUharm13();
double v14 = dayHarmrateV.getV14() / overlimit.getUharm14();
double v15 = dayHarmrateV.getV15() / overlimit.getUharm15();
double v16 = dayHarmrateV.getV16() / overlimit.getUharm16();
double v17 = dayHarmrateV.getV17() / overlimit.getUharm17();
double v18 = dayHarmrateV.getV18() / overlimit.getUharm18();
double v19 = dayHarmrateV.getV19() / overlimit.getUharm19();
double v20 = dayHarmrateV.getV20() / overlimit.getUharm20();
double v21 = dayHarmrateV.getV21() / overlimit.getUharm21();
double v22 = dayHarmrateV.getV22() / overlimit.getUharm22();
double v23 = dayHarmrateV.getV23() / overlimit.getUharm23();
double v24 = dayHarmrateV.getV24() / overlimit.getUharm24();
double v25 = dayHarmrateV.getV25() / overlimit.getUharm25();
List<Double> data = Stream.of(v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18, v19, v20, v21, v22, v23, v24, v25).collect(Collectors.toList());
double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
tem.setLineId(lineId);
tem.setValue(result);
temList.add(tem);
}
RMpPollutionDPO rMpPollutionDPO = new RMpPollutionDPO();
rMpPollutionDPO.setLineId(lineId);
rMpPollutionDPO.setValue(temList.stream().mapToDouble(RMpPollutionDPO::getValue).max().getAsDouble());
mapResult.put(lineId, rMpPollutionDPO);
});
return mapResult;
}
/**
* 谐波电流：各次谐波电流（2~25次），取各监测点最新的A、B、C三相数据。
*/
private Map<String, RMpPollutionDPO> getHarmonicCurrent(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> map = new HashMap();
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
List<DataIDto> list = dataIFeignClient.getDataI(lineCountEvaluateParam).getData();
Map<String, List<DataIDto>> dataMap = list.stream().collect(Collectors.groupingBy(DataIDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
List<RMpPollutionDPO> temList = new ArrayList<>();
for (DataIDto dayI : valList) {
double v2 = dayI.getI2() / overlimit.getIharm2();
double v3 = dayI.getI3() / overlimit.getIharm3();
double v4 = dayI.getI4() / overlimit.getIharm4();
double v5 = dayI.getI5() / overlimit.getIharm5();
double v6 = dayI.getI6() / overlimit.getIharm6();
double v7 = dayI.getI7() / overlimit.getIharm7();
double v8 = dayI.getI8() / overlimit.getIharm8();
double v9 = dayI.getI9() / overlimit.getIharm9();
double v10 = dayI.getI10() / overlimit.getIharm10();
double v11 = dayI.getI11() / overlimit.getIharm11();
double v12 = dayI.getI12() / overlimit.getIharm12();
double v13 = dayI.getI13() / overlimit.getIharm13();
double v14 = dayI.getI14() / overlimit.getIharm14();
double v15 = dayI.getI15() / overlimit.getIharm15();
double v16 = dayI.getI16() / overlimit.getIharm16();
double v17 = dayI.getI17() / overlimit.getIharm17();
double v18 = dayI.getI18() / overlimit.getIharm18();
double v19 = dayI.getI19() / overlimit.getIharm19();
double v20 = dayI.getI20() / overlimit.getIharm20();
double v21 = dayI.getI21() / overlimit.getIharm21();
double v22 = dayI.getI22() / overlimit.getIharm22();
double v23 = dayI.getI23() / overlimit.getIharm23();
double v24 = dayI.getI24() / overlimit.getIharm24();
double v25 = dayI.getI25() / overlimit.getIharm25();
List<Double> data = Stream.of(v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18, v19, v20, v21, v22, v23, v24, v25).collect(Collectors.toList());
double result = data.stream().max(Comparator.comparing(Double::doubleValue)).get();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(dayI.getLineId());
RMpPollutionDPO.setValue(result);
temList.add(RMpPollutionDPO);
}
double val = temList.stream().mapToDouble(RMpPollutionDPO::getValue).max().getAsDouble();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(val);
map.put(lineId, RMpPollutionDPO);
}
});
return map;
}
/**
* 频率偏差：各监测点最新的T相数据，取频率上下偏差的绝对值中的最大值。
*/
private Map<String, RMpPollutionDPO> getFrequencyDeviation(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setColumnName("freq_dev,line_id");
lineCountEvaluateParam.setValueType(Stream.of("MAX").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
List<DataVDto> dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData();
Map<String, List<DataVDto>> dataMap = dataVDtoList.stream().collect(Collectors.groupingBy(DataVDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double dayV = valList.stream().map(it -> Math.abs(it.getFreqDev())).max(Comparator.comparing(Double::doubleValue)).get();
double freqDev = dayV / overlimit.getFreqDev();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(freqDev);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
/**
* 电压偏差：各监测点最新的A、B、C三相数据，取电压上下偏差的绝对值中的最大值。
*/
private Map<String, RMpPollutionDPO> getVoltageDeviation(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setColumnName("vu_dev,line_id");
lineCountEvaluateParam.setValueType(Stream.of("MAX").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
List<DataVDto> dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData();
Map<String, List<DataVDto>> dataMap = dataVDtoList.stream().collect(Collectors.groupingBy(DataVDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double dayV = valList.stream().map(it -> Math.abs(it.getVuDev())).max(Comparator.comparing(Double::doubleValue)).get();
double vDev = dayV / overlimit.getUvoltageDev();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(vDev);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
/**
* 三相电压不平衡度：各监测点最新的T相数据。
*/
private Map<String, RMpPollutionDPO> getThreePhaseVoltageUnbalance(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setColumnName("v_unbalance,line_id");
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("T").collect(Collectors.toList()));
List<DataVDto> dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData();
Map<String, List<DataVDto>> dataMap = dataVDtoList.stream().collect(Collectors.groupingBy(DataVDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double dayV = valList.stream().mapToDouble(DataVDto::getVUnbalance).max().getAsDouble();
double vUb = dayV / overlimit.getUbalance();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(vUb);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
/**
* 负序电流：各监测点最新的T相数据。
*/
private Map<String, RMpPollutionDPO> getNegativeSequenceCurrent(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setColumnName("i_neg,line_id");
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("T").collect(Collectors.toList()));
List<DataIDto> dataVDtoList = dataIFeignClient.getDataI(lineCountEvaluateParam).getData();
Map<String, List<DataIDto>> dataMap = dataVDtoList.stream().collect(Collectors.groupingBy(DataIDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double dayI = valList.stream().mapToDouble(DataIDto::getINeg).max().getAsDouble();
double iNeg = dayI / overlimit.getINeg();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(iNeg);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
/**
* 间谐波电压含有率：各监测点最新的A、B、C三相数据。
*/
private Map<String, RMpPollutionDPO> getInterharmonicVoltage(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
List<DataHarmDto> dataHarmDtoList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData();
Map<String, List<DataHarmDto>> dataMap = dataHarmDtoList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
List<RMpPollutionDPO> temList = new ArrayList<>();
for (DataHarmDto dayInharmV : valList) {
double v1 = Math.abs(dayInharmV.getV1() / overlimit.getInuharm1());
double v2 = Math.abs(dayInharmV.getV2() / overlimit.getInuharm2());
double v3 = Math.abs(dayInharmV.getV3() / overlimit.getInuharm3());
double v4 = Math.abs(dayInharmV.getV4() / overlimit.getInuharm4());
double v5 = Math.abs(dayInharmV.getV5() / overlimit.getInuharm5());
double v6 = Math.abs(dayInharmV.getV6() / overlimit.getInuharm6());
double v7 = Math.abs(dayInharmV.getV7() / overlimit.getInuharm7());
double v8 = Math.abs(dayInharmV.getV8() / overlimit.getInuharm8());
double v9 = Math.abs(dayInharmV.getV9() / overlimit.getInuharm9());
double v10 = Math.abs(dayInharmV.getV10() / overlimit.getInuharm10());
double v11 = Math.abs(dayInharmV.getV11() / overlimit.getInuharm11());
double v12 = Math.abs(dayInharmV.getV12() / overlimit.getInuharm12());
double v13 = Math.abs(dayInharmV.getV13() / overlimit.getInuharm13());
double v14 = Math.abs(dayInharmV.getV14() / overlimit.getInuharm14());
double v15 = Math.abs(dayInharmV.getV15() / overlimit.getInuharm15());
double v16 = Math.abs(dayInharmV.getV16() / overlimit.getInuharm16());
double data = Stream.of(v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16).max(Comparator.comparing(Double::doubleValue)).get();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(data);
temList.add(RMpPollutionDPO);
}
double val = temList.stream().map(RMpPollutionDPO::getValue).max(Comparator.comparing(Double::doubleValue)).get();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(val);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
/**
* 长时电压闪变：各监测点最新的A、B、C三相数据。
*/
private Map<String, RMpPollutionDPO> getVoltageFlicker(LineCountEvaluateParam lineCountEvaluateParam, Map<String, Overlimit> limitMap) {
Map<String, RMpPollutionDPO> lineData = new HashMap<>();
lineCountEvaluateParam.setColumnName("plt");
lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList()));
lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList()));
List<DataPltDto> dataPltDtoList = dataPltFeignClient.getDataPlt(lineCountEvaluateParam).getData();
Map<String, List<DataPltDto>> dataMap = dataPltDtoList.stream().collect(Collectors.groupingBy(DataPltDto::getLineId));
dataMap.forEach((lineId, valList) -> {
if (limitMap.containsKey(lineId)) {
Overlimit overlimit = limitMap.get(lineId);
double dayPlt = valList.stream().mapToDouble(DataPltDto::getPlt).max().getAsDouble();
double plt = dayPlt / overlimit.getFlicker();
RMpPollutionDPO RMpPollutionDPO = new RMpPollutionDPO();
RMpPollutionDPO.setLineId(lineId);
RMpPollutionDPO.setValue(plt);
lineData.put(lineId, RMpPollutionDPO);
}
});
return lineData;
}
}

3
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIFeignClient.java
View File

@@ -32,4 +32,7 @@ public interface DataIFeignClient {
@PostMapping("/addList")
HttpResult<String> addList(@RequestBody List<DataIDto> dataIDtoList);
@PostMapping("/getDataI")
HttpResult<List<DataIDto>> getDataI(@RequestBody LineCountEvaluateParam lineParam);
}

2
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataPltFeignClient.java
View File

@@ -35,4 +35,6 @@ public interface DataPltFeignClient {
@PostMapping("/addList")
HttpResult<String> addList(@RequestBody List<DataPltDto> list);
@PostMapping("/getDataPlt")
HttpResult<List<DataPltDto>> getDataPlt(@RequestBody LineCountEvaluateParam lineParam);
}

6
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIFeignClientFallbackFactory.java
View File

@@ -64,6 +64,12 @@ public class DataIFeignClientFallbackFactory implements FallbackFactory<DataIFei
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<DataIDto>> getDataI(LineCountEvaluateParam lineParam) {
log.error("{}异常，降级处理，异常为：{}","获取电流数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

6
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataPltFeignClientFallbackFactory.java
View File

@@ -64,6 +64,12 @@ public class DataPltFeignClientFallbackFactory implements FallbackFactory<DataPl
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<DataPltDto>> getDataPlt(LineCountEvaluateParam lineParam) {
log.error("{}异常，降级处理，异常为：{}","获取闪变数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

47
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/relation/RMpPollutionDPO.java Normal file
View File

@@ -0,0 +1,47 @@
package com.njcn.dataProcess.po.relation;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.time.LocalDate;
/**
* @Author: cdf
* @CreateTime: 2025-03-19
* @Description: 监测点污区实体
*/
@Data
@TableName(value = "r_mp_pollution_d")
public class RMpPollutionDPO implements Serializable {
private static final long serialVersionUID = 1L;
@TableField(value = "line_id")
@ApiModelProperty(value = "监测点")
private String lineId;
/**
* 时间
*/
@TableField(value = "data_date")
@ApiModelProperty(value = "时间")
private LocalDate dataDate;
/**
* 污区图统计类型
*/
@TableField(value = "pollution_type")
@ApiModelProperty(value = "污区图统计类型")
private String pollutionType;
/**
* 统计值
*/
@TableField(value = "value")
@ApiModelProperty(value = "统计值")
private Double value;
}

10
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIController.java
View File

@@ -85,11 +85,13 @@ public class DataIController extends BaseController {
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getDataI")
@ApiOperation("关系型数据库插入数据")
public HttpResult<String> getDataI(@RequestBody LineCountEvaluateParam lineParam) {
@ApiOperation("获取谐波电流")
public HttpResult<List<DataIDto>> getDataI(@RequestBody LineCountEvaluateParam lineParam) {
String methodDescribe = getMethodDescribe("getDataI");
// dataIInsert.getDataI(dataIDtoList);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "", methodDescribe);
List<DataIDto> dataIDtoList = dataIInsert.getDataI(lineParam);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, dataIDtoList, methodDescribe);
}
}

10
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataPltController.java
View File

@@ -83,4 +83,14 @@ public class DataPltController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "", methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getDataPlt")
@ApiOperation("获取闪变数据")
public HttpResult<List<DataPltDto>> getDataPlt(@RequestBody LineCountEvaluateParam lineParam) {
String methodDescribe = getMethodDescribe("getDataPlt");
List<DataPltDto> data = dataPltQuery.getDataPlt(lineParam);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe);
}
}

9
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataI.java
View File

@@ -38,4 +38,13 @@ public interface IDataI extends IMppService<RStatDataID> {
* 批量插入数据
*/
void addList(List<DataIDto> dataIDtoList);
/**
* 获取谐波电流
* @param lineParam
* @return
*/
List<DataIDto> getDataI(LineCountEvaluateParam lineParam);
}

2
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataPlt.java
View File

@@ -36,4 +36,6 @@ public interface IDataPlt extends IMppService<RStatDataPltD> {
* 批量插入数据
*/
void addList(List<DataPltDto> list);
List<DataPltDto> getDataPlt(LineCountEvaluateParam lineParam);
}

10
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataIImpl.java
View File

@@ -24,10 +24,7 @@ import org.springframework.stereotype.Service;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.stream.Collectors;
/**
@@ -186,6 +183,11 @@ public class InfluxdbDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
}
@Override
public List<DataIDto> getDataI(LineCountEvaluateParam lineParam) {
return Collections.emptyList();
}
/**
* 按监测点集合、时间条件获取dataI分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理

5
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataPltImpl.java
View File

@@ -118,6 +118,11 @@ public class InfluxdbDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
}
@Override
public List<DataPltDto> getDataPlt(LineCountEvaluateParam lineParam) {
return Collections.emptyList();
}
/**
* 按监测点集合、时间条件获取dataI分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理

31
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java
View File

@@ -1,5 +1,10 @@
package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.DataIRelationMapper;
import com.njcn.dataProcess.dao.relation.mapper.RStatDataIRelationMapper;
@@ -18,7 +23,9 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -88,4 +95,28 @@ public class RelationDataIImpl extends MppServiceImpl<RStatDataIRelationMapper,
});
iDataI.saveOrUpdateBatchByMultiId(result);
}
@Override
public List<DataIDto> getDataI(LineCountEvaluateParam lineParam) {
List<DataIDto> result = new ArrayList<>();
QueryWrapper<RStatDataID> queryWrapper = new QueryWrapper<>();
if(StrUtil.isNotBlank(lineParam.getColumnName())){
queryWrapper.select(lineParam.getColumnName());
}
queryWrapper.lambda()
.in(RStatDataID::getValueType, lineParam.getValueType())
.in(RStatDataID::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataID::getTime, lineParam.getStartTime())
.le(RStatDataID::getTime, lineParam.getEndTime())
.in(RStatDataID::getLineId,lineParam.getLineId());
List<RStatDataID> list = this.list(queryWrapper);
for(RStatDataID rStatDataID:list){
DataIDto dto =BeanUtil.copyProperties(rStatDataID,DataIDto.class);
dto.setTime(rStatDataID.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
result.add(dto);
}
return result;
}
}

29
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java
View File

@@ -1,5 +1,9 @@
package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.DataPltRelationMapper;
import com.njcn.dataProcess.dao.relation.mapper.RStatDataPltRelationMapper;
@@ -8,7 +12,9 @@ import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.po.relation.DataPlt;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataPltDto;
import com.njcn.dataProcess.pojo.dto.DataVDto;
import com.njcn.dataProcess.pojo.po.RStatDataPltD;
import com.njcn.dataProcess.pojo.po.RStatDataVD;
import com.njcn.dataProcess.service.IDataPlt;
import com.njcn.dataProcess.util.BeanFeildUtils;
import com.njcn.dataProcess.util.TimeUtils;
@@ -17,6 +23,7 @@ import org.apache.commons.collections4.ListUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -85,4 +92,26 @@ public class RelationDataPltImpl extends MppServiceImpl<RStatDataPltRelationMapp
});
dataPlt.saveOrUpdateBatchByMultiId(result);
}
@Override
public List<DataPltDto> getDataPlt(LineCountEvaluateParam lineParam) {
List<DataPltDto> result = new ArrayList<>();
QueryWrapper<RStatDataPltD> queryWrapper = new QueryWrapper<>();
if(StrUtil.isNotBlank(lineParam.getColumnName())){
queryWrapper.select(lineParam.getColumnName());
}
queryWrapper.lambda()
.in(RStatDataPltD::getValueType, lineParam.getValueType())
.in(RStatDataPltD::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataPltD::getTime, lineParam.getStartTime())
.le(RStatDataPltD::getTime, lineParam.getEndTime());
List<RStatDataPltD> rStatDataVDList = this.list(queryWrapper);
for(RStatDataPltD rStatDataPltD : rStatDataVDList){
DataPltDto dto = BeanUtil.copyProperties(rStatDataPltD,DataPltDto.class);
dto.setTime(rStatDataPltD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
result.add(dto);
}
return result;
}
}

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

@@ -4,6 +4,7 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
@@ -149,12 +150,17 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
@Override
public List<DataVDto> getDataV(LineCountEvaluateParam lineParam) {
List<DataVDto> result = new ArrayList<>();
List<RStatDataVD> rStatDataVDList = iDataV.list(new LambdaQueryWrapper<RStatDataVD>()
QueryWrapper<RStatDataVD> queryWrapper = new QueryWrapper<>();
if(StrUtil.isNotBlank(lineParam.getColumnName())){
queryWrapper.select(lineParam.getColumnName());
}
queryWrapper.lambda()
.in(RStatDataVD::getValueType, lineParam.getValueType())
.in(RStatDataVD::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataVD::getTime, lineParam.getStartTime())
.le(RStatDataVD::getTime, lineParam.getEndTime())
);
.le(RStatDataVD::getTime, lineParam.getEndTime());
List<RStatDataVD> rStatDataVDList = this.list(queryWrapper);
for(RStatDataVD rStatDataVD : rStatDataVDList){
DataVDto dto = BeanUtil.copyProperties(rStatDataVD,DataVDto.class);
dto.setTime(rStatDataVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));