fix(equip):对接韶山换流站kafka数据

amos-boot-biz-common/src/main/java/com/yeejoin/amos/boot/biz/common/utils/RedisKey.java

@@ -52,7 +52,11 @@ public class RedisKey {
      * 装备指标Key值
      */
     public static final String EQUIP_INDEX_ADDRESS_KEY = "equip_index_address_key";
-    
+    /**
+     * 韶山换流站指标Key
+     */
+    public static final String EQUIP_INDEX_ADDRESS_KEY_STATION = "equip_index_address_key_station";
+
     /** 驼峰转下划线(简单写法，效率低于 ) */
 	public static String humpToLine(String str) {
 		return str.replaceAll("[A-Z]", "_$0").toLowerCase();

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/entity/vo/EquipmentIndexVO.java

@@ -72,6 +72,9 @@ public class EquipmentIndexVO {
     @ApiModelProperty(value = "信号的索引键key，用于唯一索引信号")
     private String indexAddress;
 
+    @ApiModelProperty(value = "信号的索引键pointId，用于唯一索引信号")
+    private String eventAddress;
+
     @ApiModelProperty(value = "测点类型，analog/state")
     private String dataType;
 

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/enums/MqttConstant.java

+package com.yeejoin.equipmanage.common.enums;
+
+/**
+ * @author LiuLin
+ * @date 2023年08月02日 11:02
+ */
+public interface MqttConstant {
+    String TRUE                       = "true";
+    String FALSE                      = "false";
+    String STATE                      = "state";
+    String DIS_CREATE                 = "discreate";
+    String ONE_1                      = "1";
+    String ONE_1_0                    = "1.0";
+}

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/vo/SShanMessage.java

+package com.yeejoin.equipmanage.common.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+/**
+ * @author LiuLin
+ * @date 2023年08月02日 11:02
+ */
+@Data
+@ToString
+@AllArgsConstructor
+@NoArgsConstructor
+public class SShanMessage {
+    private String eventTextL1;
+    private String pointId;
+    private String time;
+    private String deviceId;
+}

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/vo/SShanStationMessage.java

+package com.yeejoin.equipmanage.common.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import java.util.List;
+
+/**
+ * @author LiuLin
+ * @date 2023年08月02日 11:02
+ */
+@Data
+@ToString
+@AllArgsConstructor
+@NoArgsConstructor
+public class SShanStationMessage {
+    private String timestamp;
+    private List<SShanMessage> warns;
+}

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/vo/StationMessage.java

+package com.yeejoin.equipmanage.common.vo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+/**
+ * 对接苏州，绍兴换流站Kafka数据
+ * @author LiuLin
+ * @date 2023年08月02日 11:02
+ */
+@Data
+@ToString
+@AllArgsConstructor
+@NoArgsConstructor
+public class StationMessage {
+    private String dataType;
+    private String value;
+    private String timeStamp;
+    private String quality;
+    private String scadaId;
+    private String key;
+    private String disCreate;
+    private String name;
+}

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/config/EquipmentIndexCacheRunner.java

@@ -33,6 +33,7 @@ public class EquipmentIndexCacheRunner implements CommandLineRunner {
         log.info(">>服务启动执行，执行预加载数据等操作");
         redisUtils.del(RedisKey.EQUIP_INDEX_ADDRESS);
         redisUtils.del(RedisKey.EQUIP_INDEX_ADDRESS_KEY);
+        redisUtils.del(RedisKey.EQUIP_INDEX_ADDRESS_KEY_STATION);
         List<EquipmentIndexVO> equipSpecificIndexList = equipmentSpecificIndexMapper.getEquipSpecificIndexList(null);
         Map<String, Object> equipmentIndexVOMap = equipSpecificIndexList.stream()
                 .filter(v -> v.getGatewayId() != null)
@@ -40,8 +41,12 @@ public class EquipmentIndexCacheRunner implements CommandLineRunner {
         Map<String, Object> equipmentIndexKeyMap = equipSpecificIndexList.stream()
                 .filter(v -> v.getIndexAddress() != null && v.getGatewayId() == null)
                 .collect(Collectors.toMap(EquipmentIndexVO::getIndexAddress, Function.identity(),(v1, v2) -> v1));
+        Map<String, Object> equipmentIndexEventMap = equipSpecificIndexList.stream()
+                .filter(v -> v.getEventAddress() != null)
+                .collect(Collectors.toMap(EquipmentIndexVO::getEventAddress, Function.identity(),(v1, v2) -> v1));
 
         redisUtils.hmset(RedisKey.EQUIP_INDEX_ADDRESS, equipmentIndexVOMap);
         redisUtils.hmset(RedisKey.EQUIP_INDEX_ADDRESS_KEY, equipmentIndexKeyMap);
+        redisUtils.hmset(RedisKey.EQUIP_INDEX_ADDRESS_KEY_STATION, equipmentIndexEventMap);
     }
 }
\ No newline at end of file

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/config/EquipmentIotMqttReceiveConfig.java

@@ -126,6 +126,7 @@ public class EquipmentIotMqttReceiveConfig {
         list.add("+/+/event"); // 添加iot事件监听
         list.add("+/+/transmit"); // 添加交换站事件监听
         list.add("+/+/perspective"); // 添加交换站事件监听
+        list.add("+/+/shaoshan"); // 添加换流站韶山监听事件
         String[] arr = list.toArray(new String[list.size()]);
         adapter = new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttPahoClientFactory(), arr);
         adapter.setCompletionTimeout(completionTimeout);
@@ -151,8 +152,8 @@ public class EquipmentIotMqttReceiveConfig {
                     mqttEventReceiveService.handlerMqttIncrementMessage(topic, msg);
                 }else if (dataType.equals("transmit") && StringUtil.isNotEmpty(msg)){
                     mqttReceiveService.handlerMqttRomaMessage(topic,msg);
-                }else if (dataType.equals("perspective") && StringUtil.isNotEmpty(msg)){
-                    mqttReceiveService.handlerMqttIotMessage(topic,msg);
+                }else if (dataType.equals("shaoshan") && StringUtil.isNotEmpty(msg)){
+                    mqttReceiveService.handlerMqttStationMessage(topic,msg);
                 }
             }
         };

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/mapper/EquipmentSpecificIndexMapper.java

@@ -114,5 +114,5 @@ public interface EquipmentSpecificIndexMapper extends BaseMapper<EquipmentSpecif
 
     List<EquipmentSpecificIndex> getEquipIndexInIndex(@Param("list") List<String> listIndex);
 
-    EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress,String gatewayId);
+    EquipmentSpecificIndex getEquipmentSpeIndexByAddress(String indexAddress, String eventAddress, String gatewayId);
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/IEquipmentSpecificIndexService.java

@@ -40,5 +40,5 @@ public interface IEquipmentSpecificIndexService extends IService<EquipmentSpecif
      * @param indexAddress indexAddress
      * @return EquipmentSpecificIndex
      */
-    EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress,String gatewayId);
+    EquipmentSpecificIndex getEquipmentSpeIndexByAddress(String indexAddress, String eventAddress, String gatewayId);
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/MqttReceiveService.java

@@ -30,5 +30,5 @@ public interface MqttReceiveService {
      * @param topic 主题
      * @param message 消息内容
      */
-    void handlerMqttIotMessage(String topic, String message);
+    void handlerMqttStationMessage(String topic, String message);
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/EquipmentSpecificIndexServiceImpl.java

@@ -30,7 +30,7 @@ public class EquipmentSpecificIndexServiceImpl extends ServiceImpl<EquipmentSpec
     }
 
     @Override
-    public EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress,String gatewayId) {
-        return this.baseMapper.getEquipmentSpeIndexByIndexAddress(indexAddress,gatewayId);
+    public EquipmentSpecificIndex getEquipmentSpeIndexByAddress(String indexAddress,String eventAddress, String gatewayId) {
+        return this.baseMapper.getEquipmentSpeIndexByAddress(indexAddress, eventAddress, gatewayId);
     }
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -54,6 +54,8 @@ import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.stream.Collectors;
 
+import static com.yeejoin.equipmanage.common.enums.MqttConstant.*;
+
 
 /**
  * @author keyong
@@ -67,44 +69,55 @@ import java.util.stream.Collectors;
 @Slf4j
 @Service
 public class MqttReceiveServiceImpl implements MqttReceiveService {
-    private static String PUMP_JOB_GROUP_NAME = "EQUIP_PUMP_JOB_GROUP_NAME";
-    private static String PUMP_TRIGGER_NAME = "EQUIP_PUMP_TRIGGER_NAME";
-    private static String PUMP_TRIGGER_GROUP_NAME = "EQUIP_PUMP_TRIGGER_GROUP_NAME";
-
-    private static Map<String, TemperatureAlarmDto> temperatureMap = new HashMap<>();
-
+    /**
+     * 泡沫罐KEY
+     */
+    private final static String CAFS_FoamTank_FoamTankLevel = "CAFS_FoamTank_FoamTankLevel";
+    /**
+     * 泡沫罐KEY
+     */
+    private final static String FHS_PipePressureDetector_PipePressure = "FHS_PipePressureDetector_PipePressure";
+    /**
+     * 水池信息
+     */
+    private final static String FHS_FirePoolDevice_WaterLevel = "FHS_FirePoolDevice_WaterLevel";
+    /**
+     * 水池信息
+     */
+    private final static String FHS_WirelessliquidDetector_WaterLevel = "FHS_WirelessliquidDetector_WaterLevel";
+    /**
+     * 水箱液位
+     */
+    private final static String CAFS_WaterTank_WaterTankLevel = "CAFS_WaterTank_WaterTankLevel";
+    private static final String PUMP_JOB_GROUP_NAME = "EQUIP_PUMP_JOB_GROUP_NAME";
+    private static final String PUMP_TRIGGER_NAME = "EQUIP_PUMP_TRIGGER_NAME";
+    private static final String PUMP_TRIGGER_GROUP_NAME = "EQUIP_PUMP_TRIGGER_GROUP_NAME";
+    private static final Map<String, TemperatureAlarmDto> temperatureMap = new HashMap<>();
     static IEquipmentSpecificIndexService equipmentSpecificIndexService;
-
+    static EquipmentSpecificMapper equipmentSpecificMapper;
+    static IFireFightingSystemService fireFightingSystemService;
+    static MqttSendGateway mqttSendGateway;
+    private static RemoteSecurityService remoteSecurityService;
+    private static String canvasTopic;
+    private static Boolean jcsSwitch;
+    private static Boolean bool = Boolean.FALSE;
     @Autowired
-    public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService) {
-        MqttReceiveServiceImpl.equipmentSpecificIndexService = equipmentSpecificIndexService;
-    }
-
+    protected EmqKeeper emqKeeper;
     @Autowired
     ICarPropertyService carPropertyService;
-
     @Autowired
     IEquipmentSpecificAlarmService equipmentSpecificAlarmService;
-
     @Autowired
     IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService;
-
     @Autowired
     EquipmentSpecificAlarmMapper equipmentSpecificAlarmMapper;
-
     @Autowired
     EquipmentSpecificAlarmLogMapper equipmentSpecificAlarmLogMapper;
-
     @Autowired
     @Lazy
     IEquipmentIndexService equipmentIndexService;
-
     @Autowired
     EquipmentSpecificIndexMapper equipmentSpecificIndexMapper;
-
-    @Autowired
-    private IotFeign iotFeign;
-
     @Autowired
     MarqueeDataMapper marqueeDataMapper;
     //消防泵
@@ -113,161 +126,183 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     //消防炮
     @Value("${equipment.plan.monitor}")
     String monitorCodes;
-
     @Autowired
-    private InfluxDbConnection influxDbConnection;
-
-    /**
-     * 泡沫罐KEY
-     */
-    private final static String CAFS_FoamTank_FoamTankLevel = "CAFS_FoamTank_FoamTankLevel";
-    /**
-     * 泡沫罐KEY
-     */
-    private final static String FHS_PipePressureDetector_PipePressure = "FHS_PipePressureDetector_PipePressure";
-
-    /**
-     * 水池信息
-     */
-    private final static String FHS_FirePoolDevice_WaterLevel = "FHS_FirePoolDevice_WaterLevel";
-
-    /**
-     * 水池信息
-     */
-    private final static String FHS_WirelessliquidDetector_WaterLevel = "FHS_WirelessliquidDetector_WaterLevel";
-
-    /**
-     * 水箱液位
-     */
-    private final static String CAFS_WaterTank_WaterTankLevel = "CAFS_WaterTank_WaterTankLevel";
-
-
-    static EquipmentSpecificMapper equipmentSpecificMapper;
-
+    FireFightingSystemMapper FireFightingSystemMapper;
     @Autowired
-    public void setEquipmentSpecificMapper(EquipmentSpecificMapper equipmentSpecificMapper) {
-        MqttReceiveServiceImpl.equipmentSpecificMapper = equipmentSpecificMapper;
-    }
-
+    CarMapper carMapper;
     @Autowired
-    FireFightingSystemMapper FireFightingSystemMapper;
-
-    static IFireFightingSystemService fireFightingSystemService;
-
+    FireFightingSystemMapper fireFightingSystemMapper;
     @Autowired
-    public void setFireFightingSystemService(IFireFightingSystemService fireFightingSystemService) {
-        MqttReceiveServiceImpl.fireFightingSystemService = fireFightingSystemService;
-    }
-
-
-    static MqttSendGateway mqttSendGateway;
-
+    IESeqService eSeqService;
     @Autowired
-    public void setMqttSendGateway(MqttSendGateway mqttSendGateway) {
-        MqttReceiveServiceImpl.mqttSendGateway = mqttSendGateway;
-    }
-
+    private IotFeign iotFeign;
     @Autowired
-    private RedisUtils redisUtils;
-
+    private InfluxDbConnection influxDbConnection;
     @Autowired
-    CarMapper carMapper;
-
+    private RedisUtils redisUtils;
     @Autowired
     private ISyncDataService syncDataService;
-
     @Autowired
     private IEquipmentAlarmReportDayService iEquipmentAlarmReportDayService;
-
     @Autowired
     private IEquipmentSpecificSerivce iEquipmentSpecificSerivce;
-
-    @Autowired
-    FireFightingSystemMapper fireFightingSystemMapper;
-
     @Autowired
     private SystemctlFeign systemctlFeign;
-
-    private static RemoteSecurityService remoteSecurityService;
-
-    @Autowired
-    public void setRemoteSecurityService(RemoteSecurityService remoteSecurityService) {
-        MqttReceiveServiceImpl.remoteSecurityService = remoteSecurityService;
-    }
-
     @Autowired
     private TopographyService topographyService;
     @Autowired
-    IESeqService eSeqService;
-
-    @Autowired
     private IEquipmentService equipmentService;
-
     @Autowired
     private IPressurePumpService pressurePumpService;
-
     @Value("${equipManage.name}")
     private String serverName;
-
     @Value("${state.code:code}")
     private String stationCode;
-
     @Value("${system.type}")
     private String system;
-
     @Value("${state.name:name}")
     private String stationName;
-
     @Value("${mqtt.vehicle.topic}")
     private String carTopic;
-
-    private static String canvasTopic;
-
-    @Value("${equip.point.equipmentdata.topic}")
-    public void setCanvasTopic(String canvasTopic) {
-        MqttReceiveServiceImpl.canvasTopic = canvasTopic;
-    }
-
     @Value("${equip.index.topic}")
     private String indexTopic;
-
     @Value("${spring.redis.expire.time}")
     private long redisExpireTime;
-
     @Value("${systemctl.sync.switch}")
     private Boolean syncSwitch;
-
     @Value("${systemctl.amos.switch}")
     private Boolean amosSwitch;
-
     @Value("${isSendApp}")
     private Boolean isSendApp;
-
     @Value("${isSendIot:false}")
     private Boolean isSendIot;
-
     @Value("${is.open.telemetering:false}")
     private Boolean isOpenTelemetering;
+    @Value("${equipment.pressurepump.start}")
+    private String pressurePumpStart;
+    @Value("${emergency.disposal.indicators}")
+    private String emergencyDisposalIndicators;
 
-    private static Boolean jcsSwitch;
-
-    @Value("${systemctl.jcs.switch}")
-    public void setJcsSwitch(Boolean jcsSwitch) {
-        MqttReceiveServiceImpl.jcsSwitch = jcsSwitch;
+    public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey,
+                                                                   String traceId, IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService,
+                                                                   boolean flag) {
+        LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
+        queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
+        queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
+        List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper().selectList(queryWrapper);
+        if (!logs.isEmpty()) {
+            EquipmentSpecificAlarmLog log = logs.get(0);
+            EquipmentSpecific specific = equipmentSpecificMapper.selectById(log.getEquipmentSpecificId());
+            Date date = new Date();
+            SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+            String dateString = formatter.format(date);
+            logs.forEach(x -> {
+                x.setCleanTime(new Date());
+                x.setStatus(AlarmStatusEnum.HF.getCode());
+                if (!flag) {
+                    // 若已经被确警处理过，此处不再做处理
+                    if (StringUtils.isEmpty(x.getConfirmType())) {
+                        x.setConfirmType(x.getType());
+                        x.setConfirmTypeName(ConfirmAlamEnum.getTypeByCode(x.getType()));
+                        String equipmentName = StringUtil.isNotEmpty(specific.getCode()) ? specific.getName() + "(" + specific.getCode() + ")" : specific.getName();
+                        if (x.getType().equals(AlarmTypeEnum.PB.getCode())) {
+                            x.setAlarmReason(equipmentName + "频繁故障/误报，将设备报警信息屏蔽。");
+                        } else {
+                            x.setAlarmReason(equipmentName + "引起设备报警");
+                        }
+                        x.setResolveResult(dateString + "系统收到设备复归（已消除）信号，系统自动处理。");
+                        x.setConfirmUser("");
+                        x.setConfirmUserName("系统自动处理");
+                        x.setConfirmDate(date);
+                    }
+                }
+                x.setEquipmentSpecificIndexValue(TrueOrFalseEnum.fake.value);
+                if (!StringUtils.isEmpty(traceId)) {
+                    x.setTraceId(traceId);
+                }
+            });
+            equipmentSpecificAlarmLogService.updateBatchById(logs);
+        }
+        if (flag) {
+            List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService.getEquipmentSpeIndexBySpeIotCode(iotCode);
+            publishDataToCanvas(indexList);
+        }
+        return logs;
     }
 
+    /**
+     * 画布数据消息推送
+     *
+     * @param indexList
+     */
+    public static void publishDataToCanvas(List<EquipmentSpecificIndex> indexList) {
+        if (!ObjectUtils.isEmpty(indexList)) {
+            EquipmentSpecificIndex index = indexList.stream().filter(x -> x.getUpdateDate() != null)
+                    .sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
+                    .collect(Collectors.toList()).get(0);
+            EquipmentStateVo equipmentStateVo = new EquipmentStateVo();
+            equipmentStateVo.setEquipName(index.getEquipmentSpecificName());
+            equipmentStateVo.setOrgCode(index.getOrgCode());
+            equipmentStateVo.setSpecificId(index.getEquipmentSpecificId());
+            equipmentStateVo.setEquipCode(index.getQrCode());
+            equipmentStateVo.setEquipIotCode(index.getIotCode());
+            equipmentStateVo.setStatus("");
+            equipmentStateVo.setColor(index.getEmergencyLevelColor());
+            // 添加性能指标项
+            //flag 无意义 getSpeIndex 方法其他地方共用 做了额外封装 为0时走原逻辑
+            int flag = 0;
+            equipmentStateVo.setSpeindexList(fireFightingSystemService.getSpeIndex(index.getEquipmentSpecificId(), flag));
+            Map<String, Object> topicObject = new HashMap<>();
+            topicObject.put("equipCode", equipmentStateVo.getEquipCode());
+            if (TrueOrFalseEnum.real.value.equalsIgnoreCase(index.getValue())) {
+                topicObject.put("color", equipmentStateVo.getColor());
+            } else {
+                topicObject.put("color", "");
+            }
+            Map<String, Object> map = JSON.parseObject(JSON.toJSONString(equipmentStateVo));
+            topicObject.put("uuid", UUID.randomUUID().toString());
+            map.put(canvasTopic, topicObject);
+
+            // 发送数据至画布
+            mqttSendGateway.sendToMqtt(canvasTopic, JSON.toJSONString(map));
+        }
+    }
 
-    @Value("${equipment.pressurepump.start}")
-    private String pressurePumpStart;
+    @Autowired
+    public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService) {
+        MqttReceiveServiceImpl.equipmentSpecificIndexService = equipmentSpecificIndexService;
+    }
 
+    @Autowired
+    public void setEquipmentSpecificMapper(EquipmentSpecificMapper equipmentSpecificMapper) {
+        MqttReceiveServiceImpl.equipmentSpecificMapper = equipmentSpecificMapper;
+    }
 
-    @Value("${emergency.disposal.indicators}")
-    private String emergencyDisposalIndicators;
+    @Autowired
+    public void setFireFightingSystemService(IFireFightingSystemService fireFightingSystemService) {
+        MqttReceiveServiceImpl.fireFightingSystemService = fireFightingSystemService;
+    }
 
-    private static Boolean bool = Boolean.FALSE;
+    @Autowired
+    public void setMqttSendGateway(MqttSendGateway mqttSendGateway) {
+        MqttReceiveServiceImpl.mqttSendGateway = mqttSendGateway;
+    }
 
     @Autowired
-    protected EmqKeeper emqKeeper;
+    public void setRemoteSecurityService(RemoteSecurityService remoteSecurityService) {
+        MqttReceiveServiceImpl.remoteSecurityService = remoteSecurityService;
+    }
+
+    @Value("${equip.point.equipmentdata.topic}")
+    public void setCanvasTopic(String canvasTopic) {
+        MqttReceiveServiceImpl.canvasTopic = canvasTopic;
+    }
+
+    @Value("${systemctl.jcs.switch}")
+    public void setJcsSwitch(Boolean jcsSwitch) {
+        MqttReceiveServiceImpl.jcsSwitch = jcsSwitch;
+    }
 
     @Override
     @Transactional(rollbackFor = Exception.class)
@@ -343,7 +378,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void handlerMqttRomaMessage(String topic, String message) {
-        log.info("接收到Mqtt消息: {}", message);
+        log.info("接收到换流站Kafka消息: {}", message);
         TopicEntityVo topicEntity = new TopicEntityVo();
         topicEntity.setTopic(topic);
         topicEntity.setMessage(message);
@@ -353,30 +388,28 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
         List<IndexStateVo> indexStateList = new ArrayList<>();
 
-        JSONObject jsonObject = JSONObject.parseObject(message);
-        String dataType = jsonObject.getString("datatype");
-        String indexAddress = null, value, timeStamp, quality = null;
-
-        //如果消息是遥信类型，进行指标转换
-        assert dataType != null;
-        if (dataType.equals("state")) {
-            indexAddress = jsonObject.getString("scadaid");
-            value = jsonObject.getInteger("value") == 1 ? "true" : "false";
-            timeStamp = jsonObject.getString("timestamp");
-        }else if (dataType.equals("discreate")){
-            indexAddress = jsonObject.getString("key");
-            value = jsonObject.getFloat("value") == 0.0 ? "false" : "true";
-            timeStamp = jsonObject.getString("time_stamp");
-            quality = jsonObject.getString("quality");
-        }else {
-            indexAddress = jsonObject.getString("key");
-            value = jsonObject.getFloat("value")+"";
-            timeStamp = jsonObject.getString("time_stamp");
-            quality = jsonObject.getString("quality");
+        StationMessage stationMessage = JSON.parseObject(String.valueOf(message), StationMessage.class);
+        String indexAddress = null, value = null, timeStamp = null, quality = null, dataType = null;
+        if (stationMessage != null) {
+            dataType = stationMessage.getDataType();
+            timeStamp = stationMessage.getTimeStamp();
+            quality = stationMessage.getQuality();
+
+            if (dataType.equals(STATE)) {
+                indexAddress = stationMessage.getScadaId();
+                value = ONE_1.equalsIgnoreCase(stationMessage.getValue()) ? TRUE : FALSE;
+            } else if (dataType.equals(DIS_CREATE)) {
+                indexAddress = stationMessage.getKey();
+                value = ONE_1_0.equalsIgnoreCase(stationMessage.getValue()) ? TRUE : FALSE;
+            } else {
+                indexAddress = stationMessage.getKey();
+                value = stationMessage.getValue();
+            }
         }
+
         Map<Object, Object> equipmentIndexKeyMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS_KEY);
         if (equipmentIndexKeyMap.get(indexAddress) != null) {
-            EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByIndexAddress(indexAddress, null);
+            EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByAddress(indexAddress, null, null);
 
             equipmentSpeIndex.setValue(value);
             equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
@@ -439,22 +472,22 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             boolean alarmFlag = false;
             Map<String, String> messageBodyMap = new HashMap<>();
             //管网压力、泡沫罐信息、水箱液位告警处理
-            if (iotDataVO.getKey().toLowerCase().equals(CAFS_FoamTank_FoamTankLevel.toLowerCase()) ||
-                    FHS_PipePressureDetector_PipePressure.toLowerCase().equals(iotDataVO.getKey().toLowerCase()) ||
-                    iotDataVO.getKey().toLowerCase().equals(CAFS_WaterTank_WaterTankLevel.toLowerCase())) {
+            if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                    FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                    iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
                 alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
             }
             //消防水池液位处理
-            if (iotDataVO.getKey().toLowerCase().equals(FHS_FirePoolDevice_WaterLevel.toLowerCase()) ||
-                    iotDataVO.getKey().toLowerCase().equals(FHS_WirelessliquidDetector_WaterLevel.toLowerCase())) {
+            if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                    iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                 alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
             }
             // 遥测数据生成告警事件、日志处理
-            if (iotDataVO.getKey().toLowerCase().equals(CAFS_FoamTank_FoamTankLevel.toLowerCase()) ||
-                    FHS_PipePressureDetector_PipePressure.toLowerCase().equals(iotDataVO.getKey().toLowerCase()) ||
-                    iotDataVO.getKey().toLowerCase().equals(CAFS_WaterTank_WaterTankLevel.toLowerCase()) ||
-                    iotDataVO.getKey().toLowerCase().equals(FHS_FirePoolDevice_WaterLevel.toLowerCase()) ||
-                    iotDataVO.getKey().toLowerCase().equals(FHS_WirelessliquidDetector_WaterLevel.toLowerCase())) {
+            if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                    FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                    iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
+                    iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                    iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                 handlingAlarms(equipmentSpeIndex, alarmFlag);
             }
 
@@ -495,7 +528,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             jsonObjectData.put("measurementType", null == equipmentSpeIndex.getEquipmentIndexKey() ? "" : equipmentSpeIndex.getEquipmentIndexKey());
             jsonObjectData.put("dateTime", date);
             jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
-            List<JSONObject> jsonObjects = Arrays.asList(jsonObjectData);
+            List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);
 
             jsonObjectXf.put("data", jsonObjects);
 
@@ -508,7 +541,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     log.info("遥测遥信数据推送云端kafka成功");
                 } catch (MqttException e) {
                     log.error("遥测遥信数据推送云端kafka失败=====>" + e.getMessage());
-                    e.printStackTrace();
                 }
             }
 
@@ -598,7 +630,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                             //消防水源
                             if (!flag) {
                                 List<Map> lit = iEquipmentSpecificSerivce.getWater(equipmentSpecificVo.getId());
-                                if (lit != null && lit.size() > 0) {
+                                if (lit != null && !lit.isEmpty()) {
                                     topicEntity.setType("xfsy");
                                     mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
                                 }
@@ -612,128 +644,251 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     @Override
     @Transactional(rollbackFor = Exception.class)
-    public void handlerMqttIotMessage(String topic, String message) {
-        //influxdb
-        Map<String, String> tagsMap = new HashMap<>();
-        Map<String, Object> fieldsMap = new HashMap<>();
-        Map<String, Object> fieldsStrMap = new HashMap<>();
-
-        log.info("接收到iot消息: {}", message);
+    public void handlerMqttStationMessage(String topic, String message) {
+        log.info("接收到韶山Kafka消息: {}", message);
         TopicEntityVo topicEntity = new TopicEntityVo();
         topicEntity.setTopic(topic);
         topicEntity.setMessage(message);
 
+        List<IotDataVO> iotDatalist = new ArrayList<>();
         List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
+        List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
         List<IndexStateVo> indexStateList = new ArrayList<>();
 
-        JSONObject jsonObject = JSONObject.parseObject(message);
-        String dataType = jsonObject.getString("dataType");
-        String indexAddress = jsonObject.getString("address");
-        String traceId = jsonObject.getString("traceId");
-        String deviceCode = jsonObject.getString("deviceCode");
-        String gatewayId = jsonObject.getString("gatewayId");
-        String value = jsonObject.getString("value");
+        SShanStationMessage sShanStationMessage = JSON.parseObject(String.valueOf(message), SShanStationMessage.class);
+        Map<Object, Object> equipmentIndexKeyMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS_KEY_STATION);
+        sShanStationMessage.getWarns().forEach(m -> {
+            String value = m.getEventTextL1().contains("出现") ? TRUE : FALSE;
+            if (equipmentIndexKeyMap.get(m.getPointId()) != null) {
+                EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByAddress(null, m.getPointId(), null);
+                if (equipmentSpeIndex == null) {
+                    return;
+                }
+                equipmentSpeIndex.setValue(value);
+                equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
+                equipmentSpeIndex.setEquipmentType(topicEntity.getType());
+                equipmentSpeIndex.setUpdateDate(new Date());
+                equipmentSpeIndex.setTimeStamp(sShanStationMessage.getTimestamp());
+                equipmentSpeIndex.setUUID(UUIDUtils.getUUID());
+
+                IotDataVO iotDataVO = new IotDataVO();
+                iotDataVO.setKey(equipmentSpeIndex.getNameKey());
+                iotDataVO.setValue(value);
+                iotDatalist.add(iotDataVO);
+
+                QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
+                queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
+                EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
+                if (equipmentSpecific == null) {
+                    return;
+                }
+                String iotCode = equipmentSpecific.getIotCode();
+                StringBuilder endIndex = new StringBuilder(iotCode).insert(8, '/');
+                String iotTopic = "influxdb/" + endIndex;
+                JSONObject msg = new JSONObject();
+                msg.put(equipmentSpeIndex.getEquipmentIndexKey(), value);
+                mqttSendGateway.sendToMqtt(iotTopic, JSON.toJSONString(msg));
 
+                List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);
 
-        EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByIndexAddress(indexAddress, gatewayId);
-        if (equipmentSpeIndex == null) {
-            return;
-        }
+                if (eqIotCodeList.isEmpty()) {
+                    log.info("该数据{}不存在！", iotCode);
+                    return;
+                }
+                if (eqIotCodeList.size() > 1) {
+                    log.info("有重复的{}数据！", iotCode);
+                }
+                EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
+                topicEntity.setType(equipmentSpecificVo.getType());
+                topicEntity.setCode(equipmentSpecificVo.getCode());
 
-        equipmentSpeIndex.setValue(value);
-        equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
-        equipmentSpeIndex.setEquipmentType(topicEntity.getType());
-        equipmentSpeIndex.setUpdateDate(new Date());
-        equipmentSpeIndex.setGatewayId(gatewayId);
-        equipmentSpeIndex.setDataType(dataType);
-        equipmentSpeIndex.setTraceId(traceId);
-        equipmentSpeIndex.setUUID(UUIDUtils.getUUID());
-        //更新装备性能指标
-        //equipmentSpecificIndexService.updateById(equipmentSpeIndex);
-
-        tagsMap.put("key", indexAddress + "_" + gatewayId);
-
-        fieldsMap.put("traceId", traceId);
-        fieldsMap.put("address", indexAddress);
-        fieldsMap.put("value", value);
-        fieldsMap.put("gatewayId", gatewayId);
-        fieldsMap.put("dataType", dataType);
-        fieldsMap.put("equipmentId", equipmentSpeIndex.getEquipmentId());
-        fieldsMap.put("equipmentIndexName", equipmentSpeIndex.getEquipmentIndexName());
-        fieldsMap.put("equipmentIndexKey", equipmentSpeIndex.getEquipmentIndexKey());
-        fieldsMap.put("isAlarm", equipmentSpeIndex.getIsAlarm().toString());
-        fieldsMap.put("unit", equipmentSpeIndex.getUnit());
-        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-        fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
-        fieldsMap.put("equipmentIndex", JSON.toJSONString(equipmentSpeIndex));
-
-        //保存influxDB库
-        influxDbConnection.insert("iot_data", tagsMap, fieldsMap);
-
-        QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
-        queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
-        EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
-        if (equipmentSpecific == null) {
-            return;
-        }
-        String iotCode = equipmentSpecific.getIotCode();
+                //es存储数据
+                eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(equipmentSpecificVo.getSystemId()), equipmentSpecificVo.getSystemName());
+                //更新装备性能指标
+                equipmentSpecificIndexService.updateById(equipmentSpeIndex);
 
-        List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);
+                // 更新设备表指标状态
+                iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);
 
-        if (eqIotCodeList.isEmpty()) {
-            log.info("该数据{}不存在！", iotCode);
-            return;
-        }
-        if (eqIotCodeList.size() > 1) {
-            log.info("有重复的{}数据！", iotCode);
-        }
-        EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
-        topicEntity.setType(equipmentSpecificVo.getType());
-        topicEntity.setCode(equipmentSpecificVo.getCode());
+                equipmentSpecificIndexList.add(equipmentSpeIndex);
 
-        // 更新设备表指标状态
-        iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);
+                indexStateList.add(createIndexStateVo(equipmentSpeIndex));
 
-        equipmentSpecificIndexList.add(equipmentSpeIndex);
+                // 添加指标报告
+                saveEquipmentAlarmReportDay(equipmentSpeIndex);
 
-        indexStateList.add(createIndexStateVo(equipmentSpeIndex));
+                // 火眼数据构造告警指标逻辑
+                equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
 
-        // 添加指标报告
-        saveEquipmentAlarmReportDay(equipmentSpeIndex);
+                boolean alarmFlag = false;
+                Map<String, String> messageBodyMap = new HashMap<>();
+                //管网压力、泡沫罐信息、水箱液位告警处理
+                if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                        FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                        iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
+                    alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
+                }
+                //消防水池液位处理
+                if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                        iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
+                    alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
+                }
+                // 遥测数据生成告警事件、日志处理
+                if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                        FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                        iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
+                        iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                        iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
+                    handlingAlarms(equipmentSpeIndex, alarmFlag);
+                }
 
+                // 指标告警处理
+                if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
+                    equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpeIndex, messageBodyMap));
+                }
+                // 遥测遥信数据推送云端kafka
+                JSONObject jsonObjectXf = new JSONObject();
+                jsonObjectXf.put("data_class", "realdata");
 
-        // 需要在事务提交之后，否则事务隔离查询不出数据
-        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
-            @Override
-            public void afterCommit() {
+                if (equipmentSpeIndex.getIsTrend() == 1) {
+                    jsonObjectXf.put("data_type", "analog");
+                } else {
+                    jsonObjectXf.put("data_type", "state");
+                }
 
-                // 直流中心消息推送刷新
-                publishDataToDCCenterPage(equipmentSpecificIndexList);
+                String date = DateUtils.date2LongStr(new Date());
+                jsonObjectXf.put("op_type", "subscribe_emergency");
+                JSONObject jsonObjectCondition = new JSONObject();
+                jsonObjectCondition.put("station_psr_id", stationCode);
+                jsonObjectCondition.put("station_name", stationName);
+                jsonObjectCondition.put("data_upload_time", date);
+                jsonObjectXf.put("condition", jsonObjectCondition);
+
+                JSONObject jsonObjectData = new JSONObject();
+                jsonObjectData.put("psrId", stationCode);
+                jsonObjectData.put("astId", equipmentSpeIndex.getSpecificCode());
+                jsonObjectData.put("equipType", equipmentSpeIndex.getEquipmentCode());
+                jsonObjectData.put("name", equipmentSpeIndex.getEquipmentSpecificName() + "-" + equipmentSpeIndex.getEquipmentSpecificIndexName());
+                if (value.equals("true")) {
+                    jsonObjectData.put("value", "1");
+                } else {
+                    jsonObjectData.put("value", "0");
+                }
+                jsonObjectData.put("measurementType", null == equipmentSpeIndex.getEquipmentIndexKey() ? "" : equipmentSpeIndex.getEquipmentIndexKey());
+                jsonObjectData.put("dateTime", date);
+                jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
+                List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);
 
-                // 四横八纵遥测信号信息列表刷新
-                publishNormalIndexValueToPage(equipmentSpecificIndexList);
-                if ("zd".equals(system)) {
-                    System.out.println("站端系统----------------");
+                jsonObjectXf.put("data", jsonObjects);
 
-                    // 向预控系统发送消息
-                    sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);
+                // 遥测
+                if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend() == 1) {
 
-                    // 首页性能指标数据订阅
-                    mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));
+                } else {
+                    try {
+                        emqKeeper.getMqttClient().publish("emq.xf.created", jsonObjectXf.toString().getBytes(), 1, false);
+                        log.info("遥测遥信数据推送云端kafka成功");
+                    } catch (MqttException e) {
+                        log.error("遥测遥信数据推送云端kafka失败=====>" + e.getMessage());
+                        e.printStackTrace();
+                    }
+                }
 
-                    // 组态大屏消息推送，设备表实时指标修改
-                    intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);
+                // 报警数据保存
+                List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
+                if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
+                    equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
+                }
 
-                    // 数字换流站同步指标修改
-                    syncSpecificIndexsToGS(equipmentSpecificIndexList);
+                // 需要在事务提交之后，否则事务隔离查询不出数据
+                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                    @Override
+                    public void afterCommit() {
+                        equipmentSpecificAlarms.forEach(action -> {
+                            if (AlarmStatusEnum.BJ.getCode() == action.getStatus()) {
+                                alarmLogs.add(addEquipAlarmLogRecord(action));
+                                if (ValidationUtil.isEmpty(action.getAlamContent())) {
+                                    action.setAlamContent(action.getEquipmentSpecificName() + action.getEquipmentSpecificIndexName());
+                                }
+                                mqttSendGateway.sendToMqtt(TopicEnum.EQDQR.getTopic(), JSONArray.toJSON(action).toString());
+                            } else {
+                                alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
+                                        equipmentSpecificAlarmLogService, false));
+                                mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
+                                bool = Boolean.TRUE;
+                            }
+                        });
 
-                    // 则更新拓扑节点数据及告警状态
-                    updateNodeDateByEquipId(equipmentSpecificIndexList);
+                        // 直流中心消息推送刷新
+                        publishDataToDCCenterPage(equipmentSpecificIndexList);
 
-                    // 向画布推送
-                    publishDataToCanvas(equipmentSpecificIndexList);
+                        // 四横八纵遥测信号信息列表刷新
+                        publishNormalIndexValueToPage(equipmentSpecificIndexList);
+                        if ("zd".equals(system)) {
+                            System.out.println("站端系统----------------");
 
-                }
+                            // 向预控系统发送消息
+                            sendEquipSpecIndexToAutosysTopic(equipmentSpecificIndexList);
+
+                            // 首页性能指标数据订阅
+                            mqttSendGateway.sendToMqtt(indexTopic, JSON.toJSONString(indexStateList));
+
+                            // 组态大屏消息推送，设备表实时指标修改
+                            intePageSysDataRefresh(equipmentSpecificIndexList, topicEntity);
+
+                            // 数字换流站同步指标修改
+                            syncSpecificIndexsToGS(equipmentSpecificIndexList);
+
+                            // 则更新拓扑节点数据及告警状态
+                            updateNodeDateByEquipId(equipmentSpecificIndexList);
+
+                            // 向画布推送
+                            publishDataToCanvas(equipmentSpecificIndexList);
+
+                            // 向其他系统推送报警
+                            equipmentAlarmLogsToOtherSystems(alarmLogs);
+
+                            if (equipmentSpecificVo.getEcode() != null) {
+                                String ecode = equipmentSpecificVo.getEcode();
+                                boolean flag = false;
+
+                                //消防泵
+                                String[] strings = pumpCodes.split(",");
+                                for (String string : strings) {
+                                    if (ecode.startsWith(string)) {
+                                        //通知>消防应急预案
+                                        topicEntity.setType("xfb");
+                                        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
+                                        flag = true;
+                                        break;
+                                    }
+                                }
+
+                                // 消防炮
+                                String[] stringxfp = monitorCodes.split(",");
+                                if (!flag) {
+                                    for (String string1 : stringxfp) {
+                                        if (ecode.startsWith(string1)) {
+                                            //通知>消防应急预案
+                                            topicEntity.setType("xfp");
+                                            mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
+                                            flag = true;
+                                            break;
+                                        }
+                                    }
+                                }
+                                //消防水源
+                                if (!flag) {
+                                    List<Map> lit = iEquipmentSpecificSerivce.getWater(equipmentSpecificVo.getId());
+                                    if (lit != null && !lit.isEmpty()) {
+                                        topicEntity.setType("xfsy");
+                                        mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
+                                    }
+                                }
+                            }
+                        }
+                    }
+                });
             }
         });
     }
@@ -784,7 +939,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
             for (EquipmentSpecificIndex equipmentSpecificIndex : indexList) {
                 if (!ObjectUtils.isEmpty(equipmentSpecificIndex.getNameKey())
-                        && equipmentSpecificIndex.getNameKey().toLowerCase().equals(iotDataVO.getKey().toLowerCase())) {
+                        && equipmentSpecificIndex.getNameKey().equalsIgnoreCase(iotDataVO.getKey())) {
                     EquipmentSpecificIndex equipmentSpeIndex = new EquipmentSpecificIndex();
                     BeanUtils.copyProperties(equipmentSpecificIndex, equipmentSpeIndex);
                     String value = iotDataVO.getValue().toString();
@@ -816,22 +971,22 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     // 告警日志表消息内容
                     Map<String, String> messageBodyMap = new HashMap<>();
                     //管网压力、泡沫罐信息、水箱液位告警处理
-                    if (iotDataVO.getKey().toLowerCase().equals(CAFS_FoamTank_FoamTankLevel.toLowerCase()) ||
-                            FHS_PipePressureDetector_PipePressure.toLowerCase().equals(iotDataVO.getKey().toLowerCase()) ||
-                            iotDataVO.getKey().toLowerCase().equals(CAFS_WaterTank_WaterTankLevel.toLowerCase())) {
+                    if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                            FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                            iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
                         alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpecificIndex, messageBodyMap);
                     }
                     //消防水池液位处理
-                    if (iotDataVO.getKey().toLowerCase().equals(FHS_FirePoolDevice_WaterLevel.toLowerCase()) ||
-                            iotDataVO.getKey().toLowerCase().equals(FHS_WirelessliquidDetector_WaterLevel.toLowerCase())) {
+                    if (iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                            iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                         alarmFlag = doWaterPoolLevel(iotDataVO, equipmentSpecificIndex, messageBodyMap);
                     }
                     // 遥测数据生成告警事件、日志处理
-                    if (iotDataVO.getKey().toLowerCase().equals(CAFS_FoamTank_FoamTankLevel.toLowerCase()) ||
-                            FHS_PipePressureDetector_PipePressure.toLowerCase().equals(iotDataVO.getKey().toLowerCase()) ||
-                            iotDataVO.getKey().toLowerCase().equals(CAFS_WaterTank_WaterTankLevel.toLowerCase()) ||
-                            iotDataVO.getKey().toLowerCase().equals(FHS_FirePoolDevice_WaterLevel.toLowerCase()) ||
-                            iotDataVO.getKey().toLowerCase().equals(FHS_WirelessliquidDetector_WaterLevel.toLowerCase())) {
+                    if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
+                            FHS_PipePressureDetector_PipePressure.equalsIgnoreCase(iotDataVO.getKey()) ||
+                            iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel) ||
+                            iotDataVO.getKey().equalsIgnoreCase(FHS_FirePoolDevice_WaterLevel) ||
+                            iotDataVO.getKey().equalsIgnoreCase(FHS_WirelessliquidDetector_WaterLevel)) {
                         handlingAlarms(equipmentSpecificIndex, alarmFlag);
                     }
 
@@ -872,7 +1027,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     jsonObjectData.put("measurementType", null == equipmentSpecificIndex.getEquipmentIndexKey() ? "" : equipmentSpecificIndex.getEquipmentIndexKey());
                     jsonObjectData.put("dateTime", date);
                     jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
-                    List<JSONObject> jsonObjects = Arrays.asList(jsonObjectData);
+                    List<JSONObject> jsonObjects = Collections.singletonList(jsonObjectData);
 
                     jsonObjectXf.put("data", jsonObjects);
 
@@ -1003,7 +1158,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         equipmentSpecificIndex.setValue(String.valueOf(alarmFlag));
     }
 
-
     /**
      * 泡沫罐  或 者管网压力  消息发送
      *
@@ -1014,7 +1168,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         boolean alarmFlag = false;
         MessageModel model = new MessageModel();
         Map<String, Object> map = new HashMap<>();
-        if (iotDataVO.getKey().toLowerCase().equals(CAFS_FoamTank_FoamTankLevel.toLowerCase()) || iotDataVO.getKey().toLowerCase().equals(CAFS_WaterTank_WaterTankLevel.toLowerCase())) {
+        if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) || iotDataVO.getKey().equalsIgnoreCase(CAFS_WaterTank_WaterTankLevel)) {
             map = fireFightingSystemMapper.getFoamTankLevel(equipmentSpecificIndex.getEquipmentSpecificId());
         } else {
             map = fireFightingSystemMapper.getPipeNetwork(equipmentSpecificIndex.getEquipmentSpecificId());
@@ -1048,7 +1202,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             model.setRelationId(equipmentSpecificIndex.getEquipmentSpecificId().toString());
             model.setIsSendApp(false);
             model.setTerminal("WEB");
-            model.setRecivers(Arrays.asList("system"));
+            model.setRecivers(Collections.singletonList("system"));
             Map<String, String> ext = new HashMap<>();
             ext.put("content", body);
             ext.put("type", "模拟量超阈值提醒");
@@ -1084,7 +1238,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             String levelDeviceId = (String) map.get("levelDeviceId");
             BigDecimal add = nowValue;
             if (levelDeviceId.contains(",")) {
-                List<String> split = Arrays.asList(levelDeviceId.split(","));
+                String[] split = levelDeviceId.split(",");
                 int i = 1;
                 for (String s : split) {
                     if (!s.trim().equals(Long.toString(equipmentSpecificIndex.getEquipmentSpecificId()))) {
@@ -1120,7 +1274,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 model.setRelationId(map.get("id").toString());
                 model.setIsSendApp(false);
                 model.setTerminal("WEB");
-                model.setRecivers(Arrays.asList("system"));
+                model.setRecivers(Collections.singletonList("system"));
                 Map<String, String> ext = new HashMap<>();
                 ext.put("content", body);
                 ext.put("type", "模拟量超阈值提醒");
@@ -1145,10 +1299,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         if (!ObjectUtils.isEmpty(equipmentSpecificIndex) || !ObjectUtils.isEmpty(equipmentSpecificIndex.getUnit())) {
             if (UnitEnum.MM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.MM.getName().equals(equipmentSpecificIndex.getUnit())) {
                 BigDecimal divide = new BigDecimal(1000);
-                nowValue = nowValue.divide(divide, 2, BigDecimal.ROUND_HALF_UP);
+                nowValue = nowValue.divide(divide, 2, RoundingMode.HALF_UP);
             } else if (UnitEnum.CM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.CM.getName().equals(equipmentSpecificIndex.getUnit())) {
                 BigDecimal divide = new BigDecimal(100);
-                nowValue = nowValue.divide(divide, 2, BigDecimal.ROUND_HALF_UP);
+                nowValue = nowValue.divide(divide, 2, RoundingMode.HALF_UP);
             }
         }
         return nowValue;
@@ -1172,7 +1326,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         return val;
     }
 
-
     public void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties, TopicEntityVo topicEntity) {
         List<CarProperty> carIndexsList = new ArrayList<>();
         iotDatalist.forEach(iotDataVO -> {
@@ -1225,92 +1378,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         });
     }
 
-    public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey,
-                                                                   String traceId, IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService,
-                                                                   boolean flag) {
-        LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper<>();
-        queryWrapper.eq(EquipmentSpecificAlarmLog::getIotCode, iotCode);
-        queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
-        queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
-        List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper().selectList(queryWrapper);
-        if (!logs.isEmpty()) {
-            EquipmentSpecificAlarmLog log = logs.get(0);
-            EquipmentSpecific specific = equipmentSpecificMapper.selectById(log.getEquipmentSpecificId());
-            Date date = new Date();
-            SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-            String dateString = formatter.format(date);
-            logs.forEach(x -> {
-                x.setCleanTime(new Date());
-                x.setStatus(AlarmStatusEnum.HF.getCode());
-                if (!flag) {
-                    // 若已经被确警处理过，此处不再做处理
-                    if (StringUtils.isEmpty(x.getConfirmType())) {
-                        x.setConfirmType(x.getType());
-                        x.setConfirmTypeName(ConfirmAlamEnum.getTypeByCode(x.getType()));
-                        String equipmentName = StringUtil.isNotEmpty(specific.getCode()) ? specific.getName() + "(" + specific.getCode() + ")" : specific.getName();
-                        if (x.getType().equals(AlarmTypeEnum.PB.getCode())) {
-                            x.setAlarmReason(equipmentName + "频繁故障/误报，将设备报警信息屏蔽。");
-                        } else {
-                            x.setAlarmReason(equipmentName + "引起设备报警");
-                        }
-                        x.setResolveResult(dateString + "系统收到设备复归（已消除）信号，系统自动处理。");
-                        x.setConfirmUser("");
-                        x.setConfirmUserName("系统自动处理");
-                        x.setConfirmDate(date);
-                    }
-                }
-                x.setEquipmentSpecificIndexValue(TrueOrFalseEnum.fake.value);
-                if (!StringUtils.isEmpty(traceId)) {
-                    x.setTraceId(traceId);
-                }
-            });
-            equipmentSpecificAlarmLogService.updateBatchById(logs);
-        }
-        if (flag) {
-            List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService.getEquipmentSpeIndexBySpeIotCode(iotCode);
-            publishDataToCanvas(indexList);
-        }
-        return logs;
-    }
-
-    /**
-     * 画布数据消息推送
-     *
-     * @param indexList
-     */
-    public static void publishDataToCanvas(List<EquipmentSpecificIndex> indexList) {
-        if (!ObjectUtils.isEmpty(indexList)) {
-            EquipmentSpecificIndex index = indexList.stream().filter(x -> x.getUpdateDate() != null)
-                    .sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
-                    .collect(Collectors.toList()).get(0);
-            EquipmentStateVo equipmentStateVo = new EquipmentStateVo();
-            equipmentStateVo.setEquipName(index.getEquipmentSpecificName());
-            equipmentStateVo.setOrgCode(index.getOrgCode());
-            equipmentStateVo.setSpecificId(index.getEquipmentSpecificId());
-            equipmentStateVo.setEquipCode(index.getQrCode());
-            equipmentStateVo.setEquipIotCode(index.getIotCode());
-            equipmentStateVo.setStatus("");
-            equipmentStateVo.setColor(index.getEmergencyLevelColor());
-            // 添加性能指标项
-            //flag 无意义 getSpeIndex 方法其他地方共用 做了额外封装 为0时走原逻辑
-            int flag = 0;
-            equipmentStateVo.setSpeindexList(fireFightingSystemService.getSpeIndex(index.getEquipmentSpecificId(), flag));
-            Map<String, Object> topicObject = new HashMap<>();
-            topicObject.put("equipCode", equipmentStateVo.getEquipCode());
-            if (TrueOrFalseEnum.real.value.toUpperCase().equals(index.getValue().toUpperCase())) {
-                topicObject.put("color", equipmentStateVo.getColor());
-            } else {
-                topicObject.put("color", "");
-            }
-            Map<String, Object> map = JSON.parseObject(JSON.toJSONString(equipmentStateVo));
-            topicObject.put("uuid", UUID.randomUUID().toString());
-            map.put(canvasTopic, topicObject);
-
-            // 发送数据至画布
-            mqttSendGateway.sendToMqtt(canvasTopic, JSON.toJSONString(map));
-        }
-    }
-
     /**
      * 报警消息推送amos平台
      *
@@ -1594,13 +1661,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     private boolean temperatureMapIsEmpty(String traceId) {
         TemperatureAlarmDto dto = temperatureMap.get(traceId);
-        if (!ValidationUtil.isEmpty(dto) && !ValidationUtil.isEmpty(dto.getAlarmLevel())
+        return !ValidationUtil.isEmpty(dto) && !ValidationUtil.isEmpty(dto.getAlarmLevel())
                 && !ValidationUtil.isEmpty(dto.getAlarmType()) && !ValidationUtil.isEmpty(dto.getAlarmRule())
                 && !ValidationUtil.isEmpty(dto.getRuleTemperature()) && !ValidationUtil.isEmpty(dto.getTemperature())
-                && !ValidationUtil.isEmpty(dto.getThermometryUnit())) {
-            return true;
-        }
-        return false;
+                && !ValidationUtil.isEmpty(dto.getThermometryUnit());
     }
 
     /**
@@ -1622,8 +1686,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 equipmentSpecificIndex.getValue());
         equipmentSpecificAlarm.setAlamReason(equipmentSpecificAlarm.getEquipmentSpecificIndexName().replace("NB_", "")
                 + "：" + NBalarmEnum.getDescByKey(enumKey));
-        boolean flag = false;
-        if (!ValidationUtil.isEmpty(equipmentSpecificAlarm.getType())
+        boolean flag = !ValidationUtil.isEmpty(equipmentSpecificAlarm.getType())
                 && !ValidationUtil.isEmpty(equipmentSpecificAlarm.getIotCode())
                 && !ValidationUtil.isEmpty(equipmentSpecificAlarm.getEquipmentSpecificIndexKey())
                 && (NBalarmEnum.NB_liquid_level_state_0.getKey().equals(enumKey)
@@ -1632,9 +1695,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 || NBalarmEnum.NB_hydraulic_state_0.getKey().equals(enumKey)
                 || NBalarmEnum.NB_hydraulic_state_2.getKey().equals(enumKey)
                 || NBalarmEnum.NB_hydraulic_state_4.getKey().equals(enumKey)
-                || NBalarmEnum.NB_alarm_status_4.getKey().equals(enumKey))) {
-            flag = true;
-        }
+                || NBalarmEnum.NB_alarm_status_4.getKey().equals(enumKey));
         return flag;
     }
 
@@ -1941,7 +2002,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     private List<CarIndexGisVo> createCarIndexGisVo(List<CarProperty> carProperties) {
         List<CarIndexGisVo> list = new ArrayList<>();
-        long id = 0l;
+        long id = 0L;
         String iotCode = "";
         for (CarProperty action : carProperties) {
             CarIndexGisVo v = new CarIndexGisVo();

amos-boot-system-equip/src/main/resources/mapper/EquipmentSpecificIndexMapper.xml

@@ -522,7 +522,7 @@
     </select>
 
     <!-- 根据信号索引查询装备性能指标 -->
-    <select id="getEquipmentSpeIndexByIndexAddress"
+    <select id="getEquipmentSpeIndexByAddress"
             resultType="com.yeejoin.equipmanage.common.entity.EquipmentSpecificIndex">
         SELECT wesi.id      AS id,
                wei.name_key AS nameKey,
@@ -561,10 +561,16 @@
         LEFT JOIN wl_equipment_specific AS wes ON wes.id = wesi.equipment_specific_id
         LEFT JOIN wl_equipment_detail ed ON ed.id = wes.equipment_detail_id
         LEFT JOIN wl_equipment_index AS wei ON wei.id = wesi.equipment_index_id
-        WHERE
-            wesi.index_address = #{indexAddress}
-        <if test="gatewayId != null">
-            AND wesi.gateway_id = #{gatewayId}
-        </if>
+        <where>
+            <if test="indexAddress != null">
+                AND wesi.index_address = #{indexAddress}
+            </if>
+            <if test="eventAddress != null">
+                AND wesi.event_address = #{eventAddress}
+            </if>
+            <if test="gatewayId != null">
+                AND wesi.gateway_id = #{gatewayId}
+            </if>
+        </where>
     </select>
 </mapper>
\ No newline at end of file