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:
xy
2025-11-11 10:56:49 +08:00
parent 8ce1f83531
commit 89df5f3039
38 changed files with 1332 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
algorithm/algorithm-api/src/main/java/com/njcn/algorithm/pojo/bo/BaseParam.java
View File

@@ -58,6 +58,6 @@ public class BaseParam implements Serializable {
@ApiModelProperty(name = "idList",value = "索引集合")
private List<String> idList;
@ApiModelProperty(name = "type",value = "0:通用 1:省级平台 ")
@ApiModelProperty(name = "type",value = "0:通用 1:省级平台 2:物联平台")
private Integer type;
}

4
algorithm/algorithm-api/src/main/java/com/njcn/algorithm/pojo/liteflow/LiteFlowAlgorithmFeignClient.java
View File

@@ -42,4 +42,8 @@ public interface LiteFlowAlgorithmFeignClient {
@ApiOperation("新能源专项分析算法执行链")
@PostMapping("/specialAnalysis")
void specialAnalysisExecutor(@RequestBody BaseParam baseParam);
@ApiOperation("物联监测点算法执行链")
@PostMapping("/wlMeasurementPointExecutor")
void wlMeasurementPointExecutor(@RequestBody BaseParam baseParam);
}

6
algorithm/algorithm-api/src/main/java/com/njcn/algorithm/pojo/liteflow/fallback/LiteFlowAlgorithmFeignClientFallbackFactory.java
View File

@@ -52,6 +52,12 @@ public class LiteFlowAlgorithmFeignClientFallbackFactory implements FallbackFact
throw new BusinessException(finalExceptionEnum);
}
@Override
public void wlMeasurementPointExecutor(BaseParam baseParam) {
log.error("{}异常，降级处理，异常为：{}", "物联监测点算法执行链: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public void deviceExecutor(BaseParam baseParam) {
log.error("{}异常，降级处理，异常为：{}", "装置算法执行链: ", throwable.toString());

5
algorithm/algorithm-boot/pom.xml
View File

@@ -104,6 +104,11 @@
<artifactId>pq-device-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>cs-device-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>common-event</artifactId>

48
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/ExecutionCenter.java
View File

@@ -2,7 +2,10 @@ package com.njcn.algorithm;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.*;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.StrUtil;
import com.njcn.algorithm.pojo.bo.BaseParam;
@@ -13,8 +16,12 @@ import com.njcn.algorithm.utils.MemorySizeUtil;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.enums.common.LogEnum;
import com.njcn.common.pojo.exception.BusinessException;
import com.njcn.csdevice.api.CsLineFeignClient;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.dto.*;
import com.njcn.device.biz.pojo.dto.DeptGetChildrenMoreDTO;
import com.njcn.device.biz.pojo.dto.DeptGetDeviceDTO;
import com.njcn.device.biz.pojo.dto.DeptGetSubStationDTO;
import com.njcn.device.biz.pojo.dto.LineDevGetDTO;
import com.njcn.device.biz.pojo.param.DeptGetLineParam;
import com.njcn.device.pq.api.DeptLineFeignClient;
import com.njcn.user.api.DeptFeignClient;
@@ -62,6 +69,8 @@ public class ExecutionCenter extends BaseController {
private FlowExecutor flowExecutor;
@Resource
private DeptLineFeignClient deptLineFeignClient;
@Resource
private CsLineFeignClient csLineFeignClient;
/***
* 1、校验非全链执行时，tagNames节点标签集合必须为非空，否则提示---无可执行节点
@@ -155,7 +164,40 @@ public class ExecutionCenter extends BaseController {
liteflowResponse = flowExecutor.execute2Resp("measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@ApiOperation("物联监测点算法执行链")
@PostMapping("/wlMeasurementPointExecutor")
@Async("asyncExecutor")
public void wlMeasurementPointExecutor(@RequestBody BaseParam baseParam) {
String methodDescribe = getMethodDescribe("wlMeasurementPointExecutor");
//手动判断参数是否合法,
CalculatedParam calculatedParam = judgeExecuteParam(baseParam);
// 测点索引
if (CollectionUtils.isEmpty(calculatedParam.getIdList())) {
calculatedParam.setIdList(csLineFeignClient.getAllLine().getData());
}
LiteflowResponse liteflowResponse;
if (baseParam.isRepair()) {
//补招时，起始日期、截止日期必填
DateTime startDate = DateUtil.parse(baseParam.getBeginTime(), DatePattern.NORM_DATE_FORMAT);
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
liteflowResponse = flowExecutor.execute2Resp("wl_measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
} else {
//非补招
liteflowResponse = flowExecutor.execute2Resp("wl_measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)

2
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/executor/MeasurementExecutor.java
View File

@@ -231,7 +231,7 @@ public class MeasurementExecutor extends BaseExecutor {
dayDataService.dataVHandler(bindCmp.getRequestData());
}
@LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataI", nodeType = NodeTypeEnum.COMMON)
@LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataI", nodeType = NodeTypeEnum.COMMON)
public boolean dataIToDayAccess(NodeComponent bindCmp) {
return isAccess(bindCmp);
}

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

@@ -76,6 +76,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -107,7 +108,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -125,6 +125,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -155,7 +156,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -174,6 +174,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -204,7 +205,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -223,6 +223,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -253,7 +254,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -271,6 +271,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -301,7 +302,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -319,6 +319,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -349,7 +350,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -367,6 +367,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -397,7 +398,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -415,6 +415,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -445,7 +446,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -463,6 +463,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -493,7 +494,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -511,6 +511,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -541,7 +542,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -559,6 +559,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -589,7 +590,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -607,6 +607,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -637,7 +638,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -655,6 +655,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -685,7 +686,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -704,6 +704,7 @@ public class DayDataServiceImpl implements IDayDataService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(calculatedParam.getIdList(),NUM);
pendingIds.forEach(list->{
@@ -734,7 +735,6 @@ public class DayDataServiceImpl implements IDayDataService {
});
});
}
partList = null;
});
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -756,10 +756,8 @@ public class DayDataServiceImpl implements IDayDataService {
Collectors.mapping(item->TimeUtils.LocalDateTimeToString(item.getTime()), Collectors.toList())
));
lineParam.setAbnormalTime(timeMap);
pqDataVerifies.clear();
}
}
//指标处理

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

@@ -11,6 +11,7 @@ import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.service.line.IDataCrossingService;
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.dataProcess.constant.PhaseType;
import com.njcn.dataProcess.enums.DataCleanEnum;
@@ -38,7 +39,10 @@ import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -77,6 +81,8 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
private DataLimitTargetFeignClient dataLimitTargetFeignClient;
@Resource
private DataLimitQualifiedFeignClient dataLimitQualifiedFeignClient;
@Resource
private CsLineFeignClient csLineFeignClient;
@Override
@@ -105,9 +111,10 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType());
List<String> lineIds = calculatedParam.getIdList();
//获取所有监测点的限值
List<Overlimit> overLimitList = commTerminalGeneralClient.getOverLimitDataByIds(lineIds).getData();
List<Overlimit> overLimitList = csLineFeignClient.getOverLimitData(lineIds).getData();
Map<String, Overlimit> overLimitMap = overLimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
//添加异常数据时间点
getAbnormalData(lineParam);
@@ -120,7 +127,7 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
//获取电压数据
List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData();
//闪变数据
List<DataPltDto> dataFlickerAllTime = dataPltFeignClient.getRawData(lineParam).getData();
List<DataPltDto> dataPltAllTime = dataPltFeignClient.getRawData(lineParam).getData();
//谐波数据
List<DataHarmDto> dataVHarmList = dataHarmRateVFeignClient.getRawData(lineParam).getData();
//间谐波数据
@@ -133,15 +140,16 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
*/
Map<String, List<DataVDto>> allTime = dataVAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.filter(x -> InfluxDbSqlConstant.AVG_WEB.equals(x.getValueType()))
.filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_plt ->
* 闪变总计算次数(用data_plt中phasic_type=A,quality_flag=0来参与统计)
* 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/
Map<String, List<DataPltDto>> flickerAllTime = dataFlickerAllTime.stream()
Map<String, List<DataPltDto>> pltAllTime = dataPltAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
/**
@@ -215,7 +223,7 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
/**
* 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
*/
Map<String, List<DataPltDto>> dataPlt = dataFlickerAllTime.stream()
Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
@@ -224,7 +232,7 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
result.addAll(getData(calculatedParam.getDataDate(),
overLimitMap.get(item),
allTime.get(item),
flickerAllTime.get(item),
pltAllTime.get(item),
harmRateV.get(item),
dataI.get(item),
inHarmV.get(item),
@@ -240,11 +248,10 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
MemorySizeUtil.getNowMemory();
if (CollUtil.isNotEmpty(result)) {
//存储数据
List<DataLimitRateDto> dataLimitRate = result.stream().map(DataLimitDetailDto::getDataLimitRate).filter(x -> ObjectUtil.isNotNull(x)).collect(Collectors.toList());
List<DataLimitRateDto> dataLimitRate = result.stream().map(DataLimitDetailDto::getDataLimitRate).filter(ObjectUtil::isNotNull).collect(Collectors.toList());
if(CollUtil.isNotEmpty(dataLimitRate)){
dataLimitRateFeignClient.batchInsertion(dataLimitRate);
}
}
if (CollUtil.isNotEmpty(result)) {
//存储数据
@@ -252,11 +259,8 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
if(CollUtil.isNotEmpty(detail)){
dataLimitRateDetailFeignClient.batchInsertion(detail);
}
}
System.gc();
}
@Override