治理设备新增功能

异常处理

iot-access/access-api/src/main/java/com/njcn/access/pojo/RspDataDto.java

@@ -82,6 +82,10 @@ public class RspDataDto {
 
         @SerializedName("Capacity_A")
         private Double capacityA;
+
+        @SerializedName("StatCycle")
+        @ApiModelProperty("接线方式")
+        private Integer StatCycle;
     }
 
     /**

iot-access/access-api/src/main/java/com/njcn/access/pojo/dto/AutoDataDto.java

@@ -1,6 +1,7 @@
 package com.njcn.access.pojo.dto;
 
 import com.alibaba.nacos.shaded.com.google.gson.annotations.SerializedName;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
@@ -17,39 +18,49 @@ import java.util.List;
 public class AutoDataDto {
 
     @SerializedName("Mid")
+    @JsonProperty("Mid")
     private Integer mid;
 
     @SerializedName("Did")
+    @JsonProperty("Did")
     @ApiModelProperty("逻辑设备 治理逻辑设备为1 电能质量设备为2")
     private Integer did;
 
     @SerializedName("Pri")
+    @JsonProperty("Pri")
     private Integer pri;
 
     @SerializedName("Type")
+    @JsonProperty("Type")
     private Integer type;
 
     @SerializedName("Msg")
+    @JsonProperty("Msg")
     private Msg msg;
 
     @Data
     public static class Msg{
 
         @SerializedName("Cldid")
+        @JsonProperty("Cldid")
         @ApiModelProperty("逻辑子设备 治理逻辑设备为0 电能质量设备为1、2")
         private Integer clDid;
 
         @SerializedName("DataType")
+        @JsonProperty("DataType")
         private Integer dataType;
 
         @SerializedName("DataAttr")
+        @JsonProperty("DataAttr")
         @ApiModelProperty("数据属性：无-0、实时-1、统计-2")
         private Integer dataAttr;
 
         @SerializedName("DsNameIdx")
+        @JsonProperty("DsNameIdx")
         private Integer dsNameIdx;
 
         @SerializedName("DataArray")
+        @JsonProperty("DataArray")
         private List<DataArray> dataArray;
     }
 
@@ -57,95 +68,119 @@ public class AutoDataDto {
     public static class DataArray{
 
         @SerializedName("DataAttr")
+        @JsonProperty("DataAttr")
         @ApiModelProperty("数据属性 -1-无 0-Rt(实时) 1-Max 2-Min 3-Avg 4-Cp95")
         private Integer dataAttr;
 
         @SerializedName("DataTimeSec")
+        @JsonProperty("DataTimeSec")
         private Long dataTimeSec;
 
         @SerializedName("DataTimeUSec")
+        @JsonProperty("DataTimeUSec")
         private Integer dataTimeUSec;
 
         @SerializedName("DataTag")
+        @JsonProperty("DataTag")
         @ApiModelProperty("数据是否参与合格率统计")
         private Integer dataTag;
 
         @SerializedName("Code")
+        @JsonProperty("Code")
         @ApiModelProperty("事件码")
-        private Integer code;
+        private String code;
 
         @SerializedName("Data")
+        @JsonProperty("Data")
         private String data;
 
         @SerializedName("PrjName")
+        @JsonProperty("PrjName")
         @ApiModelProperty("工程名称")
         private String prjName;
 
         @SerializedName("PrjTimeStart")
+        @JsonProperty("PrjTimeStart")
         @ApiModelProperty("装置启动时间")
         private Long prjTimeStart;
 
         @SerializedName("PrjTimeEnd")
+        @JsonProperty("PrjTimeEnd")
         @ApiModelProperty("装置结束时间")
         private Long prjTimeEnd;
 
         @SerializedName("PrjDataPath")
+        @JsonProperty("PrjDataPath")
         @ApiModelProperty("装置数据路径")
         private String prjDataPath;
 
         @SerializedName("DevType")
+        @JsonProperty("DevType")
         @ApiModelProperty("装置型号")
         private String devType;
 
         @SerializedName("DevMac")
+        @JsonProperty("DevMac")
         @ApiModelProperty("装置mac地址")
         private String devMac;
 
         @SerializedName("AppVersion")
+        @JsonProperty("AppVersion")
         @ApiModelProperty("装置程序版本")
         private String appVersion;
 
         @SerializedName("Cldid")
+        @JsonProperty("Cldid")
         @ApiModelProperty("逻辑子设备id")
         private Integer clDid;
 
         @SerializedName("StatCycle")
+        @JsonProperty("StatCycle")
         @ApiModelProperty("统计间隔")
         private Integer statCycle;
 
         @SerializedName("VolGrade")
+        @JsonProperty("VolGrade")
         @ApiModelProperty("电压等级")
         private Float volGrade;
 
         @SerializedName("VolConType")
+        @JsonProperty("VolConType")
         @ApiModelProperty("电压接线方式（0-星型, 1-角型, 2-V型）")
         private Integer volConType;
 
         @SerializedName("CurConSel")
+        @JsonProperty("CurConSel")
         @ApiModelProperty("电流接线方式（0-正常, 1-合成IB, 2-合成IC）")
         private Integer curConSel;
 
         @SerializedName("PtRatio")
+        @JsonProperty("PtRatio")
         @ApiModelProperty("PT变比")
         private Integer ptRatio;
 
         @SerializedName("CtRatio")
+        @JsonProperty("CtRatio")
         @ApiModelProperty("ct变比")
         private Integer ctRatio;
 
         @SerializedName("CapacitySscb")
+        @JsonProperty("CapacitySscb")
         @ApiModelProperty("基准短路容量")
         private Float capacitySscb;
 
         @SerializedName("CapacitySscmin")
+        @JsonProperty("CapacitySscmin")
         @ApiModelProperty("最小短路容量")
         private Float capacitySscmin;
 
         @SerializedName("CapacitySt")
+        @JsonProperty("CapacitySt")
         @ApiModelProperty("供电设备容量")
         private Float capacitySt;
 
         @SerializedName("CapacitySi")
+        @JsonProperty("CapacitySi")
         @ApiModelProperty("用户协议容量")
         private Float capacitySi;
     }

iot-access/access-api/src/main/java/com/njcn/access/pojo/po/CsLineRunData.java

@@ -0,0 +1,48 @@
+package com.njcn.access.pojo.po;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * <p>
+ * 治理设备模块运行状态记录表
+ * </p>
+ *
+ * @author xy
+ * @since 2025-06-26
+ */
+@Getter
+@Setter
+@TableName("cs_line_run_data")
+public class CsLineRunData implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 监测点id
+     */
+    private String lineId;
+
+    /**
+     * 子模块编号id(没有子模块，则为0)
+     */
+    private Integer moduleId;
+
+    /**
+     * 最新数据时间
+     */
+    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private LocalDateTime timeId;
+
+    /**
+     * 子模块通讯状态(0：离线 1：连接)
+     */
+    private Integer runState;
+
+
+}

iot-access/access-boot/pom.xml

@@ -20,6 +20,12 @@
     </properties>
 
     <dependencies>
+
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
         <dependency>
             <groupId>com.njcn</groupId>
             <artifactId>access-api</artifactId>

iot-access/access-boot/src/main/java/com/njcn/access/controller/CsLineRunDataController.java

@@ -0,0 +1,52 @@
+package com.njcn.access.controller;
+
+import com.njcn.access.pojo.po.CsLineRunData;
+import com.njcn.access.service.ICsLineRunDataService;
+import com.njcn.common.pojo.annotation.OperateInfo;
+import com.njcn.common.pojo.enums.common.LogEnum;
+import com.njcn.common.pojo.enums.response.CommonResponseEnum;
+import com.njcn.common.pojo.response.HttpResult;
+import com.njcn.common.utils.HttpResultUtil;
+import com.njcn.web.controller.BaseController;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 治理设备模块运行状态记录表 前端控制器
+ * </p>
+ *
+ * @author xy
+ * @since 2025-06-26
+ */
+@RestController
+@RequestMapping("/csLineRunData")
+@Slf4j
+@Api(tags = "治理设备数据运行记录")
+@AllArgsConstructor
+public class CsLineRunDataController extends BaseController {
+
+    private final ICsLineRunDataService csLineRunDataService;
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/add")
+    @ApiOperation("新增数据")
+    @ApiImplicitParam(name = "list", value = "参数", required = true)
+    public HttpResult<String> addData(@RequestBody @Validated List<CsLineRunData> list){
+        String methodDescribe = getMethodDescribe("addData");
+        csLineRunDataService.addData(list);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
+
+}
+

iot-access/access-boot/src/main/java/com/njcn/access/handler/MqttMessageHandler.java

@@ -1,6 +1,7 @@
 package com.njcn.access.handler;
 
 import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.util.IdUtil;
 import com.alibaba.csp.sentinel.util.StringUtil;
 import com.alibaba.excel.util.CollectionUtils;
@@ -20,7 +21,6 @@ import com.njcn.access.pojo.dto.*;
 import com.njcn.access.pojo.dto.file.FileDto;
 import com.njcn.access.pojo.dto.file.FileRedisDto;
 import com.njcn.access.pojo.param.ReqAndResParam;
-import com.njcn.access.pojo.po.CsDeviceOnlineLogs;
 import com.njcn.access.pojo.po.CsLineModel;
 import com.njcn.access.pojo.po.CsTopic;
 import com.njcn.access.service.ICsDeviceOnlineLogsService;
@@ -31,6 +31,7 @@ import com.njcn.access.utils.ChannelObjectUtil;
 import com.njcn.common.pojo.dto.DeviceLogDTO;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.*;
+import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
 import com.njcn.csdevice.pojo.param.CsLineParam;
 import com.njcn.csdevice.pojo.po.*;
 import com.njcn.device.biz.pojo.po.Overlimit;
@@ -99,6 +100,7 @@ public class MqttMessageHandler {
     private final ChannelObjectUtil channelObjectUtil;
     private final WaveFeignClient waveFeignClient;
     private final RtFeignClient rtFeignClient;
+    private final CsCommunicateFeignClient csCommunicateFeignClient;
     @Autowired
     Validator validator;
 
@@ -311,34 +313,24 @@ public class MqttMessageHandler {
                         csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.ACCESS.getCode());
                         csEquipmentDeliveryService.updateRunStatusBynDid(nDid,AccessEnum.ONLINE.getCode());
                         //记录设备上线
-                        CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
+                        PqsCommunicateDto dto = new PqsCommunicateDto();
-                        CsDeviceOnlineLogs csDeviceOnlineLogs = new CsDeviceOnlineLogs();
+                        dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
-                        if(Objects.isNull(record)) {
+                        dto.setDevId(nDid);
-                            csDeviceOnlineLogs.setNdid(nDid);
+                        dto.setType(1);
-                            csDeviceOnlineLogs.setOnlineTime(LocalDateTime.now());
+                        dto.setDescription("通讯正常");
-                            onlineLogsService.save(csDeviceOnlineLogs);
+                        csCommunicateFeignClient.insertion(dto);
-                        } else {
-                            LocalDateTime time = record.getOfflineTime();
-                            if (!Objects.isNull(time)){
-                               csDeviceOnlineLogs.setNdid(nDid);
-                                csDeviceOnlineLogs.setOnlineTime(LocalDateTime.now());
-                                onlineLogsService.save(csDeviceOnlineLogs);
-                           } else {
-                                record.setOnlineTime(LocalDateTime.now());
-                                onlineLogsService.updateById(record);
-                            }
-                        }
-                        //接入后系统重置装置心跳
-                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
-                        //修改redis的mid
-                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
-                        redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L);
                         //询问设备软件信息
                         askDevData(nDid,version,1,mid);
                         //更新治理监测点信息和设备容量
                         askDevData(nDid,version,2,(res.getMid()+1));
                         //更新电网侧、负载侧监测点信息
                         askDevData(nDid,version,3,(res.getMid()+1));
+                        //接入后系统重置装置心跳
+                        redisUtil.saveByKeyWithExpire("MQTT:" + nDid, Instant.now().toEpochMilli(),180L);
+                        //修改redis的mid
+                        redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
+                        //接入成功标识
+                        redisUtil.saveByKeyWithExpire("online" + nDid,"online",10L);
                         //录波任务倒计时
                         redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
                     } else {
@@ -406,6 +398,7 @@ public class MqttMessageHandler {
                                             csLineParam.setPtRatio(item.getPtRatio());
                                             csLineParam.setCtRatio(item.getCtRatio());
                                             csLineParam.setConType(item.getConType());
+                                            csLineParam.setLineInterval(item.getStatCycle());
                                             csLineFeignClient.updateLine(csLineParam);
                                             //生成监测点限值
                                             Overlimit overlimit = COverlimitUtil.globalAssemble(item.getVolGrade().floatValue(),10f,10f,10f,0,0);
@@ -429,11 +422,19 @@ public class MqttMessageHandler {
                                             csLineParam.setPtRatio(item.getPtRatio());
                                             csLineParam.setCtRatio(item.getCtRatio());
                                             csLineParam.setConType(item.getConType());
+                                            csLineParam.setLineInterval(item.getStatCycle());
                                             csLineFeignClient.updateLine(csLineParam);
                                         });
                                     }
                                 }
                                 break;
+                            case 15:
+                                log.info("{}模块{}:处理实时数据", nDid, rspDataDto.getClDid());
+                                JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(res));
+                                AppAutoDataMessage appAutoDataMessage = JSONObject.toJavaObject(jsonObject, AppAutoDataMessage.class);
+                                appAutoDataMessage.setId(nDid);
+                                rtFeignClient.apfRtAnalysis(appAutoDataMessage);
+                                break;
                             case 48:
                                 log.info("询问装置项目列表");
                                 logDto.setUserName("运维管理员");
@@ -458,10 +459,16 @@ public class MqttMessageHandler {
                     break;
             }
             //csLogsFeignClient.addUserLog(logDto);
+        } else {
+            String result = getEnum(res.getCode());
+            log.info(result);
+            logDto.setResult(0);
+            logDto.setFailReason(result);
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(result);
         }
     }
 
-
     /**
      * 装置心跳 && 主动数据上送
      * fixme 这边由于接收文件数据时间跨度会很长，途中有其他请求进来会中断之前的程序，目前是记录中断的位置，等处理完成再继续请求接收文件
@@ -739,4 +746,41 @@ public class MqttMessageHandler {
         publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(reqAndResParam),1,false);
     }
 
+    public String getEnum(Integer code) {
+        String result = null;
+        switch (code) {
+            case 201:
+                result = AccessEnum.START_CHANNEL.getMessage();
+                break;
+            case 202:
+                result = AccessEnum.WAIT_CHANNEL.getMessage();
+                break;
+            case 400:
+                result = AccessEnum.FAIL.getMessage();
+                break;
+            case 401:
+                result = AccessEnum.ERROR.getMessage();
+                break;
+            case 402:
+                result = AccessEnum.REFUSE_WAIT.getMessage();
+                break;
+            case 403:
+                result = AccessEnum.REFUSE_UNKNOWN.getMessage();
+                break;
+            case 404:
+                result = AccessEnum.NOT_FIND.getMessage();
+                break;
+            case 405:
+                result = AccessEnum.BUSY.getMessage();
+                break;
+            case 406:
+                result = AccessEnum.TIME_OUT.getMessage();
+                break;
+            default:
+                result = AccessEnum.OTHER_ERROR.getMessage();
+                break;
+        }
+        return result;
+    }
+
 }

iot-access/access-boot/src/main/java/com/njcn/access/listener/RedisKeyExpirationListener.java

@@ -1,5 +1,6 @@
 package com.njcn.access.listener;
 
+import cn.hutool.core.date.DatePattern;
 import com.njcn.access.enums.AccessEnum;
 import com.njcn.access.pojo.dto.NoticeUserDto;
 import com.njcn.access.pojo.po.CsDeviceOnlineLogs;
@@ -10,11 +11,9 @@ import com.njcn.access.service.impl.CsDeviceServiceImpl;
 import com.njcn.access.utils.MqttUtil;
 import com.njcn.access.utils.SendMessageUtil;
 import com.njcn.common.pojo.dto.DeviceLogDTO;
-import com.njcn.csdevice.api.CsDeviceUserFeignClient;
+import com.njcn.csdevice.api.*;
-import com.njcn.csdevice.api.CsLedgerFeignClient;
-import com.njcn.csdevice.api.CsLogsFeignClient;
-import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.pojo.dto.DevDetailDTO;
+import com.njcn.csdevice.pojo.dto.PqsCommunicateDto;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
@@ -33,7 +32,6 @@ import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -76,6 +74,9 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
     private SendMessageUtil sendMessageUtil;
     @Resource
     private CsDeviceServiceImpl csDeviceServiceImpl;
+    @Resource
+    private CsCommunicateFeignClient csCommunicateFeignClient;
+
     private final Object lock = new Object();
 
     public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
@@ -98,78 +99,100 @@ public class RedisKeyExpirationListener extends KeyExpirationEventMessageListene
         String expiredKey = message.toString();
         if(expiredKey.startsWith("MQTT:")){
             String nDid = expiredKey.split(":")[1];
-            String version = csTopicService.getVersion(nDid);
+            executeMainTask(nDid);
-            ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
-            executeMainTask(scheduler,nDid,version);
         }
     }
 
     //主任务
     //1.装置心跳断连
     //2.MQTT客户端不在线
-    private void executeMainTask(ScheduledExecutorService scheduler, String nDid, String version) {
+    private void executeMainTask(String nDid) {
         log.info("{}->装置离线", nDid);
         DeviceLogDTO logDto = new DeviceLogDTO();
         logDto.setUserName("运维管理员");
         logDto.setLoginName("njcnyw");
-        //判断mqtt
+        //装置下线
-        String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+        csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
-        boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+        //装置调整为注册状态
-        //心跳异常，但是客户端在线，则发送接入请求
+        csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
-        if (mqttClient) {
+        logDto.setOperate(nDid +"装置离线");
-            csDeviceService.devAccessAskTemplate(nDid,version,1);
+        sendMessage(nDid);
-            redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
+        //记录装置掉线时间
-            try {
+//        CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
-                Thread.sleep(5000);
+//        record.setOfflineTime(LocalDateTime.now());
-                Object object = redisUtil.getObjectByKey("online" + nDid);
+//        onlineLogsService.updateById(record);
-                if (Objects.nonNull(object)) {
+        //记录装置掉线时间
-                    scheduler.shutdown();
+        PqsCommunicateDto dto = new PqsCommunicateDto();
-                    logDto.setOperate(nDid + "客户端在线重连成功");
+        dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));
-                } else {
+        dto.setDevId(nDid);
-                    //装置下线
+        dto.setType(0);
-                    csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
+        dto.setDescription("通讯中断");
-                    //装置调整为注册状态
+        csCommunicateFeignClient.insertion(dto);
-                    csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
+        csLogsFeignClient.addUserLog(logDto);
-                    //startScheduledTask(scheduler,nDid,version);
-                    //logDto.setOperate(nDid +"客户端离线进入定时任务");
-                    logDto.setOperate(nDid +"装置离线");
-                    sendMessage(nDid);
-
-                    //记录装置掉线时间
-                    CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
-                    record.setOfflineTime(LocalDateTime.now());
-                    onlineLogsService.updateById(record);
-
-                    scheduler.shutdown();
-                }
-            } catch (InterruptedException e) {
-                scheduler.shutdown();
-                throw new RuntimeException(e);
-            }
-            csLogsFeignClient.addUserLog(logDto);
-        }
-        //客户端不在线则修改装置状态，进入定时任务
-        else {
-            //装置下线
-            csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
-            //装置调整为注册状态
-            csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
-            logDto.setOperate(nDid +"主任务执行失败，装置下线");
-            //logDto.setOperate(nDid +"主任务执行失败，装置下线，进入定时任务");
-            csLogsFeignClient.addUserLog(logDto);
-            //log.info("客户端离线进入定时任务...");
-            //startScheduledTask(scheduler,nDid,version);
-            sendMessage(nDid);
-
-            //记录装置掉线时间
-            CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
-            record.setOfflineTime(LocalDateTime.now());
-            onlineLogsService.updateById(record);
-
-            scheduler.shutdown();
-        }
     }
 
+    //主任务
+    //1.装置心跳断连
+    //2.MQTT客户端不在线
+//    private void executeMainTask(String nDid, String version) {
+//        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+//        log.info("{}->装置离线", nDid);
+//        DeviceLogDTO logDto = new DeviceLogDTO();
+//        logDto.setUserName("运维管理员");
+//        logDto.setLoginName("njcnyw");
+//        //判断mqtt
+//        String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+//        boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+//        //心跳异常，但是客户端在线，则发送接入请求
+//        if (mqttClient) {
+//            csDeviceService.devAccessAskTemplate(nDid,version,1);
+//            redisUtil.saveByKey(AppRedisKey.DEVICE_MID + nDid,1);
+//            try {
+//                Thread.sleep(5000);
+//                Object object = redisUtil.getObjectByKey("online" + nDid);
+//                if (Objects.nonNull(object)) {
+//                    scheduler.shutdown();
+//                    logDto.setOperate(nDid + "客户端在线重连成功");
+//                } else {
+//                    //装置下线
+//                    csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
+//                    //装置调整为注册状态
+//                    csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
+//                    logDto.setOperate(nDid +"装置离线");
+//                    sendMessage(nDid);
+//
+//                    //记录装置掉线时间
+//                    CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
+//                    record.setOfflineTime(LocalDateTime.now());
+//                    onlineLogsService.updateById(record);
+//
+//                    scheduler.shutdown();
+//                }
+//            } catch (InterruptedException e) {
+//                scheduler.shutdown();
+//                throw new RuntimeException(e);
+//            }
+//            csLogsFeignClient.addUserLog(logDto);
+//        }
+//        //客户端不在线则修改装置状态，进入定时任务
+//        else {
+//            //装置下线
+//            csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
+//            //装置调整为注册状态
+//            csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode());
+//            logDto.setOperate(nDid +"主任务执行失败，装置下线");
+//            csLogsFeignClient.addUserLog(logDto);
+//            sendMessage(nDid);
+//
+//            //记录装置掉线时间
+//            CsDeviceOnlineLogs record = onlineLogsService.findLastData(nDid);
+//            record.setOfflineTime(LocalDateTime.now());
+//            onlineLogsService.updateById(record);
+//
+//            scheduler.shutdown();
+//        }
+//    }
+
     private void startScheduledTask(ScheduledExecutorService scheduler, String nDid, String version) {
         synchronized (lock) {
             //判断是否推送消息

iot-access/access-boot/src/main/java/com/njcn/access/mapper/CsLineRunDataMapper.java

@@ -0,0 +1,16 @@
+package com.njcn.access.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.njcn.access.pojo.po.CsLineRunData;
+
+/**
+ * <p>
+ * 治理设备模块运行状态记录表 Mapper 接口
+ * </p>
+ *
+ * @author xy
+ * @since 2025-06-26
+ */
+public interface CsLineRunDataMapper extends BaseMapper<CsLineRunData> {
+
+}

iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java

@@ -8,7 +8,7 @@ import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DictTreeFeignClient;
-import com.njcn.system.enums.DicTreeEnum;
+import com.njcn.system.enums.DicDataEnum;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
@@ -45,7 +45,7 @@ public class AccessApplicationRunner implements ApplicationRunner {
     public void run(ApplicationArguments args) {
         //项目启动60s后发起自动接入
         Runnable task = () -> {
-            log.info("系统重启,所有符合条件的装置全部接入!");
+            log.info("系统重启,所有符合条件的装置发起接入!");
             List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getOnlineDev();
             if (CollUtil.isNotEmpty(list)) {
                 ExecutorService executor = Executors.newFixedThreadPool(10);
@@ -87,20 +87,28 @@ public class AccessApplicationRunner implements ApplicationRunner {
     }
 
     public void accessDev(List<CsEquipmentDeliveryPO> list) {
-        list.forEach(item->{
+        if (CollUtil.isNotEmpty(list)) {
-            System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+            try {
-            String version = csTopicService.getVersion(item.getNdid());
+                list.forEach(item->{
-            if (!Objects.isNull(version)){
+                    System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
-                //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
-                String code = dictTreeFeignClient.queryById(item.getDevModel()).getData().getCode();
+                    String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
-                if (Objects.equals(code,DicTreeEnum.PQV_520.getCode()) && Objects.equals(item.getStatus(),1)) {
+                    if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(),1)) {
-                    csDeviceService.wlDevRegister(item.getNdid());
+                        //csDeviceService.wlDevRegister(item.getNdid());
-                } else {
+                        log.info("请先手动注册、接入");
-                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                    } else {
-                }
+                        String version = csTopicService.getVersion(item.getNdid());
-                redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+                        if (Objects.isNull(version)) {
+                            version = "V1";
+                        }
+                        csDeviceService.autoAccess(item.getNdid(),version,1);
+                    }
+                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+                });
+            } catch (Exception e) {
+                log.error(e.getMessage());
             }
-        });
+        }
     }
 
 }

iot-access/access-boot/src/main/java/com/njcn/access/runner/AutoAccessTimer.java

@@ -8,7 +8,7 @@ import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.redis.pojo.enums.AppRedisKey;
 import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.DictTreeFeignClient;
-import com.njcn.system.enums.DicTreeEnum;
+import com.njcn.system.enums.DicDataEnum;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
@@ -44,57 +44,71 @@ public class AutoAccessTimer implements ApplicationRunner {
             if (CollUtil.isNotEmpty(list)) {
                 ExecutorService executor = Executors.newFixedThreadPool(10);
                 // 将任务平均分配给10个子列表
-                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
+                List<List<CsEquipmentDeliveryPO>> subLists = CollUtil.split(list, 10);
-                int partitionSize = list.size() / 10;
+//                List<List<CsEquipmentDeliveryPO>> subLists = new ArrayList<>();
-                for (int i = 0; i < 10; i++) {
+//                int partitionSize = list.size() / 10;
-                    int start = i * partitionSize;
+//                for (int i = 0; i < 10; i++) {
-                    int end = (i == 9) ? list.size() : start + partitionSize;
+//                    int start = i * partitionSize;
-                    subLists.add(list.subList(start, end));
+//                    int end = (i == 9) ? list.size() : start + partitionSize;
-                }
+//                    subLists.add(list.subList(start, end));
-
+//                }
                 // 创建一个ExecutorService来处理这些任务
                 List<Future<Void>> futures = new ArrayList<>();
                 // 提交任务给线程池执行
-                for (int i = 0; i < 10; i++) {
+//                for (int i = 0; i < 10; i++) {
-                    int index = i;
+//                    int index = i;
-                    futures.add(executor.submit(new Callable<Void>() {
+//                    futures.add(executor.submit(() -> {
-                        @Override
+//                        accessDev(subLists.get(index));
-                        public Void call() {
+//                        return null;
-                            accessDev(subLists.get(index));
+//                    }));
-                            return null;
+//                }
-                        }
+                for (List<CsEquipmentDeliveryPO> subList : subLists) {
+                    futures.add(executor.submit(() -> {
+                        accessDev(subList);
+                        return null;
                     }));
                 }
                 // 等待所有任务完成
                 for (Future<Void> future : futures) {
                     try {
                         future.get();
-                    } catch (InterruptedException | ExecutionException e) {
+                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
+                        Thread.currentThread().interrupt();
+                        log.error("任务被中断", e);
+                    } catch (ExecutionException e) {
+                        log.error("任务执行异常", e.getCause());
                     }
                 }
                 // 关闭ExecutorService
                 executor.shutdown();
             }
         };
-        //第一次执行的时间为120s，然后每隔120s执行一次
+        //第一次执行的时间为120s，然后在前一个任务执行完毕后，等待120s再执行下一个任务
-        scheduler.scheduleAtFixedRate(task,AUTO_TIME,AUTO_TIME,TimeUnit.SECONDS);
+        scheduler.scheduleWithFixedDelay(task, AUTO_TIME, AUTO_TIME, TimeUnit.SECONDS);
     }
 
     public void accessDev(List<CsEquipmentDeliveryPO> list) {
-        list.forEach(item->{
+        if (CollUtil.isNotEmpty(list)) {
-            System.out.println(Thread.currentThread().getName() + ": reboot : nDid : " + item.getNdid());
+            try {
-            String version = csTopicService.getVersion(item.getNdid());
+                list.forEach(item -> {
-            if (!Objects.isNull(version)){
+                    System.out.println(Thread.currentThread().getName() + ": auto : nDid : " + item.getNdid());
-                //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
+                    //判断设备类型 便携式设备需要特殊处理 未注册的要先注册、再接入;已注册的直接重新接入
-                String code = dictTreeFeignClient.queryById(item.getDevModel()).getData().getCode();
+                    String code = dictTreeFeignClient.queryById(item.getDevType()).getData().getCode();
-                if (Objects.equals(code, DicTreeEnum.PQV_520.getCode()) && Objects.equals(item.getStatus(),1)) {
+                    if (Objects.equals(code, DicDataEnum.PORTABLE.getCode()) && Objects.equals(item.getStatus(), 1)) {
-                    csDeviceService.wlDevRegister(item.getNdid());
+                        //csDeviceService.wlDevRegister(item.getNdid());
-                } else {
+                        log.info("请先手动注册、接入");
-                    csDeviceService.devAccessAskTemplate(item.getNdid(),version,1);
+                    } else {
-                }
+                        String version = csTopicService.getVersion(item.getNdid());
-                redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(),1);
+                        if (Objects.isNull(version)) {
+                            version = "V1";
+                        }
+                        csDeviceService.autoAccess(item.getNdid(), version, 1);
+                    }
+                    redisUtil.saveByKey(AppRedisKey.DEVICE_MID + item.getNdid(), 1);
+                });
+            } catch (Exception e) {
+                log.error(e.getMessage());
             }
-        });
+        }
     }
 }

iot-access/access-boot/src/main/java/com/njcn/access/service/ICsLineRunDataService.java

@@ -0,0 +1,20 @@
+package com.njcn.access.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.njcn.access.pojo.po.CsLineRunData;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 治理设备模块运行状态记录表 服务类
+ * </p>
+ *
+ * @author xy
+ * @since 2025-06-26
+ */
+public interface ICsLineRunDataService extends IService<CsLineRunData> {
+
+    void addData(List<CsLineRunData> list);
+
+}

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDataSetServiceImpl.java

@@ -33,6 +33,7 @@ public class CsDataSetServiceImpl extends ServiceImpl<CsDataSetMapper, CsDataSet
     public List<CsDataSet> getDataSetData(String modelId) {
         return this.lambdaQuery()
                 .eq(CsDataSet::getPid, modelId)
+                .and(item->item.eq(CsDataSet::getConType,1).or().isNull(CsDataSet::getConType))
                 .and(item->item.eq(CsDataSet::getDataType,"Stat").or().isNull(CsDataSet::getDataType))
                 .list();
     }

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java

@@ -91,6 +91,12 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             json = JsonUtil.convertStreamToString(file.getInputStream());
             Gson gson = new Gson();
             TemplateDto templateDto = gson.fromJson(json, TemplateDto.class);
+            //判断设备型号
+            String devType = templateDto.getDevType();
+            DictTreeVO dictTreeVO = dictTreeFeignClient.queryByCode(devType).getData();
+            if (Objects.isNull(dictTreeVO)){
+                throw new BusinessException(AccessResponseEnum.DEV_TYPE_NOT_FIND);
+            }
             logDto.setOperate("新增设备模板,模板名称:" + templateDto.getDevType());
             //模板文件存入文件服务器
             String filePath = fileStorageUtil.uploadMultipart(file, OssPath.DEV_MODEL + templateDto.getDevType() + "_");
@@ -838,9 +844,6 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
         List<DataSetDto> dataSetList = templateDto.getDataSet();
         String devType = templateDto.getDevType();
         DictTreeVO dictTreeVO = dictTreeFeignClient.queryByCode(devType).getData();
-        if (Objects.isNull(dictTreeVO)){
-            throw new BusinessException(AccessResponseEnum.DEV_TYPE_NOT_FIND);
-        }
         String code = dictTreeFeignClient.queryById(dictTreeVO.getPid()).getData().getCode();
         //逻辑设备录入
         if (CollectionUtil.isNotEmpty(dataSetList)){

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDeviceServiceImpl.java

@@ -38,6 +38,7 @@ import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -370,7 +371,7 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
     }
 
     @Override
-    @Transactional(rollbackFor = {Exception.class})
+//    @Transactional(rollbackFor = Exception.class)
     public String wlDevRegister(String nDid) {
         String result = "fail";
         // 设备状态判断
@@ -439,20 +440,12 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
             csDevModelRelationAddParm.setDid(modelList.get(0).getDid());
             csDevModelRelationService.addDevModelRelation(csDevModelRelationAddParm);
             //5.发起自动接入请求
+            Thread.sleep(2000);
             devAccessAskTemplate(nDid,version,1);
             //6.修改流程,便携式设备接入成功即为实际环境
             csEquipmentDeliveryService.updateProcessBynDid(nDid,4);
             //7.存储日志
             csLogsFeignClient.addUserLog(logDto);
-            //8.存储设备调试日志表
-            CsEquipmentProcessPO csEquipmentProcess = new CsEquipmentProcessPO();
-            csEquipmentProcess.setDevId(nDid);
-            csEquipmentProcess.setOperator(RequestUtil.getUserIndex());
-            csEquipmentProcess.setStartTime(LocalDateTime.now());
-            csEquipmentProcess.setEndTime(LocalDateTime.now());
-            csEquipmentProcess.setProcess(4);
-            csEquipmentProcess.setStatus(1);
-            processFeignClient.add(csEquipmentProcess);
             //9.删除redis监测点模板信息
             redisUtil.delete(AppRedisKey.MODEL + nDid);
             redisUtil.delete(AppRedisKey.LINE + nDid);
@@ -604,6 +597,78 @@ public class CsDeviceServiceImpl implements ICsDeviceService {
         return result;
     }
 
+    /**
+     * 装置重新接入系统，需要校验所用的模板
+     * @param nDid
+     * @param version
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public boolean autoAccess(String nDid,String version,Integer mid) {
+        String clientName = "NJCN-" + nDid.substring(nDid.length() - 6);
+        boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+        if (!mqttClient) {
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "接入失败，装置客户端不在线");
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(AlgorithmResponseEnum.DEV_OFFLINE);
+        }
+        boolean result = false;
+        Map<Integer,String> modelMap = new HashMap<>();
+        try {
+            //删除缓存数据
+            redisUtil.delete(AppRedisKey.MODEL + nDid);
+            redisUtil.deleteKeysByString(AppRedisKey.DEV_MODEL);
+            //询问装置当前所用模板
+            publisher.send("/Pfm/DevCmd/"+version+"/"+nDid,new Gson().toJson(getJson(mid,TypeEnum.TYPE_3.getCode())),1,false);
+            //接收到模板，判断模板是否存在，替换模板，发起接入
+            Thread.sleep(2000);
+            List<CsModelDto> modelId = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.MODEL + nDid),CsModelDto.class);
+            if (CollUtil.isNotEmpty(modelId)) {
+                CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(nDid).getData();
+                //重新录入装置和模板关系信息
+                for (CsModelDto item : modelId) {
+                    CsDevModelRelationPO po = new CsDevModelRelationPO();
+                    po.setDevId(vo.getId());
+                    po.setModelId(item.getModelId());
+                    po.setDid(item.getDid());
+                    po.setUpdateTime(LocalDateTime.now());
+                    csDevModelRelationService.addRelation(po);
+                    modelMap.put(item.getType(),item.getModelId());
+                }
+                //修改监测点使用的模板和数据集
+                List<CsLinePO> lineList;
+                Object object = redisUtil.getObjectByKey("accessLineInfo:" + nDid);
+                if (Objects.isNull(object)) {
+                    lineList = csLineFeignClient.findByNdid(nDid).getData();
+                    for (CsLinePO item : lineList) {
+                        if (item.getClDid() == 0) {
+                            updateLineIds(modelMap.get(0),item.getClDid(),nDid);
+                        } else {
+                            updateLineIds(modelMap.get(1),item.getClDid(),nDid);
+                        }
+                    }
+                }
+                //发起接入
+                publisher.send("/Pfm/DevCmd/"+version+"/"+nDid, new Gson().toJson(getJson(mid,TypeEnum.TYPE_5.getCode())),1,false);
+                //录波任务倒计时
+                redisUtil.saveByKeyWithExpire("startFile:" + nDid,null,60L);
+                result = true;
+            }
+        } catch (Exception e) {
+            DeviceLogDTO logDto = new DeviceLogDTO();
+            logDto.setUserName("运维管理员");
+            logDto.setLoginName("njcnyw");
+            logDto.setResult(1);
+            logDto.setOperate(nDid + "装置接入失败");
+            csLogsFeignClient.addUserLog(logDto);
+            throw new BusinessException(e.getMessage());
+        }
+        return result;
+    }
+
     /**
      * 组装报文
      */

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsEquipmentDeliveryServiceImpl.java

@@ -1,5 +1,6 @@
 package com.njcn.access.service.impl;
 
+import cn.hutool.core.collection.CollUtil;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.njcn.access.enums.AccessEnum;
@@ -16,6 +17,7 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Objects;
 
@@ -97,20 +99,22 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
                 .ne(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.DEL.getCode())
                 .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
                 .list();
-        list.forEach(item->{
+        if (CollUtil.isNotEmpty(list)) {
-            String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
+            list.forEach(item->{
-            boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+                String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
-            if (mqttClient) {
+                boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
-                result.add(item);
+                if (mqttClient) {
-            } else {
+                    result.add(item);
-                DeviceLogDTO logDto = new DeviceLogDTO();
+                } else {
-                logDto.setUserName("运维管理员");
+                    DeviceLogDTO logDto = new DeviceLogDTO();
-                logDto.setLoginName("njcnyw");
+                    logDto.setUserName("运维管理员");
-                logDto.setResult(1);
+                    logDto.setLoginName("njcnyw");
-                logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
+                    logDto.setResult(1);
-                csLogsFeignClient.addUserLog(logDto);
+                    logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
-            }
+                    csLogsFeignClient.addUserLog(logDto);
-        });
+                }
+            });
+        }
         return result;
     }
 
@@ -120,22 +124,24 @@ public class CsEquipmentDeliveryServiceImpl extends ServiceImpl<CsEquipmentDeliv
         List<CsEquipmentDeliveryPO> list = this.lambdaQuery()
                 .eq(CsEquipmentDeliveryPO::getRunStatus,AccessEnum.OFFLINE.getCode())
                 .eq(CsEquipmentDeliveryPO::getUsageStatus,1)
-                .eq(CsEquipmentDeliveryPO::getStatus, AccessEnum.REGISTERED.getCode())
+                .in(CsEquipmentDeliveryPO::getStatus, Arrays.asList(2,3))
                 .list();
-        list.forEach(item->{
+        if (CollUtil.isNotEmpty(list)) {
-            String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
+            list.forEach(item->{
-            boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
+                String clientName = "NJCN-" + item.getNdid().substring(item.getNdid().length() - 6);
-            if (mqttClient) {
+                boolean mqttClient = mqttUtil.judgeClientOnline(clientName);
-                result.add(item);
+                if (mqttClient) {
-            } else {
+                    result.add(item);
-                DeviceLogDTO logDto = new DeviceLogDTO();
+                } else {
-                logDto.setUserName("运维管理员");
+                    DeviceLogDTO logDto = new DeviceLogDTO();
-                logDto.setLoginName("njcnyw");
+                    logDto.setUserName("运维管理员");
-                logDto.setResult(1);
+                    logDto.setLoginName("njcnyw");
-                logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
+                    logDto.setResult(1);
-                csLogsFeignClient.addUserLog(logDto);
+                    logDto.setOperate(item.getNdid() + "接入失败，装置客户端不在线");
-            }
+                    csLogsFeignClient.addUserLog(logDto);
-        });
+                }
+            });
+        }
         return result;
     }
 

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsLineRunDataServiceImpl.java

@@ -0,0 +1,26 @@
+package com.njcn.access.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.njcn.access.mapper.CsLineRunDataMapper;
+import com.njcn.access.pojo.po.CsLineRunData;
+import com.njcn.access.service.ICsLineRunDataService;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 治理设备模块运行状态记录表 服务实现类
+ * </p>
+ *
+ * @author xy
+ * @since 2025-06-26
+ */
+@Service
+public class CsLineRunDataServiceImpl extends ServiceImpl<CsLineRunDataMapper, CsLineRunData> implements ICsLineRunDataService {
+
+    @Override
+    public void addData(List<CsLineRunData> list) {
+        this.saveBatch(list);
+    }
+}

iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/api/RtFeignClient.java

@@ -5,7 +5,11 @@ import com.njcn.common.pojo.response.HttpResult;
 import com.njcn.mq.message.AppAutoDataMessage;
 import com.njcn.rt.api.fallback.RtClientFallbackFactory;
 import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+
+import java.util.Map;
 
 /**
  * @author xy
@@ -15,4 +19,7 @@ public interface RtFeignClient {
 
     @PostMapping("/rtAnalysis")
     HttpResult<String> analysis(AppAutoDataMessage appAutoDataMessage);
+
+    @PostMapping("/apfRtAnalysis")
+    HttpResult<Map<String,Float>> apfRtAnalysis(@RequestBody @Validated AppAutoDataMessage appAutoDataMessage);
 }

iot-analysis/analysis-rt/rt-api/src/main/java/com/njcn/rt/api/fallback/RtClientFallbackFactory.java

@@ -9,6 +9,8 @@ import feign.hystrix.FallbackFactory;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
+import java.util.Map;
+
 /**
  * @author xy
  */
@@ -27,7 +29,13 @@ public class RtClientFallbackFactory implements FallbackFactory<RtFeignClient> {
 
             @Override
             public HttpResult<String> analysis(AppAutoDataMessage appAutoDataMessage) {
-                log.error("{}异常，降级处理，异常为：{}","实时数据解析",cause.toString());
+                log.error("{}异常，降级处理，异常为：{}","便携式实时数据解析",cause.toString());
+                throw new BusinessException(finalExceptionEnum);
+            }
+
+            @Override
+            public HttpResult<Map<String,Float>> apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) {
+                log.error("{}异常，降级处理，异常为：{}","APF实时数据解析",cause.toString());
                 throw new BusinessException(finalExceptionEnum);
             }
         };

iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/controller/RtController.java

@@ -19,6 +19,8 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.Map;
+
 /**
  * 类的介绍：
  *
@@ -45,5 +47,14 @@ public class RtController extends BaseController {
         return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
     }
 
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @PostMapping("/apfRtAnalysis")
+    @ApiOperation("APF实时数据解析")
+    @ApiImplicitParam(name = "appAutoDataMessage", value = "数据实体", required = true)
+    public HttpResult<String> apfRtAnalysis(@RequestBody @Validated AppAutoDataMessage appAutoDataMessage){
+        String methodDescribe = getMethodDescribe("apfRtAnalysis");
+        rtService.apfRtAnalysis(appAutoDataMessage);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe);
+    }
 
 }

iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/IRtService.java

@@ -8,4 +8,10 @@ import com.njcn.mq.message.AppAutoDataMessage;
 public interface IRtService {
 
     void analysis(AppAutoDataMessage appAutoDataMessage);
+
+    /**
+     * APF数据解析
+     * @param appAutoDataMessage
+     */
+    void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage);
 }

iot-analysis/analysis-rt/rt-boot/src/main/java/com/njcn/rt/service/impl/RtServiceImpl.java

@@ -85,7 +85,7 @@ public class RtServiceImpl implements IRtService {
         }
         //fixme 目前实时数据只有基础数据和谐波数据，后期拓展，这边需要再判断
         else {
-            HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item);
+            HarmRealDataSet harmRealDataSet = harmData(dataArrayList,item,dataSet.getDataLevel(),po.getCtRatio());
             harmRealDataSet.setUserId(userId);
             harmRealDataSet.setLineId(lineId);
             harmRealDataSet.setPt(po.getPtRatio().floatValue());
@@ -97,6 +97,36 @@ public class RtServiceImpl implements IRtService {
         }
     }
 
+    @Override
+    public void apfRtAnalysis(AppAutoDataMessage appAutoDataMessage) {
+        List<CsDataArray> dataArrayList;
+        String lineId;
+        //监测点id
+        if (appAutoDataMessage.getDid() == 1){
+            lineId = appAutoDataMessage.getId() + "0";
+        } else {
+            lineId = appAutoDataMessage.getId() + appAutoDataMessage.getMsg().getClDid();
+        }
+        //获取监测点基础信息
+        CsLinePO po = csLineFeignClient.getById(lineId).getData();
+        //获取数据集 dataSet
+        Integer idx = appAutoDataMessage.getMsg().getDsNameIdx();
+        CsDataSet dataSet = dataSetFeignClient.getDataSetByIdx(po.getDataModelId(),idx).getData();
+
+        String key = "BaseRealData:" + lineId + idx;
+        Object object = redisUtil.getObjectByKey(key);
+        if (Objects.isNull(object)){
+            dataArrayList = saveBaseRealDataSet(key,dataSet.getId());
+        } else {
+            dataArrayList = channelObjectUtil.objectToList(object,CsDataArray.class);
+        }
+        //根据dataArray解析数据
+        AppAutoDataMessage.DataArray item = appAutoDataMessage.getMsg().getDataArray().get(0);
+        Map<String,Float> map = getData(dataArrayList,item);
+        int data = Math.round(map.get("Apf_ModWorkingSts" + "M"));
+        redisUtil.saveByKeyWithExpire("ApfRtData:" + appAutoDataMessage.getMid(),data,10L);
+    }
+
     /**
      * 时间处理
      */
@@ -228,7 +258,7 @@ public class RtServiceImpl implements IRtService {
         return baseRealDataSet;
     }
 
-    public HarmRealDataSet harmData(List<CsDataArray> dataArrayList, AppAutoDataMessage.DataArray dataArray) {
+    public HarmRealDataSet harmData(List<CsDataArray> dataArrayList, AppAutoDataMessage.DataArray dataArray, String dataLevel, Double ct) {
         HarmRealDataSet harmRealDataSet = new HarmRealDataSet();
         List<HarmData> harmDataList = new ArrayList<>();
         //解码
@@ -259,7 +289,14 @@ public class RtServiceImpl implements IRtService {
         //通过反射将数据赋值
         Class<?> clazz = HarmRealDataSet.class;
         maxDataList.forEach(item->{
-            if (Objects.equals(item.getHarmName(),"Pq_RmsFundI") || Objects.equals(item.getHarmName(),"Pq_RmsFundU")) {
+            if (Objects.equals(item.getHarmName(),"Pq_RmsFundI")) {
+                if ("Secondary".equals(dataLevel)) {
+                    double data = item.getData() * ct;
+                    harmRealDataSet.setData1(FloatUtils.get2Float((float)data));
+                } else {
+                    harmRealDataSet.setData1(FloatUtils.get2Float(item.getData()));
+                }
+            } else if (Objects.equals(item.getHarmName(),"Pq_RmsFundU")) {
                 harmRealDataSet.setData1(FloatUtils.get2Float(item.getData()));
             } else {
                 String numberStr = item.getHarmName().substring(item.getHarmName().lastIndexOf('_') + 1);
@@ -267,7 +304,16 @@ public class RtServiceImpl implements IRtService {
                 try {
                     Field field = clazz.getDeclaredField(fieldName);
                     field.setAccessible(true);
-                    field.set(harmRealDataSet,FloatUtils.get2Float(item.getData()));
+                    if (item.getHarmName().contains("Pq_HarmI_")) {
+                        if ("Secondary".equals(dataLevel)) {
+                            double data = item.getData() * ct;
+                            field.set(harmRealDataSet,FloatUtils.get2Float((float)data));
+                        } else {
+                            field.set(harmRealDataSet,FloatUtils.get2Float(item.getData()));
+                        }
+                    } else {
+                        field.set(harmRealDataSet,FloatUtils.get2Float(item.getData()));
+                    }
                 } catch (Exception e) {
                     throw new RuntimeException(e);
                 }

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsAlarmServiceImpl.java

@@ -2,14 +2,23 @@ package com.njcn.zlevent.service.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.IdUtil;
+import com.alibaba.nacos.shaded.com.google.gson.Gson;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.njcn.access.pojo.po.CsDeviceOnlineLogs;
+import com.njcn.access.utils.ChannelObjectUtil;
+import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.EquipmentFeignClient;
 import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
 import com.njcn.csharmonic.pojo.po.CsEventPO;
 import com.njcn.mq.message.AppEventMessage;
+import com.njcn.redis.pojo.enums.AppRedisKey;
+import com.njcn.redis.utils.RedisUtil;
 import com.njcn.system.api.EpdFeignClient;
+import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.param.EleEpdPqdParam;
 import com.njcn.system.pojo.po.EleEpdPqd;
+import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.zlevent.mapper.CsEventMapper;
 import com.njcn.zlevent.pojo.po.CsEventLogs;
 import com.njcn.zlevent.service.ICsAlarmService;
@@ -24,8 +33,11 @@ import org.springframework.transaction.annotation.Transactional;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
+import static com.njcn.csdevice.enums.AlgorithmResponseEnum.DATA_ERROR;
+
 /**
  * <p>
  * 告警事件表 服务实现类
@@ -43,28 +55,57 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
     private final SendEventUtils sendEventUtils;
     private final ICsEventLogsService csEventLogsService;
     private final EpdFeignClient epdFeignClient;
+    private final RedisUtil redisUtil;
+    private final ChannelObjectUtil channelObjectUtil;
 
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void analysis(AppEventMessage appEventMessage) {
         List<CsEventPO> list1 = new ArrayList<>();
         LocalDateTime eventTime = null;
-        String tag = null;
+        String tag = null, lineId = null;
         String id = IdUtil.fastSimpleUUID();
         //获取装置id
         CsEquipmentDeliveryPO po = equipmentFeignClient.findDevByNDid(appEventMessage.getId()).getData();
+        List<SysDicTreePO> dictTreeList = channelObjectUtil.objectToList(redisUtil.getObjectByKey(AppRedisKey.DICT_TREE),SysDicTreePO.class);
+            String code = dictTreeList.stream().filter(item->Objects.equals(item.getId(),po.getDevType())).findFirst().orElse(null).getCode();
+
         try {
+            //便携式设备
+            if (Objects.equals(DicDataEnum.PORTABLE.getCode(),code)) {
+                lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+            }
+            //直连设备
+            else if (Objects.equals(DicDataEnum.CONNECT_DEV.getCode(),code)) {
+                if (Objects.equals(appEventMessage.getDid(),1)){
+                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0").toString();
+                } else if (Objects.equals(appEventMessage.getDid(),2)){
+                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+                }
+            }
+
             List<AppEventMessage.DataArray> dataArray = appEventMessage.getMsg().getDataArray();
             for (AppEventMessage.DataArray item : dataArray) {
                 eventTime = eventService.timeFormat(item.getDataTimeSec(),item.getDataTimeUSec());
+                tag = item.getName();
+                //判断各模块事件，如果上次模块事件和这次一致，则不记录
+                CsEventPO csEventPO = this.lambdaQuery().eq(CsEventPO::getLineId,lineId)
+                        .eq(CsEventPO::getClDid,appEventMessage.getMsg().getClDid())
+                        .eq(CsEventPO::getProcess,po.getProcess())
+                        .orderByDesc(CsEventPO::getStartTime).last("LIMIT 1").one();
+                if (csEventPO != null) {
+                    if (Objects.equals(csEventPO.getTag(),tag)) {
+                        throw new BusinessException(DATA_ERROR);
+                    }
+                }
                 //事件入库
                 CsEventPO csEvent = new CsEventPO();
+                csEvent.setLineId(lineId);
                 csEvent.setId(id);
                 csEvent.setDeviceId(po.getId());
                 csEvent.setProcess(po.getProcess());
                 csEvent.setCode(item.getCode());
                 csEvent.setStartTime(eventTime);
-                tag = item.getName();
                 csEvent.setTag(tag);
                 csEvent.setType(3);
                 csEvent.setClDid(appEventMessage.getMsg().getClDid());
@@ -91,6 +132,7 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
             }
         } catch (Exception e) {
             CsEventLogs csEventLogs = new CsEventLogs();
+            csEventLogs.setLineId(lineId);
             csEventLogs.setDeviceId(po.getId());
             csEventLogs.setStartTime(eventTime);
             csEventLogs.setTag(tag);

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java

@@ -35,6 +35,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.influxdb.InfluxDB;
 import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
+import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -91,7 +92,19 @@ public class EventServiceImpl implements IEventService {
         //获取设备类型 true:治理设备 false:其他类型的设备
         boolean devModel = equipmentFeignClient.judgeDevModel(appEventMessage.getId()).getData();
         try {
-            lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
+//            lineId = appEventMessage.getId() + appEventMessage.getMsg().getClDid();
+            if (devModel) {
+                if (Objects.equals(appEventMessage.getDid(),1)){
+                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get("0").toString();
+                } else if (Objects.equals(appEventMessage.getDid(),2)){
+                    lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+                }
+            } else {
+                lineId = new Gson().fromJson(String.valueOf(redisUtil.getObjectByKey(AppRedisKey.LINE_POSITION+appEventMessage.getId())), Map.class).get(appEventMessage.getMsg().getClDid().toString()).toString();
+            }
+
+
+
             //处理事件数据
             List<AppEventMessage.DataArray> dataArray = appEventMessage.getMsg().getDataArray();
             for (AppEventMessage.DataArray item : dataArray) {
@@ -200,7 +213,7 @@ public class EventServiceImpl implements IEventService {
             if (!Objects.isNull(record)) {
                 if (!Objects.equals(item.getPrjTimeEnd(),-1L)) {
                     WlRecordParam.UpdateRecord wlRecord = new WlRecordParam.UpdateRecord();
-                    wlRecord.setId(record.getId());
+                    BeanUtils.copyProperties(record,wlRecord);
                     wlRecord.setProEndTime(timestampToDatetime((item.getPrjTimeEnd() - 8*3600)));
                     wlRecordFeignClient.updateTestRecord(wlRecord);
                 }
@@ -208,7 +221,7 @@ public class EventServiceImpl implements IEventService {
                 //新项目入库
                 WlRecord wlRecord = new WlRecord();
                 wlRecord.setId(IdUtil.simpleUUID());
-                wlRecord.setItemName("基础数据");
+                wlRecord.setItemName("在线监测");
                 wlRecord.setGcName(item.getPrjName());
                 wlRecord.setDevId(vo.getId());
                 wlRecord.setLineId(appEventMessage.getId() + item.getClDid().toString());
@@ -227,6 +240,9 @@ public class EventServiceImpl implements IEventService {
                 //fixme 电流接线方式 这边系统没有字典，录入字典通用性不强，采用装置上送值存储
                 wlRecord.setCurConSel(item.getCurConSel().toString());
                 wlRecord.setStartTime(timestampToDatetime((item.getPrjTimeStart() - 8*3600)));
+                if (!Objects.equals(item.getPrjTimeEnd(),-1L)) {
+                    wlRecord.setEndTime(timestampToDatetime((item.getPrjTimeEnd() - 8*3600)));
+                }
                 wlRecord.setType(1);
                 wlRecord.setState(1);
                 wlRecord.setGcDataPath(item.getPrjDataPath());

iot-message/message-boot/src/main/java/com/njcn/message/consumer/AppAutoDataConsumer.java

@@ -46,8 +46,6 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
     private RocketMqLogFeignClient rocketMqLogFeignClient;
     @Resource
     private RtFeignClient rtFeignClient;
-//    @Resource
-//    private WlRecordFeignClient wlRecordFeignClient
 
     @Override
     protected void handleMessage(AppAutoDataMessage appAutoDataMessage) {
@@ -61,7 +59,7 @@ public class AppAutoDataConsumer extends EnhanceConsumerMessageHandler<AppAutoDa
                 rtFeignClient.analysis(appAutoDataMessage);
                 break;
             case 2:
-                log.info(appAutoDataMessage.getKey() + "分发至统计数据");
+                log.info("{}分发至统计数据", appAutoDataMessage.getKey());
                 statFeignClient.analysis(appAutoDataMessage);
                 break;
             default: