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-03
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

22 Commits
2025-12 ... 2026-03

Author SHA1 Message Date
wr
09fe5ca317 Merge remote-tracking branch 'origin/master' 2026-04-02 10:48:59 +08:00
wr
7412e2beda 微调 2026-04-02 10:48:12 +08:00
wr
90baa65b71 微调 2026-04-02 10:47:05 +08:00
xy
3f9e2076d2 辽宁算法兼容 2026-03-26 10:28:33 +08:00
cdf
1bda6b1156 算法模块整合nacos鉴权 2026-03-21 15:18:25 +08:00
xy
d3b4146674 修复bug。功率T项数据缺失 2026-03-18 10:22:57 +08:00
cdf
365448d8a0 修复算法运行报错bug 2026-03-11 11:19:12 +08:00
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
33 changed files with 1245 additions and 548 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 javax.annotation.Resource;
import java.util.*; import java.util.*;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static com.njcn.algorithm.utils.MemorySizeUtil.dealResponse; 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.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class); CalculatedParam repairParam = BeanUtil.copyProperties(calculatedParam, CalculatedParam.class);
repairParam.setType(0);
flowService.execute2Resp(methodDescribe, repairParam); flowService.execute2Resp(methodDescribe, repairParam);
} }
} else { } else {
//非补招 //非补招
calculatedParam.setType(0);
liteflowResponse = flowExecutor.execute2Resp("measurement_point", calculatedParam); liteflowResponse = flowExecutor.execute2Resp("measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe); dealResponse(calculatedParam, liteflowResponse, methodDescribe);
} }
@@ -162,6 +165,7 @@ public class ExecutionCenter extends BaseController {
} }
} else { } else {
//非补招 //非补招
calculatedParam.setType(2);
liteflowResponse = flowExecutor.execute2Resp("wl_measurement_point", calculatedParam); liteflowResponse = flowExecutor.execute2Resp("wl_measurement_point", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe); dealResponse(calculatedParam, liteflowResponse, methodDescribe);
} }

17
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.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.CompletableFuture;
public interface IDataLimitRateAsync { public interface IDataLimitRateAsync {
/** /**
* limit_rate多线程算法 * limit_rate多线程算法
*
* @param dataDate * @param dataDate
* @param list * @param list
* @param phase * @param phase
* @param overLimitMap * @param overLimitMap
* @param size * @param size
* @param i * @param i
* @param type 系统类型
* @return
* @Author: wr * @Author: wr
* @Date: 2025/12/17 12:16 * @Date: 2025/12/17 12:16
*/ */
void lineDataRate(String dataDate, CompletableFuture<Void> lineDataRate(String dataDate,
List<String> list, List<String> list,
List<String> phase, List<String> phase,
Map<String, Overlimit> overLimitMap, Map<String, Overlimit> overLimitMap,
int size, 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.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate())); lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
List<String> lineIdList = calculatedParam.getIdList(); List<String> lineIdList = calculatedParam.getIdList();
getAbnormalData(lineParam); //getAbnormalData(lineParam);
for (String lineId : lineIdList) { for (String lineId : lineIdList) {
DataComassesDPO rStatComassesDpo = new DataComassesDPO(); DataComassesDPO rStatComassesDpo = new DataComassesDPO();
rStatComassesDpo.setTime(calculatedParam.getDataDate()); rStatComassesDpo.setTime(calculatedParam.getDataDate());

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

@@ -22,6 +22,7 @@ import org.springframework.stereotype.Component;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.*; import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@@ -64,9 +65,13 @@ public class DayDataServiceImpl implements IDayDataService {
private DataInharmVFeignClient dataInharmVFeignClient; private DataInharmVFeignClient dataInharmVFeignClient;
@Resource @Resource
private DataPltFeignClient dataPltFeignClient; private DataPltFeignClient dataPltFeignClient;
@Resource @Resource
private PqDataVerifyFeignClient pqDataVerifyFeignClient; private PqDataVerifyFeignClient pqDataVerifyFeignClient;
/**
* 查询配置 辽宁版本比较特殊，没有CP95值，采用平均值进行判断
*/
@Value("${version.used:master}")
private String versionUsed;
@Override @Override
public void dataVHandler(CalculatedParam calculatedParam) { public void dataVHandler(CalculatedParam calculatedParam) {
@@ -82,7 +87,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataVFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -100,9 +105,10 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataVHandler(item3,valueTypes,dto,true); //默认CP95值取 CP95的最大值；如果有特殊要求，可以根据平均值的数据取CP95
channelDataVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -131,7 +137,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataIFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -148,9 +154,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataIHandler(item3,valueTypes,dto,true); channelDataIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -180,7 +186,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataFlickerFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataFlickerFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -196,9 +202,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4); dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataFlickerHandler(item3,dto,item4,true); channelDataFlickerHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -229,7 +235,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataFlucFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataFlucFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -245,9 +251,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4); dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataFlucHandler(item3,dto,item4,true); channelDataFlucHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -277,7 +283,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmphasicIFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmphasicIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -294,9 +300,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPhasicIHandler(item3,valueTypes,dto,true); channelDataHarmPhasicIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -325,7 +331,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmphasicVFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmphasicVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -342,9 +348,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPhasicVHandler(item3,valueTypes,dto,true); channelDataHarmPhasicVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -373,7 +379,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerPFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmpowerPFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -390,9 +396,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerPHandler(item3,valueTypes,dto,true); channelDataHarmPowerPHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -421,7 +427,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerQFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmpowerQFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -438,9 +444,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerQHandler(item3,valueTypes,dto,true); channelDataHarmPowerQHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -469,7 +475,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmpowerSFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmpowerSFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -486,9 +492,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmPowerSHandler(item3,valueTypes,dto,true); channelDataHarmPowerSHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -517,7 +523,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmRateIFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmRateIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -534,9 +540,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmRateIHandler(item3,valueTypes,dto,true); channelDataHarmRateIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -565,7 +571,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataHarmRateVFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataHarmRateVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -582,9 +588,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataHarmRateVHandler(item3,valueTypes,dto,true); channelDataHarmRateVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -613,7 +619,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataInharmIFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataInharmIFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -630,9 +636,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataInHarmIHandler(item3,valueTypes,dto,true); channelDataInHarmIHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -661,7 +667,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataInharmVFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataInharmVFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -678,9 +684,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item3.getValueType()); dto.setValueType(item3.getValueType().toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataInHarmVHandler(item3,valueTypes,dto,true); channelDataInHarmVHandler(item3,valueTypes,dto,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -710,7 +716,7 @@ public class DayDataServiceImpl implements IDayDataService {
pendingIds.forEach(list->{ pendingIds.forEach(list->{
lineParam.setLineId(list); lineParam.setLineId(list);
//添加异常数据时间点 //添加异常数据时间点
getAbnormalData(lineParam); //getAbnormalData(lineParam);
//获取原始数据 //获取原始数据
List<CommonMinuteDto> partList = dataPltFeignClient.getBaseData(lineParam).getData(); List<CommonMinuteDto> partList = dataPltFeignClient.getBaseData(lineParam).getData();
if (CollUtil.isNotEmpty(partList)) { if (CollUtil.isNotEmpty(partList)) {
@@ -726,9 +732,9 @@ public class DayDataServiceImpl implements IDayDataService {
dto.setTime(item.getTime()); dto.setTime(item.getTime());
dto.setLineId(item.getLineId()); dto.setLineId(item.getLineId());
dto.setPhasicType(item2.getPhasicType()); dto.setPhasicType(item2.getPhasicType());
dto.setValueType(item4); dto.setValueType(item4.toUpperCase());
dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag()); dto.setQualityFlag(Objects.equals(item.getQualityFlag(),"null") ? "0" : item.getQualityFlag());
channelDataPltHandler(item3,dto,item4,true); channelDataPltHandler(item3,dto,item4,!Objects.equals(versionUsed, "liaoning"));
result.add(dto); result.add(dto);
}); });
}); });
@@ -1549,47 +1555,64 @@ public class DayDataServiceImpl implements IDayDataService {
valueType = valueType.toUpperCase(); valueType = valueType.toUpperCase();
if (scheme) { if (scheme) {
switch (valueType) { switch (valueType) {
case InfluxDbSqlConstant.MAX: case "MAX":
case InfluxDbSqlConstant.CP95: case "CP95":
Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare); Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
result = max.orElse(null); result = max.orElse(0.0);
break; break;
case InfluxDbSqlConstant.MIN: case "MIN":
Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare); Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
result = min.orElse(null); result = min.orElse(0.0);
break; break;
case InfluxDbSqlConstant.AVG_WEB: case "AVG":
OptionalDouble average = list.stream() OptionalDouble average = list.stream()
.filter(Objects::nonNull) .filter(Objects::nonNull)
.mapToDouble(Double::doubleValue) .mapToDouble(Double::doubleValue)
.average(); .average();
result = average.isPresent() ? average.getAsDouble() : null; result = average.isPresent() ? average.getAsDouble() : 0.0;
break; break;
default: default:
break; break;
} }
} else { } else {
switch (valueType) { switch (valueType) {
case InfluxDbSqlConstant.MAX: case "MAX":
Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare); Optional<Double> max = list.stream().filter(Objects::nonNull).max(Double::compare);
result = max.orElse(null); result = max.orElse(0.0);
break; break;
case InfluxDbSqlConstant.MIN: case "MIN":
Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare); Optional<Double> min = list.stream().filter(Objects::nonNull).min(Double::compare);
result = min.orElse(null); result = min.orElse(0.0);
break; break;
case InfluxDbSqlConstant.AVG_WEB: case "AVG":
OptionalDouble average = list.stream() OptionalDouble average = list.stream()
.filter(Objects::nonNull) .filter(Objects::nonNull)
.mapToDouble(Double::doubleValue) .mapToDouble(Double::doubleValue)
.average(); .average();
result = average.isPresent() ? average.getAsDouble() : null; result = average.isPresent() ? average.getAsDouble() : 0.0;
break; break;
case InfluxDbSqlConstant.CP95: case "CP95":
list.sort(Collections.reverseOrder()); if (list == null || list.isEmpty()) {
int discardCount = (int) Math.ceil(list.size() * 0.05); result = 0.0;
List<Double> remainingList = list.subList(discardCount, list.size()); break;
result = remainingList.isEmpty() ? null : remainingList.get(0); }
// 过滤掉 null 元素
List<Double> filteredList = list.stream()
.filter(Objects::nonNull)
.collect(Collectors.toList());
if (filteredList.isEmpty()) {
result = 0.0;
break;
}
filteredList.sort(Collections.reverseOrder());
int discardCount = (int) Math.ceil(filteredList.size() * 0.05);
if (discardCount >= filteredList.size()) {
result = 0.0;
break;
}
List<Double> remainingList = filteredList.subList(discardCount, filteredList.size());
result = remainingList.isEmpty() ? 0.0 : remainingList.get(0);
break; break;
default: default:
break; break;
@@ -1603,4 +1626,28 @@ public class DayDataServiceImpl implements IDayDataService {
return result; return result;
} }
/**
* 将平均值赋值给 CP95
*/
public void putCP95ByAvg(List<CommonMinuteDto> list) {
list.forEach(po1->{
List<CommonMinuteDto.PhasicType> phasicTypeList = po1.getPhasicTypeList();
phasicTypeList.forEach(po2 -> {
List<CommonMinuteDto.ValueType> valueTypeList = po2.getValueTypeList();
AtomicReference<List<List<Double>>> valueList = new AtomicReference<>(new ArrayList<>());
valueTypeList.forEach(po3 -> {
if (po3.getValueType().equals(InfluxDbSqlConstant.AVG_WEB)) {
valueList.set(po3.getValueList());
}
if (po3.getValueType().equals(InfluxDbSqlConstant.CP95)) {
po3.setValueList((List<List<Double>>) valueList);
}
});
});
});
}
} }

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.algorithm.utils.MemorySizeUtil;
import com.njcn.common.utils.PubUtils; import com.njcn.common.utils.PubUtils;
import com.njcn.csdevice.api.CsLineFeignClient; 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.constant.PhaseType;
import com.njcn.dataProcess.param.LineCountEvaluateParam; 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.dataProcess.util.TimeUtils;
import com.njcn.device.biz.commApi.CommTerminalGeneralClient; import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
import com.njcn.device.biz.pojo.po.Overlimit; 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.jetbrains.annotations.NotNull;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
@@ -38,6 +45,8 @@ import java.time.LocalDateTime;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function; import java.util.function.Function;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@@ -67,6 +76,8 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
private IDataLimitRateAsync dataLimitRateAsync; private IDataLimitRateAsync dataLimitRateAsync;
@Resource @Resource
private CsLineFeignClient csLineFeignClient; private CsLineFeignClient csLineFeignClient;
@Resource
private CsOverLimitClient csOverLimitClient;
@Override @Override
public void limitRateHandler(CalculatedParam calculatedParam) { public void limitRateHandler(CalculatedParam calculatedParam) {
@@ -79,18 +90,45 @@ public class IDataCrossingServiceImpl implements IDataCrossingService {
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate())); lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
lineParam.setType(calculatedParam.getType()); lineParam.setType(calculatedParam.getType());
List<String> lineIds = calculatedParam.getIdList(); 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())); Map<String, Overlimit> overLimitMap = overLimitList.stream().collect(Collectors.toMap(Overlimit::getId, Function.identity()));
//以100个监测点分片处理 //以100个监测点分片处理
List<List<String>> pendingIds = ListUtils.partition(lineIds, 1); List<List<String>> pendingIds = ListUtils.partition(lineIds, 1);
ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C); ArrayList<String> phase = ListUtil.toList(PhaseType.PHASE_A, PhaseType.PHASE_B, PhaseType.PHASE_C);
MemorySizeUtil.getNowMemory(); MemorySizeUtil.getNowMemory();
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int i = 0; i < pendingIds.size(); i++) { for (int i = 0; i < pendingIds.size(); i++) {
logger.info(calculatedParam.getDataDate()+" 总分区数据：" + pendingIds.size() + "=====》当前第"+(i + 1)+"小分区"); logger.info(calculatedParam.getDataDate()+" 总分区数据：" + pendingIds.size() + "=====》当前第"+(i + 1)+"小分区");
List<String> list = pendingIds.get(i); 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(); System.gc();
} }

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

@@ -18,16 +18,15 @@ import com.njcn.influx.pojo.constant.InfluxDBTableConstant;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.lang.reflect.Method; import java.lang.reflect.Method;
import java.util.ArrayList; import java.util.*;
import java.util.HashMap; import java.util.concurrent.CompletableFuture;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@@ -55,21 +54,28 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
private DataLimitRateFeignClient dataLimitRateFeignClient; private DataLimitRateFeignClient dataLimitRateFeignClient;
@Resource @Resource
private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient; private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient;
/**
* 查询配置 辽宁版本比较特殊，没有CP95值，采用平均值进行判断
*/
@Value("${version.used:master}")
private String versionUsed;
@Override @Override
@Async("asyncExecutor") @Async("asyncExecutor")
public void lineDataRate(String dataDate, public CompletableFuture<Void> lineDataRate(String dataDate,
List<String> list, List<String> list,
List<String> phase, List<String> phase,
Map<String, Overlimit> overLimitMap, Map<String, Overlimit> overLimitMap,
int size, int size,
int i) { int i,
int type) {
List<DataLimitDetailDto> result = new ArrayList<>(); List<DataLimitDetailDto> result = new ArrayList<>();
LineCountEvaluateParam lineParam = new LineCountEvaluateParam(); LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(dataDate)); lineParam.setStartTime(TimeUtils.getBeginOfDay(dataDate));
lineParam.setEndTime(TimeUtils.getEndOfDay(dataDate)); lineParam.setEndTime(TimeUtils.getEndOfDay(dataDate));
lineParam.setLineId(list); lineParam.setLineId(list);
lineParam.setType(type);
//获取电压数据 //获取电压数据
List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData(); List<DataVDto> dataVAllTime = dataVFeignClient.getRawData(lineParam).getData();
//闪变数据 //闪变数据
@@ -80,99 +86,183 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(lineParam).getData(); List<DataHarmDto> dataVInHarmList = dataInharmVFeignClient.getRawData(lineParam).getData();
//电流数据 //电流数据
List<DataIDto> dataIList = dataIFeignClient.getRawData(lineParam).getData(); List<DataIDto> dataIList = dataIFeignClient.getRawData(lineParam).getData();
/** /**
* 功能描述:获取influxDB -> data_v -> * 功能描述:获取influxDB -> data_v ->
* 总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计) * 总计算次数(用data_v中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/ */
//在线监测设备会有AB、BC、CA的数据，会导致数据重复计算，这边根据时间去重
Map<String, List<DataVDto>> allTime = dataVAllTime.stream() Map<String, List<DataVDto>> allTime = dataVAllTime.stream()
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType())) .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(DataVDto::getLineId)); .collect(Collectors.groupingBy(
DataVDto::getLineId,
Collectors.collectingAndThen(
Collectors.toCollection(
// 使用 TreeSet 根据时间排序并去重
() -> new TreeSet<>(
Comparator.comparing(DataVDto::getMinTime)
)
),
ArrayList::new
)
));
/** /**
* 功能描述:获取influxDB -> data_plt -> * 功能描述:获取influxDB -> data_plt ->
* 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计) * 闪变总计算次数(用data_plt中phasic_type=A,value_type=avg,quality_flag=0来参与统计)
*/ */
//fixme 冀北现场 闪变原始表没有 value_type 这个参数 //fixme 冀北现场 闪变原始表没有 value_type 这个参数
Map<String, List<DataPltDto>> pltAllTime = dataPltAllTime.stream() Map<String, List<DataPltDto>> pltAllTime;
.filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType())) if (type == 2) {
// .filter(x -> InfluxDbSqlConstant.AVG_WEB.equalsIgnoreCase(x.getValueType())) pltAllTime = dataPltAllTime.stream()
.collect(Collectors.groupingBy(DataPltDto::getLineId)); .filter(x -> PhaseType.PHASE_A.equals(x.getPhasicType()))
.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 -> *功能描述:获取influxDB -> data_harmrate_v ->
* 2-25次谐波电压含有率 -> A相||B相||C相的日95%概率值 * 2-25次谐波电压含有率 -> A相||B相||C相的日95%概率值
*/ */
Map<String, List<DataHarmDto>> harmRateV = dataVHarmList.stream() Map<String, List<DataHarmDto>> harmRateV;
.filter(x -> phase.contains(x.getPhasicType())) if (Objects.equals(versionUsed, "liaoning")) {
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType())) harmRateV = dataVHarmList.stream()
.collect(Collectors.groupingBy(DataHarmDto::getLineId)); .filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
} else {
harmRateV = dataVHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
}
/** /**
* 功能描述:获取influxDB -> data_i -> 2-25次谐波电流 -> 日95%概率值 * 功能描述:获取influxDB -> data_i -> 2-25次谐波电流 -> 日95%概率值
*/ */
Map<String, List<DataIDto>> dataI = dataIList.stream() Map<String, List<DataIDto>> dataI;
.filter(x -> phase.contains(x.getPhasicType())) if (Objects.equals(versionUsed, "liaoning")) {
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType())) dataI = dataIList.stream()
.collect(Collectors.groupingBy(DataIDto::getLineId)); .filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataIDto::getLineId));
} else {
dataI = dataIList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataIDto::getLineId));
}
/** /**
* 功能描述:获取influxDB -> data_inharm_v -> 0.5-15.5次间谐波电压含有率 -> 日95%概率值 * 功能描述:获取influxDB -> data_inharm_v -> 0.5-15.5次间谐波电压含有率 -> 日95%概率值
*/ */
Map<String, List<DataHarmDto>> inHarmV = dataVInHarmList.stream() Map<String, List<DataHarmDto>> inHarmV;
.filter(x -> phase.contains(x.getPhasicType())) if (Objects.equals(versionUsed, "liaoning")) {
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType())) inHarmV = dataVInHarmList.stream()
.collect(Collectors.groupingBy(DataHarmDto::getLineId)); .filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
} else {
inHarmV = dataVInHarmList.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataHarmDto::getLineId));
}
/** /**
* 功能描述:获取influxDB -> data_v -> 电压总谐波畸变率 -> 日95%概率值 * 功能描述:获取influxDB -> data_v -> 电压总谐波畸变率 -> 日95%概率值
*/ */
Map<String, List<DataVDto>> dataVThd = dataVAllTime.stream() Map<String, List<DataVDto>> dataVThd;
.filter(x -> phase.contains(x.getPhasicType())) if (Objects.equals(versionUsed, "liaoning")) {
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType())) dataVThd = dataVAllTime.stream()
.collect(Collectors.groupingBy(DataVDto::getLineId)); .filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataVDto::getLineId));
} else {
dataVThd = dataVAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.CP95.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId));
}
/** /**
* 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值 * 功能描述:获取influxDB -> data_v -> 负序电压不平衡度 -> 最大值 && 日95%概率值
*/ */
Map<String, List<DataVDto>> dataVUnbalance = dataVAllTime.stream() Map<String, List<DataVDto>> dataVUnbalance;
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType())) if (Objects.equals(versionUsed, "liaoning")) {
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) || dataVUnbalance = dataVAllTime.stream()
InfluxDBTableConstant.MAX.equals(x.getValueType())) .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.collect(Collectors.groupingBy(DataVDto::getLineId)); .filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataVDto::getLineId));
} else {
dataVUnbalance = dataVAllTime.stream()
.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%概率值 * 功能描述:获取influxDB -> data_i -> 负序电流 -> 最大值 && 日95%概率值
*/ */
Map<String, List<DataIDto>> dataINeg = dataIList.stream()
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType())) Map<String, List<DataIDto>> dataINeg;
.filter(x -> InfluxDBTableConstant.CP95.equals(x.getValueType()) || if (Objects.equals(versionUsed, "liaoning")) {
InfluxDBTableConstant.MAX.equals(x.getValueType())) dataINeg = dataIList.stream()
.collect(Collectors.groupingBy(DataIDto::getLineId)); .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.AVG.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
.peek(x -> x.setValueType(InfluxDBTableConstant.CP95))
.collect(Collectors.groupingBy(DataIDto::getLineId));
} else {
dataINeg = dataIList.stream()
.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 -> 频率偏差 -> 最大值 && 最小值 * 功能描述:获取influxDB -> data_v -> 频率偏差 -> 最大值 && 最小值
*/ */
Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream() Map<String, List<DataVDto>> dataVFreq = dataVAllTime.stream()
.filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equals(x.getPhasicType())) .filter(x -> InfluxDBTableConstant.PHASE_TYPE_T.equalsIgnoreCase(x.getPhasicType()))
.filter(x -> InfluxDBTableConstant.MIN.equals(x.getValueType()) || .filter(x -> InfluxDBTableConstant.MIN.equalsIgnoreCase(x.getValueType()) ||
InfluxDBTableConstant.MAX.equals(x.getValueType())) InfluxDBTableConstant.MAX.equalsIgnoreCase(x.getValueType()))
.collect(Collectors.groupingBy(DataVDto::getLineId)); .collect(Collectors.groupingBy(DataVDto::getLineId));
/** /**
* 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值 * 功能描述:获取influxDB -> data_v -> 电压偏差 -> 最大值
*/ */
Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream() Map<String, List<DataVDto>> dataVDev = dataVAllTime.stream()
.filter(x -> phase.contains(x.getPhasicType())) .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)); .collect(Collectors.groupingBy(DataVDto::getLineId));
/** /**
* 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较) * 功能描述:获取influxDB -> data_plt -> 长时间闪变 -> 注(取最大值原始算法去掉了，现没有根据最大值比较)
*/ */
Map<String, List<DataPltDto>> dataPlt = dataPltAllTime.stream() Map<String, List<DataPltDto>> dataPlt;
.filter(x -> phase.contains(x.getPhasicType())) if (type == 2) {
.collect(Collectors.groupingBy(DataPltDto::getLineId)); 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) { for (String item : list) {
if (ObjectUtil.isNotNull(overLimitMap.get(item))) { if (ObjectUtil.isNotNull(overLimitMap.get(item))) {
@@ -211,6 +301,8 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
MemorySizeUtil.getNowMemory(); MemorySizeUtil.getNowMemory();
} }
System.gc(); System.gc();
return CompletableFuture.completedFuture(null);
} }
/** /**
@@ -277,6 +369,7 @@ public class IDataLimitRateAsyncImpl implements IDataLimitRateAsync {
result.add(b); result.add(b);
result.add(c); result.add(c);
result.add(t); result.add(t);
return result; return result;
} }

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

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

@@ -27,7 +27,7 @@ import com.njcn.event.api.TransientFeignClient;
import com.njcn.event.file.pojo.dto.WaveDataDTO; import com.njcn.event.file.pojo.dto.WaveDataDTO;
import com.njcn.event.pojo.param.EventCountParam; import com.njcn.event.pojo.param.EventCountParam;
import com.njcn.event.pojo.po.RmpEventDetailPO; import com.njcn.event.pojo.po.RmpEventDetailPO;
import com.njcn.supervision.api.UserLedgerFeignClient; import com.njcn.supervision.api.UserLedgerOldFeignClient;
import com.njcn.supervision.pojo.vo.user.NewUserReportVO; import com.njcn.supervision.pojo.vo.user.NewUserReportVO;
import com.njcn.system.api.DicDataFeignClient; import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.api.DictTreeFeignClient; import com.njcn.system.api.DictTreeFeignClient;
@@ -84,7 +84,7 @@ public class SpecialAnalysisServiceImpl implements ISpecialAnalysisService {
@Resource @Resource
private CommLineClient commLineClient; private CommLineClient commLineClient;
@Resource @Resource
private UserLedgerFeignClient userLedgerFeignClient; private UserLedgerOldFeignClient userLedgerFeignClient;
@Resource @Resource
private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient; private DataLimitRateDetailFeignClient dataLimitRateDetailFeignClient;

76
algorithm/algorithm-boot/src/main/resources/bootstrap-jb.yml Normal file
View File

@@ -0,0 +1,76 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10401
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
compression:
request:
enabled: true
# 配置压缩支持的MIME TYPE
mime-types: text/xml,application/xml,application/json
# 配置压缩数据大小的下限
min-request-size: 1024
response:
# 配置响应GZIP压缩
enabled: true
spring:
application:
name: @artifactId@
cloud:
nacos:
discovery:
server-addr: @nacos.url@
namespace: @nacos.namespace@
config:
username: @nacos.username@
password: @nacos.password@
server-addr: @nacos.url@
namespace: @nacos.namespace@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
liteflow:
rule-source-ext-data-map:
serverAddr: @nacos.url@
dataId: prepare_liteflow
group: DEFAULT_GROUP
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
when-max-wait-time: 3600000
print-banner: false
#项目日志的配置
logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: info
com.alibaba.nacos.client: INFO
#mybatis配置信息
mybatis-plus:
#别名扫描
type-aliases-package: com.njcn.harmonic.pojo
line:
num: 20
mqtt:
client-id: @artifactId@${random.value}

79
algorithm/algorithm-boot/src/main/resources/bootstrap-sjzx.yml Normal file
View File

@@ -0,0 +1,79 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10401
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
compression:
request:
enabled: true
# 配置压缩支持的MIME TYPE
mime-types: text/xml,application/xml,application/json
# 配置压缩数据大小的下限
min-request-size: 1024
response:
# 配置响应GZIP压缩
enabled: true
spring:
application:
name: @artifactId@
cloud:
nacos:
discovery:
ip: @service.server.url@
server-addr: @nacos.url@
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
config:
server-addr: @nacos.url@
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-Id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
liteflow:
rule-source-ext-data-map:
serverAddr: @nacos.url@
dataId: prepare_liteflow
group: DEFAULT_GROUP
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
when-max-wait-time: 3600000
print-banner: false
#项目日志的配置
logging:
#config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: info
com.alibaba.nacos.client: INFO
#mybatis配置信息
mybatis-plus:
#别名扫描
type-aliases-package: com.njcn.harmonic.pojo
line:
num: 20
#mqtt:
# client-id: @artifactId@${random.value}

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

@@ -1,72 +1,3 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10401
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
compression:
request:
enabled: true
# 配置压缩支持的MIME TYPE
mime-types: text/xml,application/xml,application/json
# 配置压缩数据大小的下限
min-request-size: 1024
response:
# 配置响应GZIP压缩
enabled: true
spring: spring:
application: profiles:
name: @artifactId@ active: @spring.profiles.active@
cloud:
nacos:
discovery:
ip: @service.server.url@
server-addr: @nacos.url@
namespace: @nacos.namespace@
config:
server-addr: @nacos.url@
namespace: @nacos.namespace@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-Id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
liteflow:
rule-source-ext-data-map:
serverAddr: @nacos.url@
dataId: prepare_liteflow
group: DEFAULT_GROUP
namespace: @nacos.namespace@
when-max-wait-time: 600000
print-banner: false
#项目日志的配置
logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: error
#mybatis配置信息
mybatis-plus:
#别名扫描
type-aliases-package: com.njcn.harmonic.pojo
line:
num: 10
mqtt:
client-id: @artifactId@${random.value}

10
analysis/stat/stat-boot/src/main/resources/bootstrap.yml
View File

@@ -22,16 +22,20 @@ spring:
ip: @service.server.url@ ip: @service.server.url@
server-addr: @nacos.url@ server-addr: @nacos.url@
namespace: @nacos.namespace@ namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
config: config:
server-addr: @nacos.url@ server-addr: @nacos.url@
namespace: @nacos.namespace@ namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
file-extension: yaml file-extension: yaml
shared-configs: shared-configs:
- data-id: share-config.yaml - data-id: share-config.yaml
refresh: true refresh: true
- data-Id: share-config-datasource-db.yaml - data-id: share-config-datasource-db.yaml
refresh: true refresh: true
- data-Id: data-platform.yaml - data-id: data-platform.yaml
refresh: true refresh: true
main: main:
allow-bean-definition-overriding: true allow-bean-definition-overriding: true
@@ -39,7 +43,7 @@ spring:
#项目日志的配置 #项目日志的配置
logging: logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level: level:
root: info root: info

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") @PostMapping("/getGroupByTimeDataV")
HttpResult<List<DataVDto>> getGroupByTimeDataV(@RequestBody LineCountEvaluateParam lineParam); 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()); log.error("{}异常，降级处理，异常为：{}","DataV按时间分组获取原始数据",cause.toString());
throw new BusinessException(finalExceptionEnum); 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") @Column(name = "p_50")
private Double p50; 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) { public static List<DataHarmpowerP> relationToInfluxDB(DataHarmpowerPDTO dataHarmpowerP) {
if (dataHarmpowerP == null) { if (dataHarmpowerP == null) {
return 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") @Column(name = "q_50")
private Double q50; 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) { public static List<DataHarmpowerQ> relationToInfluxDB(DataHarmpowerQDTO dataHarmpowerQ) {
if (dataHarmpowerQ == null) { if (dataHarmpowerQ == null) {
return 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") @Column(name = "s_50")
private Double s50; 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) { public static List<DataHarmpowerS> relationToInfluxDB(DataHarmpowerSDTO dataHarmpowerS) {
if (dataHarmpowerS == null) { if (dataHarmpowerS == null) {
return 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 String description;
private Integer type; 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); 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> getDataV(LineCountEvaluateParam lineParam);
List<DataVDto> getGroupByTimeDataV(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.setTime(LocalDateTimeUtil.format(dataVDTO.getTimeid(), DatePattern.NORM_DATETIME_PATTERN));
pqsCommunicateDto.setDevId(temp); pqsCommunicateDto.setDevId(temp);
pqsCommunicateDto.setType(1); pqsCommunicateDto.setType(1);
pqsCommunicateDto.setFlag(1);
iPqsCommunicate.insertion(pqsCommunicateDto); 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.dao.relation.mapper.RStatDataHarmPowerPRelationMapper;
import com.njcn.dataProcess.dto.DataHarmpowerPDTO; import com.njcn.dataProcess.dto.DataHarmpowerPDTO;
import com.njcn.dataProcess.param.LineCountEvaluateParam; 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.po.influx.DataHarmpowerP;
import com.njcn.dataProcess.pojo.dto.CommonMinuteDto; import com.njcn.dataProcess.pojo.dto.CommonMinuteDto;
import com.njcn.dataProcess.pojo.dto.DataHarmPowerPDto; import com.njcn.dataProcess.pojo.dto.DataHarmPowerPDto;
@@ -29,6 +27,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
@@ -128,68 +127,12 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{ valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType(); CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(valueType); value.setValueType(valueType);
//规定好集合指标参数 List<List<Double>> lists;
List<Double> data1 = valueTypeList.stream().map(DataHarmpowerP::getDf).collect(Collectors.toList()); if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
List<Double> data2 = valueTypeList.stream().map(DataHarmpowerP::getPf).collect(Collectors.toList()); lists = extractDataLists(valueTypeList, "Tot");
List<Double> data3 = valueTypeList.stream().map(DataHarmpowerP::getP).collect(Collectors.toList()); } else {
lists = extractDataLists(valueTypeList, "");
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);
value.setValueList(lists); value.setValueList(lists);
valueTypes.add(value); valueTypes.add(value);
}); });
@@ -203,6 +146,43 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
return result; 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 @Override
public void addList(List<DataHarmPowerPDto> list) { public void addList(List<DataHarmPowerPDto> list) {
@@ -298,20 +278,31 @@ public class InfluxdbDataHarmpowerPImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId(); String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId); CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerP.class); InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerP.class);
//A、B、C谐波有功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmP_", "p_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1)); influxQueryWrapper.samePrefixAndSuffix("Pq_HarmP_", "p_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
influxQueryWrapper.eq(DataHarmpowerP::getLineId, lineId) //三相总 谐波有功功率有效值
.eq(DataHarmpowerP::getProcess,Integer.toString(devsMap.get(devId).getProcess())) 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::getLineId)
.select(DataHarmpowerP::getPhasicType) .select(DataHarmpowerP::getPhasicType)
.select(DataHarmpowerP::getValueType) .select(DataHarmpowerP::getValueType)
//位移功率因数 //位移功率因数
.select("Pq_DF","df") .select("Pq_DF", "df")
//视在功率因素 //视在功率因素
.select("Pq_PF","pf") .select("Pq_PF", "pf")
//总功功率 //A、B、C三相有功功率
.select("Pq_P","p") .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) .between(DataHarmpowerP::getTime, startTime, endTime)
.eq(DataHarmpowerP::getQualityFlag, "0"); .eq(DataHarmpowerP::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) { 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 org.springframework.stereotype.Service;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
@@ -124,66 +125,13 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{ valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType(); CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(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<List<Double>> lists;
List<Double> data13 = valueTypeList.stream().map(DataHarmpowerQ::getQ2).collect(Collectors.toList()); if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
List<Double> data14 = valueTypeList.stream().map(DataHarmpowerQ::getQ3).collect(Collectors.toList()); lists = extractDataLists(valueTypeList, "Tot");
List<Double> data15 = valueTypeList.stream().map(DataHarmpowerQ::getQ4).collect(Collectors.toList()); } else {
List<Double> data16 = valueTypeList.stream().map(DataHarmpowerQ::getQ5).collect(Collectors.toList()); lists = extractDataLists(valueTypeList, "");
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);
value.setValueList(lists); value.setValueList(lists);
valueTypes.add(value); valueTypes.add(value);
}); });
@@ -197,6 +145,39 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
return result; 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 @Override
public void addList(List<DataHarmPowerQDto> list) { public void addList(List<DataHarmPowerQDto> list) {
@@ -290,16 +271,23 @@ public class InfluxdbDataHarmpowerQImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId(); String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId); CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerQ.class); InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerQ.class);
//A、B、C谐波无功功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmQ_", "q_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1)); 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) influxQueryWrapper.eq(DataHarmpowerQ::getLineId, lineId)
.eq(DataHarmpowerQ::getProcess,Integer.toString(devsMap.get(devId).getProcess())) .eq(DataHarmpowerQ::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerQ::getLineId) .select(DataHarmpowerQ::getLineId)
.select(DataHarmpowerQ::getPhasicType) .select(DataHarmpowerQ::getPhasicType)
.select(DataHarmpowerQ::getValueType) .select(DataHarmpowerQ::getValueType)
//总功功率 //A、B、C三相无功功率
.select("Pq_Q","q") .select("Pq_Q", "q")
//基波有功功率 //基波无功功率
.select("Pq_FundQ","q_1") .select("Pq_FundQ","q_1")
//总无功功率
.select("Pq_TotQ","totQ")
//基波无功功率(T)
.select("Pq_TotHarmQ","totQ1")
.between(DataHarmpowerQ::getTime, startTime, endTime) .between(DataHarmpowerQ::getTime, startTime, endTime)
.eq(DataHarmpowerQ::getQualityFlag, "0"); .eq(DataHarmpowerQ::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) { 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 org.springframework.stereotype.Service;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.ZoneId; import java.time.ZoneId;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
@@ -124,66 +125,12 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
valueTypeMap.forEach((valueType,valueTypeList)->{ valueTypeMap.forEach((valueType,valueTypeList)->{
CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType(); CommonMinuteDto.ValueType value = new CommonMinuteDto.ValueType();
value.setValueType(valueType); value.setValueType(valueType);
//规定好集合指标参数 List<List<Double>> lists;
List<Double> data1 = valueTypeList.stream().map(DataHarmpowerS::getS).collect(Collectors.toList()); if (Objects.equals(phasicType, "T") && Objects.equals(lineParam.getType(), 2)) {
lists = extractDataLists(valueTypeList, "Tot");
List<Double> data12 = valueTypeList.stream().map(DataHarmpowerS::getS1).collect(Collectors.toList()); } else {
List<Double> data13 = valueTypeList.stream().map(DataHarmpowerS::getS2).collect(Collectors.toList()); lists = extractDataLists(valueTypeList, "");
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);
value.setValueList(lists); value.setValueList(lists);
valueTypes.add(value); valueTypes.add(value);
}); });
@@ -197,6 +144,39 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
return result; 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 @Override
public void addList(List<DataHarmPowerSDto> list) { public void addList(List<DataHarmPowerSDto> list) {
@@ -290,16 +270,23 @@ public class InfluxdbDataHarmpowerSImpl extends MppServiceImpl<RStatDataHarmPowe
String devId = lineMap.get(lineId).getDeviceId(); String devId = lineMap.get(lineId).getDeviceId();
CsLinePO po = lineMap.get(lineId); CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerS.class); InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmpowerS.class);
//A、B、C谐波视在功率有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmS_", "s_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1)); 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) influxQueryWrapper.eq(DataHarmpowerS::getLineId, lineId)
.eq(DataHarmpowerS::getProcess,Integer.toString(devsMap.get(devId).getProcess())) .eq(DataHarmpowerS::getProcess,Integer.toString(devsMap.get(devId).getProcess()))
.select(DataHarmpowerS::getLineId) .select(DataHarmpowerS::getLineId)
.select(DataHarmpowerS::getPhasicType) .select(DataHarmpowerS::getPhasicType)
.select(DataHarmpowerS::getValueType) .select(DataHarmpowerS::getValueType)
//总功功率 //A、B、C视在功率
.select("Pq_S","s") .select("Pq_S","s")
//基波有功功率 //基波视在功率
.select("Pq_FundS","s_1") .select("Pq_FundS","s_1")
//总视在功率
.select("Pq_TotS","totS")
//基波视在功率（T）
.select("Pq_TotHarmS","totS1")
.between(DataHarmpowerS::getTime, startTime, endTime) .between(DataHarmpowerS::getTime, startTime, endTime)
.eq(DataHarmpowerS::getQualityFlag, "0"); .eq(DataHarmpowerS::getQualityFlag, "0");
if (Objects.isNull(po.getLineNo())) { 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); List<DataV> list = dataVMapper.selectByQueryWrapper(influxQueryWrapper);
if (CollUtil.isNotEmpty(list)) { if (CollUtil.isNotEmpty(list)) {
Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime(); Map<String, List<String>> abnormalTime = lineParam.getAbnormalTime();
if(CollUtil.isNotEmpty(abnormalTime)){ if (CollUtil.isNotEmpty(abnormalTime)) {
if (abnormalTime.containsKey(lineParam.getLineId().get(0))) { if (abnormalTime.containsKey(lineParam.getLineId().get(0))) {
List<String> timeList = abnormalTime.get(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())); result.addAll(list.stream().filter(item -> !timeList.contains(DATE_TIME_FORMATTER.format(item.getTime()))).collect(Collectors.toList()));
} }
} }
}else{ } else {
result.addAll(list); result.addAll(list);
} }
} }
@@ -390,8 +390,8 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
.groupBy(DataV::getLineId) .groupBy(DataV::getLineId)
.between(DataV::getTime, startTime, endTime); .between(DataV::getTime, startTime, endTime);
List<MeasurementCountDTO> measurementCountDTOList = dataVMapper.getMeasurementCount(influxQueryWrapper); List<MeasurementCountDTO> measurementCountDTOList = dataVMapper.getMeasurementCount(influxQueryWrapper);
if(CollUtil.isNotEmpty(measurementCountDTOList)){ if (CollUtil.isNotEmpty(measurementCountDTOList)) {
for(MeasurementCountDTO m : measurementCountDTOList){ for (MeasurementCountDTO m : measurementCountDTOList) {
MeasurementCountDTO.MeasurementCountResultDTO p = new MeasurementCountDTO.MeasurementCountResultDTO(); MeasurementCountDTO.MeasurementCountResultDTO p = new MeasurementCountDTO.MeasurementCountResultDTO();
p.setFreq(m.getFreq()); p.setFreq(m.getFreq());
p.setLineId(m.getLineId()); p.setLineId(m.getLineId());
@@ -425,6 +425,27 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
return result; 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分钟数据 * 按监测点集合、时间条件获取dataV分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理 * timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理
@@ -483,53 +504,53 @@ public class InfluxdbDataVImpl extends MppServiceImpl<RStatDataVRelationMapper,
CsLinePO po = lineMap.get(lineId); CsLinePO po = lineMap.get(lineId);
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class); InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataV.class);
influxQueryWrapper.eq(DataV::getLineId, lineId) 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::getLineId)
.select(DataV::getPhasicType) .select(DataV::getPhasicType)
.select(DataV::getValueType) .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()) .between(DataV::getTime, lineParam.getStartTime(), lineParam.getEndTime())
.eq(DataV::getQualityFlag, "0"); .eq(DataV::getQualityFlag, "0");
if (CollUtil.isNotEmpty(lineParam.getPhasicType())) { if (CollUtil.isNotEmpty(lineParam.getPhasicType())) {
influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType()); influxQueryWrapper.regular(DataV::getPhasicType, lineParam.getPhasicType());
} }
if (Objects.isNull(po.getLineNo())) { if (Objects.isNull(po.getLineNo())) {
influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getClDid())); influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getClDid()));
} else { } else {
influxQueryWrapper.eq(DataV::getCldid,Integer.toString(po.getLineNo())); influxQueryWrapper.eq(DataV::getCldid, Integer.toString(po.getLineNo()));
} }
//判断接线方式 (0-星型 1-角型 2-V型) 星型是相电压 角型或者v型是线电压 //判断接线方式 (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次 相电压谐波有效值 //2-50次 相电压谐波有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1)); influxQueryWrapper.samePrefixAndSuffix("Pq_HarmUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
} else { } 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次 线电压谐波有效值 //2-50次 线电压谐波有效值
influxQueryWrapper.samePrefixAndSuffix("Pq_HarmLUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1)); influxQueryWrapper.samePrefixAndSuffix("Pq_HarmLUV_", "v_", HarmonicTimesUtil.harmonicTimesList(2, 50, 1));
} }

4
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数据 //更新mysql数据
DevComFlagDTO devComFlagDTO = new DevComFlagDTO(); DevComFlagDTO devComFlagDTO = new DevComFlagDTO();
devComFlagDTO.setId(pqsCommunicateDto.getDevId()); devComFlagDTO.setId(pqsCommunicateDto.getDevId());
devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER)); if(Objects.equals(pqsCommunicateDto.getFlag(),1)){
devComFlagDTO.setDate(LocalDateTime.parse(pqsCommunicateDto.getTime(), DATE_TIME_FORMATTER));
}
devComFlagDTO.setStatus(pqsCommunicateDto.getType()); devComFlagDTO.setStatus(pqsCommunicateDto.getType());
deviceFeignClient.updateDevComFlag(devComFlagDTO); 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(); return Collections.emptyList();
} }
@Override
public Boolean excludeZeroData(LineCountEvaluateParam lineParam) {
return null;
}
private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) { private List<DataVDto> quality(List<DataVDto> list, LineCountEvaluateParam lineParam) {
List<DataVDto> result = new ArrayList<>(); List<DataVDto> result = new ArrayList<>();
Map<String, List<DataVDto>> lineMap = list.stream().collect(Collectors.groupingBy(DataVDto::getLineId)); Map<String, List<DataVDto>> lineMap = list.stream().collect(Collectors.groupingBy(DataVDto::getLineId));

78
data-processing/data-processing-boot/src/main/resources/bootstrap-jb.yml Normal file
View File

@@ -0,0 +1,78 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10405
compression:
enabled: true
mime-types: application/json,application/xml,text/html,text/xml,text/plain
min-response-size: 1024
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
spring:
application:
name: @artifactId@
#nacos注册中心以及配置中心的指定
cloud:
nacos:
discovery:
server-addr: @nacos.url@
namespace: @nacos.namespace@
config:
username: @nacos.username@
password: @nacos.password@
server-addr: @nacos.url@
namespace: @nacos.namespace@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
servlet:
multipart:
max-file-size: 100MB
max-request-size: 100MB
jackson:
time-zone: GMT+8
#项目日志的配置
logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: error
mqtt:
client-id: @artifactId@${random.value}
data:
source:
query: Influxdb
# insert: Influxdb
insert: Relation
#mybatis配置信息
mybatis-plus:
configuration:
#配置sql日志输出
log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
# type-aliases-package: com.njcn.harmonic.pojo
# type-handlers-package: com.njcn.db.handler
# global-config:
# db-config:
# date-format: yyyy-MM-dd HH:mm:ss
# global-config:
# enable-sql-runner: true

81
data-processing/data-processing-boot/src/main/resources/bootstrap-sjzx.yml Normal file
View File

@@ -0,0 +1,81 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10405
compression:
enabled: true
mime-types: application/json,application/xml,text/html,text/xml,text/plain
min-response-size: 1024
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
spring:
application:
name: @artifactId@
#nacos注册中心以及配置中心的指定
cloud:
nacos:
discovery:
ip: @service.server.url@
server-addr: @nacos.url@
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
config:
server-addr: @nacos.url@
namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
servlet:
multipart:
max-file-size: 100MB
max-request-size: 100MB
jackson:
time-zone: GMT+8
#项目日志的配置
logging:
#config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: error
#mqtt:
# client-id: @artifactId@${random.value}
data:
source:
query: Influxdb
# insert: Influxdb
insert: Relation
#mybatis配置信息
mybatis-plus:
configuration:
#配置sql日志输出
log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
# type-aliases-package: com.njcn.harmonic.pojo
# type-handlers-package: com.njcn.db.handler
# global-config:
# db-config:
# date-format: yyyy-MM-dd HH:mm:ss
# global-config:
# enable-sql-runner: true

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

@@ -1,77 +1,3 @@
#当前服务的基本信息
microservice:
ename: @artifactId@
name: '@name@'
version: @version@
sentinel:
url: @sentinel.url@
gateway:
url: @gateway.url@
server:
port: 10405
compression:
enabled: true
mime-types: application/json,application/xml,text/html,text/xml,text/plain
min-response-size: 1024
#feign接口开启服务熔断降级处理
feign:
sentinel:
enabled: true
spring: spring:
application: profiles:
name: @artifactId@ active: @spring.profiles.active@
#nacos注册中心以及配置中心的指定
cloud:
nacos:
discovery:
ip: @service.server.url@
server-addr: @nacos.url@
namespace: @nacos.namespace@
config:
server-addr: @nacos.url@
namespace: @nacos.namespace@
file-extension: yaml
shared-configs:
- data-id: share-config.yaml
refresh: true
- data-Id: share-config-datasource-db.yaml
refresh: true
main:
allow-bean-definition-overriding: true
servlet:
multipart:
max-file-size: 100MB
max-request-size: 100MB
jackson:
time-zone: GMT+8
#项目日志的配置
logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level:
root: info
mqtt:
client-id: @artifactId@${random.value}
data:
source:
query: Influxdb
# insert: Influxdb
insert: Relation
#mybatis配置信息
mybatis-plus:
configuration:
#配置sql日志输出
log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
# type-aliases-package: com.njcn.harmonic.pojo
# type-handlers-package: com.njcn.db.handler
# global-config:
# db-config:
# date-format: yyyy-MM-dd HH:mm:ss
# global-config:
# enable-sql-runner: true

12
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.api.RocketMqLogFeignClient;
import com.njcn.system.pojo.po.RocketmqMsgErrorLog; import com.njcn.system.pojo.po.RocketmqMsgErrorLog;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@@ -82,7 +83,16 @@ public class DeviceRunFlagDataConsumer extends EnhanceConsumerMessageHandler<Dev
@Override @Override
protected void handleMessage(DevComFlagDTO message) { protected void handleMessage(DevComFlagDTO message) {
messAnalysisFeignClient.handleDevRunflag(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()+"");
}
} }

10
message/message-boot/src/main/resources/bootstrap.yml
View File

@@ -22,16 +22,20 @@ spring:
ip: @service.server.url@ ip: @service.server.url@
server-addr: @nacos.url@ server-addr: @nacos.url@
namespace: @nacos.namespace@ namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
config: config:
server-addr: @nacos.url@ server-addr: @nacos.url@
namespace: @nacos.namespace@ namespace: @nacos.namespace@
username: @nacos.username@
password: @nacos.password@
file-extension: yaml file-extension: yaml
shared-configs: shared-configs:
- data-id: share-config.yaml - data-id: share-config.yaml
refresh: true refresh: true
- data-Id: share-config-datasource-db.yaml - data-id: share-config-datasource-db.yaml
refresh: true refresh: true
- data-Id: data-platform.yaml - data-id: data-platform.yaml
refresh: true refresh: true
main: main:
allow-bean-definition-overriding: true allow-bean-definition-overriding: true
@@ -39,7 +43,7 @@ spring:
#项目日志的配置 #项目日志的配置
logging: logging:
config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml #config: http://@nacos.url@/nacos/v1/cs/configs?tenant=@nacos.namespace@&group=DEFAULT_GROUP&dataId=logback.xml
level: level:
root: info root: info

54
pom.xml
View File

@@ -15,7 +15,7 @@
<module>algorithm</module> <module>algorithm</module>
<module>log</module> <module>log</module>
<module>data-processing</module> <module>data-processing</module>
<module>migration-influxdb</module> <!--<module>migration-influxdb</module>-->
</modules> </modules>
<distributionManagement> <distributionManagement>
@@ -32,47 +32,13 @@
</distributionManagement> </distributionManagement>
<properties> <properties>
<spring.profiles.active>sjzx</spring.profiles.active>
<!--内网--> <!--内网-->
<!-- <middle.server.url>192.168.1.22</middle.server.url>--> <middle.server.url>192.168.1.65</middle.server.url>
<!-- <service.server.url>192.168.1.121</service.server.url>--> <service.server.url>192.168.1.65</service.server.url>
<!-- <docker.server.url>192.168.1.22</docker.server.url>--> <docker.server.url>192.168.1.22</docker.server.url>
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>--> <nacos.url>${middle.server.url}:18848</nacos.url>
<!-- <nacos.namespace>07ec981d-1377-4920-8760-b5a98ff7e668</nacos.namespace>--> <nacos.namespace></nacos.namespace>
<!-- <middle.server.url>192.168.21.129</middle.server.url>-->
<!-- <service.server.url>192.168.21.1</service.server.url>-->
<!-- <docker.server.url>192.168.1.22</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
<!-- <middle.server.url>127.0.0.1</middle.server.url>-->
<!-- &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
<!-- <service.server.url>127.0.0.1</service.server.url>-->
<!-- &lt;!&ndash;docker仓库地址&ndash;&gt;-->
<!-- <docker.server.url>127.0.0.1</docker.server.url>-->
<!-- &lt;!&ndash;nacos的ip:port&ndash;&gt;-->
<!-- <nacos.url>${middle.server.url}:8848</nacos.url>-->
<!-- <nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>-->
<!-- <middle.server.url>24.43.102.139</middle.server.url>-->
<!-- <service.server.url>24.43.102.139</service.server.url>-->
<!-- <docker.server.url>24.43.102.139</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace></nacos.namespace>-->
<middle.server.url>127.0.0.1</middle.server.url>
<service.server.url>127.0.0.1</service.server.url>
<docker.server.url>127.0.0.1</docker.server.url>
<nacos.url>${middle.server.url}:8848</nacos.url>
<nacos.namespace>37a0e99f-eb96-4d2f-9b47-75bffa765827</nacos.namespace>
<!-- <middle.server.url>192.168.21.129</middle.server.url>-->
<!-- &lt;!&ndash;微服务模块发布地址&ndash;&gt;-->
<!-- <service.server.url>192.168.21.1</service.server.url>-->
<!-- &lt;!&ndash;docker仓库地址&ndash;&gt;-->
<!-- <docker.server.url>192.168.1.22</docker.server.url>-->
<!-- <nacos.url>${middle.server.url}:18848</nacos.url>-->
<!-- <nacos.namespace>0c19477c-444e-413c-9782-e6ebc2eedac1</nacos.namespace>-->
<!--sentinel:port--> <!--sentinel:port-->
<sentinel.url>${middle.server.url}:8080</sentinel.url> <sentinel.url>${middle.server.url}:8080</sentinel.url>
<!--网关地址，主要用于配置swagger中认证token--> <!--网关地址，主要用于配置swagger中认证token-->
@@ -87,6 +53,12 @@
<micro.sdk>common-microservice</micro.sdk> <micro.sdk>common-microservice</micro.sdk>
<!-- <micro.sdk>common-huawei</micro.sdk>--> <!-- <micro.sdk>common-huawei</micro.sdk>-->
<docker.operate>site</docker.operate> <docker.operate>site</docker.operate>
<!--nacos开启鉴权后的用户名-->
<nacos.username>nacos</nacos.username>
<!--nacos的ip:port-->
<nacos.password>njcnpqs</nacos.password>
<java.version>1.8</java.version> <java.version>1.8</java.version>
<maven.compiler.source>8</maven.compiler.source> <maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target> <maven.compiler.target>8</maven.compiler.target>