实时数据接口转化；根据配置模拟数据

service-cdi/service-cdi-biz/src/main/java/com/usky/cdi/controller/IotDataController.java

@@ -29,8 +29,8 @@ public class IotDataController {
      * 上报水浸状态
      */
     @PostMapping("/flooded")
-    public String sendWaterLeak(@RequestBody WaterLeakVO vo) {
-        boolean success = iotDataTransferService.sendWaterLeak(vo);
+    public String sendWaterLeak() {
+        boolean success = iotDataTransferService.sendWaterLeak();
         return success ? "上报成功" : "上报失败";
     }
 

service-cdi/service-cdi-biz/src/main/java/com/usky/cdi/domain/DmpDevice.java

@@ -0,0 +1,132 @@
+package com.usky.cdi.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * <p>
+ * 设备信息表
+ * </p>
+ *
+ * @author han
+ * @since 2023-09-22
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+public class DmpDevice implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 主键id
+     */
+    @TableId(value = "id", type = IdType.AUTO)
+    private Integer id;
+
+    /**
+     * 设备ID;设备注册时系统自动生成一个唯一编号
+     */
+    private String deviceId;
+
+    /**
+     * 设备名称
+     */
+    private String deviceName;
+
+    /**
+     * 设备类型（501、监控系统  502、门禁系统  503、梯控系统  504、机房系统  509、环境系统  510、照明系统）
+     */
+    private Integer deviceType;
+
+    /**
+     * 产品ID
+     */
+    private Integer productId;
+
+    /**
+     * 物联网卡号
+     */
+    private String simCode;
+
+    /**
+     * 国际移动用户识别码
+     */
+    private String imsiCode;
+
+    /**
+     * 自动订阅标识（0：否，1：是）
+     */
+    private Integer subscribeFlag;
+
+    /**
+     * 节点类型
+     */
+    private Integer nodeType;
+
+    /**
+     * 分组id
+     */
+    private Integer groupId;
+
+    /**
+     * 删除标识
+     */
+    private Integer deleteFlag;
+
+    /**
+     * 创建人
+     */
+    private String createdBy;
+
+    /**
+     * 创建时间
+     */
+    private LocalDateTime createdTime;
+
+    /**
+     * 更新人
+     */
+    private String updatedBy;
+
+    /**
+     * 更新时间
+     */
+    private LocalDateTime updatedTime;
+
+    /**
+     * 租户号
+     */
+    private Integer tenantId;
+
+    /**
+     * 单位编号
+     */
+    private String companyCode;
+
+    /**
+     * 安装位置
+     */
+    private String installAddress;
+
+    /**
+     * 业务状态;1：未激活，2：已激活，3：禁用
+     */
+    private Integer serviceStatus;
+
+    /**
+     * 产品编码
+     */
+    private String productCode;
+
+    /**
+     * 设备UUID
+     */
+    private String deviceUuid;
+
+
+}

service-cdi/service-cdi-biz/src/main/java/com/usky/cdi/mapper/DmpDeviceMapper.java

@@ -0,0 +1,16 @@
+package com.usky.cdi.mapper;
+
+import com.usky.cdi.domain.DmpDevice;
+import com.usky.common.mybatis.core.CrudMapper;
+
+/**
+ * <p>
+ * 设备信息表 Mapper 接口
+ * </p>
+ *
+ * @author han
+ * @since 2023-09-22
+ */
+public interface DmpDeviceMapper extends CrudMapper<DmpDevice> {
+
+}

service-cdi/service-cdi-biz/src/main/java/com/usky/cdi/service/impl/IotDataTransferService.java

@@ -1,18 +1,24 @@
 package com.usky.cdi.service.impl;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.usky.cdi.service.config.mqtt.MqttGateway;
 import com.usky.cdi.service.enums.EnvMonitorMqttTopic;
+import com.usky.cdi.service.util.DeviceDataQuery;
 import com.usky.cdi.service.util.SnowflakeIdGenerator;
 import com.usky.cdi.service.vo.base.*;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.tomcat.jni.Local;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Service;
 
 import java.text.SimpleDateFormat;
+import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.Date;
+import java.util.List;
 
 /**
  *
@@ -30,6 +36,10 @@ public class IotDataTransferService {
 
     private final SnowflakeIdGenerator idGenerator;
 
+    @Autowired
+    private DeviceDataQuery deviceDataQuery;
+
+
     public IotDataTransferService() {
         // 使用默认的workerId和datacenterId，实际项目中可以从配置读取
         this.idGenerator = new SnowflakeIdGenerator(1L, 1L);
@@ -53,27 +63,41 @@ public class IotDataTransferService {
      * 发送水浸状态
      * Topic: iotInfo/flooded
      *
-     * @param vo 水浸状态
      * @return 是否发送成功
      */
-    public boolean sendWaterLeak(WaterLeakVO vo) {
+    public boolean sendWaterLeak() {
         if (mqttGateway == null) {
             log.warn("MQTT Gateway未初始化，无法发送消息");
             return false;
         }
         try {
-            if (vo.getDataPacketID() == null) {
-                vo.setDataPacketID(generateDataPacketID());
+            List<JSONObject> deviceData = deviceDataQuery.getDeviceData(702);
+
+            if (deviceData.isEmpty()) {
+                log.warn("没有获取到水浸数据！");
+                return false;
             }
-            if (vo.getPublishTime() == null) {
+
+            for (JSONObject deviceDataItem : deviceData) {
+                Integer leachStatus = Integer.valueOf(deviceDataItem.getString("leach_status"));
+                Long dataTime = deviceDataItem.getLong("time");
+                LocalDateTime dataEndTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(dataTime), ZoneId.systemDefault());
+
+                WaterLeakVO vo = new WaterLeakVO();
+                vo.setDataPacketID(generateDataPacketID());
+                vo.setSensorID(4399L);
+                vo.setEngineeringID(9527L);
                 vo.setPublishTime(getCurrentTime());
-            }
+                vo.setSensorValue(leachStatus);
+                vo.setDataEndTime(dataEndTime);
 
-            String json = JSON.toJSONString(vo);
-            String topic = EnvMonitorMqttTopic.WATER_LEAK.getTopic();
+                String json = JSON.toJSONString(vo);
+                String topic = EnvMonitorMqttTopic.WATER_LEAK.getTopic();
+
+                log.info("发送水浸状态信息，Topic: {}, Data: {}", topic, json);
+                mqttGateway.sendToMqtt(topic, json);
+            }
 
-            log.info("发送水浸状态信息，Topic: {}, Data: {}", topic, json);
-            mqttGateway.sendToMqtt(topic, json);
             return true;
         } catch (Exception e) {
             log.error("发送水浸状态信息失败", e);

service-cdi/service-cdi-biz/src/main/java/com/usky/cdi/service/util/DeviceDataQuery.java

@@ -0,0 +1,202 @@
+package com.usky.cdi.service.util;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.usky.cdi.domain.DmpDevice;
+import com.usky.cdi.mapper.DmpDeviceMapper;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ *
+ * @author fyc
+ * @email yuchuan.fu@chinausky.com
+ * @date 2025/11/27
+ */
+@Data
+@Slf4j
+@Component
+@ConfigurationProperties(prefix = "device")
+public class DeviceDataQuery {
+    @Autowired
+    private DmpDeviceMapper dmpDeviceMapper;
+    @Value("${device.data.base-url}")
+    private String baseUrl;
+    @Value("${device.data.simulation}")
+    private boolean simulation;
+    private Map<String, String> deviceFieldMapping;
+
+    /**
+     * 获取指定设备类型的设备数据
+     */
+    public List<JSONObject> getDeviceData(Integer deviceType) {
+        List<String> deviceUuids = getDeviceUuids(deviceType);
+        if (deviceUuids.isEmpty()) {
+            log.warn("该租户下没有注册设备！");
+            return Collections.emptyList();
+        }
+
+        JSONObject requestBody = new JSONObject();
+        requestBody.put("deviceUuids", deviceUuids);
+
+        String response = HttpClientUtils.doPostJson(baseUrl, String.valueOf(requestBody));
+
+        List<JSONObject> resultList = parseResponseData(response, deviceType);
+
+        if (resultList.isEmpty() && simulation) {
+            resultList = generateSimulationData(deviceType, deviceUuids);
+        }
+
+        return resultList;
+
+    }
+
+    /**
+     * 获取指定设备类型的设备UUID列表
+     */
+    private List<String> getDeviceUuids(Integer deviceType) {
+        LambdaQueryWrapper<DmpDevice> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(DmpDevice::getTenantId, 1208)
+                .eq(deviceType != null, DmpDevice::getDeviceType, deviceType);
+        List<DmpDevice> dmpDevices = dmpDeviceMapper.selectList(queryWrapper);
+        return dmpDevices.stream().map(DmpDevice::getDeviceUuid).collect(Collectors.toList());
+    }
+
+    /**
+     * 解析接口响应数据，提取指定字段
+     */
+    private List<JSONObject> parseResponseData(String responseJson, Integer deviceType) {
+        List<JSONObject> resultList = new ArrayList<>();
+        if (responseJson == null) {
+            return resultList;
+        }
+
+        JSONObject responseObj = JSON.parseObject(responseJson);
+        if (!"SUCCESS".equals(responseObj.getString("status")) || !"0".equals(responseObj.getString("code"))) {
+            System.err.println("接口返回失败：" + responseObj.getString("msg"));
+            return resultList;
+        }
+
+        JSONArray dataArray = responseObj.getJSONArray("data");
+        if (dataArray == null || dataArray.isEmpty()) {
+            return resultList;
+        }
+
+        List<String> targetFields = getTargetFieldsByDeviceType(deviceType);
+
+        for (int i = 0; i < dataArray.size(); i++) {
+            JSONObject deviceData = dataArray.getJSONObject(i);
+            JSONObject metrics = deviceData.getJSONObject("metrics");
+            String deviceUuid = deviceData.getString("deviceuuid");
+
+            if (metrics == null) {
+                continue;
+            }
+
+            JSONObject targetData = new JSONObject();
+            boolean hasValidData = false;
+            for (String field : targetFields) {
+                Object value = metrics.get(field);
+                if (value != null) {
+                    targetData.put(field, value);
+                    hasValidData = true;
+                }
+            }
+
+            targetData.put("deviceuuid", deviceUuid);
+            if (hasValidData) {
+                resultList.add(targetData);
+            }
+        }
+
+        return resultList;
+    }
+
+    /**
+     * 根据设备类型获取目标字段（自动包含time）
+     */
+    private List<String> getTargetFieldsByDeviceType(Integer deviceType) {
+        String fieldsStr = deviceFieldMapping.get(deviceType.toString());
+        List<String> fields = Arrays.stream(fieldsStr.split(","))
+                .map(String::trim)
+                .collect(Collectors.toList());
+        if (!fields.contains("time")) {
+            fields.add("time");
+        }
+        return fields;
+    }
+
+    /**
+     * 生成模拟数据
+     */
+    private List<JSONObject> generateSimulationData(Integer deviceType, List<String> deviceUuids) {
+        List<JSONObject> simulationList = new ArrayList<>();
+        long currentTime = System.currentTimeMillis();
+
+        for (String deviceUuid : deviceUuids) {
+            JSONObject simulationData = new JSONObject();
+            simulationData.put("deviceuuid", deviceUuid);
+            simulationData.put("time", currentTime);
+
+            switch (deviceType) {
+                // 空气质量
+                case 701:
+                    simulationData.put("wd", new Random().nextInt(50) - 10); // 温度：-10~40℃
+                    simulationData.put("sd", new Random().nextInt(101)); // 湿度：0~100%
+                    simulationData.put("o2", new Random().nextDouble() * 21); // 氧气：0~21%
+                    simulationData.put("co", new Random().nextDouble() * 100); // 一氧化碳：0~100ppm
+                    simulationData.put("co2", new Random().nextDouble() * 2000); // 二氧化碳：0~2000ppm
+                    break;
+                // 水浸
+                case 702:
+                    // 渗漏状态：0-正常，1-渗漏
+                    simulationData.put("leach_status", new Random().nextInt(2));
+                    break;
+                // 人员统计
+                case 703:
+                    // 流量数据：模拟正数
+                    simulationData.put("amount_into", new Random().nextDouble() * 1000);
+                    simulationData.put("amount_out", new Random().nextDouble() * 1000);
+                    simulationData.put("day_into", new Random().nextDouble() * 10000);
+                    simulationData.put("day_out", new Random().nextDouble() * 10000);
+                    break;
+                // 电气火灾
+                case 704:
+                    // 电气参数：模拟合理范围
+                    simulationData.put("voltage_a", 220 + new Random().nextDouble() * 10); // 电压A：220~230V
+                    simulationData.put("voltage_b", 220 + new Random().nextDouble() * 10);
+                    simulationData.put("voltage_c", 220 + new Random().nextDouble() * 10);
+                    simulationData.put("current_a", new Random().nextDouble() * 50); // 电流A：0~50A
+                    simulationData.put("current_b", new Random().nextDouble() * 50);
+                    simulationData.put("current_c", new Random().nextDouble() * 50);
+                    simulationData.put("temperature_a", 20 + new Random().nextDouble() * 30); // 温度A：20~50℃
+                    simulationData.put("temperature_b", 20 + new Random().nextDouble() * 30);
+                    simulationData.put("temperature_c", 20 + new Random().nextDouble() * 30);
+                    simulationData.put("current_residual", new Random().nextDouble() * 1); // 剩余电流：0~1A
+                    break;
+                // 电能采集
+                case 705:
+                    // 电能：模拟正数
+                    simulationData.put("electrical_energy", new Random().nextDouble() * 10000);
+                    break;
+                default:
+                    break;
+            }
+
+            simulationList.add(simulationData);
+        }
+
+        System.out.println("生成模拟数据，设备类型：" + deviceType + "，数量：" + simulationList.size());
+        return simulationList;
+    }
+
+}

service-cdi/service-cdi-biz/src/main/resources/mapper.cdi/DmpDeviceMapper.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.usky.alarm.mapper.DmpDeviceMapper">
+
+    <!-- 通用查询映射结果 -->
+    <resultMap id="BaseResultMap" type="com.usky.cdi.domain.DmpDevice">
+        <id column="id" property="id" />
+        <result column="device_id" property="deviceId" />
+        <result column="device_name" property="deviceName" />
+        <result column="device_type" property="deviceType" />
+        <result column="product_id" property="productId" />
+        <result column="sim_code" property="simCode" />
+        <result column="imsi_code" property="imsiCode" />
+        <result column="subscribe_flag" property="subscribeFlag" />
+        <result column="node_type" property="nodeType" />
+        <result column="group_id" property="groupId" />
+        <result column="delete_flag" property="deleteFlag" />
+        <result column="created_by" property="createdBy" />
+        <result column="created_time" property="createdTime" />
+        <result column="updated_by" property="updatedBy" />
+        <result column="updated_time" property="updatedTime" />
+        <result column="tenant_id" property="tenantId" />
+        <result column="company_code" property="companyCode" />
+        <result column="install_address" property="installAddress" />
+        <result column="service_status" property="serviceStatus" />
+        <result column="product_code" property="productCode" />
+        <result column="device_uuid" property="deviceUuid" />
+    </resultMap>
+
+</mapper>