Microservice/pqs
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

1.谐波污区图算法修改

Browse Source
This commit is contained in:
wr
2023-09-20 13:37:57 +08:00
parent 0843e11dc5
commit 243088415d
1 changed files with 40 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/service/mysql/Impl/line/PollutionServiceImpl.java
View File

@@ -14,13 +14,11 @@ import com.njcn.common.utils.LogUtil;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.dto.DeptGetChildrenDTO;
import com.njcn.device.biz.pojo.param.DeptGetLineParam;
import com.njcn.device.pq.api.DeptLineFeignClient;
import com.njcn.device.pq.api.LineFeignClient;
import com.njcn.device.biz.pojo.po.Overlimit;
import com.njcn.harmonic.pojo.dto.PublicDTO;
import com.njcn.harmonic.pojo.po.*;
import com.njcn.harmonic.pojo.po.day.*;
import com.njcn.influxdb.utils.InfluxDbUtils;
import com.njcn.prepare.harmonic.mapper.mysql.line.*;
import com.njcn.prepare.harmonic.pojo.dto.PollutionDTO;
import com.njcn.prepare.harmonic.pojo.param.LineParam;
@@ -38,7 +36,6 @@ import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.function.BinaryOperator;
@@ -58,12 +55,9 @@ import java.util.stream.Stream;
@AllArgsConstructor
public class PollutionServiceImpl implements PollutionService {
private final InfluxDbUtils influxDbUtils;
private final LineFeignClient lineFeignClient;
private final DeptLineFeignClient deptLineFeignClient;
private final DicDataFeignClient dicDataFeignClient;
private final RStatPollutionOrgDMapper rStatPollutionOrgDMapper;
@@ -97,18 +91,13 @@ public class PollutionServiceImpl implements PollutionService {
@Async("asyncExecutor")
public void processPollutionData(LineParam lineParam) {
log.info("参数起始时间："+lineParam.getBeginTime());
log.info("参数结束时间："+lineParam.getEndTime());
TimeInterval timer = new TimeInterval();
List<PollutionDTO> pollutionList;
LocalDateTime local = LocalDateTimeUtil.now();
if (StrUtil.isNotBlank(lineParam.getDataDate())){
local = LocalDateTimeUtil.parse(lineParam.getBeginTime());
local = DateUtil.parseLocalDateTime(lineParam.getBeginTime());
}
List<DictData> dictData = dicDataFeignClient.getDicDataByTypeCode(DicDataTypeEnum.POLLUTION_STATIS.getCode()).getData();
//时间类型为日的情况下执行influxdb计算
if (Integer.valueOf(BizParamConstant.STAT_BIZ_DAY).equals(lineParam.getType())){
@@ -123,14 +112,14 @@ public class PollutionServiceImpl implements PollutionService {
}
List<String> lineList = overLimitList.stream().map(Overlimit::getId).collect(Collectors.toList());
//指标数据获取
List<PublicDTO> harmonicVoltageList = getHarmonicVoltage(overLimitList,lineParam.getDataDate());
List<PublicDTO> harmonicCurrentList = getHarmonicCurrent(overLimitList,lineParam.getDataDate());
List<PublicDTO> frequencyDeviationList = getFrequencyDeviation(overLimitList,lineParam.getDataDate());
List<PublicDTO> voltageDeviationList = getVoltageDeviation(overLimitList,lineParam.getDataDate());
List<PublicDTO> threePhaseVoltageList = getThreePhaseVoltageUnbalance(overLimitList,lineParam.getDataDate());
List<PublicDTO> negativeSequenceList = getNegativeSequenceCurrent(overLimitList,lineParam.getDataDate());
List<PublicDTO> interHarmonicVoltageList = getInterharmonicVoltage(overLimitList,lineParam.getDataDate());
List<PublicDTO> voltageFlickerList = getVoltageFlicker(overLimitList,lineParam.getDataDate());
List<PublicDTO> harmonicVoltageList = getHarmonicVoltage(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> harmonicCurrentList = getHarmonicCurrent(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> frequencyDeviationList = getFrequencyDeviation(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> voltageDeviationList = getVoltageDeviation(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> threePhaseVoltageList = getThreePhaseVoltageUnbalance(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> negativeSequenceList = getNegativeSequenceCurrent(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> interHarmonicVoltageList = getInterharmonicVoltage(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
List<PublicDTO> voltageFlickerList = getVoltageFlicker(overLimitList,lineParam.getBeginTime(),lineParam.getEndTime());
LogUtil.njcnDebug(log, "监测点污染指标数据查询耗时：{}", timer.intervalRestart());
List<PollutionDTO> lineIdList = new ArrayList<>();
@@ -147,11 +136,10 @@ public class PollutionServiceImpl implements PollutionService {
}
LocalDateTime localEnd = LocalDateTimeUtil.parse(lineParam.getEndTime());
LocalDateTime localEnd = DateUtil.parseLocalDateTime(lineParam.getEndTime());
LambdaQueryWrapper<RMpPollutionDPO> lambdaQuery = new LambdaQueryWrapper<>();
log.info("污区数据统计间隔"+local+"----->"+localEnd);
List<RMpPollutionDPO> pollutionDayList =new ArrayList<>();
if(Integer.valueOf(BizParamConstant.STAT_BIZ_DAY).equals(lineParam.getType())){
lambdaQuery.ge(RMpPollutionDPO::getDataDate, local).le(RMpPollutionDPO::getDataDate,localEnd);
@@ -399,10 +387,10 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 谐波电压：取监测点最新的A、B、C三相数据，再取电压总谐波畸变率、各次谐波电压含有率（2~25次）中的最大值，作为结果
*/
private List<PublicDTO> getHarmonicVoltage(List<Overlimit> overLimitList,String dataDate){
private List<PublicDTO> getHarmonicVoltage(List<Overlimit> overLimitList,String beginTime,String endTime){
List<PublicDTO> list = new ArrayList<>();
Map<String, Optional<PublicDTO>> threePhase = getThreePhaseData(overLimitList,dataDate);
Map<String, Optional<PublicDTO>> distortionRate = getDistortionRateData(overLimitList,dataDate);
Map<String, Optional<PublicDTO>> threePhase = getThreePhaseData(overLimitList,beginTime,endTime);
Map<String, Optional<PublicDTO>> distortionRate = getDistortionRateData(overLimitList,beginTime,endTime);
for (String key : threePhase.keySet()) {
list.add(threePhase.get(key).get());
}
@@ -414,13 +402,12 @@ public class PollutionServiceImpl implements PollutionService {
return process(outMap);
}
private Map<String, Optional<PublicDTO>> getThreePhaseData(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
private Map<String, Optional<PublicDTO>> getThreePhaseData(List<Overlimit> overLimitList,String beginTime,String endTime){
List<RStatDataVDPO> threePhaseList = dataVDService.list(new LambdaQueryWrapper<RStatDataVDPO>()
.eq(RStatDataVDPO::getValueType, "CP95")
.in(RStatDataVDPO::getPhaseType, Arrays.asList("A", "B", "C"))
.ge(RStatDataVDPO::getTime, stringStringMap.get("startTime"))
.le(RStatDataVDPO::getTime, stringStringMap.get("endTime"))
.ge(RStatDataVDPO::getTime, beginTime)
.le(RStatDataVDPO::getTime, endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -447,13 +434,12 @@ public class PollutionServiceImpl implements PollutionService {
return lineData.stream().collect(Collectors.groupingBy(PublicDTO::getId,Collectors.reducing(BinaryOperator.maxBy(comparator))));
}
private Map<String, Optional<PublicDTO>> getDistortionRateData(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
private Map<String, Optional<PublicDTO>> getDistortionRateData(List<Overlimit> overLimitList,String beginTime,String endTime){
List<RStatDataHarmrateVDPO> distortionRateList = harmRateVDService.list(new LambdaQueryWrapper<RStatDataHarmrateVDPO>()
.eq(RStatDataHarmrateVDPO::getValueType, "CP95")
.in(RStatDataHarmrateVDPO::getPhaseType, Arrays.asList("A", "B", "C"))
.ge(RStatDataHarmrateVDPO::getTime, stringStringMap.get("startTime"))
.le(RStatDataHarmrateVDPO::getTime, stringStringMap.get("endTime"))
.ge(RStatDataHarmrateVDPO::getTime, beginTime)
.le(RStatDataHarmrateVDPO::getTime, endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -506,13 +492,12 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 谐波电流：各次谐波电流（2~25次），取各监测点最新的A、B、C三相数据。
*/
private List<PublicDTO> getHarmonicCurrent(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
private List<PublicDTO> getHarmonicCurrent(List<Overlimit> overLimitList,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, stringStringMap.get("startTime"))
.le(RStatDataIDPO::getTime, stringStringMap.get("endTime"))
.ge(RStatDataIDPO::getTime, beginTime)
.le(RStatDataIDPO::getTime, endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -561,14 +546,13 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 频率偏差：各监测点最新的T相数据，取频率上下偏差的绝对值中的最大值。
*/
private List<PublicDTO> getFrequencyDeviation(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time",endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -599,14 +583,13 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 电压偏差：各监测点最新的A、B、C三相数据，取电压上下偏差的绝对值中的最大值。
*/
private List<PublicDTO> getVoltageDeviation(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time", endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -638,14 +621,13 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 三相电压不平衡度：各监测点最新的T相数据。
*/
private List<PublicDTO> getThreePhaseVoltageUnbalance(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time", endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -676,14 +658,13 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 负序电流：各监测点最新的T相数据。
*/
private List<PublicDTO> getNegativeSequenceCurrent(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time", endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -717,13 +698,12 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 间谐波电压含有率：各监测点最新的A、B、C三相数据。
*/
private List<PublicDTO> getInterharmonicVoltage(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time", endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -769,13 +749,12 @@ public class PollutionServiceImpl implements PollutionService {
/**
* 长时电压闪变：各监测点最新的A、B、C三相数据。
*/
private List<PublicDTO> getVoltageFlicker(List<Overlimit> overLimitList,String dataDate){
Map<String, String> stringStringMap = processMysqlDate(dataDate, Integer.valueOf(BizParamConstant.STAT_BIZ_DAY));
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", stringStringMap.get("startTime"))
.le("time", stringStringMap.get("endTime"))
.ge("time", beginTime)
.le("time", endTime)
);
List<Double> data;
PublicDTO publicDTO;
@@ -865,70 +844,6 @@ public class PollutionServiceImpl implements PollutionService {
return outList;
}
/**
* MYSQL 时间返回
* @param date
* @param type
* @return
*/
private Map<String,String> processMysqlDate(String date,Integer type){
Map<String,String> mapDate=new HashMap<>();
if (StrUtil.isNotBlank(date)){
if (Integer.valueOf(BizParamConstant.STAT_BIZ_DAY).equals(type)){
mapDate.put("startTime",date+" 00:00:00");
mapDate.put("endTime",date+" 23:59:59");
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_MONTH).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfMonth(dateOut);
Date dateOute = DateUtil.endOfMonth(dateOut);
mapDate.put("startTime",dateOutb+" 00:00:00");
mapDate.put("endTime",dateOute+" 23:59:59");
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_QUARTER).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfQuarter(dateOut);
Date dateOute = DateUtil.endOfQuarter(dateOut);
mapDate.put("startTime",dateOutb+" 00:00:00");
mapDate.put("endTime",dateOute+" 23:59:59");
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_YEAR).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfYear(dateOut);
Date dateOute = DateUtil.endOfYear(dateOut);
mapDate.put("startTime",dateOutb+" 00:00:00");
mapDate.put("endTime",dateOute+" 23:59:59");
}
}
return mapDate;
}
/**
* influxDb时间条件处理
*/
private String processDate(String date,Integer type){
String processSql = "";
if (StrUtil.isNotBlank(date)){
if (Integer.valueOf(BizParamConstant.STAT_BIZ_DAY).equals(type)){
processSql = " and time >= '"+date+" 00:00:00' and time <= '"+date+" 23:59:59' ";
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_MONTH).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfMonth(dateOut);
Date dateOute = DateUtil.endOfMonth(dateOut);
processSql =" and time >= '"+DateUtil.formatDate(dateOutb)+" 00:00:00' and time <= '"+DateUtil.formatDate(dateOute)+" 23:59:59' ";
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_QUARTER).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfQuarter(dateOut);
Date dateOute = DateUtil.endOfQuarter(dateOut);
processSql =" and time >= '"+DateUtil.formatDate(dateOutb)+" 00:00:00' and time <= '"+DateUtil.formatDate(dateOute)+" 23:59:59' ";
}else if (Integer.valueOf(BizParamConstant.STAT_BIZ_YEAR).equals(type)){
Date dateOut = DateUtil.parse(date);
Date dateOutb = DateUtil.beginOfYear(dateOut);
Date dateOute = DateUtil.endOfYear(dateOut);
processSql =" and time >= '"+DateUtil.formatDate(dateOutb)+" 00:00:00' and time <= '"+DateUtil.formatDate(dateOute)+" 23:59:59' ";
}
}
return processSql;
}
/**
* 获取限值表中的所有监测点信息
*/
@@ -936,74 +851,4 @@ public class PollutionServiceImpl implements PollutionService {
return lineFeignClient.getAllLineOverLimit("harmonic-boot","").getData();
}
/*private List<PollutionDTO> getDataPolluction(LineParam lineParam){
List<String> lineList;
if (CollUtil.isEmpty(lineParam.getLineIds())){
List<Overlimit> overLimitList = getAllLinesLimitData();
lineList = overLimitList.stream().map(Overlimit::getId).collect(Collectors.toList());
}else {
lineList = new ArrayList<>(lineParam.getLineIds());
}
List<PollutionDTO> pollutionDTOList = new ArrayList<>();
InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
for (String lineId : lineList){
String sql="SELECT * FROM harmonic_pollution where line_id = '" + lineId +"' "+processDate(lineParam.getDataDate(),lineParam.getType());
QueryResult dataPolluctionResult = influxDbUtils.query(sql);
List<DataPolluctionPO> threePhaseList = resultMapper.toPOJO(dataPolluctionResult, DataPolluctionPO.class);
for (DataPolluctionPO dataPolluction : threePhaseList){
PollutionDTO pollutionDTO = new PollutionDTO();
pollutionDTO.setLineId(dataPolluction.getLineId());
pollutionDTO.setFreqDev(dataPolluction.getFreqDev());
pollutionDTO.setVDev(dataPolluction.getVDev());
pollutionDTO.setVUnbalance(dataPolluction.getVUnbalance());
pollutionDTO.setVAll(dataPolluction.getVAll());
pollutionDTO.setPlt(dataPolluction.getPlt());
pollutionDTO.setIAll(dataPolluction.getIAll());
pollutionDTO.setINeg(dataPolluction.getINeg());
pollutionDTO.setVInharm(dataPolluction.getVInharm());
pollutionDTOList.add(pollutionDTO);
}
}
return pollutionDTOList;
}*/
/*MySql入表 r_mp_pollution_d*/
/*private void insertLinePollution(List<PollutionDTO> pollutionList, LocalDateTime local){
for (PollutionDTO pollution : pollutionList){
RMpPollutionD rMpPollution = new RMpPollutionD();
rMpPollution.setLineId(pollution.getLineId());
rMpPollution.setFreqDev(pollution.getFreqDev().floatValue());
rMpPollution.setvDev(pollution.getVDev().floatValue());
rMpPollution.setvUnbalance(pollution.getVUnbalance().floatValue());
rMpPollution.setvAll(pollution.getVAll().floatValue());
rMpPollution.setPlt(pollution.getPlt().floatValue());
rMpPollution.setiAll(pollution.getIAll().floatValue());
rMpPollution.setiNeg(pollution.getINeg().floatValue());
rMpPollution.setvInharm(pollution.getVInharm().floatValue());
rMpPollution.setDataDate(local);
rMpPollutionDMapper.insertPollution(rMpPollution);
}
}*/
/*private void insertPolluction(List<PollutionDTO> list, long time){
List<String> records = new ArrayList<String>();
list.forEach(item->{
Map<String, String> tags = new HashMap<>();
Map<String, Object> fields = new HashMap<>();
tags.put("line_id",item.getLineId());
fields.put("freq_dev",item.getFreqDev());
fields.put("v_dev",item.getVDev());
fields.put("v_unbalance",item.getVUnbalance());
fields.put("i_neg",item.getINeg());
fields.put("v_all",item.getVAll());
fields.put("i_all",item.getIAll());
fields.put("v_inharm",item.getVInharm());
fields.put("plt",item.getPlt());
Point point = influxDbUtils.pointBuilder("harmonic_pollution", time, TimeUnit.MILLISECONDS,tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).tag("line_id", item.getLineId()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
records.add(batchPoints.lineProtocol());
});
influxDbUtils.batchInsert(influxDbUtils.getDbName(),"", InfluxDB.ConsistencyLevel.ALL, records);
}*/
}