更新采集状态

 iailab-module-data/iailab-module-data-biz/db/mysql/tenant.sql

@@ -399,3 +399,13 @@
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT = '累计点表';
INSERT INTO `t_da_sequence_num` (`id`, `code`, `name`, `sequence_num`, `prefix`) VALUES ('8', 'POINT_L', '累计点编码', 100001, 'L');
INSERT INTO `iailab_plat_system`.`system_dict_data` (`id`, `sort`, `label`, `value`, `dict_type`, `status`, `color_type`, `css_class`, `remark`, `creator`, `create_time`, `updater`, `update_time`, `deleted`) VALUES (1673, 4, '累计点', 'CUMULATE', 'data_point_type', 0, '', '', '', '142', '2024-12-10 10:13:12', '142', '2024-12-10 10:13:12', b'0');

CREATE TABLE t_da_point_collect_status(
                           `id` VARCHAR(36) NOT NULL  COMMENT 'ID' ,
                           `point_id` VARCHAR(36) NOT NULL   COMMENT '测点ID',
                           `collect_value` VARCHAR(36)   COMMENT '采集值',
                           `collect_quality` VARCHAR(36)   COMMENT '采集质量',
                           `collect_time` DATETIME   COMMENT '采集时间' ,
                           PRIMARY KEY (id) USING BTREE,
                           UNIQUE KEY `uk_point_id` (`point_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT = '测点采集状态表';

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/channel/http/collector/ihdb/HttpCollectorForIhd.java

@@ -12,6 +12,7 @@
import com.iailab.module.data.common.utils.HttpRequest;
import com.iailab.module.data.common.utils.TagUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
@@ -265,7 +266,12 @@
            StringBuilder tagSb = new StringBuilder();
            tagSb.append("[");
            for (int i = 0; i < params.size(); i++) {
                Map<String, Object> queryParams = new HashMap<>();
                if (StringUtils.isBlank(params.get(i)[1].toString()) ||
                        StringUtils.isBlank(params.get(i)[2].toString()) ||
                        StringUtils.isBlank(params.get(i)[3].toString())) {
                    continue;
                }
                Map<String, Object> queryParams = new HashMap<>(3);
                queryParams.put(N, params.get(i)[1]);
                queryParams.put(D, params.get(i)[2]);
                queryParams.put(P, params.get(i)[3]);

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/collection/PointCollector.java

@@ -3,10 +3,13 @@
import com.iailab.module.data.common.enums.DataSourceType;
import com.iailab.module.data.common.utils.R;
import com.iailab.module.data.channel.kio.collector.KingIOCollector;
import com.iailab.module.data.influxdb.pojo.InfluxPointValueDigPOJO;
import com.iailab.module.data.influxdb.pojo.InfluxPointValueSimPOJO;
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.DaPointCollectStatusService;
import com.iailab.module.data.point.service.DaPointService;
import com.iailab.module.data.influxdb.pojo.InfluxPointValuePOJO;
import com.iailab.module.data.channel.modbus.collector.ModBusCollector;
@@ -16,7 +19,10 @@
import com.iailab.module.data.point.dto.DaPointWriteValueDTO;
import com.iailab.module.data.influxdb.service.InfluxDBService;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

@@ -59,6 +65,9 @@
    @Resource
    private CumulateHandle cumulateHandle;

    @Autowired
    private DaPointCollectStatusService daPointCollectStatusService;

    /**
     * 采集
     *
@@ -89,6 +98,8 @@
            log.info("存入数据库");
            influxDBService.asyncWritePointValues(pointValues);

            log.info("更新采集状态");
            updateCollectStatus(pointValues, collectTime);
            log.info("采集完成");
        } catch (Exception ex)  {
            log.info("采集异常！");
@@ -96,6 +107,22 @@
        }
    }

    private void updateCollectStatus(List<InfluxPointValuePOJO> pointValues, Date collectTime) {
        try {
            for (InfluxPointValuePOJO pointValue : pointValues) {
                if (pointValue instanceof InfluxPointValueSimPOJO) {
                    InfluxPointValueSimPOJO pvo = (InfluxPointValueSimPOJO) pointValue;
                    daPointCollectStatusService.recordStatus(pvo.getPoint(), pvo.getValue().toString(), collectTime);
                } else if (pointValue instanceof InfluxPointValueDigPOJO) {
                    InfluxPointValueDigPOJO pvo = (InfluxPointValueDigPOJO) pointValue;
                    daPointCollectStatusService.recordStatus(pvo.getPoint(), pvo.getValue().toString(), collectTime);
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public Map<String, Object> getCurrentValue(List<String> pointNos) {
        try {
            Map<String, Object> data = new HashMap<>();

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/collection/handler/MeasureHandle.java

@@ -133,9 +133,11 @@
    private Object handleData(DaPointDTO dto, Object value) {
        Object result = value;
        try {
            if (value == null) {
                return CommonConstant.BAD_VALUE;
            }
            if (DataTypeEnum.FLOAT.getCode().equals(dto.getDataType()) || DataTypeEnum.INT.getCode().equals(dto.getDataType())) {
                BigDecimal rawValue = new BigDecimal(value.toString());

                // 异常值处理
                if (rawValue.compareTo(maxValue) > 0 || rawValue.compareTo(minValue) < 0) {
                    rawValue = CommonConstant.BAD_VALUE;
@@ -157,7 +159,6 @@
            } else if (DataTypeEnum.BOOLEAN.getCode().equals(dto.getDataType())) {
                result = Boolean.parseBoolean(value.toString());
            }

        } catch (Exception ex) {
            log.warn("handleData异常，PointNo=" + dto.getPointNo());
            ex.printStackTrace();

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/dao/DaPointCollectStatusDao.java

对比新文件
@@ -0,0 +1,16 @@
package com.iailab.module.data.point.dao;

import com.iailab.framework.common.dao.BaseDao;
import com.iailab.framework.tenant.core.db.dynamic.TenantDS;
import com.iailab.module.data.point.entity.DaPointCollectStatusEntity;
import org.apache.ibatis.annotations.Mapper;

/**
 * @author PanZhibao
 * @Description
 * @createTime 2024年12月13日
 */
@TenantDS
@Mapper
public interface DaPointCollectStatusDao extends BaseDao<DaPointCollectStatusEntity> {
}

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/entity/DaPointCollectStatusEntity.java

对比新文件
@@ -0,0 +1,46 @@
package com.iailab.module.data.point.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

/**
 * @author PanZhibao
 * @Description
 * @createTime 2024年12月13日
 */
@Data
@TableName("t_da_point_collect_status")
public class DaPointCollectStatusEntity implements Serializable {
    private static final long serialVersionUID = 1L;

    /**
     * id
     */
    @TableId(type = IdType.ASSIGN_UUID)
    private String id;

    /**
     * 测点ID
     */
    private String pointId;

    /**
     * 采集值
     */
    private String collectValue;

    /**
     * 采集质量
     */
    private String collectQuality;

    /**
     * 采集时间
     */
    private Date collectTime;
}

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/service/DaPointCollectStatusService.java

对比新文件
@@ -0,0 +1,18 @@
package com.iailab.module.data.point.service;

import com.iailab.framework.common.service.BaseService;
import com.iailab.module.data.point.entity.DaPointCollectStatusEntity;
import org.springframework.scheduling.annotation.Async;

import java.util.Date;

/**
 * @author PanZhibao
 * @Description
 * @createTime 2024年12月13日
 */
public interface DaPointCollectStatusService extends BaseService<DaPointCollectStatusEntity> {

    @Async
    void recordStatus(String pointId, String collectValue, Date collectTime);
}

 iailab-module-data/iailab-module-data-biz/src/main/java/com/iailab/module/data/point/service/impl/DaPointCollectStatusServiceImpl.java

对比新文件
@@ -0,0 +1,43 @@
package com.iailab.module.data.point.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.iailab.framework.common.service.impl.BaseServiceImpl;
import com.iailab.module.data.common.enums.DataQualityEnum;
import com.iailab.module.data.point.dao.DaPointCollectStatusDao;
import com.iailab.module.data.point.entity.DaPointCollectStatusEntity;
import com.iailab.module.data.point.service.DaPointCollectStatusService;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.UUID;

/**
 * @author PanZhibao
 * @Description
 * @createTime 2024年12月13日
 */
@Service
public class DaPointCollectStatusServiceImpl extends BaseServiceImpl<DaPointCollectStatusDao, DaPointCollectStatusEntity>
        implements DaPointCollectStatusService {

    public void recordStatus(String pointId, String collectValue, Date collectTime) {
        QueryWrapper<DaPointCollectStatusEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("point_id", pointId);
        DaPointCollectStatusEntity entity = baseDao.selectOne(queryWrapper);
        if (entity == null) {
            entity = new DaPointCollectStatusEntity();
            entity.setId(UUID.randomUUID().toString());
            entity.setPointId(pointId);
            entity.setCollectValue(collectValue);
            entity.setCollectQuality(DataQualityEnum.getEumByValue(collectValue).getCode());
            entity.setCollectTime(collectTime);
            baseDao.insert(entity);
        } else {
            entity.setCollectValue(collectValue);
            entity.setCollectQuality(DataQualityEnum.getEumByValue(collectValue).getCode());
            entity.setCollectTime(collectTime);
            baseDao.updateById(entity);
        }

    }
}