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-12-15 17:11:34 +08:00
parent 074c020b10
commit d8d516f63d
3 changed files with 164 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -151,7 +151,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
@@ -185,7 +185,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
@@ -220,7 +220,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
@@ -263,7 +263,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATETIME_FORMAT);
long betweenHour = DateUtil.between(startDate, endDate, DateUnit.HOUR);
//递增日期执行算法链
for (int i = 0; i < betweenHour; i++) {
for (int i = 0; i <= betweenHour; i++) {
if (i != 0) {
startDate = DateUtil.offsetHour(startDate, 1);
}
@@ -305,7 +305,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
@@ -344,7 +344,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}
@@ -432,7 +432,7 @@ public class ExecutionCenter extends BaseController {
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
//递增日期执行算法链
for (int i = 0; i < betweenDay; i++) {
for (int i = 0; i <= betweenDay; i++) {
if (i != 0) {
startDate = DateUtil.offsetDay(startDate, 1);
}

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);
}

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

@@ -58,7 +58,7 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
private static final Logger logger = LoggerFactory.getLogger(DayDataServiceImpl.class);
@Value("${line.num}")
private Integer NUM = 100;
private Integer NUM;
@Resource
private DataVFeignClient dataVFeignClient;
@Resource
@@ -105,162 +105,162 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
System.out.println("已使用的内存: " + usedMemory / (1024 * 1024) + " MB");
System.out.println("第一次分析结束-----------------------------------------");
logger.info("{},limitRate表转r_stat_limit_rate_d算法开始=====》", LocalDateTime.now());
List<DataLimitDetailDto> result = new ArrayList<>();
//远程接口获取分钟数据
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();
Map<String, Overlimit> overLimitMap = overLimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
//添加异常数据时间点
getAbnormalData(lineParam);
//以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
MemorySizeUtil.getNowMemory();
pendingIds.forEach(list -> {
lineParam.setLineId(list);
//获取电压数据
List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData();
//闪变数据
List<DataPltDto> dataPltAllTime = dataPltFeignClient.getRawData(lineParam).getData();
//谐波数据
List<DataHarmDto> dataVHarmList = dataHarmRateVFeignClient.getRawData(lineParam).getData();
//间谐波数据
List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(lineParam).getData();
//电流数据
List<DataIDto> dataIList = dataIFeignClient.getRawData(lineParam).getData();
/**
* 功能描述:获取influxDB -> data_v ->
* 总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/
Map<String, List<DataVDto>> allTime = dataVAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_plt ->
* 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/
//fixme 冀北现场 闪变原始表没有 value_type 这个参数
Map<String, List<DataPltDto>> pltAllTime = dataPltAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
logger.info("limitRate表转r_stat_limit_rate_d算法开始，执行日期为{}=====》", calculatedParam.getDataDate());
// List<DataLimitDetailDto> result = new ArrayList<>();
// //远程接口获取分钟数据
// 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();
// Map<String, Overlimit> overLimitMap = overLimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
// //添加异常数据时间点
// getAbnormalData(lineParam);
// //以100个监测点分片处理
// List<List<String>> pendingIds = ListUtils.partition(lineIds, NUM);
// ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
// MemorySizeUtil.getNowMemory();
// pendingIds.forEach(list -> {
// lineParam.setLineId(list);
// //获取电压数据
// List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData();
// //闪变数据
// List<DataPltDto> dataPltAllTime = dataPltFeignClient.getRawData(lineParam).getData();
// //谐波数据
// List<DataHarmDto> dataVHarmList = dataHarmRateVFeignClient.getRawData(lineParam).getData();
// //间谐波数据
// List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(lineParam).getData();
// //电流数据
// List<DataIDto> dataIList = dataIFeignClient.getRawData(lineParam).getData();
// /**
// * 功能描述:获取influxDB -> data_v ->
// * 总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
// */
// Map<String, List<DataVDto>> allTime = dataVAllTime.stream()
// .filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
// .filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
/**
*功能描述:获取influxDB -> data_harmrate_v ->
* 2-25次谐波电压含有率 -> A相||B相||C相的日95%概率值
*/
Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
/**
* 功能描述:获取influxDB -> data_i -> 2-25次谐波电流 -> 日95%概率值
*/
Map<String, List<DataIDto>> dataI = dataIList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataIDto::getLineId));
/**
* 功能描述:获取influxDB -> data_inharm_v -> 0.5-15.5次间谐波电压含有率 -> 日95%概率值
*/
Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
/**
* 功能描述:获取influxDB -> data_v -> 电压总谐波畸变率 -> 日95%概率值
*/
Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值
*/
Map<String, List<DataVDto>> dataVUnbalance = dataVAllTime.stream()
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
InfluxDBTableConstant.MAX.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_i -> 负序电流 -> 最大值 && 日95%概率值
*/
Map<String, List<DataIDto>> dataINeg = dataIList.stream()
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
InfluxDBTableConstant.MAX.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataIDto::getLineId));
/**
* 功能描述:获取influxDB -> data_v -> 频率偏差 -> 最大值 && 最小值
*/
Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream()
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.MIN.equals(x.getValueType()) ||
InfluxDBTableConstant.MAX.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值
*/
Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.MAX.equals(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
*/
Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
for (String item : list) {
if(ObjectUtil.isNotNull(overLimitMap.get(item))){
result.addAll(getData(calculatedParam.getDataDate(),
overLimitMap.get(item),
allTime.get(item),
pltAllTime.get(item),
harmRateV.get(item),
dataI.get(item),
inHarmV.get(item),
dataVThd.get(item),
dataVUnbalance.get(item),
dataINeg.get(item),
dataVFreq.get(item),
dataVDev.get(item),
dataPlt.get(item)));
}
}
});
MemorySizeUtil.getNowMemory();
if (CollUtil.isNotEmpty(result)) {
//存储数据
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)) {
//存储数据
List<DataLimitRateDetailDto> detail = result.stream().map(DataLimitDetailDto::getDataLimitRateDetail).filter(x -> ObjectUtil.isNotNull(x)).collect(Collectors.toList());
if(CollUtil.isNotEmpty(detail)){
dataLimitRateDetailFeignClient.batchInsertion(detail);
}
}
// .collect(Collectors.groupingBy(DataVDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_plt ->
// * 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
// */
// //fixme 冀北现场 闪变原始表没有 value_type 这个参数
// 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));
//
// /**
// *功能描述:获取influxDB -> data_harmrate_v ->
// * 2-25次谐波电压含有率 -> A相||B相||C相的日95%概率值
// */
// Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataHarmDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_i -> 2-25次谐波电流 -> 日95%概率值
// */
// Map<String, List<DataIDto>> dataI = dataIList.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataIDto::getLineId));
//
//
// /**
// * 功能描述:获取influxDB -> data_inharm_v -> 0.5-15.5次间谐波电压含有率 -> 日95%概率值
// */
// Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataHarmDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_v -> 电压总谐波畸变率 -> 日95%概率值
// */
// Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataVDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值
// */
// Map<String, List<DataVDto>> dataVUnbalance = dataVAllTime.stream()
// .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
// InfluxDBTableConstant.MAX.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataVDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_i -> 负序电流 -> 最大值 && 日95%概率值
// */
// Map<String, List<DataIDto>> dataINeg = dataIList.stream()
// .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) ||
// InfluxDBTableConstant.MAX.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataIDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_v -> 频率偏差 -> 最大值 && 最小值
// */
// Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream()
// .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.MIN.equals(x.getValueType()) ||
// InfluxDBTableConstant.MAX.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataVDto::getLineId));
// /**
// * 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值
// */
// Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .filter(x -> InfluxDBTableConstant.MAX.equals(x.getValueType()))
// .collect(Collectors.groupingBy(DataVDto::getLineId));
//
// /**
// * 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
// */
// Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
// .filter(x -> phase.contains(x.getPhasicType()))
// .collect(Collectors.groupingBy(DataPltDto::getLineId));
//
// for (String item : list) {
// if(ObjectUtil.isNotNull(overLimitMap.get(item))){
// result.addAll(getData(calculatedParam.getDataDate(),
// overLimitMap.get(item),
// allTime.get(item),
// pltAllTime.get(item),
// harmRateV.get(item),
// dataI.get(item),
// inHarmV.get(item),
// dataVThd.get(item),
// dataVUnbalance.get(item),
// dataINeg.get(item),
// dataVFreq.get(item),
// dataVDev.get(item),
// dataPlt.get(item)));
// }
// }
// });
// MemorySizeUtil.getNowMemory();
// if (CollUtil.isNotEmpty(result)) {
// //存储数据
// 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)) {
// //存储数据
// List<DataLimitRateDetailDto> detail = result.stream().map(DataLimitDetailDto::getDataLimitRateDetail).filter(x -> ObjectUtil.isNotNull(x)).collect(Collectors.toList());
// if(CollUtil.isNotEmpty(detail)) {
// dataLimitRateDetailFeignClient.batchInsertion(detail);
// }
// }
System.gc();
}