diff --git a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java index f63b894..f56303e 100644 --- a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java +++ b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java @@ -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 calculatedParam) { -// List pollutionList ; -// -// List 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 idList = calculatedParam.getIdList(); -// List overlimitList = commTerminalGeneralClient.getOverLimitDataByIds(idList).getData(); -// Map limitMap = overlimitList.stream().collect(Collectors.toMap(Overlimit::getId,Function.identity())); -// -// -// -// //指标数据获取 -// List harmonicVoltageList = getHarmonicVoltage(idList,limitMap,beginDay,endDay); -// List harmonicCurrentList = getHarmonicCurrent(idList,limitMap,beginDay,endDay); -// List frequencyDeviationList = getFrequencyDeviation(idList,limitMap,beginDay,endDay); -// List voltageDeviationList = getVoltageDeviation(idList,limitMap,beginDay,endDay); -// List threePhaseVoltageList = getThreePhaseVoltageUnbalance(idList,limitMap,beginDay,endDay); -// List negativeSequenceList = getNegativeSequenceCurrent(idList,limitMap,beginDay,endDay); -// List interHarmonicVoltageList = getInterharmonicVoltage(idList,limitMap,beginDay,endDay); -// List voltageFlickerList = getVoltageFlicker(idList,limitMap,beginDay,endDay); -// -// List 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 getHarmonicVoltage(List ids,Map limitMap,String beginTime,String endTime){ -// List list = new ArrayList<>(); -// -// Map threePhase = getThreePhaseData(ids,limitMap,beginTime,endTime); -// Map 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = list.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 出参处理 -// */ -// private List process(Map> outMap){ -// List outList = new ArrayList<>(); -// for (String key : outMap.keySet()) { -// outList.add(outMap.get(key).get()); -// } -// return outList; -// } -// -// private Map getThreePhaseData(List ids,Map 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 threePhaseList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData(); -// Map> dataVMap = threePhaseList.stream().collect(Collectors.groupingBy(DataVDto::getLineId)); -// -// Map 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 getDistortionRateData(List ids,Map limitMap,String beginTime,String endTime){ -// Map 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 distortionRateList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData(); -// -// -// -// Map> harmDataMap = distortionRateList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId)); -// harmDataMap.forEach((lineId,vList)->{ -// Overlimit overlimit = limitMap.get(lineId); -// List 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 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 getHarmonicCurrent(List ids,Map limitMap,String beginTime,String endTime){ -// List list = dataIDService.list(new LambdaQueryWrapper() -// .eq(RStatDataIDPO::getValueType, "CP95") -// .in(RStatDataIDPO::getPhaseType, Arrays.asList("A", "B", "C")) -// .ge(RStatDataIDPO::getTime, beginTime) -// .le(RStatDataIDPO::getTime, endTime) -// ); -// List data; -// PublicDTO publicDTO; -// List 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 频率偏差：各监测点最新的T相数据，取频率上下偏差的绝对值中的最大值。 -// */ -// private List getFrequencyDeviation(List overLimitList,String beginTime,String endTime){ -// List threePhaseList = dataVDService.list(new QueryWrapper() -// .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 data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 电压偏差：各监测点最新的A、B、C三相数据，取电压上下偏差的绝对值中的最大值。 -// */ -// private List getVoltageDeviation(List overLimitList,String beginTime,String endTime){ -// List list = dataVDService.list(new QueryWrapper() -// .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 data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 三相电压不平衡度：各监测点最新的T相数据。 -// */ -// private List getThreePhaseVoltageUnbalance(List overLimitList,String beginTime,String endTime){ -// List list = dataVDService.list(new QueryWrapper() -// .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 data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 负序电流：各监测点最新的T相数据。 -// */ -// private List getNegativeSequenceCurrent(List overLimitList,String beginTime,String endTime){ -// List list = dataIDService.list(new QueryWrapper() -// .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 data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 间谐波电压含有率：各监测点最新的A、B、C三相数据。 -// */ -// private List getInterharmonicVoltage(List overLimitList,String beginTime,String endTime){ -// List list = inharmVDService.list(new QueryWrapper() -// .in("value_type", Arrays.asList("CP95")) -// .in("phasic_type", Arrays.asList("A","B","C")) -// .ge("time", beginTime) -// .le("time", endTime) -// ); -// List data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> outMap = lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator)))); -// return process(outMap); -// } -// -// /** -// * 长时电压闪变：各监测点最新的A、B、C三相数据。 -// */ -// private List getVoltageFlicker(List overLimitList,String beginTime,String endTime){ -// List list = pltDService.list(new QueryWrapper() -// .in("value_type", Arrays.asList("CP95")) -// .in("phasic_type", Arrays.asList("A","B","C")) -// .ge("time", beginTime) -// .le("time", endTime) -// ); -// List data; -// PublicDTO publicDTO; -// List 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 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 comparator = Comparator.comparing(PublicDTO::getData); -// Map> 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 calculatedParam) { + List pollutionList; + + List dictDataList = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.POLLUTION_STATIS.getCode()).getData(); + Map 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 idList = calculatedParam.getIdList(); + List overlimitList = commTerminalGeneralClient.getOverLimitDataByIds(idList).getData(); + Map limitMap = overlimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity())); + + LineCountEvaluateParam lineCountEvaluateParam = new LineCountEvaluateParam(); + lineCountEvaluateParam.setLineId(idList); + + lineCountEvaluateParam.setStartTime(beginDay); + lineCountEvaluateParam.setEndTime(endDay); + + //指标数据获取 + Map harmonicVoltageList = getHarmonicVoltage(lineCountEvaluateParam, limitMap); + Map harmonicCurrentList = getHarmonicCurrent(lineCountEvaluateParam, limitMap); + Map frequencyDeviationList = getFrequencyDeviation(lineCountEvaluateParam, limitMap); + Map voltageDeviationList = getVoltageDeviation(lineCountEvaluateParam, limitMap); + Map threePhaseVoltageList = getThreePhaseVoltageUnbalance(lineCountEvaluateParam, limitMap); + Map negativeSequenceList = getNegativeSequenceCurrent(lineCountEvaluateParam, limitMap); + Map interHarmonicVoltageList = getInterharmonicVoltage(lineCountEvaluateParam, limitMap); + Map 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 map,List 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 processPollutionList(LocalDate local, List lineIds, Map dictData, + Map harmonicVoltageMap, Map harmonicCurrentMap, + Map frequencyDeviationMap, Map voltageDeviationMap, + Map threePhaseVoltageMap, Map negativeSequenceMap, + Map interharmonicVoltageMap, Map voltageFlickerMap) { + List 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 getHarmonicVoltage(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + List list = new ArrayList<>(); + lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList())); + lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList())); + Map threePhase = getThreePhaseData(lineCountEvaluateParam, limitMap); + Map 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 comparator = Comparator.comparing(RMpPollutionDPO::getValue); + Map> outMap = list.stream().collect(Collectors.groupingBy(RMpPollutionDPO::getLineId, Collectors.reducing(BinaryOperator.maxBy(comparator)))); + return process(outMap); + } + + /** + * 出参处理 + */ + private Map process(Map> outMap) { + Map map = new HashMap<>(); + for (String key : outMap.keySet()) { + map.put(key, outMap.get(key).get()); + } + return map; + } + + private Map getThreePhaseData(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + + List threePhaseList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData(); + Map> dataVMap = threePhaseList.stream().collect(Collectors.groupingBy(DataVDto::getLineId)); + + Map 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 getDistortionRateData(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map mapResult = new HashMap<>(); + List distortionRateList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData(); + + Map> harmDataMap = distortionRateList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId)); + harmDataMap.forEach((lineId, vList) -> { + Overlimit overlimit = limitMap.get(lineId); + List 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 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 getHarmonicCurrent(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map map = new HashMap(); + lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList())); + lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList())); + List list = dataIFeignClient.getDataI(lineCountEvaluateParam).getData(); + Map> dataMap = list.stream().collect(Collectors.groupingBy(DataIDto::getLineId)); + dataMap.forEach((lineId, valList) -> { + if (limitMap.containsKey(lineId)) { + Overlimit overlimit = limitMap.get(lineId); + List 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 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 getFrequencyDeviation(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map 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 dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData(); + Map> 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 getVoltageDeviation(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map 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 dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData(); + Map> 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 getThreePhaseVoltageUnbalance(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map 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 dataVDtoList = dataVFeignClient.getDataV(lineCountEvaluateParam).getData(); + Map> 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 getNegativeSequenceCurrent(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map 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 dataVDtoList = dataIFeignClient.getDataI(lineCountEvaluateParam).getData(); + Map> 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 getInterharmonicVoltage(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map lineData = new HashMap<>(); + lineCountEvaluateParam.setPhasicType(Stream.of("A", "B", "C").collect(Collectors.toList())); + lineCountEvaluateParam.setValueType(Stream.of("CP95").collect(Collectors.toList())); + List dataHarmDtoList = dataHarmRateVFeignClient.getHarmRateVData(lineCountEvaluateParam).getData(); + Map> dataMap = dataHarmDtoList.stream().collect(Collectors.groupingBy(DataHarmDto::getLineId)); + + dataMap.forEach((lineId, valList) -> { + if (limitMap.containsKey(lineId)) { + Overlimit overlimit = limitMap.get(lineId); + List 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 getVoltageFlicker(LineCountEvaluateParam lineCountEvaluateParam, Map limitMap) { + Map 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 dataPltDtoList = dataPltFeignClient.getDataPlt(lineCountEvaluateParam).getData(); + Map> 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; + + } + + +} diff --git a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIFeignClient.java b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIFeignClient.java index f1312c1..74b01c5 100644 --- a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIFeignClient.java +++ b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIFeignClient.java @@ -32,4 +32,7 @@ public interface DataIFeignClient { @PostMapping("/addList") HttpResult addList(@RequestBody List dataIDtoList); + + @PostMapping("/getDataI") + HttpResult> getDataI(@RequestBody LineCountEvaluateParam lineParam); } diff --git a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataPltFeignClient.java b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataPltFeignClient.java index 7abe494..7d51b0c 100644 --- a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataPltFeignClient.java +++ b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataPltFeignClient.java @@ -35,4 +35,6 @@ public interface DataPltFeignClient { @PostMapping("/addList") HttpResult addList(@RequestBody List list); + @PostMapping("/getDataPlt") + HttpResult> getDataPlt(@RequestBody LineCountEvaluateParam lineParam); } diff --git a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIFeignClientFallbackFactory.java b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIFeignClientFallbackFactory.java index 1c11019..c207748 100644 --- a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIFeignClientFallbackFactory.java +++ b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIFeignClientFallbackFactory.java @@ -64,6 +64,12 @@ public class DataIFeignClientFallbackFactory implements FallbackFactory> getDataI(LineCountEvaluateParam lineParam) { + log.error("{}异常，降级处理，异常为：{}","获取电流数据",cause.toString()); + throw new BusinessException(finalExceptionEnum); + } + }; } } diff --git a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataPltFeignClientFallbackFactory.java b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataPltFeignClientFallbackFactory.java index 443738b..01d0875 100644 --- a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataPltFeignClientFallbackFactory.java +++ b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataPltFeignClientFallbackFactory.java @@ -64,6 +64,12 @@ public class DataPltFeignClientFallbackFactory implements FallbackFactory> getDataPlt(LineCountEvaluateParam lineParam) { + log.error("{}异常，降级处理，异常为：{}","获取闪变数据",cause.toString()); + throw new BusinessException(finalExceptionEnum); + } + }; } } diff --git a/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/relation/RMpPollutionDPO.java b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/relation/RMpPollutionDPO.java new file mode 100644 index 0000000..f41a4cd --- /dev/null +++ b/data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/relation/RMpPollutionDPO.java @@ -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; + + +} diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIController.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIController.java index 26d96e7..2c81264 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIController.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIController.java @@ -85,11 +85,13 @@ public class DataIController extends BaseController { @OperateInfo(info = LogEnum.BUSINESS_COMMON) @PostMapping("/getDataI") - @ApiOperation("关系型数据库插入数据") - public HttpResult getDataI(@RequestBody LineCountEvaluateParam lineParam) { + @ApiOperation("获取谐波电流") + public HttpResult> getDataI(@RequestBody LineCountEvaluateParam lineParam) { String methodDescribe = getMethodDescribe("getDataI"); - // dataIInsert.getDataI(dataIDtoList); - return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "", methodDescribe); + List dataIDtoList = dataIInsert.getDataI(lineParam); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, dataIDtoList, methodDescribe); } + + } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataPltController.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataPltController.java index eaccf94..301f96c 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataPltController.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataPltController.java @@ -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> getDataPlt(@RequestBody LineCountEvaluateParam lineParam) { + String methodDescribe = getMethodDescribe("getDataPlt"); + List data = dataPltQuery.getDataPlt(lineParam); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe); + } + + } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataI.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataI.java index 2f36331..4dc1f0f 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataI.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataI.java @@ -38,4 +38,13 @@ public interface IDataI extends IMppService { * 批量插入数据 */ void addList(List dataIDtoList); + + + /** + * 获取谐波电流 + * @param lineParam + * @return + */ + List getDataI(LineCountEvaluateParam lineParam); + } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataPlt.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataPlt.java index 8c3d499..08f95b6 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataPlt.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataPlt.java @@ -36,4 +36,6 @@ public interface IDataPlt extends IMppService { * 批量插入数据 */ void addList(List list); + + List getDataPlt(LineCountEvaluateParam lineParam); } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataIImpl.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataIImpl.java index 99705b6..94a1770 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataIImpl.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataIImpl.java @@ -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 getDataI(LineCountEvaluateParam lineParam) { + return Collections.emptyList(); + } + /** * 按监测点集合、时间条件获取dataI分钟数据 * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataPltImpl.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataPltImpl.java index 4ebb76a..5505351 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataPltImpl.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/influxdb/InfluxdbDataPltImpl.java @@ -118,6 +118,11 @@ public class InfluxdbDataPltImpl extends MppServiceImpl getDataPlt(LineCountEvaluateParam lineParam) { + return Collections.emptyList(); + } + /** * 按监测点集合、时间条件获取dataI分钟数据 * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java index 2ea5656..061992e 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataIImpl.java @@ -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 getDataI(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + QueryWrapper 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 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; + } } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java index 09c3d1d..4eaf5c2 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataPltImpl.java @@ -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 getDataPlt(LineCountEvaluateParam lineParam) { + List result = new ArrayList<>(); + QueryWrapper 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 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; + } } diff --git a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java index 42e0229..8a4d57b 100644 --- a/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java +++ b/data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RelationDataVImpl.java @@ -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 getDataV(LineCountEvaluateParam lineParam) { List result = new ArrayList<>(); - List rStatDataVDList = iDataV.list(new LambdaQueryWrapper() + QueryWrapper 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 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)));