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

CVT代码

Browse Source
This commit is contained in:
hzj
2025-03-07 09:26:40 +08:00
parent 12c387d66e
commit 7b65b79662
12 changed files with 204 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
algorithm/algorithm-api/src/main/java/com/njcn/algorithm/pojo/bo/HourParam.java Normal file
View File

@@ -0,0 +1,62 @@
package com.njcn.algorithm.pojo.bo;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import java.io.Serializable;
import java.util.List;
import java.util.Set;
/**
*
* ** 按小时纬度执行算法参数 **
* @author hongawen
* @version 1.0.0
* @date 2023年11月01日 16:17
*/
@Data
public class HourParam implements Serializable {
/***
* 是否全链路执行算法
* 非全链路执行时，tag集合必须非空
*/
@ApiModelProperty(name = "fullChain",value = "是否全链执行")
private boolean fullChain;
/**
* 目前仅监测点日统计存在补招功能 by yxb
* 是否补招标识,默认不补招
*/
@ApiModelProperty(name = "repair",value = "是否补招")
private boolean repair;
@ApiModelProperty(name = "beginTime",value = "补招起始日期_yyyy-MM-dd hh::mm::ss")
private String beginTime;
@ApiModelProperty(name = "endTime",value = "补招截止日期_yyyy-MM-dd hh::mm::ss")
private String endTime;
@ApiModelProperty(name = "dataDate",value = "时间日期_yyyy-MM-dd hh::mm::ss")
private String dataDate;
/***
* 需要执行的组件
* 当不需要全链路执行时，通过tag名称动态指定执行某个算法组件
*/
@ApiModelProperty(name = "tagNames",value = "待执行链节点的tag集合")
private Set<String> tagNames;
/**
* 待计算的对象索引集合，监测点、设备、母线、变电站、单位等等
*/
@ApiModelProperty(name = "idList",value = "索引集合")
private List<String> idList;
}

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

@@ -4,11 +4,13 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.StrUtil;
import com.njcn.algorithm.pojo.bo.BaseParam;
import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.pojo.bo.HourParam;
import com.njcn.algorithm.pojo.enums.PrepareResponseEnum;
import com.njcn.common.pojo.annotation.OperateInfo;
import com.njcn.common.pojo.enums.common.LogEnum;
@@ -147,6 +149,49 @@ public class ExecutionCenter extends BaseController {
}
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON)
@ApiOperation("监测点算法执行链(按小时执行的算法)")
@PostMapping("/measurementPointExecutorByHour")
@Async("asyncExecutor")
public void measurementPointExecutorByHour(@RequestBody HourParam baseParam) {
String methodDescribe = getMethodDescribe("measurementPointExecutorByHour");
//手动判断参数是否合法,
if (!baseParam.isFullChain() && CollectionUtil.isEmpty(baseParam.getTagNames())) {
throw new BusinessException(PrepareResponseEnum.NO_EXECUTOR_NODE);
}
if (baseParam.isRepair() && StrUtil.isAllEmpty(baseParam.getBeginTime(), baseParam.getEndTime())) {
throw new BusinessException(PrepareResponseEnum.NO_REPAIR_DATE);
}
CalculatedParam calculatedParam = new CalculatedParam();
BeanUtil.copyProperties(baseParam, calculatedParam);
// 测点索引
if (CollectionUtils.isEmpty(calculatedParam.getIdList())) {
calculatedParam.setIdList(commTerminalGeneralClient.getRunMonitorIds().getData());
}
LiteflowResponse liteflowResponse;
if (baseParam.isRepair()) {
//补招时，起始日期、截止日期必填
DateTime startDate = DateUtil.parse(baseParam.getBeginTime(), DatePattern.NORM_DATETIME_FORMAT);
DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATETIME_FORMAT);
long betweenHour = DateUtil.between(startDate, endDate, DateUnit.HOUR);
//递增日期执行算法链
for (int i = 0; i < betweenHour; i++) {
if (i != 0) {
startDate = DateUtil.offsetHour(startDate, 1);
}
calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATETIME_FORMATTER));
liteflowResponse = flowExecutor.execute2Resp("measurement_point_hour", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
} else {
//非补招
calculatedParam.setDataDate(DateUtil.format(DateUtil.offsetHour( DateUtil.parse(baseParam.getDataDate(), DatePattern.NORM_DATETIME_FORMAT),-1), DatePattern.NORM_DATETIME_FORMATTER));
liteflowResponse = flowExecutor.execute2Resp("measurement_point_hour", calculatedParam);
dealResponse(calculatedParam, liteflowResponse, methodDescribe);
}
}
// @OperateInfo(info = LogEnum.BUSINESS_COMMON)
// @ApiOperation("装置算法执行链")
// @PostMapping("/deviceExecutor")

16
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/executor/MeasurementExecutor.java
View File

@@ -1,7 +1,6 @@
package com.njcn.algorithm.executor;
import com.njcn.algorithm.service.line.IDataCleanService;
import com.njcn.algorithm.service.line.IDataVCvtService;
import com.njcn.algorithm.service.line.IDayDataService;
import com.yomahub.liteflow.annotation.LiteflowComponent;
import com.yomahub.liteflow.annotation.LiteflowMethod;
@@ -29,8 +28,7 @@ public class MeasurementExecutor extends BaseExecutor {
@Resource
private IDataCleanService dataCleanService;
@Resource
private IDataVCvtService dataVCvtService;
/**
* 数据清洗 电压表
@@ -290,17 +288,5 @@ public class MeasurementExecutor extends BaseExecutor {
}
/**
* 监测点cvt转换算法()
* @author hzj
*/
@LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataVCvt", nodeType = NodeTypeEnum.COMMON)
public boolean dataVCvtAccess(NodeComponent bindCmp) {
return isAccess(bindCmp);
}
@LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataVCvt", nodeType = NodeTypeEnum.COMMON)
public void dataVCvtProcess(NodeComponent bindCmp) {
dataVCvtService.dataVCvtHandler(bindCmp.getRequestData());
}
}

44
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/executor/MeasurementHourExecutor.java Normal file
View File

@@ -0,0 +1,44 @@
package com.njcn.algorithm.executor;
import com.njcn.algorithm.service.line.IDataHarmRateVCvtService;
import com.yomahub.liteflow.annotation.LiteflowComponent;
import com.yomahub.liteflow.annotation.LiteflowMethod;
import com.yomahub.liteflow.core.NodeComponent;
import com.yomahub.liteflow.enums.LiteFlowMethodEnum;
import com.yomahub.liteflow.enums.NodeTypeEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import javax.annotation.Resource;
/**
* @author xy
* @version 1.0.0
* @date 2025年1月16日
*/
@Slf4j
@LiteflowComponent
@RequiredArgsConstructor
public class MeasurementHourExecutor extends BaseExecutor {
@Resource
private IDataHarmRateVCvtService dataHarmRateVCvtService;
/**
* 监测点cvt转换算法()
* @author hzj
*/
@LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "dataHarmRateVCvt", nodeType = NodeTypeEnum.COMMON)
public boolean dataVCvtAccess(NodeComponent bindCmp) {
return isAccess(bindCmp);
}
@LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "dataHarmRateVCvt", nodeType = NodeTypeEnum.COMMON)
public void dataVCvtProcess(NodeComponent bindCmp) {
dataHarmRateVCvtService.dataHarmRateVCvtHandler(bindCmp.getRequestData());
}
}

4
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IDataVCvtService.java → algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/service/line/IDataHarmRateVCvtService.java
View File

@@ -9,6 +9,6 @@ import com.njcn.algorithm.pojo.bo.CalculatedParam;
* @author clam
* @version V1.0.0
*/
public interface IDataVCvtService {
void dataVCvtHandler(CalculatedParam requestData);
public interface IDataHarmRateVCvtService {
void dataHarmRateVCvtHandler(CalculatedParam requestData);
}

21
algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataVCvtServiceImpl.java
View File

@@ -1,11 +1,15 @@
package com.njcn.algorithm.serviceimpl.line;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.njcn.algorithm.pojo.bo.CalculatedParam;
import com.njcn.algorithm.service.line.IDataVCvtService;
import com.njcn.algorithm.service.line.IDataHarmRateVCvtService;
import com.njcn.dataProcess.api.DataHarmRateVCvtFeignClient;
import com.njcn.dataProcess.api.DataHarmRateVFeignClient;
import com.njcn.dataProcess.constant.InfluxDBTableConstant;
import com.njcn.dataProcess.param.LineCountEvaluateParam;
import com.njcn.dataProcess.pojo.dto.DataHarmDto;
import com.njcn.dataProcess.pojo.dto.DataHarmRateVCvtDto;
import com.njcn.dataProcess.pojo.dto.DataHarmRateVDto;
import com.njcn.dataProcess.util.TimeUtils;
@@ -34,7 +38,7 @@ import java.util.stream.Collectors;
@Slf4j
@Component
@RequiredArgsConstructor
public class DataVCvtServiceImpl implements IDataVCvtService {
public class DataVCvtServiceImpl implements IDataHarmRateVCvtService {
@Resource
private CvtRelationFeignClient cvtRelationFeignClient;
@Resource
@@ -43,15 +47,22 @@ public class DataVCvtServiceImpl implements IDataVCvtService {
private DataHarmRateVFeignClient dataHarmRateVFeignClient;
@Override
public void dataVCvtHandler(CalculatedParam calculatedParam) {
public void dataHarmRateVCvtHandler(CalculatedParam calculatedParam) {
LineCountEvaluateParam lineParam = new LineCountEvaluateParam();
lineParam.setStartTime(TimeUtils.getBeginOfDay(calculatedParam.getDataDate()));
lineParam.setEndTime(TimeUtils.getEndOfDay(calculatedParam.getDataDate()));
List idList = calculatedParam.getIdList();
LineCountEvaluateParam lineCountEvaluateParam = new LineCountEvaluateParam();
lineCountEvaluateParam.setLineId(calculatedParam.getIdList());
lineCountEvaluateParam.setEndTime(calculatedParam.getDataDate()+ InfluxDBTableConstant.END_TIME);
lineCountEvaluateParam.setStartTime(calculatedParam.getDataDate()+ InfluxDBTableConstant.START_TIME);
DateTime beginTime1= DateUtil.beginOfHour(DateUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATETIME_FORMAT));
//获取最新一条数据时间
DataHarmDto topData = dataHarmRateVFeignClient.getTopData().getData();
String minTime = topData.getMinTime();
DateTime beginTime2= DateUtil.beginOfHour(DateUtil.parse(minTime, DatePattern.NORM_DATETIME_FORMAT));
DateTime beginTime = beginTime1.isBefore(beginTime2) ? beginTime1 : beginTime2;
DateTime endTime = DateUtil.endOfHour(beginTime1);
lineCountEvaluateParam.setEndTime(DateUtil.format(endTime,DatePattern.NORM_DATETIME_FORMAT));
lineCountEvaluateParam.setStartTime(DateUtil.format(beginTime,DatePattern.NORM_DATETIME_FORMAT));
List<DataHarmRateVCvtDto> data = dataHarmRateVCvtFeignClient.getRawData(lineCountEvaluateParam).getData();

2
data-processing/data-processing-api/src/main/java/com/njcn/dataProcess/api/DataHarmRateVFeignClient.java
View File

@@ -27,5 +27,7 @@ public interface DataHarmRateVFeignClient {
@PostMapping("/addList")
HttpResult<String> addList(@RequestBody List<DataHarmRateVDto> list);
@PostMapping("/getTopData")
HttpResult<DataHarmDto> getTopData();
}

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

@@ -55,6 +55,12 @@ public class DataHarmRateVFeignClientFallbackFactory implements FallbackFactory<
throw new BusinessException(finalExceptionEnum);
}
@Override
public HttpResult<DataHarmDto> getTopData() {
log.error("{}异常，降级处理，异常为：{}","获取最新原始数据",cause.toString());
throw new BusinessException(finalExceptionEnum);
}
};
}
}

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

@@ -55,6 +55,15 @@ public class DataHarmRateVController extends BaseController {
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.QUERY)
@PostMapping("/getTopData")
@ApiOperation("获取数据最新时间点")
public HttpResult<DataHarmDto> getTopData() {
String methodDescribe = getMethodDescribe("getTopData");
DataHarmDto data = dataHarmRateVQuery.getTopData();
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, data, methodDescribe);
}
@OperateInfo(info = LogEnum.BUSINESS_COMMON,operateType = OperateType.ADD)
@PostMapping("/batchInsertion")
@ApiOperation("批量插入")

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

@@ -35,4 +35,6 @@ IDataHarmRateV extends IMppService<RStatDataHarmRateVD> {
* 批量插入数据
*/
void addList(List<DataHarmRateVDto> list);
DataHarmDto getTopData();
}

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

@@ -183,6 +183,16 @@ public class InfluxdbDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
dataHarmRateVMapper.insertBatch(result);
}
@Override
public DataHarmDto getTopData() {
DataHarmDto dto = new DataHarmDto();
InfluxQueryWrapper influxQueryWrapper = new InfluxQueryWrapper(DataHarmrateV.class);
influxQueryWrapper.timeDesc().limit(1);
List<DataHarmrateV> dataHarmrateVList = dataHarmRateVMapper.selectByQueryWrapper(influxQueryWrapper);
dto.setMinTime(DATE_TIME_FORMATTER.format(dataHarmrateVList.get(0).getTime()));
return dto;
}
/**
* 按监测点集合、时间条件获取分钟数据
* timeMap参数来判断是否进行数据处理 timeMap为空则不进行数据处理

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

@@ -89,4 +89,9 @@ public class RelationDataHarmRateVImpl extends MppServiceImpl<RStatDataHarmRateV
});
dataHarmRateV.saveOrUpdateBatchByMultiId(result);
}
@Override
public DataHarmDto getTopData() {
return null;
}
}