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-09-14 13:57:10 +08:00
parent 2cf664276a
commit 36c5de878a
5 changed files with 215 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/param/FrequencyStatisticalQueryParam.java Normal file
View File

@@ -0,0 +1,41 @@
package com.njcn.csharmonic.param;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import javax.validation.constraints.NotBlank;
import java.util.List;
/**
* Description:
* Date: 2023/6/14 15:09【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Data
public class FrequencyStatisticalQueryParam {
@ApiModelProperty(value = "监测点集")
private List<String> lineList;
@ApiModelProperty(value = "设备id")
private String DevId;
@ApiModelProperty(value = "指标id")
@NotBlank(message="指标id不能为空")
private String statisticalId;
@ApiModelProperty(value = "取值类型（Max，Min，cp95，avg）")
private String valueType;
@ApiModelProperty(value = "频次开始")
private Integer frequencyStart;
@ApiModelProperty(value = "频次结束")
private Integer frequencyEnd;
@ApiModelProperty(value = "开始时间")
private String startTime;
@ApiModelProperty(value = "结束时间")
private String endTime;
}

12
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/controller/StableDataController.java
View File

@@ -6,6 +6,7 @@ import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.csharmonic.param.CommonStatisticalQueryParam;
import com.njcn.csharmonic.param.FrequencyStatisticalQueryParam;
import com.njcn.csharmonic.param.ThdDataQueryParm;
import com.njcn.csharmonic.pojo.vo.ThdDataVO;
import com.njcn.csharmonic.service.StableDataService;
@@ -112,4 +113,15 @@ public class StableDataController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, result, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@PostMapping("/QuerySqlData")
@ApiOperation("查询(2-50)的指标")
@ApiImplicitParam(name = "frequencyStatisticalQueryParam", value = "查询参数", required = true)
public HttpResult<List<ThdDataVO>> QuerySqlData(@RequestBody FrequencyStatisticalQueryParam frequencyStatisticalQueryParam) {
String methodDescribe = getMethodDescribe("queryFisrtCommonStatistical");
List<ThdDataVO> thdDataVOList = stableDataService.QuerySqlData(frequencyStatisticalQueryParam);
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, thdDataVOList, methodDescribe);
}
}

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

@@ -7,6 +7,7 @@ import com.github.tocrhz.mqtt.annotation.NamedValue;
import com.github.tocrhz.mqtt.annotation.Payload;
import com.github.tocrhz.mqtt.publisher.MqttPublisher;
import com.njcn.csharmonic.param.CommonStatisticalQueryParam;
import com.njcn.csharmonic.param.FrequencyStatisticalQueryParam;
import com.njcn.csharmonic.pojo.vo.CsRtDataVO;
import com.njcn.csharmonic.pojo.vo.ThdDataVO;
import com.njcn.csharmonic.service.ILineTargetService;
@@ -71,57 +72,65 @@ public class MqttMessageHandler {
Gson gson = new Gson();
publisher.send("/zl/TemperData/"+devId,gson.toJson(statisticalDataDTOS),1,false);
}
/**
* 实时数据应答
/*
* 谐波数据
*/
// @MqttSubscribe(value = "/zl/askDevData/{devId}",qos = 1)
// public void responseTopoData(String topic, @NamedValue("devId") String devId, MqttMessage message, @Payload String payload) {
//
// }
/**
* 实时数据应答
*/
@MqttSubscribe(value = "/zl/askDevData/{devId}",qos = 1)
public void responseTopoData(String topic, @NamedValue("devId") String devId, MqttMessage message, @Payload String payload) {
String topoDataJson =redisUtil.getStringByKey (devId);
if(StringUtils.isEmpty(topoDataJson)){
List<ThdDataVO> result = new ArrayList<>();
List<ThdDataVO> tempList = new ArrayList<>();
ExecutorService executorService = new ThreadPoolExecutor(10, 100,
1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100, true),
Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
List<Future<List<ThdDataVO>>> resultList = new ArrayList< Future<List<ThdDataVO>>>();
// ExecutorService executorService = new ThreadPoolExecutor(10, Integer.MAX_VALUE,
// 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(),
// Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
// List<Future<List<ThdDataVO>>> resultList = new ArrayList< Future<List<ThdDataVO>>>();
//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+"");
Future<List<ThdDataVO>> listFuture= executorService.submit(new TaskWithResult(commonStatisticalQueryParam));
resultList.add(listFuture);
}
FrequencyStatisticalQueryParam frequencyStatisticalQueryParam = new FrequencyStatisticalQueryParam();
frequencyStatisticalQueryParam.setDevId(devId);
frequencyStatisticalQueryParam.setStatisticalId(temp.getId());
frequencyStatisticalQueryParam.setValueType("avg");
frequencyStatisticalQueryParam.setFrequencyStart(temp.getHarmStart());
frequencyStatisticalQueryParam.setFrequencyEnd(temp.getHarmEnd());
List<ThdDataVO> thdDataVOList = stableDataService.QuerySqlData(frequencyStatisticalQueryParam);
tempList.addAll(thdDataVOList);
}else {
CommonStatisticalQueryParam commonStatisticalQueryParam = new CommonStatisticalQueryParam();
commonStatisticalQueryParam.setDevId(devId);
commonStatisticalQueryParam.setStatisticalId(temp.getId());
commonStatisticalQueryParam.setValueType("avg");
Future<List<ThdDataVO>> listFuture= executorService.submit(new TaskWithResult(commonStatisticalQueryParam));
resultList.add(listFuture);
List<ThdDataVO> listFuture= stableDataService.queryFisrtCommonStatistical(commonStatisticalQueryParam);
tempList.addAll(listFuture);
}
});
executorService.shutdown();
resultList.forEach(temp->{
try {
tempList.addAll(temp.get());
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
}
});
// executorService.shutdown();
//
// resultList.forEach(temp->{
// try {
// tempList.addAll(temp.get());
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// } catch (ExecutionException e) {
// throw new RuntimeException(e);
// }
// });
//过滤M相
List<ThdDataVO> m = tempList.stream().filter(temp -> Objects.equals(temp.getPhase(), "M")).collect(Collectors.toList());
m.stream().forEach(temp->{
@@ -147,19 +156,19 @@ public class MqttMessageHandler {
}
});
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)){
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> 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)){
// 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> 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)->{
@@ -225,25 +234,24 @@ public class MqttMessageHandler {
return "";
}
class TaskWithResult implements Callable<List<ThdDataVO>> {
private CommonStatisticalQueryParam commonStatisticalQueryParam;
public TaskWithResult(CommonStatisticalQueryParam commonStatisticalQueryParam) {
this.commonStatisticalQueryParam = commonStatisticalQueryParam;
}
/**
* 任务的具体过程，一旦任务传给ExecutorService的submit方法，则该方法自动在一个线程上执行。
*
* @return
* @throws Exception
*/
@Override
public List<ThdDataVO> call() throws Exception {
List<ThdDataVO> thdDataVOList = new ArrayList<>();
thdDataVOList = stableDataService.queryFisrtCommonStatistical(commonStatisticalQueryParam);
return thdDataVOList;
}
}
// class TaskWithResult implements Callable<List<ThdDataVO>> {
// private CommonStatisticalQueryParam commonStatisticalQueryParam;
//
// public TaskWithResult(CommonStatisticalQueryParam commonStatisticalQueryParam) {
// this.commonStatisticalQueryParam = commonStatisticalQueryParam;
// }
//
// /**
// * 任务的具体过程，一旦任务传给ExecutorService的submit方法，则该方法自动在一个线程上执行。
// *
// * @return
// * @throws Exception
// */
// @Override
// public List<ThdDataVO> call() throws Exception {
// List<ThdDataVO> thdDataVOList = new ArrayList<>();
// thdDataVOList = stableDataService.queryFisrtCommonStatistical(commonStatisticalQueryParam);
// return thdDataVOList;
// }
// }
}

3
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/StableDataService.java
View File

@@ -2,6 +2,7 @@ package com.njcn.csharmonic.service;
import com.njcn.csharmonic.param.CommonStatisticalQueryParam;
import com.njcn.csharmonic.param.FrequencyStatisticalQueryParam;
import com.njcn.csharmonic.param.ThdDataQueryParm;
import com.njcn.csharmonic.pojo.vo.ThdDataVO;
@@ -34,4 +35,6 @@ public interface StableDataService {
List<ThdDataVO> queryCommonStatisticalByTime(CommonStatisticalQueryParam commonStatisticalQueryParam);
List<ThdDataVO> queryLineCommonStatistical(CommonStatisticalQueryParam commonStatisticalQueryParam);
public List<ThdDataVO> QuerySqlData(FrequencyStatisticalQueryParam frequencyStatisticalQueryParam);
}

89
cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StableDataServiceImpl.java
View File

@@ -8,6 +8,7 @@ import com.njcn.csdevice.enums.AlgorithmResponseEnum;
import com.njcn.csdevice.pojo.po.CsLinePO;
import com.njcn.csdevice.utils.ReflectUtils;
import com.njcn.csharmonic.param.CommonStatisticalQueryParam;
import com.njcn.csharmonic.param.FrequencyStatisticalQueryParam;
import com.njcn.csharmonic.param.ThdDataQueryParm;
import com.njcn.csharmonic.pojo.vo.ThdDataVO;
import com.njcn.csharmonic.service.StableDataService;
@@ -17,9 +18,14 @@ import com.njcn.influx.pojo.po.PowerQualityData;
import com.njcn.influx.service.CommonService;
import com.njcn.influx.service.HaronicRatioService;
import com.njcn.influx.service.PowerQualityService;
import com.njcn.influxdb.param.InfluxDBPublicParam;
import com.njcn.influxdb.utils.InfluxDbUtils;
import com.njcn.system.api.DicDataFeignClient;
import com.njcn.system.api.EpdFeignClient;
import com.njcn.system.enums.DicDataEnum;
import com.njcn.system.pojo.po.EleEpdPqd;
import lombok.RequiredArgsConstructor;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
@@ -52,6 +58,9 @@ public class StableDataServiceImpl implements StableDataService {
private final EpdFeignClient epdFeignClient;
private final CommonService commonService;
private final DecimalFormat df = new DecimalFormat("#0.0000");
private final DicDataFeignClient dicDataFeignClient;
private final InfluxDbUtils influxDbUtils;
@Override
public List<ThdDataVO> queryThdData(String devId, String statisticalName) {
@@ -105,6 +114,8 @@ public class StableDataServiceImpl implements StableDataService {
return thdDataVOList;
}
@Override
public List<ThdDataVO> queryFisrtThdContent(String devId, String statisticalName) {
List<ThdDataVO> thdDataVOList = new ArrayList<>();
@@ -170,8 +181,10 @@ public class StableDataServiceImpl implements StableDataService {
csLinePOList = csLineFeignClient.queryLineByDevId(commonStatisticalQueryParam.getDevId()).getData();
}
String areaId = dicDataFeignClient.getDicDataByCode(DicDataEnum.OUTPUT_SIDE.getCode()).getData().getId();
Optional.ofNullable(csLinePOList).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR));
collect = csLinePOList.stream().map(CsLinePO::getLineId).collect(Collectors.toList());
collect = csLinePOList.stream().filter(temp->Objects.equals(areaId,temp.getPosition())).map(CsLinePO::getLineId).collect(Collectors.toList());
@@ -208,6 +221,7 @@ public class StableDataServiceImpl implements StableDataService {
return collect1;
}
@Override
public List<ThdDataVO> queryCommonStatisticalByTime(CommonStatisticalQueryParam commonStatisticalQueryParam) {
Optional.ofNullable(commonStatisticalQueryParam.getDevId()).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.DEVICE_LOSE));
@@ -307,4 +321,77 @@ public class StableDataServiceImpl implements StableDataService {
}
}
@Override
public List<ThdDataVO> QuerySqlData(FrequencyStatisticalQueryParam frequencyStatisticalQueryParam) {
List<String> collect = new ArrayList<>();
List<CsLinePO> csLinePOList = new ArrayList<>();
List<ThdDataVO> thdDataVOList = new ArrayList<>();
if(Objects.isNull(frequencyStatisticalQueryParam.getDevId())){
if(!CollectionUtil.isEmpty(frequencyStatisticalQueryParam.getLineList())){
collect =frequencyStatisticalQueryParam.getLineList();
csLinePOList = csLineFeignClient.queryLineById(collect).getData();
}else {
throw new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR);
}
}else {
csLinePOList = csLineFeignClient.queryLineByDevId(frequencyStatisticalQueryParam.getDevId()).getData();
}
String areaId = dicDataFeignClient.getDicDataByCode(DicDataEnum.OUTPUT_SIDE.getCode()).getData().getId();
Optional.ofNullable(csLinePOList).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.LINE_DATA_ERROR));
List<CsLinePO> collect1 = csLinePOList.stream().filter(temp -> Objects.equals(areaId, temp.getPosition())).collect(Collectors.toList());
/*治理侧监测点*/
CsLinePO csLinePO = collect1.get(0);
EleEpdPqd data = epdFeignClient.selectById(frequencyStatisticalQueryParam.getStatisticalId()).getData();
Optional.ofNullable(data).orElseThrow(()-> new BusinessException(AlgorithmResponseEnum.ELEEPDPQD_DATA_ERROR));
StringBuilder stringBuilder1 = new StringBuilder();
StringBuilder stringBuilder2 = new StringBuilder();
for (int i = frequencyStatisticalQueryParam.getFrequencyStart(); i <=frequencyStatisticalQueryParam.getFrequencyEnd() ; i++) {
if (i==frequencyStatisticalQueryParam.getFrequencyEnd()){
stringBuilder1.append("last("+data.getName()).append("_"+i).append(") AS "+data.getName()).append("_"+i);
} else {
stringBuilder1.append("last("+data.getName()).append("_"+i).append(") AS "+data.getName()).append("_"+i).append(",");
}
}
stringBuilder2.append ("line_id='").append (csLinePO.getLineId()).append("' and value_type = 'avg' group by phasic_type ").append(InfluxDBPublicParam.TIME_ZONE);
String sql1 = "select "+stringBuilder1+" from "+"apf_data"+" where "+stringBuilder2;
QueryResult sqlData = influxDbUtils.query(sql1);
List<QueryResult.Result> queryResults = sqlData.getResults();
if (CollectionUtil.isNotEmpty(queryResults)) {
for (QueryResult.Result result : queryResults) {
if (CollectionUtil.isNotEmpty(result.getSeries())) {
List<QueryResult.Series> seriesList = result.getSeries();
if (CollectionUtil.isNotEmpty(seriesList)) {
//如果查询语句中存在group by，会存在多个series
for (QueryResult.Series series : seriesList) {
String key = data.getName();
for (int i = frequencyStatisticalQueryParam.getFrequencyStart(); i <=frequencyStatisticalQueryParam.getFrequencyEnd() ; i++) {
int i1 = i - frequencyStatisticalQueryParam.getFrequencyStart()+1;
Double value = Double.parseDouble(series.getValues().get(0).get(i1).toString());
ThdDataVO thdDataVO = new ThdDataVO();
thdDataVO.setLineId(csLinePO.getLineId());
thdDataVO.setLineName(csLinePO.getName());
thdDataVO.setPhase(series.getTags().get("phasic_type"));
thdDataVO.setStatMethod("avg");
thdDataVO.setStatisticalData(Double.valueOf(df.format(value)));
thdDataVO.setStatisticalIndex(data.getId());
thdDataVO.setStatisticalName(data.getName()+ i +"("+data.getUnit()+")");
thdDataVO.setAnotherName(data.getShowName());
thdDataVO.setUnit(data.getUnit());
thdDataVOList.add(thdDataVO);
}
}
}
}
}
}
return thdDataVOList;
}
}