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

算法调整

Browse Source
This commit is contained in:
xy
2026-01-07 18:23:38 +08:00
parent 4c239b1604
commit 91ff704f0a
7 changed files with 81 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -82,7 +82,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -100,7 +100,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataVHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -131,7 +131,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -148,7 +148,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataIHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -180,7 +180,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataFlickerFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -196,7 +196,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4);
dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataFlickerHandler(item3,dto,item4,true);
result.add(dto);
@@ -229,7 +229,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataFlucFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -245,7 +245,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4);
dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataFlucHandler(item3,dto,item4,true);
result.add(dto);
@@ -277,7 +277,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmphasicIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -294,7 +294,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPhasicIHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -325,7 +325,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmphasicVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -342,7 +342,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPhasicVHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -373,7 +373,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerPFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -390,7 +390,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerPHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -421,7 +421,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerQFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -438,7 +438,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerQHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -469,7 +469,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerSFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -486,7 +486,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerSHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -517,7 +517,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmRateIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -534,7 +534,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmRateIHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -565,7 +565,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataHarmRateVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -582,7 +582,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmRateVHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -613,7 +613,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataInharmIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -630,7 +630,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataInHarmIHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -661,7 +661,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataInharmVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -678,7 +678,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType());
dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataInHarmVHandler(item3,valueTypes,dto,true);
result.add(dto);
@@ -710,7 +710,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{
lineParam.setLineId(list);
//添加异常数据时间点
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
//获取原始数据
List<CommonMinuteDto> partList = dataPltFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) {
@@ -726,7 +726,7 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime());
dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4);
dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataPltHandler(item3,dto,item4,true);
result.add(dto);
@@ -1552,18 +1552,21 @@ public class DayDataServiceImpl implements IDayDataService {
case InfluxDbSqlConstant.MAX:
case InfluxDbSqlConstant.CP95:
Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
result = max.orElse(null);
// result = max.orElse(null);
result = max.orElse(0.0);
break;
case InfluxDbSqlConstant.MIN:
Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
result = min.orElse(null);
// result = min.orElse(null);
result = min.orElse(0.0);
break;
case InfluxDbSqlConstant.AVG_WEB:
OptionalDouble average = list.stream()
.filter(Objects::nonNull)
.mapToDouble(Double::doubleValue)
.average();
result = average.isPresent() ? average.getAsDouble() : null;
// result = average.isPresent() ? average.getAsDouble() : null;
result = average.isPresent() ? average.getAsDouble() : 0.0;
break;
default:
break;
@@ -1572,24 +1575,28 @@ public class DayDataServiceImpl implements IDayDataService {
switch (valueType) {
case InfluxDbSqlConstant.MAX:
Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
result = max.orElse(null);
// result = max.orElse(null);
result = max.orElse(0.0);
break;
case InfluxDbSqlConstant.MIN:
Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
result = min.orElse(null);
// result = min.orElse(null);
result = min.orElse(0.0);
break;
case InfluxDbSqlConstant.AVG_WEB:
OptionalDouble average = list.stream()
.filter(Objects::nonNull)
.mapToDouble(Double::doubleValue)
.average();
result = average.isPresent() ? average.getAsDouble() : null;
// result = average.isPresent() ? average.getAsDouble() : null;
result = average.isPresent() ? average.getAsDouble() : 0.0;
break;
case InfluxDbSqlConstant.CP95:
list.sort(Collections.reverseOrder());
int discardCount = (int) Math.ceil(list.size() * 0.05);
List<Double> remainingList = list.subList(discardCount, list.size());
result = remainingList.isEmpty() ? null : remainingList.get(0);
// result = remainingList.isEmpty() ? null : remainingList.get(0);
result = remainingList.isEmpty() ? 0.0 : remainingList.get(0);
break;
default:
break;

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

@@ -12,10 +12,16 @@ import com.njcn.algorithm.service.line.IDataLimitRateAsync;
import com.njcn.algorithm.utils.MemorySizeUtil;
import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.dataProcess.api.*;
import com.njcn.csdevice.api.CsOverLimitClient;
import com.njcn.dataProcess.api.DataLimitQualifiedFeignClient;
import com.njcn.dataProcess.api.DataLimitRateFeignClient;
import com.njcn.dataProcess.api.DataLimitTargetFeignClient;
import com.njcn.dataProcess.constant.PhaseType;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.*;
import com.njcn.dataProcess.pojo.dto.DataLimitDetailDto;
import com.njcn.dataProcess.pojo.dto.DataLimitRateDto;
import com.njcn.dataProcess.pojo.dto.DataLimitTargetDto;
import com.njcn.dataProcess.pojo.dto.DataQualifiedDetail;
import com.njcn.dataProcess.util.TimeUtils;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.po.Overlimit;
@@ -25,6 +31,7 @@ import org.apache.commons.collections4.ListUtils;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
@@ -38,6 +45,7 @@ import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -67,6 +75,8 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
private IDataLimitRateAsync dataLimitRateAsync;
@Resource
private CsLineFeignClient csLineFeignClient;
@Resource
private CsOverLimitClient csOverLimitClient;
@Override
public void limitRateHandler(CalculatedParam calculatedParam) {
@@ -79,8 +89,20 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
List<String> lineIds = calculatedParam.getIdList();
List<Overlimit> overLimitList;
//获取所有监测点的限值
List<Overlimit> overLimitList = commTerminalGeneralClient.getOverLimitDataByIds(lineIds).getData();
if (Objects.equals(lineParam.getType(), 2)) {
List<com.njcn.csdevice.pojo.po.Overlimit> list = csOverLimitClient.getOverLimitByLineIds(lineIds).getData();
overLimitList = list.stream()
.map(item -> {
Overlimit overlimit = new Overlimit();
BeanUtils.copyProperties(item, overlimit);
return overlimit;
})
.collect(Collectors.toList());
} else {
overLimitList = commTerminalGeneralClient.getOverLimitDataByIds(lineIds).getData();
}
Map<String, Overlimit> overLimitMap = overLimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(lineIds, 1);

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

@@ -277,6 +277,10 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
result.add(b);
result.add(c);
result.add(t);
return result;
}

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

@@ -299,6 +299,7 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerP.class);
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmP_", "p_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmP_", "p_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.eq(DataHarmpowerP::getLineId, lineId)
.eq(DataHarmpowerP::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerP::getLineId)
@@ -308,8 +309,8 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
.select("Pq_DF","df")
//视在功率因素
.select("Pq_PF","pf")
//总功功率
.select("Pq_P","p")
//三相总功功率
.select("Pq_TotP","p")
//基波有功功率
.select("Pq_FundP","p_1")
.between(DataHarmpowerP::getTime, startTime, endTime)

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

@@ -296,9 +296,9 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
.select(DataHarmpowerQ::getLineId)
.select(DataHarmpowerQ::getPhasicType)
.select(DataHarmpowerQ::getValueType)
//总功功率
.select("Pq_Q","q")
//基波有功功率
//三相总无功功率
.select("Pq_TotQ","q")
//基波无功功率
.select("Pq_FundQ","q_1")
.between(DataHarmpowerQ::getTime, startTime, endTime)
.eq(DataHarmpowerQ::getQualityFlag, "0");

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

@@ -296,9 +296,9 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
.select(DataHarmpowerS::getLineId)
.select(DataHarmpowerS::getPhasicType)
.select(DataHarmpowerS::getValueType)
//总功功率
.select("Pq_S","s")
//基波有功功率
//三相总视在功率
.select("Pq_TotS","s")
//基波视在功率
.select("Pq_FundS","s_1")
.between(DataHarmpowerS::getTime, startTime, endTime)
.eq(DataHarmpowerS::getQualityFlag, "0");

2
data-processing/data-processing-boot/src/main/resources/bootstrap.yml
View File

@@ -50,7 +50,7 @@ spring:
logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: info
root: error
mqtt: