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

1.调整暂态算法

Browse Source
This commit is contained in:
wr
2025-12-11 14:56:14 +08:00
parent 5afb860467
commit b0972ba896
5 changed files with 549 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java
View File

@@ -26,6 +26,10 @@ import com.njcn.device.pq.pojo.vo.LineDetailDataVO;
import com.njcn.device.pq.pojo.vo.LineDetailVO; import com.njcn.device.pq.pojo.vo.LineDetailVO;
import com.njcn.oss.constant.OssPath; import com.njcn.oss.constant.OssPath;
import com.njcn.oss.utils.FileStorageUtil; import com.njcn.oss.utils.FileStorageUtil;
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.RequiredArgsConstructor;
import org.apache.commons.collections4.ListUtils; import org.apache.commons.collections4.ListUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
@@ -104,6 +108,8 @@ public class DataCleanServiceImpl implements IDataCleanService {
private PqDataVerifyCountFeignClient pqDataVerifyCountFeignClient; private PqDataVerifyCountFeignClient pqDataVerifyCountFeignClient;
@Resource @Resource
private FileStorageUtil fileStorageUtil; private FileStorageUtil fileStorageUtil;
@Resource
private DicDataFeignClient dicDataFeignClient;
@Override @Override
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
@@ -296,6 +302,8 @@ public class DataCleanServiceImpl implements IDataCleanService {
@Override @Override
public void dataCleanHandler(CalculatedParam calculatedParam) { public void dataCleanHandler(CalculatedParam calculatedParam) {
DictData dip = dicDataFeignClient.getDicDataByCodeAndType(DicDataEnum.VOLTAGE_DIP.getCode(), DicDataTypeEnum.EVENT_STATIS.getCode()).getData();
DictData rise = dicDataFeignClient.getDicDataByCodeAndType(DicDataEnum.VOLTAGE_RISE.getCode(), DicDataTypeEnum.EVENT_STATIS.getCode()).getData();
MemorySizeUtil.getNowMemory(); MemorySizeUtil.getNowMemory();
logger.info("{},原始表数据清洗=====》", LocalDateTime.now()); logger.info("{},原始表数据清洗=====》", LocalDateTime.now());
//获取标准 //获取标准
@@ -526,7 +534,7 @@ public class DataCleanServiceImpl implements IDataCleanService {
Map<String, PqReasonableRangeDto> pqReasonableRangeDtoMap = list.stream().collect(Collectors.toMap(PqReasonableRangeDto::getIndexCode, Function.identity())); Map<String, PqReasonableRangeDto> pqReasonableRangeDtoMap = list.stream().collect(Collectors.toMap(PqReasonableRangeDto::getIndexCode, Function.identity()));
eventList.forEach(data -> { eventList.forEach(data -> {
data.setFeatureAmplitude(data.getFeatureAmplitude() * 100.0); data.setFeatureAmplitude(data.getFeatureAmplitude() * 100.0);
List<PqDataVerify> pqDataVerifies = judgeDataEvent(pqReasonableRangeDtoMap,data); List<PqDataVerify> pqDataVerifies = judgeDataEvent(pqReasonableRangeDtoMap, data, dip, rise);
result.addAll(pqDataVerifies); result.addAll(pqDataVerifies);
}); });
if (CollUtil.isNotEmpty(result)) { if (CollUtil.isNotEmpty(result)) {
@@ -552,65 +560,65 @@ public class DataCleanServiceImpl implements IDataCleanService {
.collect(Collectors.toList()); .collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataV)) { if (CollUtil.isNotEmpty(dataV)) {
verifyCount.setTotal(dataV.size()); verifyCount.setTotal(dataV.size());
dataV.forEach(item1-> item1.setAbnormalFlag(1)); // dataV.forEach(item1-> item1.setAbnormalFlag(1));
dataVFeignClient.addInfluxDbList(dataV); // dataVFeignClient.addInfluxDbList(dataV);
}
List<DataIDto> dataI = data2.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataI)) {
dataI.forEach(item1-> item1.setAbnormalFlag(1));
dataIFeignClient.addInfluxDbList(dataI);
}
List<DataPltDto> dataPlt = data3.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataPlt)) {
dataPlt.forEach(item1-> item1.setAbnormalFlag(1));
dataPltFeignClient.addInfluxDbList(dataPlt);
}
List<DataHarmDto> dataInHarmV = data4.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataInHarmV)) {
dataInHarmV.forEach(item1-> item1.setAbnormalFlag(1));
dataInharmVFeignClient.addInfluxDbList(dataInHarmV);
}
List<DataHarmDto> dataHarmRateV = data5.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmRateV)) {
dataHarmRateV.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmRateVFeignClient.addInfluxDbList(dataHarmRateV);
}
List<DataPowerPDto> dataHarmPowerP = data6.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmPowerP)) {
dataHarmPowerP.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmpowerPFeignClient.addInfluxDbList(dataHarmPowerP);
}
List<DataHarmDto> dataHarmPhasicV = data7.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataHarmPhasicV)) {
dataHarmPhasicV.forEach(item1-> item1.setAbnormalFlag(1));
dataHarmphasicVFeignClient.addInfluxDbList(dataHarmPhasicV);
}
List<DataFlucDto> dataFluc = data8.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataFluc)) {
dataFluc.forEach(item1-> item1.setAbnormalFlag(1));
dataFlucFeignClient.addInfluxDbList(dataFluc);
} }
// List<DataIDto> dataI = data2.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataI)) {
// dataI.forEach(item1-> item1.setAbnormalFlag(1));
// dataIFeignClient.addInfluxDbList(dataI);
// }
// List<DataPltDto> dataPlt = data3.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataPlt)) {
// dataPlt.forEach(item1-> item1.setAbnormalFlag(1));
// dataPltFeignClient.addInfluxDbList(dataPlt);
// }
// List<DataHarmDto> dataInHarmV = data4.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataInHarmV)) {
// dataInHarmV.forEach(item1-> item1.setAbnormalFlag(1));
// dataInharmVFeignClient.addInfluxDbList(dataInHarmV);
// }
// List<DataHarmDto> dataHarmRateV = data5.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataHarmRateV)) {
// dataHarmRateV.forEach(item1-> item1.setAbnormalFlag(1));
// dataHarmRateVFeignClient.addInfluxDbList(dataHarmRateV);
// }
// List<DataPowerPDto> dataHarmPowerP = data6.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataHarmPowerP)) {
// dataHarmPowerP.forEach(item1-> item1.setAbnormalFlag(1));
// dataHarmpowerPFeignClient.addInfluxDbList(dataHarmPowerP);
// }
// List<DataHarmDto> dataHarmPhasicV = data7.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataHarmPhasicV)) {
// dataHarmPhasicV.forEach(item1-> item1.setAbnormalFlag(1));
// dataHarmphasicVFeignClient.addInfluxDbList(dataHarmPhasicV);
// }
// List<DataFlucDto> dataFluc = data8.stream()
// .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
// .collect(Collectors.toList());
// if (CollUtil.isNotEmpty(dataFluc)) {
// dataFluc.forEach(item1-> item1.setAbnormalFlag(1));
// dataFlucFeignClient.addInfluxDbList(dataFluc);
// }
List<DataFlickerDto> dataFlicker = data9.stream() List<DataFlickerDto> dataFlicker = data9.stream()
.filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19))) .filter(item1 -> allTimeSet.contains(item1.getMinTime().substring(11, 19)))
.collect(Collectors.toList()); .collect(Collectors.toList());
if (CollUtil.isNotEmpty(dataFlicker)) { if (CollUtil.isNotEmpty(dataFlicker)) {
verifyCount.setFlicker(dataFlicker.size()); verifyCount.setFlicker(dataFlicker.size());
dataFlicker.forEach(item1-> item1.setAbnormalFlag(1)); // dataFlicker.forEach(item1-> item1.setAbnormalFlag(1));
dataFlickerFeignClient.addInfluxDbList(dataFlicker); // dataFlickerFeignClient.addInfluxDbList(dataFlicker);
} }
pqDataVerifyCountFeignClient.insertData(verifyCount); pqDataVerifyCountFeignClient.insertData(verifyCount);
Map<String, Object> dtoMap = new HashMap<>(1); Map<String, Object> dtoMap = new HashMap<>(1);
@@ -1890,10 +1898,11 @@ public class DataCleanServiceImpl implements IDataCleanService {
/** /**
* 获取数据的标准范围，进行数据比对 * 获取数据的标准范围，进行数据比对
*/ */
public List<PqDataVerify> judgeDataEvent(Map<String, PqReasonableRangeDto> map, RmpEventDetailDTO dto) { public List<PqDataVerify> judgeDataEvent(Map<String, PqReasonableRangeDto> map, RmpEventDetailDTO dto, DictData dip, DictData rise) {
List<PqDataVerify> list = new ArrayList<>(); List<PqDataVerify> list = new ArrayList<>();
if (ObjectUtil.isNotNull(dip)) {
PqReasonableRangeDto pqReasonableRangeDto = map.get(DataCleanEnum.VoltageDip.getCode()); PqReasonableRangeDto pqReasonableRangeDto = map.get(DataCleanEnum.VoltageDip.getCode());
if(ObjectUtil.isNotNull(dto.getFeatureAmplitude())){ if (ObjectUtil.isNotNull(dto.getFeatureAmplitude()) && dip.getId().equals(dto.getEventType())) {
if (dto.getFeatureAmplitude() < pqReasonableRangeDto.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto.getMaxValue()) { if (dto.getFeatureAmplitude() < pqReasonableRangeDto.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId() PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId()
, dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN)) , dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN))
@@ -1908,8 +1917,10 @@ public class DataCleanServiceImpl implements IDataCleanService {
list.add(pqDataVerify); list.add(pqDataVerify);
} }
} }
}
if (ObjectUtil.isNotNull(rise)) {
PqReasonableRangeDto pqReasonableRangeDto2 = map.get(DataCleanEnum.VoltageRise.getCode()); PqReasonableRangeDto pqReasonableRangeDto2 = map.get(DataCleanEnum.VoltageRise.getCode());
if(ObjectUtil.isNotNull(dto.getFeatureAmplitude())){ if (ObjectUtil.isNotNull(dto.getFeatureAmplitude()) && rise.getId().equals(dto.getEventType())) {
if (dto.getFeatureAmplitude() < pqReasonableRangeDto2.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto2.getMaxValue()) { if (dto.getFeatureAmplitude() < pqReasonableRangeDto2.getMinValue() || dto.getFeatureAmplitude() > pqReasonableRangeDto2.getMaxValue()) {
PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId() PqDataVerify pqDataVerify = getPqDataVerify(dto.getMeasurementPointId()
, dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN)) , dto.getStartTime().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_MS_PATTERN))
@@ -1924,6 +1935,7 @@ public class DataCleanServiceImpl implements IDataCleanService {
list.add(pqDataVerify); list.add(pqDataVerify);
} }
} }
}
return list; return list;
} }

4
algorithm/algorithm-boot/src/main/resources/bootstrap.yml
View File

@@ -58,7 +58,7 @@ liteflow:
logging: logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level: level:
root: root: error
#mybatis配置信息 #mybatis配置信息
@@ -66,7 +66,7 @@ mybatis-plus:
#别名扫描 #别名扫描
type-aliases-package: com.njcn.harmonic.pojo type-aliases-package: com.njcn.harmonic.pojo
line: line:
num: 20 num: 10
mqtt: mqtt:
client-id: @artifactId@${random.value} client-id: @artifactId@${random.value}

4
migration-influxdb/migration-influxdb-read-boot/src/main/java/com/njcn/migration/read/controller/MigrationInfluxDBController.java
View File

@@ -82,14 +82,14 @@ public class MigrationInfluxDBController {
long betweenDay = LocalDateTimeUtil.between(startDate, endDate, ChronoUnit.HOURS); long betweenDay = LocalDateTimeUtil.between(startDate, endDate, ChronoUnit.HOURS);
param.setStartTime(startDate.format(dateTimeFormatter)); param.setStartTime(startDate.format(dateTimeFormatter));
param.setEndTime(startDate.with(LocalTime.of(startDate.getHour(), 59, 59)).format(dateTimeFormatter)); param.setEndTime(startDate.with(LocalTime.of(startDate.getHour(), 59, 59)).format(dateTimeFormatter));
migrationService.hourseLineEventBacthSysc(param); migrationService.hourseDevDataBacthSysc(param);
for (int i = 0; i <betweenDay; i++) { for (int i = 0; i <betweenDay; i++) {
LineCountEvaluateParam countEvaluateParam=new LineCountEvaluateParam(); LineCountEvaluateParam countEvaluateParam=new LineCountEvaluateParam();
countEvaluateParam.setIsManual(true); countEvaluateParam.setIsManual(true);
startDate = LocalDateTimeUtil.offset(startDate, 1, ChronoUnit.HOURS); startDate = LocalDateTimeUtil.offset(startDate, 1, ChronoUnit.HOURS);
countEvaluateParam.setStartTime(startDate.format(dateTimeFormatter)); countEvaluateParam.setStartTime(startDate.format(dateTimeFormatter));
countEvaluateParam.setEndTime(startDate.with(LocalTime.of(startDate.getHour(), 59, 59)).format(dateTimeFormatter)); countEvaluateParam.setEndTime(startDate.with(LocalTime.of(startDate.getHour(), 59, 59)).format(dateTimeFormatter));
migrationService.hourseLineEventBacthSysc(countEvaluateParam); migrationService.hourseDevDataBacthSysc(countEvaluateParam);
} }
} catch (Exception exception) { } catch (Exception exception) {
exception.printStackTrace(); exception.printStackTrace();

2
migration-influxdb/migration-influxdb-read-boot/src/test/java/BaseJunitTest.java
View File

@@ -98,7 +98,7 @@ public class BaseJunitTest {
lineParam.setStartTime("2025-04-23 00:00:00"); lineParam.setStartTime("2025-04-23 00:00:00");
lineParam.setEndTime("2025-04-23 23:59:00"); lineParam.setEndTime("2025-04-23 23:59:00");
lineParam.setIsManual(true); lineParam.setIsManual(true);
migrationService.hourseLineDataBacthSysc(lineParam); // migrationService.hourseLineDataBacthSysc(lineParam);
System.out.println(""); System.out.println("");

45
pom.xml
View File

@@ -33,11 +33,46 @@
<properties> <properties>
<!--内网--> <!--内网-->
<middle.server.url>192.168.1.22</middle.server.url> <!-- <middle.server.url>192.168.1.22</middle.server.url>-->
<service.server.url>192.168.1.121</service.server.url> <!-- <service.server.url>192.168.1.121</service.server.url>-->
<docker.server.url>192.168.1.22</docker.server.url> <!-- <docker.server.url>192.168.1.22</docker.server.url>-->
<nacos.url>${middle.server.url}:18848</nacos.url> <!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<nacos.namespace>07ec981d-1377-4920-8760-b5a98ff7e668</nacos.namespace> <!-- <nacos.namespace>07ec981d-1377-4920-8760-b5a98ff7e668</nacos.namespace>-->
<!-- <middle.server.url>192.168.21.129</middle.server.url>-->
<!-- <service.server.url>192.168.21.1</service.server.url>-->
<!-- <docker.server.url>192.168.1.22</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
<!-- <middle.server.url>127.0.0.1</middle.server.url>-->
<!-- &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
<!-- <service.server.url>127.0.0.1</service.server.url>-->
<!-- &lt;!&ndash;docker仓库地址&ndash;&gt;-->
<!-- <docker.server.url>127.0.0.1</docker.server.url>-->
<!-- &lt;!&ndash;nacos的ip:port&ndash;&gt;-->
<!-- <nacos.url>${middle.server.url}:8848</nacos.url>-->
<!-- <nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>-->
<!-- <middle.server.url>24.43.102.139</middle.server.url>-->
<!-- <service.server.url>24.43.102.139</service.server.url>-->
<!-- <docker.server.url>24.43.102.139</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace></nacos.namespace>-->
<middle.server.url>127.0.0.1</middle.server.url>
<service.server.url>127.0.0.1</service.server.url>
<docker.server.url>127.0.0.1</docker.server.url>
<nacos.url>${middle.server.url}:8848</nacos.url>
<nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>
<!-- <middle.server.url>192.168.21.129</middle.server.url>-->
<!-- &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
<!-- <service.server.url>192.168.21.1</service.server.url>-->
<!-- &lt;!&ndash;docker仓库地址&ndash;&gt;-->
<!-- <docker.server.url>192.168.1.22</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
<!--sentinel:port--> <!--sentinel:port-->
<sentinel.url>${middle.server.url}:8080</sentinel.url> <sentinel.url>${middle.server.url}:8080</sentinel.url>
<!--网关地址，主要用于配置swagger中认证token--> <!--网关地址，主要用于配置swagger中认证token-->