新增 性诱测报III设备对接

src/main/java/com/yunfeiyun/agmp/iots/core/mqtt/DeviceTopicService.java

@@ -52,6 +52,13 @@ public class DeviceTopicService {
         return deviceIds;
     }
 
+    private String[] getYfXycbIIIBatchSubTopic(String[] deviceId) {
+        String[] topicArray = {
+                IotMqttConstant.YFXycbIIITopic.TOPIC_XYCB_III_REPORT_PREFIX
+        };
+        return getTopics(deviceId, topicArray);
+    }
+
 
     /**
      * 根据实现类获取topic
@@ -67,33 +74,28 @@ public class DeviceTopicService {
             deviceId[i]=mqttTopicValues.get(i).getDevCode();
         }
         switch (serviceName) {
-            case ServiceNameConst.SERVICE_YF_CBD: {
+            case ServiceNameConst.SERVICE_YF_CBD:
                 return getYfCbdBatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_BY_SF: {
+            case ServiceNameConst.SERVICE_BY_SF:
                 return getBySfBatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_YF_SCD: {
+            case ServiceNameConst.SERVICE_YF_SCD:
                 return getYfScdBatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_YF_BZY: {
+            case ServiceNameConst.SERVICE_YF_BZY:
                 return getYfBzyBatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_YF_QXZ: {
+            case ServiceNameConst.SERVICE_YF_QXZ:
                 return getYfQxzBatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_YF_XYCB_2: {
+            case ServiceNameConst.SERVICE_YF_XYCB_2:
                 return getYfXycb2BatchSubTopic(deviceId);
-            }
-            case ServiceNameConst.SERVICE_EZVIZ_MINITOR: {
+            case ServiceNameConst.SERVICE_EZVIZ_MINITOR:
                 return getHKBatchSubTopic();
-            }
             case ServiceNameConst.SERVICE_XPH_YF_QXZ:
                 return getXphDeviceBatchSubTopic(deviceId);
             case ServiceNameConst.SERVICE_HPF_ZNKG:
                 return getHpfZnkgDeviceBatchSubTopic(deviceId);
             case ServiceNameConst.SERVICE_YR_SF_SERVICE:
                 return getYrSfDeviceBatchSubTopic(deviceId);
+            case ServiceNameConst.SERVICE_YF_XYCB_III:
+                return getYfXycbIIIBatchSubTopic(deviceId);
             default: {
                 throw new IotBizException(IotErrorCode.FAILURE.getCode(), serviceName + "不存在对应topic 解析");
             }

src/main/java/com/yunfeiyun/agmp/iots/device/service/IIotYfXycbIIIdataService.java

@@ -0,0 +1,15 @@
+package com.yunfeiyun.agmp.iots.device.service;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
+import com.yunfeiyun.agmp.iot.common.domain.IotYfXycbIIIdata;
+
+import java.util.Map;
+
+public interface IIotYfXycbIIIdataService {
+    public void insertXycbIIIdata(IotDevice iotDevice, JSONObject jsonObject);
+
+    public IotYfXycbIIIdata selectIotYfXycbIIIdataByDevBid(String devBid);
+
+    public void updateIotYfXycbIIIdata(Map<String, Object> updateField);
+}

src/main/java/com/yunfeiyun/agmp/iots/device/service/IYfXycbIIIDevice.java

@@ -0,0 +1,8 @@
+package com.yunfeiyun.agmp.iots.device.service;
+
+import com.yunfeiyun.agmp.iots.device.common.Device;
+
+/** 宁录性诱设备 */
+public interface IYfXycbIIIDevice extends Device {
+
+}

src/main/java/com/yunfeiyun/agmp/iots/device/serviceImp/IotXycbIIIdataServiceImpl.java

@@ -0,0 +1,58 @@
+package com.yunfeiyun.agmp.iots.device.serviceImp;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
+import com.yunfeiyun.agmp.iot.common.domain.IotYfXycbIIIdata;
+import com.yunfeiyun.agmp.iot.common.service.MongoService;
+import com.yunfeiyun.agmp.iots.device.service.IIotYfXycbIIIdataService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Service
+public class IotXycbIIIdataServiceImpl implements IIotYfXycbIIIdataService {
+    @Autowired
+    private MongoService mongoService;
+
+
+
+    /**
+     * @param iotDevice
+     * @param jsonObject
+     */
+    @Override
+    public void insertXycbIIIdata(IotDevice iotDevice, JSONObject jsonObject) {
+        IotYfXycbIIIdata iotYfXycb2data = new IotYfXycbIIIdata();
+        iotYfXycb2data.setCId(iotDevice.getTid());
+        iotYfXycb2data.setXycbdataBid(iotYfXycb2data.getUUId());
+        iotYfXycb2data.setDevBid(iotDevice.getDevBid());
+        iotYfXycb2data.setXycbdataCreatedDate(iotDevice.getDevUpdateddate());
+        iotYfXycb2data.setXycbdataContent(jsonObject);
+
+        mongoService.saveOne(iotYfXycb2data);
+    }
+
+    /**
+     * @param devBid
+     * @return
+     */
+    @Override
+    public IotYfXycbIIIdata selectIotYfXycbIIIdataByDevBid(String devBid) {
+        Map<String, Object> map = new HashMap<>();
+        map.put("devBid", devBid);
+        IotYfXycbIIIdata iotYfXycb2data = (IotYfXycbIIIdata) mongoService.findOne(
+                IotYfXycbIIIdata.class, map, "xycbdataCreatedDate", "desc"
+        );
+        return iotYfXycb2data;
+    }
+
+    /**
+     * @param updateField
+     */
+    @Override
+    public void updateIotYfXycbIIIdata(Map<String, Object> updateField) {
+        mongoService.update(IotYfXycbIIIdata.class, updateField);
+    }
+}

src/main/java/com/yunfeiyun/agmp/iots/device/serviceImp/YfXycbIIIDeviceImpl.java

@@ -0,0 +1,242 @@
+package com.yunfeiyun.agmp.iots.device.serviceImp;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.yunfeiyun.agmp.common.utils.DateUtils;
+import com.yunfeiyun.agmp.common.utils.JSONUtils;
+import com.yunfeiyun.agmp.common.utils.StringUtils;
+import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceDictEnum;
+import com.yunfeiyun.agmp.iot.common.constant.devicetype.ServiceNameConst;
+import com.yunfeiyun.agmp.iot.common.constant.mqtt.IotMqttConstant;
+import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
+import com.yunfeiyun.agmp.iot.common.enums.IotDeviceStatusTypeEnum;
+import com.yunfeiyun.agmp.iots.core.manager.MqttManager;
+import com.yunfeiyun.agmp.iots.device.common.DeviceAbstractImpl;
+import com.yunfeiyun.agmp.iots.device.service.IIotYfXycbIIIdataService;
+import com.yunfeiyun.agmp.iots.device.service.IYfXycbIIIDevice;
+import com.yunfeiyun.agmp.iots.service.IIotDeviceService;
+import com.yunfeiyun.agmp.iots.service.IIotDeviceconfigService;
+import com.yunfeiyun.agmp.iots.service.IIotDevicelasteddataService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.util.TextUtils;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 云飞性诱 III：不带拍照
+ */
+@Component(ServiceNameConst.SERVICE_YF_XYCB_III)
+@Slf4j
+public class YfXycbIIIDeviceImpl extends DeviceAbstractImpl implements IYfXycbIIIDevice {
+    private final String tag = IotDeviceDictEnum.TYPE_YF_XYCB_III.getName();
+
+    @Autowired
+    private MqttManager mqttManager;
+
+    @Autowired
+    private IIotDeviceService iIotDeviceService;
+
+    @Autowired
+    private IIotDeviceconfigService iIotDeviceconfigService;
+
+    @Autowired
+    private IIotYfXycbIIIdataService iIotYfXycbIIIdataService;
+
+    @Autowired
+    private IIotDevicelasteddataService iIotDevicelasteddataService;
+
+    /**
+     * 根据topic、设备发来的消息，查询对应设备实体
+     *
+     * @param topic
+     * @param jobjMsg
+     * @return
+     */
+    @Override
+    public IotDevice findIotDevice(String topic, JSONObject jobjMsg,String connectionId) {
+        String devId = mqttManager.getDevIdByTopic(connectionId,topic);
+        return iIotDeviceService.selectIotDeviceByDevBid(devId);
+    }
+
+    public void cmdData(JSONObject dataJson, String topic, String connectionId, String devUpdateddate) throws Exception {
+        log.info("【{}】数据解析 {}，topic:{}", tag, dataJson.toString(),topic);
+
+        IotDevice oldIotDevice = findIotDevice(topic, dataJson, connectionId);
+        if (oldIotDevice == null) {
+            log.error("未取到 iotDevice");
+            return;
+        }
+
+        IotDevice newIotDevice = new IotDevice();
+        newIotDevice.setDevBid(oldIotDevice.getDevBid());
+        newIotDevice.setDevVersion("YF-XY-III");
+        newIotDevice.setDevPositionstatus("0");
+        newIotDevice.setDevPositiontype("0");
+        newIotDevice.setDevUpdateddate(devUpdateddate);
+        newIotDevice.setDevStatus(IotDeviceStatusTypeEnum.ONLINE.getCode());
+
+        String[] keyArrays = {
+                "ts",
+                "st",
+                "et",
+                "ds",
+                "clt_t",
+                "dat_f"
+        };
+
+        JSONObject extConf = new JSONObject();
+        for (String k : keyArrays) {
+            String v = "0";
+            if (dataJson.containsKey(k)) {
+                v = dataJson.getString(k);
+            }
+            extConf.put(k, v);
+        }
+        String devConfig = JSONUtils.toJSONString(extConf);
+
+        // 更新设备基础信息数据库 mysql
+        iIotDeviceService.updateIotDevice(newIotDevice);
+        // 创建或更新设备配置信息
+        if (StringUtils.isNotEmpty(devConfig)) {
+            iIotDeviceconfigService.createOrUpdateDevConfig(oldIotDevice, devConfig, devUpdateddate);
+        }
+
+        oldIotDevice.setDevUpdateddate(devUpdateddate);
+        // 更新设备数据到mongodb
+        iIotYfXycbIIIdataService.insertXycbIIIdata(oldIotDevice, dataJson);
+
+        // 保存 设备最新数据 到redis
+        iIotDevicelasteddataService.createOrUpdateDeviceLastedData(
+                dataJson, oldIotDevice, devUpdateddate, 60 * 60 * 24L);
+
+    }
+
+    public void cmdDataConfig(JSONObject dataJson, String topic, String connectionId, String devUpdateddate) throws Exception {
+        log.info("【{}】配置 数据解析 {}，topic:{}", tag, dataJson.toString(),topic);
+
+        IotDevice oldIotDevice = findIotDevice(topic, dataJson, connectionId);
+        if (oldIotDevice == null) {
+            log.error("未取到 iotDevice");
+            return;
+        }
+
+        JSONObject transformedObject = new JSONObject();
+
+        // 定义默认值
+        String defaultValue = "0"; // 可以根据需要调整默认值
+
+        // 提取并转换字段
+        if (dataJson.containsKey("work_tim")) {
+            JSONObject workTim = dataJson.getJSONObject("work_tim");
+            transformedObject.put("timctrl", workTim.getOrDefault("timctrl", defaultValue));
+            transformedObject.put("st", workTim.getOrDefault("st", defaultValue));
+            transformedObject.put("et", workTim.getOrDefault("st", defaultValue)); // 注意这里将et值改为10
+        }
+
+        if (dataJson.containsKey("data_tim")) {
+            JSONObject dataTim = dataJson.getJSONObject("data_tim");
+            transformedObject.put("dat_f",  dataTim.getOrDefault("dat_f", defaultValue)); // 注意这里将dat_f值改为1
+        }
+
+        if (dataJson.containsKey("power")) {
+            JSONObject power = dataJson.getJSONObject("power");
+            transformedObject.put("ds", power.getOrDefault("ds", defaultValue));
+        }
+
+        if (dataJson.containsKey("clt_tim")) {
+            JSONObject cltTim = dataJson.getJSONObject("clt_tim");
+            transformedObject.put("clt_t",  cltTim.getOrDefault("clt_t", defaultValue)); // 注意这里将clt_t值改为10
+        }
+
+        IotDevice newIotDevice = new IotDevice();
+        newIotDevice.setDevBid(oldIotDevice.getDevBid());
+        newIotDevice.setDevStatus(IotDeviceStatusTypeEnum.ONLINE.getCode());
+
+        // 更新设备基础信息数据库 mysql
+        iIotDeviceService.updateIotDevice(newIotDevice);
+        // 创建或更新设备配置信息
+        iIotDeviceconfigService.createOrUpdateDevConfig(oldIotDevice, transformedObject.toString(), devUpdateddate);
+
+    }
+
+    private void publish(IotDevice iotDevice, String mqttMsgContent) {
+        String devCode = iotDevice.getDevCode();
+        List<String> topicList = new ArrayList<>();
+        topicList.add(IotMqttConstant.YFXycbIIITopic.TOPIC_XYCB_III_CMD_PREFIX + devCode);
+        for(String topic:topicList){
+            try{
+                mqttManager.publishMsg(iotDevice.getDevconnBid(), topic, mqttMsgContent);
+                log.info("【{}】发送指令完毕！connectionId:{},topic :{} mqttMsgContent: {}", tag, iotDevice.getDevconnBid(),topic, mqttMsgContent);
+            }catch (Exception e){
+                log.error("【{}】发送指令失败！connectionId:{},topic :{} mqttMsgContent: {}", tag, iotDevice.getDevconnBid(),topic, mqttMsgContent);
+            }
+        }
+    }
+
+    public void cmdOffline(JSONObject dataJson, String topic, String connectionId, String devUpdateddate) throws MqttException {
+        log.info("【{}】离线 数据解析 {}，topic:{}", tag, dataJson.toString(),topic);
+        IotDevice oldIotDevice = findIotDevice(topic, dataJson, connectionId);
+        if (oldIotDevice == null) {
+            log.error("未取到 iotDevice");
+            return;
+        }
+
+        IotDevice newIotDevice = new IotDevice();
+        newIotDevice.setDevBid(oldIotDevice.getDevBid());
+        newIotDevice.setDevStatus("0");
+        newIotDevice.setDevOfflinedate(devUpdateddate);
+        iIotDeviceService.updateIotDevice(newIotDevice);
+
+        /**
+         * 下发刷新指令，检测设备是否真离线
+         */
+        JSONObject payload = new JSONObject();
+        payload.put("cmd", "read");
+        payload.put("ext", "data");
+        String mqttMsgContent = JSONUtils.toJSONString(payload);
+        publish(oldIotDevice, mqttMsgContent);
+
+        log.info("[{}] 下发刷新指令，检测设备是否真离线: {}", tag, oldIotDevice);
+    }
+
+
+    /**
+     * 接收上报数据
+     *
+     * @param topic
+     * @param dataJson
+     * @return
+     */
+    @Override
+    public Object receiveData(String topic, JSONObject dataJson, String connectionId) throws Exception {
+        log.info("【{}】 处理收到的 设备上报数据 {}", tag, dataJson.toString());
+        // 接收设备上报数据后的处理逻辑
+        String cmd = dataJson.getString("cmd");
+        if (TextUtils.isEmpty(cmd)) {
+            log.error("未取到cmd");
+            return false;
+        }
+        JSONObject ext = dataJson.getJSONObject("ext");
+        String devUpdateddate = dataJson.getString("devUpdateddate");
+        if(StringUtils.isEmpty(devUpdateddate)){
+            devUpdateddate= DateUtils.dateTimeNow();
+        }
+        if (ext == null) {
+            log.error("未取到ext");
+            return false;
+        }
+
+        if ("data".equals(cmd)) {
+            this.cmdData(ext, topic, connectionId, devUpdateddate);
+        } else if ("offline".equals(cmd)) {
+            this.cmdOffline(ext, topic, connectionId, devUpdateddate);
+        }
+//        else if ("paramconf".equals(cmd)){
+//            this.cmdDataConfig(ext, topic, connectionId, devUpdateddate);
+//        }
+        return null;
+    }
+}