iailab-module-data/iailab-module-data-biz/db/mysql/tenant.sql
@@ -391,7 +391,7 @@ CREATE TABLE t_da_cumulate_point( `id` VARCHAR(36) NOT NULL COMMENT 'ID' , `point_id` VARCHAR(36) COMMENT '测点ID' , `point_no` VARCHAR(36) COMMENT '瞬时测点' , `moment_point` VARCHAR(36) COMMENT '瞬时测点' , `length` int COMMENT '累计长度', `divisor` int COMMENT '除数', PRIMARY KEY (id) USING BTREE, iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/collection/PointCollector.java
@@ -4,6 +4,7 @@ import com.iailab.module.data.common.utils.R; import com.iailab.module.data.channel.kio.collector.KingIOCollector; import com.iailab.module.data.point.collection.handler.CalculateHandle; import com.iailab.module.data.point.collection.handler.CumulateHandle; import com.iailab.module.data.point.common.PointTypeEnum; import com.iailab.module.data.point.dto.DaPointDTO; import com.iailab.module.data.point.service.DaPointService; @@ -55,6 +56,9 @@ @Resource private OpcUaCollector opcUaCollector; @Resource private CumulateHandle cumulateHandle; /** * 采集 * @@ -78,6 +82,10 @@ List<DaPointDTO> pointCalculateList = daPointService.getMathPoint(minfreq); pointValues.addAll(calculateHandle.handle(collectTime, pointCalculateList, dataMap)); log.info("读取累计点"); List<DaPointDTO> pointCumulateList = daPointService.getCumulatePoint(minfreq); pointValues.addAll(cumulateHandle.handle(collectTime, pointCumulateList)); log.info("存入数据库"); influxDBService.asyncWritePointValues(pointValues);
对比新文件 @@ -0,0 +1,138 @@ package com.iailab.module.data.point.collection.handler; import com.iailab.module.data.api.point.DataPointApi; import com.iailab.module.data.api.point.dto.ApiPointDTO; import com.iailab.module.data.api.point.dto.ApiPointValueDTO; import com.iailab.module.data.api.point.dto.ApiPointValueQueryDTO; import com.iailab.module.data.common.enums.CommonConstant; import com.iailab.module.data.enums.DataPointFreqEnum; import com.iailab.module.data.influxdb.pojo.InfluxPointValuePOJO; import com.iailab.module.data.point.collection.utils.GenInfluxPointValueUtils; import com.iailab.module.data.point.dto.DaPointDTO; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; import java.math.BigDecimal; import java.util.*; /** * 累计点处理 * * @author PanZhibao * @Description * @createTime 2024年11月29日 */ @Slf4j @Component public class CumulateHandle { @Autowired private DataPointApi dataPointApi; public List<InfluxPointValuePOJO> handle(Date collectTime, List<DaPointDTO> dtos) { List<InfluxPointValuePOJO> result = new ArrayList<>(); try { log.info("累计点处理开始"); if (CollectionUtils.isEmpty(dtos)) { return result; } dtos.forEach(dto -> { try { Object value = singleCompute(dto, collectTime); InfluxPointValuePOJO pojo = GenInfluxPointValueUtils.getByPoint(dto, value); pojo.setTimestamp(collectTime.toInstant()); result.add(pojo); } catch (Exception ex) { ex.printStackTrace(); log.info("累计点异常！PointNo=" + dto.getPointNo()); } }); } catch (Exception ex) { ex.printStackTrace(); log.info("累计点处理异常！"); } return result; } private Object singleCompute(DaPointDTO dto, Date collectTime) { ApiPointDTO pointDTO = dataPointApi.getInfoByNo(dto.getMomentPoint()); if (pointDTO == null) { return CommonConstant.BAD_VALUE; } Calendar calendar = Calendar.getInstance(); calendar.setTime(collectTime); calendar.add(Calendar.MINUTE, -1); Date endTime = calendar.getTime(); calendar.add(Calendar.MINUTE, dto.getLength() * -1); Date startTime = calendar.getTime(); ApiPointValueQueryDTO queryDto = new ApiPointValueQueryDTO(); queryDto.setStart(startTime); queryDto.setEnd(endTime); queryDto.setPointNo(dto.getMomentPoint()); List<ApiPointValueDTO> dataList = dataPointApi.queryPointHistoryValue(queryDto); if (CollectionUtils.isEmpty(dataList)) { return BigDecimal.ZERO; } else if (dataList.size() < dto.getLength()) { // 补全数据 dataList = completionData(dto.getLength(), dataList, startTime, endTime, pointDTO); } double total = dataList.stream().mapToDouble(ApiPointValueDTO::getV).sum(); return new BigDecimal(total).divide(new BigDecimal(dto.getDivisor()), 2, BigDecimal.ROUND_HALF_UP); } private List<ApiPointValueDTO> completionData(int length, List<ApiPointValueDTO> dataList, Date startTime, Date endTime, ApiPointDTO pointDTO) { if (CollectionUtils.isEmpty(dataList) || length == dataList.size()) { return dataList; } else if (length < dataList.size()) { return dataList.subList(dataList.size() - length, dataList.size()); } List<ApiPointValueDTO> result = new ArrayList<>(); long start = startTime.getTime(); long end = endTime.getTime(); long oneMin = 1000L * DataPointFreqEnum.getEumByCode(pointDTO.getMinfreqid()).getValue(); long mins = (end - start) / oneMin; //找出缺少项 Map<Long, Double> sourceDataMap = new HashMap<>(dataList.size()); Map<Long, Double> dataMap = new LinkedHashMap<>(); for (int i = 0; i < mins; i++) { Long key = start + oneMin * i; Double value = sourceDataMap.get(key); dataMap.put(key, value); } //补充缺少项 int k = 0; Map.Entry<Long, Double> lastItem = null; for (Map.Entry<Long, Double> item : dataMap.entrySet()) { if (k == 0 && item.getValue() == null) { item.setValue(getFirstValue(dataMap)); } else if (item.getValue() == null) { item.setValue(lastItem.getValue()); } k++; lastItem = item; ApiPointValueDTO dataEntity = new ApiPointValueDTO(); dataEntity.setT(new Date(item.getKey())); dataEntity.setV(item.getValue()); result.add(dataEntity); } return result; } private Double getFirstValue(Map<Long, Double> dataMap) { for (Map.Entry<Long, Double> item : dataMap.entrySet()) { if (item.getValue() != null) { return item.getValue(); } } return null; } } iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/dao/DaPointDao.java
@@ -35,6 +35,8 @@ List<DaPointDTO> getMathPoint(Map<String, Object> params); List<DaPointDTO> getCumulatePoint(Map<String, Object> params); default IPage<DaPointDTO> selectPageList(DaPointPageReqVO reqVO) { return getPageList(getPage(reqVO), reqVO); } iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/dto/DaCumulatePointDTO.java
@@ -29,7 +29,7 @@ private String pointId; @Schema(description = "瞬时测点", required = true) private String pointNo; private String momentPoint; @Schema(description = "累计长度", required = true) private Integer length; iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/dto/DaPointDTO.java
@@ -107,6 +107,15 @@ @Schema(description = "计算公式", required = true) private String expression; @Schema(description = "瞬时测点") private String momentPoint; @Schema(description = "累计长度") private Integer length; @Schema(description = "除数") private Integer divisor; @Schema(description = "数据源选项") private List<String> sourceOption;
@@ -33,7 +33,7 @@ /** * 累计测点 */ private String pointNo; private String momentPoint; /** * 累计长度 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/service/DaPointService.java
@@ -43,6 +43,8 @@ List<DaPointDTO> getMathPoint(List<String> pointNos); List<DaPointDTO> getCumulatePoint(String freq); DaPointDTO getByNo(String pointNo); List<DaPointDTO> getByNos(List<String> pointNos); iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/service/impl/DaPointServiceImpl.java
@@ -318,6 +318,15 @@ } @Override public List<DaPointDTO> getCumulatePoint(String freq) { Map<String, Object> params = new HashMap<>(); params.put("pointType", PointTypeEnum.CUMULATE.getCode()); params.put("isEnable", CommonConstant.IS_ENABLE); params.put("minfreqid", freq); return daPointDao.getCumulatePoint(params); } @Override public DaPointDTO getByNo(String pointNo) { if (pointNoMap.containsKey(pointNo)) { return pointNoMap.get(pointNo); iailab-module-data/iailab-module-data-biz/src/main/resources/mapper/point/DaPointDao.xml
@@ -236,4 +236,29 @@ </where> </select> <select id="getCumulatePoint" resultType="com.iailab.module.data.point.dto.DaPointDTO"> SELECT t1.point_no, t1.point_name, t1.default_value, t1.point_type, t1.data_type, t1.store_type, t1.minfreqid, t2.moment_point, t2.length, t2.divisor FROM t_da_point t1 LEFT JOIN t_da_cumulate_point t2 ON t2.point_id = t1.id <where> t1.point_type = #{pointType} <if test="isEnable != null"> AND t1.is_enable = #{isEnable} </if> <if test="minfreqid != null and minfreqid != ''"> AND t1.minfreqid = #{minfreqid} </if> </where> </select> </mapper>
