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-13 10:16:37 +08:00
parent 7d503a42b1
commit dc178ff920
28 changed files with 919 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IPollutionService.java Normal file
View File

@@ -0,0 +1,11 @@
package com.njcn.algorithm.service.line;
import com.njcn.algorithm.pojo.bo.CalculatedParam;
/**
* 测点污区数据
*/
public interface IPollutionService {
void handleDay(CalculatedParam<String> calculatedParam);
}

33
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/IDataIntegrityServiceImpl.java
View File

@@ -1,12 +1,16 @@
package com.njcn.algorithm.serviceimpl.line;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.service.line.IDataIntegrityService;
import com.njcn.dataProcess.api.DataIntegrityFeignClient;
import com.njcn.dataProcess.api.DataVFeignClient;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.pojo.po.RStatIntegrityD;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.dto.LineDevGetDTO;
import com.njcn.device.pq.pojo.po.RStatIntegrityD;
import com.njcn.influx.constant.InfluxDbSqlConstant;
import com.njcn.influx.pojo.bo.MeasurementCount;
import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
@@ -33,6 +37,10 @@ public class IDataIntegrityServiceImpl implements IDataIntegrityService {
@Resource
private CommTerminalGeneralClient commTerminalGeneralClient;
private final DataVFeignClient dataVFeignClient;
private final DataIntegrityFeignClient dataIntegrityFeignClient;
@Override
public void dataIntegrity(CalculatedParam<String> calculatedParam) {
@@ -50,7 +58,7 @@ public class IDataIntegrityServiceImpl implements IDataIntegrityService {
List<List<String>> pendingIds = ListUtils.partition(lineIds,5);
for (List<String> pendingId : pendingIds) {
List<LineDevGetDTO> lineDevGetDTOList = commTerminalGeneralClient.getMonitorDetailList(pendingId).getData();
List<MeasurementCount> countList = this.getMeasurementCount(pendingId,beginDay,endDay);
List<MeasurementCountDTO> countList = dataVFeignClient.getMeasurementCount(pendingId,beginDay,endDay).getData();
poList.addAll(
lineDevGetDTOList.stream()
.map(item -> {
@@ -68,27 +76,12 @@ public class IDataIntegrityServiceImpl implements IDataIntegrityService {
.collect(Collectors.toList())
);
}
if(CollUtil.isNotEmpty(poList)){
dataIntegrityFeignClient.batchInsertion(poList);
}
}
/**
* 获取data_v中各个监测点的数据总数
* @param lineIndex 监测点索引
* @param startTime 起始时间
* @param endTime 结束时间
*/
public List<MeasurementCount> getMeasurementCount(List<String> lineIndex, String startTime, String endTime) {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class,MeasurementCount.class);
influxQueryWrapper.regular(DataV::getLineId, lineIndex)
.eq(DataV::getValueType, InfluxDbSqlConstant.MAX)
.eq(DataV::getPhaseType, InfluxDBTableConstant.PHASE_TYPE_A)
.count(DataV::getFreq)
.groupBy(DataV::getLineId)
.between(DataV::getTime, startTime, endTime);
//return dataVMapper.getMeasurementCount(influxQueryWrapper);
//TODO 调用插入数据库
return null;
}
}

496
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionServiceImpl.java Normal file
View File

@@ -0,0 +1,496 @@
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);
}
}

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

@@ -30,4 +30,7 @@ public interface DataHarmRateVFeignClient {
@PostMapping("/getTopData")
HttpResult<DataHarmDto> getTopData();
@PostMapping("/getHarmRateVData")
HttpResult<List<DataHarmDto>> getHarmRateVData(@RequestBody LineCountEvaluateParam lineParam);
}

26
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataIntegrityFeignClient.java Normal file
View File

@@ -0,0 +1,26 @@
package com.njcn.dataProcess.api;
import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.fallback.DataOnlineRateFeignClientFallbackFactory;
import com.njcn.dataProcess.pojo.dto.DataIntegrityDto;
import com.njcn.dataProcess.pojo.dto.DataOnlineRateDto;
import com.njcn.dataProcess.pojo.po.RStatIntegrityD;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import java.util.List;
/**
* @Description:
* @Author: wr
* @Date: 2025/3/7 9:30
*/
@FeignClient(value = ServerInfo.PLATFORM_DATA_PROCESSING_BOOT, path = "/dataIntegrity", fallbackFactory = DataIntegrityFeignClient.class, contextId = "dataIntegrity")
public interface DataIntegrityFeignClient {
@PostMapping("/batchInsertion")
HttpResult<String> batchInsertion(@RequestBody List<RStatIntegrityD> dataList);
}

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

@@ -4,6 +4,7 @@ import com.njcn.common.pojo.constant.ServerInfo;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.fallback.DataVFeignClientFallbackFactory;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataVCvtDto;
@@ -53,4 +54,10 @@ public interface DataVFeignClient {
@PostMapping("/batchInsertionCvtDTO")
HttpResult<String> batchInsertionCvtDTO(@RequestBody List<DataVCvtDto> cvtDTOList);
@PostMapping("/getMeasurementCount")
HttpResult<List<MeasurementCountDTO>> getMeasurementCount(@RequestParam("lineIndex")List<String> lineIndex, @RequestParam("startTime")String startTime, @RequestParam("endTime")String endTime);
//获取原始数据
@PostMapping("/getDataV")
HttpResult<List<DataVDto>> getDataV(@RequestBody LineCountEvaluateParam lineParam);
}

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

@@ -61,6 +61,12 @@ public class DataHarmRateVFeignClientFallbackFactory implements FallbackFactory<
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<DataHarmDto>> getHarmRateVData(LineCountEvaluateParam lineParam) {
log.error("{}异常，降级处理，异常为：{}","获取谐波含有率数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

50
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataIntegrityFeignClientFallbackFactory.java Normal file
View File

@@ -0,0 +1,50 @@
package com.njcn.dataProcess.api.fallback;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.DataIntegrityFeignClient;
import com.njcn.dataProcess.api.DataOnlineRateFeignClient;
import com.njcn.dataProcess.pojo.dto.DataIntegrityDto;
import com.njcn.dataProcess.pojo.dto.DataOnlineRateDto;
import com.njcn.dataProcess.pojo.po.RStatIntegrityD;
import com.njcn.dataProcess.util.DataProcessingEnumUtil;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @Description:
* @Author: wr
* @Date: 2025/3/7 9:30
*/
@Slf4j
@Component
public class DataIntegrityFeignClientFallbackFactory implements FallbackFactory<DataIntegrityFeignClient> {
/**
* 输出远程请求接口异常日志
* @param cause RPC请求异常
*/
@Override
public DataIntegrityFeignClient create(Throwable cause) {
//判断抛出异常是否为解码器抛出的业务异常
Enum<?> exceptionEnum = CommonResponseEnum.SERVICE_FALLBACK;
if(cause.getCause() instanceof BusinessException){
BusinessException businessException = (BusinessException) cause.getCause();
exceptionEnum = DataProcessingEnumUtil.getExceptionEnum(businessException.getResult());
}
Enum<?> finalExceptionEnum = exceptionEnum;
return new DataIntegrityFeignClient() {
@Override
public HttpResult<String> batchInsertion(List<RStatIntegrityD> dataList) {
log.error("{}异常，降级处理，异常为：{}","批量插入数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

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

@@ -5,6 +5,7 @@ import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.dataProcess.api.DataVFeignClient;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataVCvtDto;
@@ -13,6 +14,8 @@ import com.njcn.dataProcess.util.DataProcessingEnumUtil;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import java.time.LocalDateTime;
import java.util.List;
@@ -89,6 +92,18 @@ public class DataVFeignClientFallbackFactory implements FallbackFactory<DataVFei
log.error("{}异常，降级处理，异常为：{}","cvt数据插入DataV",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<MeasurementCountDTO>> getMeasurementCount(List<String> lineIndex,String startTime,String endTime){
log.error("{}异常，降级处理，异常为：{}","cvt数据插入DataV",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<List<DataVDto>> getDataV(@RequestBody LineCountEvaluateParam lineParam) {
log.error("{}异常，降级处理，异常为：{}","cvt数据插入DataV",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

29
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/dto/MeasurementCountDTO.java Normal file
View File

@@ -0,0 +1,29 @@
package com.njcn.dataProcess.dto;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.njcn.influx.utils.InstantDateSerializer;
import lombok.Data;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Measurement;
import java.time.Instant;
/**
* @Author: cdf
* @CreateTime: 2025-03-12
* @Description:
*/
@Data
@Measurement(name = "data_v")
public class MeasurementCountDTO {
@Column(name = "time")
@JsonSerialize(using = InstantDateSerializer.class)
private Instant time;
@Column(name = "line_id")
private String lineId;
@Column(name = "freq")
private String freq;
}

64
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/dto/PollutionDTO.java Normal file
View File

@@ -0,0 +1,64 @@
package com.njcn.dataProcess.dto;
import lombok.Data;
/**
* @Author: cdf
* @CreateTime: 2025-03-12
* @Description:
*/
@Data
public class PollutionDTO {
/**
* 监测点id
*/
private String lineId;
/**
* 频率偏差污染值
*/
private Double freqDev = 0.0;
/**
* 电压偏差污染值
*/
private Double vDev = 0.0;
/**
* 三相电压不平衡度污染值
*/
private Double vUnbalance = 0.0;
/**
* 谐波电压污染值
*/
private Double vAll = 0.0;
/**
* 长时电压闪变污染值
*/
private Double plt = 0.0;
/**
* 谐波电流污染值
*/
private Double iAll = 0.0;
/**
* 负序电流污染值
*/
private Double iNeg = 0.0;
/**
* 间谐波电压含有率污染值
*/
private Double vInharm = 0.0;
/**
* 谐波电压含有率污染值
*/
private Double vHarmonic = 0.0;
}

5
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/param/LineCountEvaluateParam.java
View File

@@ -30,6 +30,11 @@ public class LineCountEvaluateParam extends BaseParam implements Serializable {
*/
private List<String> phasicType;
/**
* 值类型
*/
private List<String> valueType;
/**
* 异常数据时间集合
* Map<String,List<String>> key:监测点id value:异常时间集合

2
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/po/RStatIntegrityD.java
View File

@@ -29,7 +29,7 @@ public class RStatIntegrityD {
@MppMultiId
@TableField(value = "time_id")
private String timeId;
private LocalDate timeId;
@MppMultiId
@TableField(value = "line_index")

9
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataHarmRateVController.java
View File

@@ -92,4 +92,13 @@ public class DataHarmRateVController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, "", methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getHarmRateVData")
@ApiOperation("获取谐波含有率数据")
public HttpResult<List<DataHarmDto>> getHarmRateVData(@RequestBody LineCountEvaluateParam lineParam) {
String methodDescribe = getMethodDescribe("getHarmRateVData");
List<DataHarmDto> data = dataHarmRateVInsert.getHarmRateVData(lineParam);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe);
}
}

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

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

48
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataIntegrityController.java Normal file
View File

@@ -0,0 +1,48 @@
package com.njcn.dataProcess.controller;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.constant.OperateType;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.dataProcess.annotation.InsertBean;
import com.njcn.dataProcess.pojo.dto.DataIntegrityDto;
import com.njcn.dataProcess.pojo.dto.DataLimitRateDto;
import com.njcn.dataProcess.pojo.po.RStatIntegrityD;
import com.njcn.dataProcess.service.IDataIntegrity;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @Author: cdf
* @CreateTime: 2025-03-12
* @Description: 数据完整性
*/
@RestController
@RequestMapping("/dataIntegrity")
@Api(tags = "数据完整性")
@RequiredArgsConstructor
public class DataIntegrityController extends BaseController {
@InsertBean
private IDataIntegrity iDataIntegrity;
@OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD)
@PostMapping("/batchInsertion")
@ApiOperation("批量插入")
public HttpResult<String> batchInsertion(@RequestBody List<RStatIntegrityD> poList) {
String methodDescribe = getMethodDescribe("batchInsertion");
iDataIntegrity.batchInsertion(poList);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
}
}

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

@@ -10,6 +10,7 @@ import com.njcn.dataProcess.annotation.InsertBean;
import com.njcn.dataProcess.annotation.QueryBean;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.DataVFiveItemDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataVCvtDto;
@@ -133,4 +134,15 @@ public class DataVController extends BaseController {
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/getMeasurementCount")
@ApiOperation("获取算法基础数据")
public HttpResult<List<MeasurementCountDTO>> getMeasurementCount(List<String> lineIndex, String startTime, String endTime) {
String methodDescribe = getMethodDescribe("getMeasurementCount");
List<MeasurementCountDTO> data = dataVQuery.getMeasurementCount(lineIndex,startTime,endTime);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe);
}
}

8
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/dao/imapper/DataVMapper.java
View File

@@ -1,10 +1,12 @@
package com.njcn.dataProcess.dao.imapper;
import com.njcn.dataProcess.dto.LineDataVFiveItemDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.po.influx.DataV;
import com.njcn.influx.base.InfluxDbBaseMapper;
import com.njcn.influx.query.InfluxQueryWrapper;
import java.util.List;
/**
@@ -17,4 +19,10 @@ public interface DataVMapper extends InfluxDbBaseMapper<DataV> {
List<LineDataVFiveItemDTO> queryDataValue(InfluxQueryWrapper dataVQueryWrapper);
List<MeasurementCountDTO> getMeasurementCount(InfluxQueryWrapper influxQueryWrapper);
}

3
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataHarmRateV.java
View File

@@ -37,4 +37,7 @@ IDataHarmRateV extends IMppService<RStatDataHarmRateVD> {
void addList(List<DataHarmRateVDto> list);
DataHarmDto getTopData();
List<DataHarmDto> getHarmRateVData(LineCountEvaluateParam lineParam);
}

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

@@ -18,6 +18,6 @@ public interface IDataIntegrity extends IMppService<RStatIntegrityD>{
* 批量插入数据
* @param onlineRateList
*/
void batchInsertion(List<DataIntegrityDto> onlineRateList);
void batchInsertion(List<RStatIntegrityD> onlineRateList);
}

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

@@ -3,6 +3,7 @@ package com.njcn.dataProcess.service;
import com.github.jeffreyning.mybatisplus.service.IMppService;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.DataVFiveItemDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataVCvtDto;
@@ -59,4 +60,10 @@ public interface IDataV extends IMppService<RStatDataVD> {
void addInfluxDbList(List<DataVDto> dataVList);
void batchInsertionCvtDTO(List<DataVCvtDto> cvtDTOList);
List<MeasurementCountDTO> getMeasurementCount(List<String> lineIndex, String startTime, String endTime);
List<DataVDto> getDataV(LineCountEvaluateParam lineParam);
}

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

@@ -26,10 +26,7 @@ import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
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;
/**
@@ -193,6 +190,11 @@ public class InfluxdbDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
return dto;
}
@Override
public List<DataHarmDto> getHarmRateVData(LineCountEvaluateParam lineParam) {
return Collections.emptyList();
}
/**
* 按监测点集合、时间条件获取分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理

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

@@ -23,7 +23,7 @@ public class InfluxdbDataIntegrityImpl extends MppServiceImpl<RStatIntegrityDMap
@Override
public void batchInsertion(List<DataIntegrityDto> onlineRateList) {
public void batchInsertion(List<RStatIntegrityD> onlineRateList) {
}
}

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

@@ -12,6 +12,7 @@ import com.njcn.dataProcess.dao.relation.mapper.RStatDataVRelationMapper;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.DataVFiveItemDTO;
import com.njcn.dataProcess.dto.LineDataVFiveItemDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.po.influx.DataV;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
@@ -319,6 +320,23 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
}
}
@Override
public List<MeasurementCountDTO> getMeasurementCount(List<String> lineIndex, String startTime, String endTime) {
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class,MeasurementCountDTO.class);
influxQueryWrapper.regular(DataV::getLineId, lineIndex)
.eq(DataV::getValueType, InfluxDbSqlConstant.MAX)
.eq(DataV::getPhasicType, InfluxDBTableConstant.PHASE_TYPE_A)
.count(DataV::getFreq)
.groupBy(DataV::getLineId)
.between(DataV::getTime, startTime, endTime);
return dataVMapper.getMeasurementCount(influxQueryWrapper);
}
@Override
public List<DataVDto> getDataV(LineCountEvaluateParam lineParam) {
return Collections.emptyList();
}
/**
* 按监测点集合、时间条件获取dataV分钟数据
* timeMap参数来判断是否进行数据出来 timeMap为空则不进行数据处理

9
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/relation/RStatIntegrityDImpl.java Normal file
View File

@@ -0,0 +1,9 @@
package com.njcn.dataProcess.service.impl.relation;
/**
* @Author: cdf
* @CreateTime: 2025-03-12
* @Description: 数据完整性
*/
public class RStatIntegrityDImpl {
}

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

@@ -1,5 +1,8 @@
package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DatePattern;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.DataHarmrateVRelationMapper;
import com.njcn.dataProcess.dao.relation.mapper.DataIRelationMapper;
@@ -10,7 +13,9 @@ import com.njcn.dataProcess.po.relation.DataHarmrateV;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataHarmDto;
import com.njcn.dataProcess.pojo.dto.DataHarmRateVDto;
import com.njcn.dataProcess.pojo.dto.DataVDto;
import com.njcn.dataProcess.pojo.po.RStatDataHarmRateVD;
import com.njcn.dataProcess.pojo.po.RStatDataVD;
import com.njcn.dataProcess.service.IDataHarmRateV;
import com.njcn.dataProcess.util.BeanFeildUtils;
import com.njcn.dataProcess.util.TimeUtils;
@@ -20,6 +25,7 @@ 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.Collections;
import java.util.List;
@@ -94,4 +100,22 @@ public class RelationDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
public DataHarmDto getTopData() {
return null;
}
@Override
public List<DataHarmDto> getHarmRateVData(LineCountEvaluateParam lineParam) {
List<DataHarmDto> result = new ArrayList<>();
List<RStatDataHarmRateVD> rStatDataHarmRateVDList = dataHarmRateV.list(new LambdaQueryWrapper<RStatDataHarmRateVD>()
.in(RStatDataHarmRateVD::getLineId,lineParam.getLineId())
.in(RStatDataHarmRateVD::getValueType, lineParam.getValueType())
.in(RStatDataHarmRateVD::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataHarmRateVD::getTime, lineParam.getStartTime())
.le(RStatDataHarmRateVD::getTime, lineParam.getEndTime())
);
for(RStatDataHarmRateVD rStatDataHarmRateVD : rStatDataHarmRateVDList){
DataHarmDto dto = BeanUtil.copyProperties(rStatDataHarmRateVD,DataHarmDto.class);
dto.setTime(rStatDataHarmRateVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
result.add(dto);
}
return result;
}
}

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

@@ -1,6 +1,7 @@
package com.njcn.dataProcess.service.impl.relation;
import cn.hutool.core.bean.BeanUtil;
import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
import com.njcn.dataProcess.dao.relation.mapper.RStatIntegrityDMapper;
import com.njcn.dataProcess.pojo.dto.DataIntegrityDto;
@@ -23,7 +24,7 @@ public class RelationDataIntegrityImpl extends MppServiceImpl<RStatIntegrityDMap
@Override
public void batchInsertion(List<DataIntegrityDto> onlineRateList) {
public void batchInsertion(List<RStatIntegrityD> dataIntegrityDtoList) {
this.saveOrUpdateBatchByMultiId(dataIntegrityDtoList);
}
}

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

@@ -1,11 +1,18 @@
package com.njcn.dataProcess.service.impl.relation;
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 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.constant.InfluxDBTableConstant;
import com.njcn.dataProcess.dao.relation.mapper.DataVRelationMapper;
import com.njcn.dataProcess.dao.relation.mapper.RStatDataVRelationMapper;
import com.njcn.dataProcess.dto.DataVDTO;
import com.njcn.dataProcess.dto.DataVFiveItemDTO;
import com.njcn.dataProcess.dto.MeasurementCountDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.po.relation.DataV;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
@@ -15,17 +22,19 @@ import com.njcn.dataProcess.pojo.po.RStatDataVD;
import com.njcn.dataProcess.service.IDataV;
import com.njcn.dataProcess.util.BeanFeildUtils;
import com.njcn.dataProcess.util.TimeUtils;
import com.njcn.influx.constant.InfluxDbSqlConstant;
import com.njcn.influx.query.InfluxQueryWrapper;
import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.ListUtils;
import org.ehcache.core.util.CollectionUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
/**
@@ -125,5 +134,27 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
}
@Override
public List<MeasurementCountDTO> getMeasurementCount(List<String> lineIndex, String startTime, String endTime) {
return Collections.emptyList();
}
@Override
public List<DataVDto> getDataV(LineCountEvaluateParam lineParam) {
List<DataVDto> result = new ArrayList<>();
List<RStatDataVD> rStatDataVDList = iDataV.list(new LambdaQueryWrapper<RStatDataVD>()
.in(RStatDataVD::getValueType, lineParam.getValueType())
.in(RStatDataVD::getPhasicType, lineParam.getPhasicType())
.ge(RStatDataVD::getTime, lineParam.getStartTime())
.le(RStatDataVD::getTime, lineParam.getEndTime())
);
for(RStatDataVD rStatDataVD : rStatDataVDList){
DataVDto dto = BeanUtil.copyProperties(rStatDataVD,DataVDto.class);
dto.setTime(rStatDataVD.getTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATE_PATTERN)));
result.add(dto);
}
return result;
}
}