Microservice/govern
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

提交

Browse Source
This commit is contained in:
huangzj
2023-08-21 10:54:16 +08:00
parent 9e56e47f1b
commit 62058c41ca
6 changed files with 86 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/handler/MqttMessageHandler.java
View File

@@ -15,6 +15,7 @@ import com.njcn.system.api.CsStatisticalSetFeignClient;
import com.njcn.system.pojo.po.EleEpdPqd;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
@@ -42,7 +43,7 @@ public class MqttMessageHandler {
private final ILineTargetService lineTargetService;
private final CsStatisticalSetFeignClient csStatisticalSetFeignClient;
private final StableDataService stableDataService;
private final DecimalFormat df = new DecimalFormat("#0.00");
private final DecimalFormat df = new DecimalFormat("#0.000");
/**
* 实时数据应答
*/
@@ -64,12 +65,27 @@ public class MqttMessageHandler {
//1.查询拓扑图配置的指标:拓扑图扑图配置：7677f94c749dedaff30f911949cbd724
List<EleEpdPqd> data = csStatisticalSetFeignClient.queryStatisticalSelect("7677f94c749dedaff30f911949cbd724").getData();
data.forEach(temp->{
if(Objects.nonNull(temp.getHarmStart())&&Objects.nonNull(temp.getHarmEnd())){
for (int i = temp.getHarmStart(); i < temp.getHarmEnd()+1; i++) {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);
commonStatisticalQueryParam.setStatisticalId(temp.getId());
commonStatisticalQueryParam.setValueType("avg");
commonStatisticalQueryParam.setFrequency(i+"");
List<ThdDataVO> thdDataVOS = stableDataService.queryFisrtCommonStatistical(commonStatisticalQueryParam);
tempList.addAll(thdDataVOS);
}
}else {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);
commonStatisticalQueryParam.setStatisticalId(temp.getId());
commonStatisticalQueryParam.setValueType("avg");
List<ThdDataVO> thdDataVOS = stableDataService.queryFisrtCommonStatistical(commonStatisticalQueryParam);
tempList.addAll(thdDataVOS);
}
});
//过滤M相
List<ThdDataVO> m = tempList.stream().filter(temp -> Objects.equals(temp.getPhase(), "M")).collect(Collectors.toList());
@@ -82,7 +98,7 @@ public class MqttMessageHandler {
});
});
//过滤谐波电流，谐波电压畸变率求平均值
List<ThdDataVO> thdI = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "ThdPhI(%)")).collect(Collectors.toList());
List<ThdDataVO> thdI = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "Pq_ThdU(%)")).collect(Collectors.toList());
Map<String, List<ThdDataVO>> collect = thdI.stream().collect(Collectors.groupingBy(ThdDataVO::getLineId));
collect.forEach((k,v)->{
if(!CollectionUtil.isEmpty(v)){
@@ -96,7 +112,7 @@ public class MqttMessageHandler {
}
});
List<ThdDataVO> thdV = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "ThdPhV(%)")).collect(Collectors.toList());
List<ThdDataVO> thdV = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "Pq_ThdI(%)")).collect(Collectors.toList());
Map<String, List<ThdDataVO>> collect1 = thdV.stream().collect(Collectors.groupingBy(ThdDataVO::getLineId));
collect1.forEach((k,v)->{
if(!CollectionUtil.isEmpty(v)){
@@ -109,6 +125,32 @@ public class MqttMessageHandler {
}
});
List<ThdDataVO> apfThdI = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "Apf_ThdA_Load(%)")).collect(Collectors.toList());
Map<String, List<ThdDataVO>> collect3 = apfThdI.stream().collect(Collectors.groupingBy(ThdDataVO::getLineId));
collect3.forEach((k,v)->{
if(!CollectionUtil.isEmpty(v)){
double asDouble = v.stream().mapToDouble(ThdDataVO::getStatisticalData).average().getAsDouble();
ThdDataVO thdDataVO = new ThdDataVO();
BeanUtils.copyProperties(v.get(0),thdDataVO);
thdDataVO.setStatisticalData(Double.valueOf(df.format(asDouble)));
thdDataVO.setPhase("avg");
result.add(thdDataVO);
}
});
List<ThdDataVO> apfThdISys = tempList.stream().filter(temp -> Objects.equals(temp.getStatisticalName(), "Apf_ThdA_Sys(%)")).collect(Collectors.toList());
Map<String, List<ThdDataVO>> collect4 = apfThdISys.stream().collect(Collectors.groupingBy(ThdDataVO::getLineId));
collect4.forEach((k,v)->{
if(!CollectionUtil.isEmpty(v)){
double asDouble = v.stream().mapToDouble(ThdDataVO::getStatisticalData).average().getAsDouble();
ThdDataVO thdDataVO = new ThdDataVO();
BeanUtils.copyProperties(v.get(0),thdDataVO);
thdDataVO.setStatisticalData(Double.valueOf(df.format(asDouble)));
thdDataVO.setPhase("avg");
result.add(thdDataVO);
}
});
List<ThdDataVO> notM = tempList.stream().filter(temp -> !Objects.equals(temp.getPhase(), "M")).collect(Collectors.toList());