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

Compare commits

base: Middleware:2025-12
Branches Tags
Middleware:master Middleware:master-jb
Middleware:2026-04 Middleware:2026-03 Middleware:2026-02 Middleware:2026-01 Middleware:2025-12 Middleware:2025-11
..
compare: Middleware:2026-02
Branches Tags
Middleware:master Middleware:master-jb
Middleware:2026-04 Middleware:2026-03 Middleware:2026-02 Middleware:2026-01 Middleware:2025-12 Middleware:2025-11

15 Commits
2025-12 ... 2026-02

Author SHA1 Message Date
xy
d3e46f5ba4 修复bug。算法内部使用新线程导致LiteFlow层面的process()方法瞬间返回，但实际计算还在进行，会影响有上下层关系的算法计算 2026-02-28 16:07:36 +08:00
xy
35939e6f8f 长时闪变越限兼容判断 2026-02-03 16:45:05 +08:00
xy
ba1f5a2e00 长时间闪变新增最大值限值判断 2026-02-03 16:37:36 +08:00
hzj
97e3386d53 优化装置状态翻转不更新updateTime 2026-01-30 10:18:18 +08:00
xy
4f8c34d21b 微调 2026-01-28 08:40:51 +08:00
wr
fd398a85de Merge remote-tracking branch 'origin/master' 2026-01-26 14:58:12 +08:00
wr
1aae135e83 优化污染值算法增加是否零票判断 2026-01-22 09:20:45 +08:00
xy
60bad4bc06 优化算法 2026-01-20 14:10:13 +08:00
xy
fe3c0363fb 算法优化 2026-01-20 13:32:55 +08:00
cdf
718fd012c8 修复算法运行报错bug 2026-01-15 21:52:52 +08:00
hzj
7bac8f5b14 优化装置状态翻转 2026-01-15 11:42:27 +08:00
hzj
fec6e6c27d 优化装置状态翻转 2026-01-15 11:34:16 +08:00
xy
fe2e3b118d 微调 2026-01-09 09:54:15 +08:00
xy
dbf0027c65 算法调整 2026-01-07 18:52:03 +08:00
xy
91ff704f0a 算法调整 2026-01-07 18:23:38 +08:00
24 changed files with 754 additions and 308 deletions
Expand all files Collapse all files

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

@@ -44,6 +44,7 @@ import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;
import static com.njcn.algorithm.utils.MemorySizeUtil.dealResponse;
/**
@@ -123,10 +124,12 @@ public class ExecutionCenter extends BaseController {
}
calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
repairParam.setType(0);
flowService.execute2Resp(methodDescribe, repairParam);
}
} else {
//非补招
calculatedParam.setType(0);
liteflowResponse = flowExecutor.execute2Resp("measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
@@ -162,6 +165,7 @@ public class ExecutionCenter extends BaseController {
}
} else {
//非补招
calculatedParam.setType(2);
liteflowResponse = flowExecutor.execute2Resp("wl_measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}

9
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IDataLimitRateAsync.java
View File

@@ -4,24 +4,29 @@ import com.njcn.device.biz.pojo.po.Overlimit;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
public interface IDataLimitRateAsync {
/**
* limit_rate多线程算法
*
* @param dataDate
* @param list
* @param phase
* @param overLimitMap
* @param size
* @param i
* @param type 系统类型
* @return
* @Author: wr
* @Date: 2025/12/17 12:16
*/
void lineDataRate(String dataDate,
CompletableFuture<Void> lineDataRate(String dataDate,
List<String> list,
List<String> phase,
Map<String, Overlimit> overLimitMap,
int size,
int i);
int i,
int type);
}

2
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataComAssServiceImpl.java
View File

@@ -58,7 +58,7 @@ public class DataComAssServiceImpl implements IDataComAssService {
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
List<String> lineIdList = calculatedParam.getIdList();
getAbnormalData(lineParam);
//getAbnormalData(lineParam);
for (String lineId : lineIdList) {
DataComassesDPO rStatComassesDpo = new DataComassesDPO();
rStatComassesDpo.setTime(calculatedParam.getDataDate());

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;

46
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,8 @@ import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -67,6 +76,8 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
private IDataLimitRateAsync dataLimitRateAsync;
@Resource
private CsLineFeignClient csLineFeignClient;
@Resource
private CsOverLimitClient csOverLimitClient;
@Override
public void limitRateHandler(CalculatedParam calculatedParam) {
@@ -79,18 +90,45 @@ 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<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);
ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
MemorySizeUtil.getNowMemory();
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int i = 0; i < pendingIds.size(); i++) {
logger.info(calculatedParam.getDataDate()+" 总分区数据：" + pendingIds.size() + "=====》当前第"+(i + 1)+"小分区");
List<String> list = pendingIds.get(i);
dataLimitRateAsync.lineDataRate(calculatedParam.getDataDate(), list, phase, overLimitMap, pendingIds.size(), (i + 1));
// 获取Future
CompletableFuture<Void> future = dataLimitRateAsync.lineDataRate(
calculatedParam.getDataDate(),
list,
phase,
overLimitMap,
pendingIds.size(),
(i + 1),
lineParam.getType()
);
futures.add(future);
}
// 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
System.gc();
}

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

@@ -24,10 +24,8 @@ import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
/**
@@ -59,17 +57,19 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
@Override
@Async("asyncExecutor")
public void lineDataRate(String dataDate,
public CompletableFuture<Void> lineDataRate(String dataDate,
List<String> list,
List<String> phase,
Map<String, Overlimit> overLimitMap,
int size,
int i) {
int i,
int type) {
List<DataLimitDetailDto> result = new ArrayList<>();
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(dataDate));
lineParam.setEndTime(TimeUtils.getEndOfDay(dataDate));
lineParam.setLineId(list);
lineParam.setType(type);
//获取电压数据
List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData();
//闪变数据
@@ -84,20 +84,39 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
* 功能描述:获取influxDB -> data_v ->
* 总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/
//在线监测设备会有AB、BC、CA的数据，会导致数据重复计算，这边根据时间去重
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));
.collect(Collectors.groupingBy(
DataVDto::getLineId,
Collectors.collectingAndThen(
Collectors.toCollection(
// 使用 TreeSet 根据时间排序并去重
() -> new TreeSet<>(
Comparator.comparing(DataVDto::getMinTime)
)
),
ArrayList::new
)
));
/**
* 功能描述:获取influxDB -> data_plt ->
* 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/
//fixme 冀北现场 闪变原始表没有 value_type 这个参数
Map<String, List<DataPltDto>> pltAllTime = dataPltAllTime.stream()
Map<String, List<DataPltDto>> pltAllTime;
if (type == 2) {
pltAllTime = dataPltAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
// .filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
} else {
pltAllTime = dataPltAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
}
/**
*功能描述:获取influxDB -> data_harmrate_v ->
@@ -105,7 +124,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
*/
Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
/**
@@ -113,7 +132,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
*/
Map<String, List<DataIDto>> dataI = dataIList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataIDto::getLineId));
@@ -122,7 +141,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
*/
Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
/**
@@ -130,49 +149,57 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
*/
Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(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()))
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equalsIgnoreCase(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()))
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equalsIgnoreCase(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()))
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.MIN.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equalsIgnoreCase(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()))
.filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
/**
* 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
*/
Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream()
Map<String, List<DataPltDto>> dataPlt;
if (type == 2) {
dataPlt = dataPltAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
} else {
dataPlt = dataPltAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.collect(Collectors.groupingBy(DataPltDto::getLineId));
}
for (String item : list) {
if (ObjectUtil.isNotNull(overLimitMap.get(item))) {
@@ -211,6 +238,8 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
MemorySizeUtil.getNowMemory();
}
System.gc();
return CompletableFuture.completedFuture(null);
}
/**
@@ -277,6 +306,10 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
result.add(b);
result.add(c);
result.add(t);
return result;
}

10
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/PollutionCalcImpl.java
View File

@@ -80,19 +80,21 @@ public class PollutionCalcImpl implements IPollutionCalc {
LineDevGetDTO line = lineDetailMap.get(id);
if (limitMap.containsKey(id)) {
Overlimit overlimit = limitMap.get(id);
lineParam.setValueType(Arrays.asList(line.getTimeInterval() + ""));
lineParam.setLineId(Arrays.asList(id));
if(!dataVFeignClient.excludeZeroData(lineParam).getData()){
dataPollutionD = new DataPollutionD();
dataPollutionD.setLineId(id);
dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));
dataPollutionD.setPollutionType(vHarmonicLimit);
lineParam.setValueType(Arrays.asList(line.getTimeInterval() + ""));
lineParam.setLineId(Arrays.asList(id));
List<DataVDto> dataVDtoList = dataVFeignClient.getGroupByTimeDataV(lineParam).getData();
List<DataHarmDto> dataHarmDtoList = dataHarmRateVFeignClient.getGroupByTimeHarmRateV(lineParam).getData();
//计算谐波电压污染值
dataPollutionD.setValue(PubUtils.doubleRound(2, calcVAllPollutionValue(dataVDtoList, dataHarmDtoList, overlimit) * line.getTimeInterval()));
list.add(dataPollutionD);
}
dataPollutionD = new DataPollutionD();
dataPollutionD.setLineId(id);
dataPollutionD.setDataDate(LocalDateTimeUtil.parseDate(calculatedParam.getDataDate()));

4
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataVFeignClient.java
View File

@@ -68,4 +68,8 @@ public interface DataVFeignClient {
//按时间分组获取原始数据
@PostMapping("/getGroupByTimeDataV")
HttpResult<List<DataVDto>> getGroupByTimeDataV(@RequestBody LineCountEvaluateParam lineParam);
@PostMapping("/excludeZeroData")
HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam);
}

6
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/fallback/DataVFeignClientFallbackFactory.java
View File

@@ -116,6 +116,12 @@ public class DataVFeignClientFallbackFactory implements FallbackFactory<DataVFei
log.error("{}异常，降级处理，异常为：{}","DataV按时间分组获取原始数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<Boolean> excludeZeroData(LineCountEvaluateParam lineParam) {
log.error("{}异常，降级处理，异常为：{}","判断是否存在是零飘数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

109
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/influx/DataHarmpowerP.java
View File

@@ -213,6 +213,115 @@ public class DataHarmpowerP {
@Column(name = "p_50")
private Double p50;
//在线监测添加的字段
@Column(name = "totPf")
private Double totPf;
@Column(name = "totDf")
private Double totDf;
@Column(name = "totP")
private Double totP;
@Column(name = "totP1")
private Double totP1;
@Column(name = "totP2")
private Double totP2;
@Column(name = "totP3")
private Double totP3;
@Column(name = "totP4")
private Double totP4;
@Column(name = "totP5")
private Double totP5;
@Column(name = "totP6")
private Double totP6;
@Column(name = "totP7")
private Double totP7;
@Column(name = "totP8")
private Double totP8;
@Column(name = "totP9")
private Double totP9;
@Column(name = "totP10")
private Double totP10;
@Column(name = "totP11")
private Double totP11;
@Column(name = "totP12")
private Double totP12;
@Column(name = "totP13")
private Double totP13;
@Column(name = "totP14")
private Double totP14;
@Column(name = "totP15")
private Double totP15;
@Column(name = "totP16")
private Double totP16;
@Column(name = "totP17")
private Double totP17;
@Column(name = "totP18")
private Double totP18;
@Column(name = "totP19")
private Double totP19;
@Column(name = "totP20")
private Double totP20;
@Column(name = "totP21")
private Double totP21;
@Column(name = "totP22")
private Double totP22;
@Column(name = "totP23")
private Double totP23;
@Column(name = "totP24")
private Double totP24;
@Column(name = "totP25")
private Double totP25;
@Column(name = "totP26")
private Double totP26;
@Column(name = "totP27")
private Double totP27;
@Column(name = "totP28")
private Double totP28;
@Column(name = "totP29")
private Double totP29;
@Column(name = "totP30")
private Double totP30;
@Column(name = "totP31")
private Double totP31;
@Column(name = "totP32")
private Double totP32;
@Column(name = "totP33")
private Double totP33;
@Column(name = "totP34")
private Double totP34;
@Column(name = "totP35")
private Double totP35;
@Column(name = "totP36")
private Double totP36;
@Column(name = "totP37")
private Double totP37;
@Column(name = "totP38")
private Double totP38;
@Column(name = "totP39")
private Double totP39;
@Column(name = "totP40")
private Double totP40;
@Column(name = "totP41")
private Double totP41;
@Column(name = "totP42")
private Double totP42;
@Column(name = "totP43")
private Double totP43;
@Column(name = "totP44")
private Double totP44;
@Column(name = "totP45")
private Double totP45;
@Column(name = "totP46")
private Double totP46;
@Column(name = "totP47")
private Double totP47;
@Column(name = "totP49")
private Double totP48;
@Column(name = "totP49")
private Double totP49;
@Column(name = "totP50")
private Double totP50;
public static List<DataHarmpowerP> relationToInfluxDB(DataHarmpowerPDTO dataHarmpowerP) {
if (dataHarmpowerP == null) {
return null;

105
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/influx/DataHarmpowerQ.java
View File

@@ -207,6 +207,111 @@ public class DataHarmpowerQ {
@Column(name = "q_50")
private Double q50;
//在线监测添加的字段
@Column(name = "totQ")
private Double totQ;
@Column(name = "totQ1")
private Double totQ1;
@Column(name = "totQ2")
private Double totQ2;
@Column(name = "totQ3")
private Double totQ3;
@Column(name = "totQ4")
private Double totQ4;
@Column(name = "totQ5")
private Double totQ5;
@Column(name = "totQ6")
private Double totQ6;
@Column(name = "totQ7")
private Double totQ7;
@Column(name = "totQ8")
private Double totQ8;
@Column(name = "totQ9")
private Double totQ9;
@Column(name = "totQ10")
private Double totQ10;
@Column(name = "totQ11")
private Double totQ11;
@Column(name = "totQ12")
private Double totQ12;
@Column(name = "totQ13")
private Double totQ13;
@Column(name = "totQ14")
private Double totQ14;
@Column(name = "totQ15")
private Double totQ15;
@Column(name = "totQ16")
private Double totQ16;
@Column(name = "totQ17")
private Double totQ17;
@Column(name = "totQ18")
private Double totQ18;
@Column(name = "totQ19")
private Double totQ19;
@Column(name = "totQ20")
private Double totQ20;
@Column(name = "totQ21")
private Double totQ21;
@Column(name = "totQ22")
private Double totQ22;
@Column(name = "totQ23")
private Double totQ23;
@Column(name = "totQ24")
private Double totQ24;
@Column(name = "totQ25")
private Double totQ25;
@Column(name = "totQ26")
private Double totQ26;
@Column(name = "totQ27")
private Double totQ27;
@Column(name = "totQ28")
private Double totQ28;
@Column(name = "totQ29")
private Double totQ29;
@Column(name = "totQ30")
private Double totQ30;
@Column(name = "totQ31")
private Double totQ31;
@Column(name = "totQ32")
private Double totQ32;
@Column(name = "totQ33")
private Double totQ33;
@Column(name = "totQ34")
private Double totQ34;
@Column(name = "totQ35")
private Double totQ35;
@Column(name = "totQ36")
private Double totQ36;
@Column(name = "totQ37")
private Double totQ37;
@Column(name = "totQ38")
private Double totQ38;
@Column(name = "totQ39")
private Double totQ39;
@Column(name = "totQ40")
private Double totQ40;
@Column(name = "totQ41")
private Double totQ41;
@Column(name = "totQ42")
private Double totQ42;
@Column(name = "totQ43")
private Double totQ43;
@Column(name = "totQ44")
private Double totQ44;
@Column(name = "totQ45")
private Double totQ45;
@Column(name = "totQ46")
private Double totQ46;
@Column(name = "totQ47")
private Double totQ47;
@Column(name = "totQ49")
private Double totQ48;
@Column(name = "totQ49")
private Double totQ49;
@Column(name = "totQ50")
private Double totQ50;
public static List<DataHarmpowerQ> relationToInfluxDB(DataHarmpowerQDTO dataHarmpowerQ) {
if (dataHarmpowerQ == null) {
return null;

105
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/po/influx/DataHarmpowerS.java
View File

@@ -207,6 +207,111 @@ public class DataHarmpowerS {
@Column(name = "s_50")
private Double s50;
//在线监测添加的字段
@Column(name = "totS")
private Double totS;
@Column(name = "totS1")
private Double totS1;
@Column(name = "totS2")
private Double totS2;
@Column(name = "totS3")
private Double totS3;
@Column(name = "totS4")
private Double totS4;
@Column(name = "totS5")
private Double totS5;
@Column(name = "totS6")
private Double totS6;
@Column(name = "totS7")
private Double totS7;
@Column(name = "totS8")
private Double totS8;
@Column(name = "totS9")
private Double totS9;
@Column(name = "totS10")
private Double totS10;
@Column(name = "totS11")
private Double totS11;
@Column(name = "totS12")
private Double totS12;
@Column(name = "totS13")
private Double totS13;
@Column(name = "totS14")
private Double totS14;
@Column(name = "totS15")
private Double totS15;
@Column(name = "totS16")
private Double totS16;
@Column(name = "totS17")
private Double totS17;
@Column(name = "totS18")
private Double totS18;
@Column(name = "totS19")
private Double totS19;
@Column(name = "totS20")
private Double totS20;
@Column(name = "totS21")
private Double totS21;
@Column(name = "totS22")
private Double totS22;
@Column(name = "totS23")
private Double totS23;
@Column(name = "totS24")
private Double totS24;
@Column(name = "totS25")
private Double totS25;
@Column(name = "totS26")
private Double totS26;
@Column(name = "totS27")
private Double totS27;
@Column(name = "totS28")
private Double totS28;
@Column(name = "totS29")
private Double totS29;
@Column(name = "totS30")
private Double totS30;
@Column(name = "totS31")
private Double totS31;
@Column(name = "totS32")
private Double totS32;
@Column(name = "totS33")
private Double totS33;
@Column(name = "totS34")
private Double totS34;
@Column(name = "totS35")
private Double totS35;
@Column(name = "totS36")
private Double totS36;
@Column(name = "totS37")
private Double totS37;
@Column(name = "totS38")
private Double totS38;
@Column(name = "totS39")
private Double totS39;
@Column(name = "totS40")
private Double totS40;
@Column(name = "totS41")
private Double totS41;
@Column(name = "totS42")
private Double totS42;
@Column(name = "totS43")
private Double totS43;
@Column(name = "totS44")
private Double totS44;
@Column(name = "totS45")
private Double totS45;
@Column(name = "totS46")
private Double totS46;
@Column(name = "totS47")
private Double totS47;
@Column(name = "totS49")
private Double totS48;
@Column(name = "totS49")
private Double totS49;
@Column(name = "totS50")
private Double totS50;
public static List<DataHarmpowerS> relationToInfluxDB(DataHarmpowerSDTO dataHarmpowerS) {
if (dataHarmpowerS == null) {
return null;

2
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/pojo/dto/PqsCommunicateDto.java
View File

@@ -18,6 +18,8 @@ public class PqsCommunicateDto {
private String description;
private Integer type;
//是否更新updateTime标志；数据上送更新1，状态翻转不更新0
private Integer flag=0;
}

9
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/controller/DataVController.java
View File

@@ -166,6 +166,15 @@ public class DataVController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, dataV, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/excludeZeroData")
@ApiOperation("判断是否存在是零飘数据")
public HttpResult<Boolean> excludeZeroData(@RequestBody LineCountEvaluateParam lineParam) {
String methodDescribe = getMethodDescribe("excludeZeroData");
Boolean b = dataVQuery.excludeZeroData(lineParam);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, b, methodDescribe);
}

13
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/IDataV.java
View File

@@ -81,4 +81,17 @@ public interface IDataV extends IMppService<RStatDataVD> {
List<DataVDto> getDataV(LineCountEvaluateParam lineParam);
List<DataVDto> getGroupByTimeDataV(LineCountEvaluateParam lineParam);
/**
* 查看监测点的电压，判断是否存在是零飘数据
* 如果零飘数据占比一半多，则抛弃该监测点数据
* @param lineParam
* @return: java.lang.Boolean
* @Author: wr
* @Date: 2026/1/15 9:26
*/
Boolean excludeZeroData(LineCountEvaluateParam lineParam);
}

2
data-processing/data-processing-boot/src/main/java/com/njcn/dataProcess/service/impl/LnDataDealServiceImpl.java
View File

@@ -97,7 +97,7 @@ public class LnDataDealServiceImpl implements LnDataDealService {
pqsCommunicateDto.setTime(LocalDateTimeUtil.format(dataVDTO.getTimeid(), DatePattern.NORM_DATETIME_PATTERN));
pqsCommunicateDto.setDevId(temp);
pqsCommunicateDto.setType(1);
pqsCommunicateDto.setFlag(1);
iPqsCommunicate.insertion(pqsCommunicateDto);
});

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

@@ -12,8 +12,6 @@ import com.njcn.dataProcess.dao.imapper.DataHarmpowerPMapper;
import com.njcn.dataProcess.dao.relation.mapper.RStatDataHarmPowerPRelationMapper;
import com.njcn.dataProcess.dto.DataHarmpowerPDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.po.influx.DataHarmphasicI;
import com.njcn.dataProcess.po.influx.DataHarmphasicV;
import com.njcn.dataProcess.po.influx.DataHarmpowerP;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataHarmPowerPDto;
@@ -29,6 +27,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@@ -128,68 +127,12 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(valueType);
//规定好集合指标参数
List<Double> data1 = valueTypeList.stream().map(DataHarmpowerP::getDf).collect(Collectors.toList());
List<Double> data2 = valueTypeList.stream().map(DataHarmpowerP::getPf).collect(Collectors.toList());
List<Double> data3 = valueTypeList.stream().map(DataHarmpowerP::getP).collect(Collectors.toList());
List<Double> data12 = valueTypeList.stream().map(DataHarmpowerP::getP1).collect(Collectors.toList());
List<Double> data13 = valueTypeList.stream().map(DataHarmpowerP::getP2).collect(Collectors.toList());
List<Double> data14 = valueTypeList.stream().map(DataHarmpowerP::getP3).collect(Collectors.toList());
List<Double> data15 = valueTypeList.stream().map(DataHarmpowerP::getP4).collect(Collectors.toList());
List<Double> data16 = valueTypeList.stream().map(DataHarmpowerP::getP5).collect(Collectors.toList());
List<Double> data17 = valueTypeList.stream().map(DataHarmpowerP::getP6).collect(Collectors.toList());
List<Double> data18 = valueTypeList.stream().map(DataHarmpowerP::getP7).collect(Collectors.toList());
List<Double> data19 = valueTypeList.stream().map(DataHarmpowerP::getP8).collect(Collectors.toList());
List<Double> data20 = valueTypeList.stream().map(DataHarmpowerP::getP9).collect(Collectors.toList());
List<Double> data21 = valueTypeList.stream().map(DataHarmpowerP::getP10).collect(Collectors.toList());
List<Double> data22 = valueTypeList.stream().map(DataHarmpowerP::getP11).collect(Collectors.toList());
List<Double> data23 = valueTypeList.stream().map(DataHarmpowerP::getP12).collect(Collectors.toList());
List<Double> data24 = valueTypeList.stream().map(DataHarmpowerP::getP13).collect(Collectors.toList());
List<Double> data25 = valueTypeList.stream().map(DataHarmpowerP::getP14).collect(Collectors.toList());
List<Double> data26 = valueTypeList.stream().map(DataHarmpowerP::getP15).collect(Collectors.toList());
List<Double> data27 = valueTypeList.stream().map(DataHarmpowerP::getP16).collect(Collectors.toList());
List<Double> data28 = valueTypeList.stream().map(DataHarmpowerP::getP17).collect(Collectors.toList());
List<Double> data29 = valueTypeList.stream().map(DataHarmpowerP::getP18).collect(Collectors.toList());
List<Double> data30 = valueTypeList.stream().map(DataHarmpowerP::getP19).collect(Collectors.toList());
List<Double> data31 = valueTypeList.stream().map(DataHarmpowerP::getP20).collect(Collectors.toList());
List<Double> data32 = valueTypeList.stream().map(DataHarmpowerP::getP21).collect(Collectors.toList());
List<Double> data33 = valueTypeList.stream().map(DataHarmpowerP::getP22).collect(Collectors.toList());
List<Double> data34 = valueTypeList.stream().map(DataHarmpowerP::getP23).collect(Collectors.toList());
List<Double> data35 = valueTypeList.stream().map(DataHarmpowerP::getP24).collect(Collectors.toList());
List<Double> data36 = valueTypeList.stream().map(DataHarmpowerP::getP25).collect(Collectors.toList());
List<Double> data37 = valueTypeList.stream().map(DataHarmpowerP::getP26).collect(Collectors.toList());
List<Double> data38 = valueTypeList.stream().map(DataHarmpowerP::getP27).collect(Collectors.toList());
List<Double> data39 = valueTypeList.stream().map(DataHarmpowerP::getP28).collect(Collectors.toList());
List<Double> data40 = valueTypeList.stream().map(DataHarmpowerP::getP29).collect(Collectors.toList());
List<Double> data41 = valueTypeList.stream().map(DataHarmpowerP::getP30).collect(Collectors.toList());
List<Double> data42 = valueTypeList.stream().map(DataHarmpowerP::getP31).collect(Collectors.toList());
List<Double> data43 = valueTypeList.stream().map(DataHarmpowerP::getP32).collect(Collectors.toList());
List<Double> data44 = valueTypeList.stream().map(DataHarmpowerP::getP33).collect(Collectors.toList());
List<Double> data45 = valueTypeList.stream().map(DataHarmpowerP::getP34).collect(Collectors.toList());
List<Double> data46 = valueTypeList.stream().map(DataHarmpowerP::getP35).collect(Collectors.toList());
List<Double> data47 = valueTypeList.stream().map(DataHarmpowerP::getP36).collect(Collectors.toList());
List<Double> data48 = valueTypeList.stream().map(DataHarmpowerP::getP37).collect(Collectors.toList());
List<Double> data49 = valueTypeList.stream().map(DataHarmpowerP::getP38).collect(Collectors.toList());
List<Double> data50 = valueTypeList.stream().map(DataHarmpowerP::getP39).collect(Collectors.toList());
List<Double> data51 = valueTypeList.stream().map(DataHarmpowerP::getP40).collect(Collectors.toList());
List<Double> data52 = valueTypeList.stream().map(DataHarmpowerP::getP41).collect(Collectors.toList());
List<Double> data53 = valueTypeList.stream().map(DataHarmpowerP::getP42).collect(Collectors.toList());
List<Double> data54 = valueTypeList.stream().map(DataHarmpowerP::getP43).collect(Collectors.toList());
List<Double> data55 = valueTypeList.stream().map(DataHarmpowerP::getP44).collect(Collectors.toList());
List<Double> data56 = valueTypeList.stream().map(DataHarmpowerP::getP45).collect(Collectors.toList());
List<Double> data57 = valueTypeList.stream().map(DataHarmpowerP::getP46).collect(Collectors.toList());
List<Double> data58 = valueTypeList.stream().map(DataHarmpowerP::getP47).collect(Collectors.toList());
List<Double> data59 = valueTypeList.stream().map(DataHarmpowerP::getP48).collect(Collectors.toList());
List<Double> data60 = valueTypeList.stream().map(DataHarmpowerP::getP49).collect(Collectors.toList());
List<Double> data61 = valueTypeList.stream().map(DataHarmpowerP::getP50).collect(Collectors.toList());
List<List<Double>> lists = Arrays.asList(data1,data2,data3,data12,data13,data14
,data15,data16,data17,data18,data19,data20,data21,data22,data23,data24
,data25,data26,data27,data28,data29,data30,data31,data32,data33,data34
,data35,data36,data37,data38,data39,data40,data41,data42,data43,data44
,data45,data46,data47,data48,data49,data50,data51,data52,data53,data54
,data55,data56,data57,data58,data59,data60,data61);
List<List<Double>> lists;
if (Objects.equals(phasicType, "T")) {
lists = extractDataLists(valueTypeList, "Tot");
} else {
lists = extractDataLists(valueTypeList, "");
}
value.setValueList(lists);
valueTypes.add(value);
});
@@ -203,6 +146,43 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
return result;
}
private List<List<Double>> extractDataLists(List<DataHarmpowerP> dataList, String prefix) {
List<List<Double>> result = new ArrayList<>();
// 前三个特殊字段
result.add(extractField(dataList, prefix + "Df"));
result.add(extractField(dataList, prefix + "Pf"));
result.add(extractField(dataList, prefix + "P"));
// P1到P50
for (int i = 1; i <= 50; i++) {
String fieldName = prefix + (i == 1 ? "P1" : "P" + i);
result.add(extractField(dataList, fieldName));
}
return result;
}
private List<Double> extractField(List<DataHarmpowerP> dataList, String fieldName) {
try {
Method method = DataHarmpowerP.class.getMethod("get" + capitalize(fieldName));
return dataList.stream()
.map(data -> {
try {
return (Double) method.invoke(data);
} catch (Exception e) {
throw new RuntimeException("Failed to get field: " + fieldName, e);
}
})
.collect(Collectors.toList());
} catch (NoSuchMethodException e) {
throw new RuntimeException("Field not found: " + fieldName, e);
}
}
private String capitalize(String str) {
if (str == null || str.isEmpty()) return str;
return str.substring(0, 1).toUpperCase() + str.substring(1);
}
@Override
public void addList(List<DataHarmPowerPDto> list) {
@@ -298,20 +278,31 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerP.class);
//A、B、C谐波有功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmP_", "p_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.eq(DataHarmpowerP::getLineId, lineId)
.eq(DataHarmpowerP::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
//三相总 谐波有功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmTP_", "totP_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
InfluxQueryWrapper eq = influxQueryWrapper.eq(DataHarmpowerP::getLineId, lineId)
.eq(DataHarmpowerP::getProcess, Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerP::getLineId)
.select(DataHarmpowerP::getPhasicType)
.select(DataHarmpowerP::getValueType)
//位移功率因数
.select("Pq_DF","df")
.select("Pq_DF", "df")
//视在功率因素
.select("Pq_PF","pf")
//总功功率
.select("Pq_P","p")
.select("Pq_PF", "pf")
//A、B、C三相有功功率
.select("Pq_P", "p")
//基波有功功率
.select("Pq_FundP","p_1")
.select("Pq_FundP", "p_1")
//三相总视在功率因数
.select("Pq_TotPF", "totPf")
//三相总位移功率因数
.select("Pq_TotDF", "totDf")
//三相总功率因数
.select("Pq_TotP", "totP")
//三相总基波有功功率
.select("Pq_TotHarmP", "totP1")
.between(DataHarmpowerP::getTime, startTime, endTime)
.eq(DataHarmpowerP::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) {

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

@@ -26,6 +26,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@@ -124,66 +125,13 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(valueType);
//规定好集合指标参数
List<Double> data1 = valueTypeList.stream().map(DataHarmpowerQ::getQ).collect(Collectors.toList());
List<Double> data12 = valueTypeList.stream().map(DataHarmpowerQ::getQ1).collect(Collectors.toList());
List<Double> data13 = valueTypeList.stream().map(DataHarmpowerQ::getQ2).collect(Collectors.toList());
List<Double> data14 = valueTypeList.stream().map(DataHarmpowerQ::getQ3).collect(Collectors.toList());
List<Double> data15 = valueTypeList.stream().map(DataHarmpowerQ::getQ4).collect(Collectors.toList());
List<Double> data16 = valueTypeList.stream().map(DataHarmpowerQ::getQ5).collect(Collectors.toList());
List<Double> data17 = valueTypeList.stream().map(DataHarmpowerQ::getQ6).collect(Collectors.toList());
List<Double> data18 = valueTypeList.stream().map(DataHarmpowerQ::getQ7).collect(Collectors.toList());
List<Double> data19 = valueTypeList.stream().map(DataHarmpowerQ::getQ8).collect(Collectors.toList());
List<Double> data20 = valueTypeList.stream().map(DataHarmpowerQ::getQ9).collect(Collectors.toList());
List<Double> data21 = valueTypeList.stream().map(DataHarmpowerQ::getQ10).collect(Collectors.toList());
List<Double> data22 = valueTypeList.stream().map(DataHarmpowerQ::getQ11).collect(Collectors.toList());
List<Double> data23 = valueTypeList.stream().map(DataHarmpowerQ::getQ12).collect(Collectors.toList());
List<Double> data24 = valueTypeList.stream().map(DataHarmpowerQ::getQ13).collect(Collectors.toList());
List<Double> data25 = valueTypeList.stream().map(DataHarmpowerQ::getQ14).collect(Collectors.toList());
List<Double> data26 = valueTypeList.stream().map(DataHarmpowerQ::getQ15).collect(Collectors.toList());
List<Double> data27 = valueTypeList.stream().map(DataHarmpowerQ::getQ16).collect(Collectors.toList());
List<Double> data28 = valueTypeList.stream().map(DataHarmpowerQ::getQ17).collect(Collectors.toList());
List<Double> data29 = valueTypeList.stream().map(DataHarmpowerQ::getQ18).collect(Collectors.toList());
List<Double> data30 = valueTypeList.stream().map(DataHarmpowerQ::getQ19).collect(Collectors.toList());
List<Double> data31 = valueTypeList.stream().map(DataHarmpowerQ::getQ20).collect(Collectors.toList());
List<Double> data32 = valueTypeList.stream().map(DataHarmpowerQ::getQ21).collect(Collectors.toList());
List<Double> data33 = valueTypeList.stream().map(DataHarmpowerQ::getQ22).collect(Collectors.toList());
List<Double> data34 = valueTypeList.stream().map(DataHarmpowerQ::getQ23).collect(Collectors.toList());
List<Double> data35 = valueTypeList.stream().map(DataHarmpowerQ::getQ24).collect(Collectors.toList());
List<Double> data36 = valueTypeList.stream().map(DataHarmpowerQ::getQ25).collect(Collectors.toList());
List<Double> data37 = valueTypeList.stream().map(DataHarmpowerQ::getQ26).collect(Collectors.toList());
List<Double> data38 = valueTypeList.stream().map(DataHarmpowerQ::getQ27).collect(Collectors.toList());
List<Double> data39 = valueTypeList.stream().map(DataHarmpowerQ::getQ28).collect(Collectors.toList());
List<Double> data40 = valueTypeList.stream().map(DataHarmpowerQ::getQ29).collect(Collectors.toList());
List<Double> data41 = valueTypeList.stream().map(DataHarmpowerQ::getQ30).collect(Collectors.toList());
List<Double> data42 = valueTypeList.stream().map(DataHarmpowerQ::getQ31).collect(Collectors.toList());
List<Double> data43 = valueTypeList.stream().map(DataHarmpowerQ::getQ32).collect(Collectors.toList());
List<Double> data44 = valueTypeList.stream().map(DataHarmpowerQ::getQ33).collect(Collectors.toList());
List<Double> data45 = valueTypeList.stream().map(DataHarmpowerQ::getQ34).collect(Collectors.toList());
List<Double> data46 = valueTypeList.stream().map(DataHarmpowerQ::getQ35).collect(Collectors.toList());
List<Double> data47 = valueTypeList.stream().map(DataHarmpowerQ::getQ36).collect(Collectors.toList());
List<Double> data48 = valueTypeList.stream().map(DataHarmpowerQ::getQ37).collect(Collectors.toList());
List<Double> data49 = valueTypeList.stream().map(DataHarmpowerQ::getQ38).collect(Collectors.toList());
List<Double> data50 = valueTypeList.stream().map(DataHarmpowerQ::getQ39).collect(Collectors.toList());
List<Double> data51 = valueTypeList.stream().map(DataHarmpowerQ::getQ40).collect(Collectors.toList());
List<Double> data52 = valueTypeList.stream().map(DataHarmpowerQ::getQ41).collect(Collectors.toList());
List<Double> data53 = valueTypeList.stream().map(DataHarmpowerQ::getQ42).collect(Collectors.toList());
List<Double> data54 = valueTypeList.stream().map(DataHarmpowerQ::getQ43).collect(Collectors.toList());
List<Double> data55 = valueTypeList.stream().map(DataHarmpowerQ::getQ44).collect(Collectors.toList());
List<Double> data56 = valueTypeList.stream().map(DataHarmpowerQ::getQ45).collect(Collectors.toList());
List<Double> data57 = valueTypeList.stream().map(DataHarmpowerQ::getQ46).collect(Collectors.toList());
List<Double> data58 = valueTypeList.stream().map(DataHarmpowerQ::getQ47).collect(Collectors.toList());
List<Double> data59 = valueTypeList.stream().map(DataHarmpowerQ::getQ48).collect(Collectors.toList());
List<Double> data60 = valueTypeList.stream().map(DataHarmpowerQ::getQ49).collect(Collectors.toList());
List<Double> data61 = valueTypeList.stream().map(DataHarmpowerQ::getQ50).collect(Collectors.toList());
List<List<Double>> lists = Arrays.asList(data1,data12,data13,data14
,data15,data16,data17,data18,data19,data20,data21,data22,data23,data24
,data25,data26,data27,data28,data29,data30,data31,data32,data33,data34
,data35,data36,data37,data38,data39,data40,data41,data42,data43,data44
,data45,data46,data47,data48,data49,data50,data51,data52,data53,data54
,data55,data56,data57,data58,data59,data60,data61);
List<List<Double>> lists;
if (Objects.equals(phasicType, "T")) {
lists = extractDataLists(valueTypeList, "Tot");
} else {
lists = extractDataLists(valueTypeList, "");
}
value.setValueList(lists);
valueTypes.add(value);
});
@@ -197,6 +145,39 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
return result;
}
private List<List<Double>> extractDataLists(List<DataHarmpowerQ> dataList, String prefix) {
List<List<Double>> result = new ArrayList<>();
result.add(extractField(dataList, prefix + "Q"));
// Q1到Q50
for (int i = 1; i <= 50; i++) {
String fieldName = prefix + (i == 1 ? "Q1" : "Q" + i);
result.add(extractField(dataList, fieldName));
}
return result;
}
private List<Double> extractField(List<DataHarmpowerQ> dataList, String fieldName) {
try {
Method method = DataHarmpowerQ.class.getMethod("get" + capitalize(fieldName));
return dataList.stream()
.map(data -> {
try {
return (Double) method.invoke(data);
} catch (Exception e) {
throw new RuntimeException("Failed to get field: " + fieldName, e);
}
})
.collect(Collectors.toList());
} catch (NoSuchMethodException e) {
throw new RuntimeException("Field not found: " + fieldName, e);
}
}
private String capitalize(String str) {
if (str == null || str.isEmpty()) return str;
return str.substring(0, 1).toUpperCase() + str.substring(1);
}
@Override
public void addList(List<DataHarmPowerQDto> list) {
@@ -290,16 +271,23 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerQ.class);
//A、B、C谐波无功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmQ_", "q_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
//三相总 谐波有功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmTQ_", "totQ_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.eq(DataHarmpowerQ::getLineId, lineId)
.eq(DataHarmpowerQ::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerQ::getLineId)
.select(DataHarmpowerQ::getPhasicType)
.select(DataHarmpowerQ::getValueType)
//总功功率
.select("Pq_Q","q")
//基波有功功率
//A、B、C三相无功功率
.select("Pq_Q", "q")
//基波无功功率
.select("Pq_FundQ","q_1")
//总无功功率
.select("Pq_TotQ","totQ")
//基波无功功率(T)
.select("Pq_TotHarmQ","totQ1")
.between(DataHarmpowerQ::getTime, startTime, endTime)
.eq(DataHarmpowerQ::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) {

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

@@ -26,6 +26,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@@ -124,66 +125,12 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(valueType);
//规定好集合指标参数
List<Double> data1 = valueTypeList.stream().map(DataHarmpowerS::getS).collect(Collectors.toList());
List<Double> data12 = valueTypeList.stream().map(DataHarmpowerS::getS1).collect(Collectors.toList());
List<Double> data13 = valueTypeList.stream().map(DataHarmpowerS::getS2).collect(Collectors.toList());
List<Double> data14 = valueTypeList.stream().map(DataHarmpowerS::getS3).collect(Collectors.toList());
List<Double> data15 = valueTypeList.stream().map(DataHarmpowerS::getS4).collect(Collectors.toList());
List<Double> data16 = valueTypeList.stream().map(DataHarmpowerS::getS5).collect(Collectors.toList());
List<Double> data17 = valueTypeList.stream().map(DataHarmpowerS::getS6).collect(Collectors.toList());
List<Double> data18 = valueTypeList.stream().map(DataHarmpowerS::getS7).collect(Collectors.toList());
List<Double> data19 = valueTypeList.stream().map(DataHarmpowerS::getS8).collect(Collectors.toList());
List<Double> data20 = valueTypeList.stream().map(DataHarmpowerS::getS9).collect(Collectors.toList());
List<Double> data21 = valueTypeList.stream().map(DataHarmpowerS::getS10).collect(Collectors.toList());
List<Double> data22 = valueTypeList.stream().map(DataHarmpowerS::getS11).collect(Collectors.toList());
List<Double> data23 = valueTypeList.stream().map(DataHarmpowerS::getS12).collect(Collectors.toList());
List<Double> data24 = valueTypeList.stream().map(DataHarmpowerS::getS13).collect(Collectors.toList());
List<Double> data25 = valueTypeList.stream().map(DataHarmpowerS::getS14).collect(Collectors.toList());
List<Double> data26 = valueTypeList.stream().map(DataHarmpowerS::getS15).collect(Collectors.toList());
List<Double> data27 = valueTypeList.stream().map(DataHarmpowerS::getS16).collect(Collectors.toList());
List<Double> data28 = valueTypeList.stream().map(DataHarmpowerS::getS17).collect(Collectors.toList());
List<Double> data29 = valueTypeList.stream().map(DataHarmpowerS::getS18).collect(Collectors.toList());
List<Double> data30 = valueTypeList.stream().map(DataHarmpowerS::getS19).collect(Collectors.toList());
List<Double> data31 = valueTypeList.stream().map(DataHarmpowerS::getS20).collect(Collectors.toList());
List<Double> data32 = valueTypeList.stream().map(DataHarmpowerS::getS21).collect(Collectors.toList());
List<Double> data33 = valueTypeList.stream().map(DataHarmpowerS::getS22).collect(Collectors.toList());
List<Double> data34 = valueTypeList.stream().map(DataHarmpowerS::getS23).collect(Collectors.toList());
List<Double> data35 = valueTypeList.stream().map(DataHarmpowerS::getS24).collect(Collectors.toList());
List<Double> data36 = valueTypeList.stream().map(DataHarmpowerS::getS25).collect(Collectors.toList());
List<Double> data37 = valueTypeList.stream().map(DataHarmpowerS::getS26).collect(Collectors.toList());
List<Double> data38 = valueTypeList.stream().map(DataHarmpowerS::getS27).collect(Collectors.toList());
List<Double> data39 = valueTypeList.stream().map(DataHarmpowerS::getS28).collect(Collectors.toList());
List<Double> data40 = valueTypeList.stream().map(DataHarmpowerS::getS29).collect(Collectors.toList());
List<Double> data41 = valueTypeList.stream().map(DataHarmpowerS::getS30).collect(Collectors.toList());
List<Double> data42 = valueTypeList.stream().map(DataHarmpowerS::getS31).collect(Collectors.toList());
List<Double> data43 = valueTypeList.stream().map(DataHarmpowerS::getS32).collect(Collectors.toList());
List<Double> data44 = valueTypeList.stream().map(DataHarmpowerS::getS33).collect(Collectors.toList());
List<Double> data45 = valueTypeList.stream().map(DataHarmpowerS::getS34).collect(Collectors.toList());
List<Double> data46 = valueTypeList.stream().map(DataHarmpowerS::getS35).collect(Collectors.toList());
List<Double> data47 = valueTypeList.stream().map(DataHarmpowerS::getS36).collect(Collectors.toList());
List<Double> data48 = valueTypeList.stream().map(DataHarmpowerS::getS37).collect(Collectors.toList());
List<Double> data49 = valueTypeList.stream().map(DataHarmpowerS::getS38).collect(Collectors.toList());
List<Double> data50 = valueTypeList.stream().map(DataHarmpowerS::getS39).collect(Collectors.toList());
List<Double> data51 = valueTypeList.stream().map(DataHarmpowerS::getS40).collect(Collectors.toList());
List<Double> data52 = valueTypeList.stream().map(DataHarmpowerS::getS41).collect(Collectors.toList());
List<Double> data53 = valueTypeList.stream().map(DataHarmpowerS::getS42).collect(Collectors.toList());
List<Double> data54 = valueTypeList.stream().map(DataHarmpowerS::getS43).collect(Collectors.toList());
List<Double> data55 = valueTypeList.stream().map(DataHarmpowerS::getS44).collect(Collectors.toList());
List<Double> data56 = valueTypeList.stream().map(DataHarmpowerS::getS45).collect(Collectors.toList());
List<Double> data57 = valueTypeList.stream().map(DataHarmpowerS::getS46).collect(Collectors.toList());
List<Double> data58 = valueTypeList.stream().map(DataHarmpowerS::getS47).collect(Collectors.toList());
List<Double> data59 = valueTypeList.stream().map(DataHarmpowerS::getS48).collect(Collectors.toList());
List<Double> data60 = valueTypeList.stream().map(DataHarmpowerS::getS49).collect(Collectors.toList());
List<Double> data61 = valueTypeList.stream().map(DataHarmpowerS::getS50).collect(Collectors.toList());
List<List<Double>> lists = Arrays.asList(data1,data12,data13,data14
,data15,data16,data17,data18,data19,data20,data21,data22,data23,data24
,data25,data26,data27,data28,data29,data30,data31,data32,data33,data34
,data35,data36,data37,data38,data39,data40,data41,data42,data43,data44
,data45,data46,data47,data48,data49,data50,data51,data52,data53,data54
,data55,data56,data57,data58,data59,data60,data61);
List<List<Double>> lists;
if (Objects.equals(phasicType, "T")) {
lists = extractDataLists(valueTypeList, "Tot");
} else {
lists = extractDataLists(valueTypeList, "");
}
value.setValueList(lists);
valueTypes.add(value);
});
@@ -197,6 +144,39 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
return result;
}
private List<List<Double>> extractDataLists(List<DataHarmpowerS> dataList, String prefix) {
List<List<Double>> result = new ArrayList<>();
result.add(extractField(dataList, prefix + "S"));
// Q1到Q50
for (int i = 1; i <= 50; i++) {
String fieldName = prefix + (i == 1 ? "S1" : "S" + i);
result.add(extractField(dataList, fieldName));
}
return result;
}
private List<Double> extractField(List<DataHarmpowerS> dataList, String fieldName) {
try {
Method method = DataHarmpowerS.class.getMethod("get" + capitalize(fieldName));
return dataList.stream()
.map(data -> {
try {
return (Double) method.invoke(data);
} catch (Exception e) {
throw new RuntimeException("Failed to get field: " + fieldName, e);
}
})
.collect(Collectors.toList());
} catch (NoSuchMethodException e) {
throw new RuntimeException("Field not found: " + fieldName, e);
}
}
private String capitalize(String str) {
if (str == null || str.isEmpty()) return str;
return str.substring(0, 1).toUpperCase() + str.substring(1);
}
@Override
public void addList(List<DataHarmPowerSDto> list) {
@@ -290,16 +270,23 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerS.class);
//A、B、C谐波视在功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmS_", "s_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
//三相总 谐波视在功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmTS_", "totS_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.eq(DataHarmpowerS::getLineId, lineId)
.eq(DataHarmpowerS::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerS::getLineId)
.select(DataHarmpowerS::getPhasicType)
.select(DataHarmpowerS::getValueType)
//总功功率
//A、B、C视在功率
.select("Pq_S","s")
//基波有功功率
//基波视在功率
.select("Pq_FundS","s_1")
//总视在功率
.select("Pq_TotS","totS")
//基波视在功率（T）
.select("Pq_TotHarmS","totS1")
.between(DataHarmpowerS::getTime, startTime, endTime)
.eq(DataHarmpowerS::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) {

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

@@ -208,7 +208,7 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
List<DataV> list = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
if (CollUtil.isNotEmpty(list)) {
Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime();
if(CollUtil.isNotEmpty(abnormalTime)){
if (CollUtil.isNotEmpty(abnormalTime)) {
if (abnormalTime.containsKey(lineParam.getLineId().get(0))) {
List<String> timeList = abnormalTime.get(lineParam.getLineId().get(0));
//有异常数据,当前监测点自身的异常数据
@@ -216,7 +216,7 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
result.addAll(list.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()));
}
}
}else{
} else {
result.addAll(list);
}
}
@@ -390,8 +390,8 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
.groupBy(DataV::getLineId)
.between(DataV::getTime, startTime, endTime);
List<MeasurementCountDTO> measurementCountDTOList = dataVMapper.getMeasurementCount(influxQueryWrapper);
if(CollUtil.isNotEmpty(measurementCountDTOList)){
for(MeasurementCountDTO m : measurementCountDTOList){
if (CollUtil.isNotEmpty(measurementCountDTOList)) {
for (MeasurementCountDTO m : measurementCountDTOList) {
MeasurementCountDTO.MeasurementCountResultDTO p = new MeasurementCountDTO.MeasurementCountResultDTO();
p.setFreq(m.getFreq());
p.setLineId(m.getLineId());
@@ -425,6 +425,27 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
return result;
}
@Override
public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
InfluxQueryWrapper dataVQueryWrapper = new InfluxQueryWrapper(DataV.class);
dataVQueryWrapper.eq(DataV::getLineId, lineParam.getLineId().get(0))
.eq(DataV::getValueType, InfluxDbSqlConstant.AVG_WEB)
.ne(DataV::getPhasicType, InfluxDBTableConstant.PHASE_TYPE_T)
.select(DataV::getRms)
.between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime());
List<DataV> rmsResult = dataVMapper.selectByQueryWrapper(dataVQueryWrapper);
if (CollUtil.isNotEmpty(rmsResult)) {
List<DataV> exceptionData = rmsResult.stream().filter(dataV -> dataV.getRms() < 1.0).collect(Collectors.toList());
if (CollUtil.isNotEmpty(exceptionData)) {
if (exceptionData.size() * 2 >= rmsResult.size()) {
return true;
}
}
}
return false;
}
/**
* 按监测点集合、时间条件获取dataV分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理
@@ -483,53 +504,53 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class);
influxQueryWrapper.eq(DataV::getLineId, lineId)
.eq(DataV::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.eq(DataV::getProcess, Integer.toString(devsMap.get(devId).getProcess()))
.select(DataV::getLineId)
.select(DataV::getPhasicType)
.select(DataV::getValueType)
//频率
.select("Pq_Freq","freq")
.select("Pq_Freq", "freq")
//频率偏差
.select("Pq_FreqDev","freq_dev")
.select("Pq_FreqDev", "freq_dev")
//相电压有效值
.select("Pq_RmsU","rms")
.select("Pq_RmsU", "rms")
//线电压有效值
.select("Pq_RmsLU","rms_lvr")
.select("Pq_RmsLU", "rms_lvr")
//电压负序
.select("Pq_SeqNegU","v_neg")
.select("Pq_SeqNegU", "v_neg")
//电压正序
.select("Pq_SeqPosU","v_pos")
.select("Pq_SeqPosU", "v_pos")
//电压零序
.select("Pq_SeqZeroU","v_zero")
.select("Pq_SeqZeroU", "v_zero")
//电压负序不平衡度
.select("Pq_UnbalNegU","v_unbalance")
.select("Pq_UnbalNegU", "v_unbalance")
.between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime())
.eq(DataV::getQualityFlag, "0");
if (CollUtil.isNotEmpty(lineParam.getPhasicType())) {
influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType());
}
if (Objects.isNull(po.getLineNo())) {
influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getClDid()));
influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getClDid()));
} else {
influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getLineNo()));
influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getLineNo()));
}
//判断接线方式 (0-星型 1-角型 2-V型) 星型是相电压 角型或者v型是线电压
if (Objects.equals(po.getConType(),0)) {
if (Objects.equals(po.getConType(), 0)) {
//相电压偏差
influxQueryWrapper.select("Pq_UDev","vu_dev");
influxQueryWrapper.select("Pq_UDev", "vu_dev");
//相电压谐波总畸变率
influxQueryWrapper.select("Pq_ThdU","v_thd");
influxQueryWrapper.select("Pq_ThdU", "v_thd");
//相电压基波有效值
influxQueryWrapper.select("Pq_RmsFundU","v_1");
influxQueryWrapper.select("Pq_RmsFundU", "v_1");
//2-50次 相电压谐波有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
} else {
//线电压偏差
influxQueryWrapper.select("Pq_LUDev","vu_dev");
influxQueryWrapper.select("Pq_LUDev", "vu_dev");
//线电压谐波总畸变率
influxQueryWrapper.select("Pq_ThdLU","v_thd");
influxQueryWrapper.select("Pq_ThdLU", "v_thd");
//线电压基波有效值
influxQueryWrapper.select("Pq_RmsFundLU","v_1");
influxQueryWrapper.select("Pq_RmsFundLU", "v_1");
//2-50次 线电压谐波有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmLUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
}

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

@@ -129,7 +129,9 @@ public class InfluxdbPqsCommunicateImpl implements IPqsCommunicate {
//更新mysql数据
DevComFlagDTO devComFlagDTO = new DevComFlagDTO();
devComFlagDTO.setId(pqsCommunicateDto.getDevId());
if(Objects.equals(pqsCommunicateDto.getFlag(),1)){
devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER));
}
devComFlagDTO.setStatus(pqsCommunicateDto.getType());
deviceFeignClient.updateDevComFlag(devComFlagDTO);

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

@@ -167,6 +167,11 @@ public class RelationDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
return Collections.emptyList();
}
@Override
public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
return null;
}
private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) {
List<DataVDto> result = new ArrayList<>();
Map<String, List<DataVDto>> lineMap = list.stream().collect(Collectors.groupingBy(DataVDto::getLineId));

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:

10
message/message-boot/src/main/java/com/njcn/message/consumer/DeviceRunFlagDataConsumer.java
View File

@@ -13,6 +13,7 @@ import com.njcn.stat.api.MessAnalysisFeignClient;
import com.njcn.system.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
@@ -82,7 +83,16 @@ public class DeviceRunFlagDataConsumer extends EnhanceConsumerMessageHandler<Dev
@Override
protected void handleMessage(DevComFlagDTO message) {
//获取之前设备状态
String devFalg =redisUtil.getStringByKey(RedisKeyPrefix.DEVICE_RUN_FLAG.concat(message.getId()));
if(StringUtils.isBlank(devFalg)||(!Objects.equals(Integer.valueOf(devFalg),message.getStatus()))){
//状态翻转
messAnalysisFeignClient.handleDevRunflag(message);
redisUtil.saveByKey(RedisKeyPrefix.DEVICE_RUN_FLAG.concat(message.getId()),message.getStatus()+"");
}
}