fix(equip):解析交换站下发数据，更新指标状态

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

@@ -6,6 +6,7 @@ import com.yeejoin.equipmanage.common.entity.publics.BaseEntity;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 
 import java.util.Date;
 
@@ -15,6 +16,7 @@ import java.util.Date;
  * @since v2.0
  */
 @Data
+@EqualsAndHashCode(callSuper = true)
 @TableName("wl_equipment_specific_index")
 @ApiModel(value = "EquipmentSpecificIndex对象", description = "性能指标参数")
 public class EquipmentSpecificIndex extends BaseEntity {
@@ -126,9 +128,6 @@ public class EquipmentSpecificIndex extends BaseEntity {
     @TableField(exist = false)
     private String alamReason;
 
-    @TableField(exist = false)
-    private String valueEnum;
-
     @ApiModelProperty(value = "设备CODE")
     @TableField(exist = false)
     private String equipmentSpecificCode;
@@ -182,4 +181,26 @@ public class EquipmentSpecificIndex extends BaseEntity {
     @TableField(value = "unit")
     private String unit;
 
+    @ApiModelProperty(value = "信号的索引键key，用于唯一索引信号")
+    @TableField(value = "index_address")
+    private String indexAddress;
+
+    @ApiModelProperty(value = "品质，0为有效，1为无效")
+    @TableField(value = "quality")
+    private String quality;
+
+    @ApiModelProperty(value = "测点类型，analog/state")
+    @TableField(value = "data_type")
+    private String dataType;
+
+    @ApiModelProperty(value = "时间")
+    @TableField(value = "time_stamp")
+    private String timeStamp;
+
+    /**
+     * 指标值枚举
+     */
+    @ApiModelProperty(value = "指标值枚举")
+    @TableField("value_enum")
+    private String valueEnum;
 }

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

@@ -124,6 +124,7 @@ public class EquipmentIotMqttReceiveConfig {
         list.addAll(ConfigPageTopicEnum.getEnumTopicList());          //大屏数据推送接口订阅
         list.add("+/+/property"); // 添加iot車輛裝備數據上報事件监听
         list.add("+/+/event"); // 添加iot事件监听
+        list.add("+/+/transmit"); // 添加交换站事件监听
         String[] arr = list.toArray(new String[list.size()]);
         adapter = new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttPahoClientFactory(), arr);
         adapter.setCompletionTimeout(completionTimeout);
@@ -147,6 +148,8 @@ public class EquipmentIotMqttReceiveConfig {
                     mqttReceiveService.handlerMqttIncrementMessage(topic, msg);
                 } else if (dataType.equals("event") && StringUtil.isNotEmpty(msg)) {
                     mqttEventReceiveService.handlerMqttIncrementMessage(topic, msg);
+                }else if (dataType.equals("transmit") && StringUtil.isNotEmpty(msg)){
+                    mqttReceiveService.handlerMqttRomaMessage(topic,msg);
                 }
             }
         };

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/controller/EquipmentIndexController.java

@@ -199,6 +199,7 @@ public class EquipmentIndexController {
                         for (EquipmentSpecificIndex y : indexList) {
                             y.setEquipmentIndexKey(equipmentIndex.getPerfQuotaDefinitionId());
                             y.setEquipmentIndexName(equipmentIndex.getPerfQuotaName());
+                            y.setValueEnum(equipmentIndex.getValueEnum());
                             if (bool.get() && !ObjectUtils.isEmpty(signalClassify)) {
                                 y.setEmergencyLevelColor(signalClassify.getEmergencyLevelColor());
                                 y.setIsAlarm(signalClassify.getIsAlarm());

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

@@ -113,4 +113,6 @@ public interface EquipmentSpecificIndexMapper extends BaseMapper<EquipmentSpecif
     List<Map<String, Object>> getEquipSpecificScrap();
 
     List<EquipmentSpecificIndex> getEquipIndexInIndex(@Param("list") List<String> listIndex);
+
+    EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress);
 }

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

@@ -34,4 +34,11 @@ public interface IEquipmentSpecificIndexService extends IService<EquipmentSpecif
      * @return 列表
      */
     List<EquipmentSpecificIndex> getEquipmentSpeIndexByIndex(List<String> listIndex);
+
+    /**
+     * 根据信号索引查询装备性能指标
+     * @param indexAddress indexAddress
+     * @return EquipmentSpecificIndex
+     */
+    EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress);
 }

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

@@ -17,4 +17,11 @@ public interface MqttReceiveService {
      * @param message 消息内容
      */
     void handlerMqttIncrementMessage(String topic, String message);
+
+    /**
+     * 处理交换站消息数据
+     * @param topic 主题
+     * @param message 消息内容
+     */
+    void handlerMqttRomaMessage(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

@@ -28,4 +28,9 @@ public class EquipmentSpecificIndexServiceImpl extends ServiceImpl<EquipmentSpec
     public List<EquipmentSpecificIndex> getEquipmentSpeIndexByIndex(List<String> listIndex) {
         return this.baseMapper.getEquipIndexInIndex(listIndex);
     }
+
+    @Override
+    public EquipmentSpecificIndex getEquipmentSpeIndexByIndexAddress(String indexAddress) {
+        return this.baseMapper.getEquipmentSpeIndexByIndexAddress(indexAddress);
+    }
 }

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

@@ -1598,6 +1598,7 @@ public class EquipmentSpecificSerivceImpl extends ServiceImpl<EquipmentSpecificM
             equipmentSpecificIndex.setEmergencyLevel(index.getEmergencyLevel());
             equipmentSpecificIndex.setEmergencyLevelDescribe(index.getEmergencyLevelDescribe());
             equipmentSpecificIndex.setUnit(index.getUnit());
+            equipmentSpecificIndex.setValueEnum(index.getValueEnum());
             equipmentSpecificIndices.add(equipmentSpecificIndex);
         });
     }

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

@@ -72,8 +72,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     private static Map<String, TemperatureAlarmDto> temperatureMap = new HashMap<>();
 
     static IEquipmentSpecificIndexService equipmentSpecificIndexService;
+
     @Autowired
-    public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService){
+    public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService) {
         MqttReceiveServiceImpl.equipmentSpecificIndexService = equipmentSpecificIndexService;
     }
 
@@ -136,8 +137,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
 
     static EquipmentSpecificMapper equipmentSpecificMapper;
+
     @Autowired
-    public void setEquipmentSpecificMapper(EquipmentSpecificMapper equipmentSpecificMapper){
+    public void setEquipmentSpecificMapper(EquipmentSpecificMapper equipmentSpecificMapper) {
         MqttReceiveServiceImpl.equipmentSpecificMapper = equipmentSpecificMapper;
     }
 
@@ -145,15 +147,17 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     FireFightingSystemMapper FireFightingSystemMapper;
 
     static IFireFightingSystemService fireFightingSystemService;
+
     @Autowired
-    public void setFireFightingSystemService(IFireFightingSystemService fireFightingSystemService){
+    public void setFireFightingSystemService(IFireFightingSystemService fireFightingSystemService) {
         MqttReceiveServiceImpl.fireFightingSystemService = fireFightingSystemService;
     }
 
 
     static MqttSendGateway mqttSendGateway;
+
     @Autowired
-    public void setMqttSendGateway(MqttSendGateway mqttSendGateway){
+    public void setMqttSendGateway(MqttSendGateway mqttSendGateway) {
         MqttReceiveServiceImpl.mqttSendGateway = mqttSendGateway;
     }
 
@@ -182,8 +186,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     private SystemctlFeign systemctlFeign;
 
     private static RemoteSecurityService remoteSecurityService;
+
     @Autowired
-    public void setRemoteSecurityService(RemoteSecurityService remoteSecurityService){
+    public void setRemoteSecurityService(RemoteSecurityService remoteSecurityService) {
         MqttReceiveServiceImpl.remoteSecurityService = remoteSecurityService;
     }
 
@@ -214,8 +219,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     private String carTopic;
 
     private static String canvasTopic;
+
     @Value("${equip.point.equipmentdata.topic}")
-    public void setCanvasTopic(String canvasTopic){
+    public void setCanvasTopic(String canvasTopic) {
         MqttReceiveServiceImpl.canvasTopic = canvasTopic;
     }
 
@@ -241,8 +247,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     private Boolean isOpenTelemetering;
 
     private static Boolean jcsSwitch;
+
     @Value("${systemctl.jcs.switch}")
-    public void setJcsSwitch(Boolean jcsSwitch){
+    public void setJcsSwitch(Boolean jcsSwitch) {
         MqttReceiveServiceImpl.jcsSwitch = jcsSwitch;
     }
 
@@ -310,10 +317,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         // 发送emq消息转kafka
         JSONObject jsonObject = new JSONObject();
         jsonObject.put("topic", topic);
-        jsonObject.put("data",message);
+        jsonObject.put("data", message);
 
         try {
-            emqKeeper.getMqttClient().publish("emq.iot.created",jsonObject.toString().getBytes(),1,false);
+            emqKeeper.getMqttClient().publish("emq.iot.created", jsonObject.toString().getBytes(), 1, false);
         } catch (MqttException e) {
             log.info(String.format("发送eqm转kafka消息失败：%s", e.getMessage()));
         }
@@ -326,8 +333,259 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             }).collect(Collectors.toList());
             realTimeDateProcessing(topicEntity, collect, vo);
         } else {
-            realTimeDateProcessing(topicEntity, iotDatalist,vo);
+            realTimeDateProcessing(topicEntity, iotDatalist, vo);
+        }
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public void handlerMqttRomaMessage(String topic, String 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 = dataType != null ? jsonObject.getString("key") : jsonObject.getString("scadaid");
+        String quality = jsonObject.getString("quality");
+        String value = jsonObject.getInteger("value") ==1 ? "true":"false";
+        String timeStamp = dataType != null ? jsonObject.getString("time_stamp") : jsonObject.getString("timestamp");
+
+        EquipmentSpecificIndex equipmentSpeIndex = equipmentSpecificIndexService.getEquipmentSpeIndexByIndexAddress(indexAddress);
+        if (equipmentSpeIndex == null){
+            return;
+        }
+        equipmentSpeIndex.setValue(value);
+        equipmentSpeIndex.setValueLabel(valueTranslate(value, equipmentSpeIndex.getValueEnum()));
+        equipmentSpeIndex.setEquipmentType(topicEntity.getType());
+        equipmentSpeIndex.setUpdateDate(new Date());
+        equipmentSpeIndex.setQuality(quality);
+        equipmentSpeIndex.setDataType(dataType);
+        equipmentSpeIndex.setTimeStamp(timeStamp);
+        equipmentSpeIndex.setUUID(UUIDUtils.getUUID());
+
+        IotDataVO iotDataVO = new IotDataVO();
+        iotDataVO.setKey(equipmentSpeIndex.getNameKey());
+        iotDataVO.setValue(value);
+        iotDatalist.add(iotDataVO);
+
+//        iEquipmentSpecificSerivce.getEquipSpecificDetailsByEquipmentId(equipmentSpeIndex.getEquipmentSpecificId());
+        QueryWrapper<EquipmentSpecific> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("id", equipmentSpeIndex.getEquipmentSpecificId());
+        EquipmentSpecific equipmentSpecific = iEquipmentSpecificSerivce.getOne(queryWrapper);
+        if (equipmentSpecific == null) {
+            return;
+        }
+        String iotCode = equipmentSpecific.getIotCode();
+
+        List<EquipmentSpecificVo> eqIotCodeList = iEquipmentSpecificSerivce.getEquipAndCarIotcodeByIotcode(iotCode);
+
+        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());
+
+        //es存储数据
+        eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(equipmentSpecificVo.getSystemId()), equipmentSpecificVo.getSystemName());
+        //更新装备性能指标
+        equipmentSpecificIndexService.updateById(equipmentSpeIndex);
+
+        // 更新设备表指标状态
+        iEquipmentSpecificSerivce.updateEquipmentSpecIndexRealtimeData(equipmentSpeIndex);
+
+        equipmentSpecificIndexList.add(equipmentSpeIndex);
+
+        indexStateList.add(createIndexStateVo(equipmentSpeIndex));
+
+        // 添加指标报告
+        saveEquipmentAlarmReportDay(equipmentSpeIndex);
+
+        // 火眼数据构造告警指标逻辑
+        equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
+
+        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())) {
+            alarmFlag = doFoamTankLevel(iotDataVO, equipmentSpeIndex, messageBodyMap);
+        }
+        //消防水池液位处理
+        if (iotDataVO.getKey().toLowerCase().equals(FHS_FirePoolDevice_WaterLevel.toLowerCase()) ||
+                iotDataVO.getKey().toLowerCase().equals(FHS_WirelessliquidDetector_WaterLevel.toLowerCase())) {
+            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())) {
+            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");
+
+        if (equipmentSpeIndex.getIsTrend() == 1) {
+            jsonObjectXf.put("data_type", "analog");
+        } else {
+            jsonObjectXf.put("data_type", "state");
         }
+
+        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 if (value.equals("false")) {
+            jsonObjectData.put("value", "0");
+        } else {
+            jsonObjectData.put("value", value);
+        }
+        jsonObjectData.put("measurementType", null == equipmentSpeIndex.getEquipmentIndexKey() ? "" : equipmentSpeIndex.getEquipmentIndexKey());
+        jsonObjectData.put("dateTime", date);
+        jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
+        List<JSONObject> jsonObjects = Arrays.asList(jsonObjectData);
+
+        jsonObjectXf.put("data", jsonObjects);
+
+        // 遥测
+        if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend() == 1) {
+
+        } 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();
+            }
+        }
+
+        // 报警数据保存
+        List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
+        if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
+            equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
+        }
+
+        // 需要在事务提交之后，否则事务隔离查询不出数据
+        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;
+                    }
+                });
+
+                // 直流中心消息推送刷新
+                publishDataToDCCenterPage(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.size() > 0) {
+                                topicEntity.setType("xfsy");
+                                mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
     }
 
     /**
@@ -336,7 +594,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
      * @param topicEntity
      * @param iotDatalist
      */
-    public void realTimeDateProcessing(TopicEntityVo topicEntity, List<IotDataVO> iotDatalist,EquipmentSpecificVo vo) {
+    public void realTimeDateProcessing(TopicEntityVo topicEntity, List<IotDataVO> iotDatalist, EquipmentSpecificVo vo) {
         String iotCode = topicEntity.getIotCode();
         if (EquipAndCarEnum.equip.type.equals(topicEntity.getType())) {
             List<EquipmentSpecificIndex> indexList = equipmentSpecificIndexService
@@ -344,7 +602,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             if (ObjectUtils.isEmpty(indexList)) {
                 return;
             }
-            equipRealTimeDate(iotDatalist, indexList, topicEntity,vo);
+            equipRealTimeDate(iotDatalist, indexList, topicEntity, vo);
             String bizOrgCode = indexList.get(0).getBizOrgCode();
             //  redis缓存指定指标、指定时长物联数据
             pressurePumpService.saveDataToRedis(iotDatalist, iotCode, bizOrgCode);
@@ -353,7 +611,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             if (ObjectUtils.isEmpty(carProperties)) {
                 return;
             }
-            carRealTimeDate(iotDatalist, carProperties,topicEntity);
+            carRealTimeDate(iotDatalist, carProperties, topicEntity);
         }
     }
 
@@ -365,7 +623,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
      * @param topicEntity
      */
     public void equipRealTimeDate(List<IotDataVO> iotDatalist, List<EquipmentSpecificIndex> indexList,
-                                  TopicEntityVo topicEntity,EquipmentSpecificVo vo) {
+                                  TopicEntityVo topicEntity, EquipmentSpecificVo vo) {
         List<EquipmentSpecificIndex> equipmentSpecificIndexList = new ArrayList<>();
         List<EquipmentSpecificAlarm> equipmentSpecificAlarms = new ArrayList<>();
         List<IndexStateVo> indexStateList = new ArrayList<>();
@@ -377,7 +635,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             String indexValue = iotDataVO.getValue().toString();
             // 稳压泵启停信号处理
             if (indexKey.equals(pressurePumpStart)) {
-                pressurePump(indexKey,indexValue, iotDatalist, topicEntity);
+                pressurePump(indexKey, indexValue, iotDatalist, topicEntity);
             }
             for (EquipmentSpecificIndex equipmentSpecificIndex : indexList) {
                 if (!ObjectUtils.isEmpty(equipmentSpecificIndex.getNameKey())
@@ -394,8 +652,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     equipmentSpeIndex.setUUID(UUIDUtils.getUUID());
 
                     //es存储数据
-                    eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex,String.valueOf(vo.getSystemId()),vo.getSystemName());
-
+                    eSeqService.saveESEquiplistSpecificBySystemESVO(equipmentSpeIndex, String.valueOf(vo.getSystemId()), vo.getSystemName());
 
                     equipmentSpecificIndexService.updateById(equipmentSpeIndex);
 
@@ -441,7 +698,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     JSONObject jsonObjectXf = new JSONObject();
                     jsonObjectXf.put("data_class", "realdata");
 
-                    if(equipmentSpeIndex.getIsTrend() == 1) {
+                    if (equipmentSpeIndex.getIsTrend() == 1) {
                         jsonObjectXf.put("data_type", "analog");
                     } else {
                         jsonObjectXf.put("data_type", "state");
@@ -452,36 +709,37 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     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);
+                    jsonObjectCondition.put("data_upload_time", date);
+                    jsonObjectXf.put("condition", jsonObjectCondition);
 
                     JSONObject jsonObjectData = new JSONObject();
                     jsonObjectData.put("psrId", stationCode);
                     jsonObjectData.put("astId", equipmentSpecificIndex.getSpecificCode());
                     jsonObjectData.put("equipType", equipmentSpecificIndex.getEquipmentCode());
-                    jsonObjectData.put("name", equipmentSpecificIndex.getEquipmentSpecificName()+"-"+equipmentSpecificIndex.getEquipmentSpecificIndexName());
-                    if(value.equals("true")) {
-                        jsonObjectData.put("value","1");
+                    jsonObjectData.put("name", equipmentSpecificIndex.getEquipmentSpecificName() + "-" + equipmentSpecificIndex.getEquipmentSpecificIndexName());
+                    if (value.equals("true")) {
+                        jsonObjectData.put("value", "1");
                     } else if (value.equals("false")) {
-                        jsonObjectData.put("value","0");
+                        jsonObjectData.put("value", "0");
                     } else {
                         jsonObjectData.put("value", value);
                     }
-                    jsonObjectData.put("measurementType", null == equipmentSpecificIndex.getEquipmentIndexKey() ? "" :equipmentSpecificIndex.getEquipmentIndexKey());
-                    jsonObjectData.put("dateTime",date);
-                    jsonObjectData.put("quality","0"); // 量测质量码：0 有效，1 无效
+                    jsonObjectData.put("measurementType", null == equipmentSpecificIndex.getEquipmentIndexKey() ? "" : equipmentSpecificIndex.getEquipmentIndexKey());
+                    jsonObjectData.put("dateTime", date);
+                    jsonObjectData.put("quality", "0"); // 量测质量码：0 有效，1 无效
                     List<JSONObject> jsonObjects = Arrays.asList(jsonObjectData);
 
                     jsonObjectXf.put("data", jsonObjects);
 
                     // 遥测
-                    if(!isOpenTelemetering && equipmentSpeIndex.getIsTrend() == 1) {
+                    if (!isOpenTelemetering && equipmentSpeIndex.getIsTrend() == 1) {
+
                     } else {
                         try {
-                            emqKeeper.getMqttClient().publish("emq.xf.created",jsonObjectXf.toString().getBytes(),1,false);
+                            emqKeeper.getMqttClient().publish("emq.xf.created", jsonObjectXf.toString().getBytes(), 1, false);
                             log.info("遥测遥信数据推送云端kafka成功");
                         } catch (MqttException e) {
-                            log.error("遥测遥信数据推送云端kafka失败=====>" +  e.getMessage());
+                            log.error("遥测遥信数据推送云端kafka失败=====>" + e.getMessage());
                             e.printStackTrace();
                         }
                     }
@@ -491,8 +749,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
         // 报警数据保存
         List<EquipmentSpecificAlarmLog> alarmLogs = new ArrayList<>();
-        if(!ObjectUtils.isEmpty(equipmentSpecificAlarms)){
-        	equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
+        if (!ObjectUtils.isEmpty(equipmentSpecificAlarms)) {
+            equipmentSpecificAlarmService.saveOrUpdateBatch(equipmentSpecificAlarms);
         }
         // 需要在事务提交之后，否则事务隔离查询不出数据
         TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@@ -518,7 +776,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
                 // 四横八纵遥测信号信息列表刷新
                 publishNormalIndexValueToPage(equipmentSpecificIndexList);
-                if("zd".equals(system)){
+                if ("zd".equals(system)) {
                     System.out.println("站端系统----------------");
 
                     // 向预控系统发送消息
@@ -539,54 +797,54 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     // 向画布推送
                     publishDataToCanvas(equipmentSpecificIndexList);
 
-	                // 向其他系统推送报警
-	                equipmentAlarmLogsToOtherSystems(alarmLogs);
-	
-	                if(vo.getEcode()!=null){
-	                    String ecode= vo.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(vo.getId());
-	                        if(lit!=null&& lit.size()>0){
-	                            topicEntity.setType("xfsy");
-	                            mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
-	                        }
-	                    }
-	                }
+                    // 向其他系统推送报警
+                    equipmentAlarmLogsToOtherSystems(alarmLogs);
+
+                    if (vo.getEcode() != null) {
+                        String ecode = vo.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(vo.getId());
+                            if (lit != null && lit.size() > 0) {
+                                topicEntity.setType("xfsy");
+                                mqttSendGateway.sendToMqtt(emergencyDisposalIndicators, JSONObject.toJSONString(topicEntity));
+                            }
+                        }
+                    }
                 }
             }
         });
 
     }
 
-    private void handlingAlarms(EquipmentSpecificIndex equipmentSpecificIndex, boolean alarmFlag){
+    private void handlingAlarms(EquipmentSpecificIndex equipmentSpecificIndex, boolean alarmFlag) {
         equipmentSpecificIndex.setIsAlarm(1);
         equipmentSpecificIndex.setValue(String.valueOf(alarmFlag));
     }
@@ -594,8 +852,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     /**
      * 泡沫罐  或 者管网压力  消息发送
-     * @param iotDataVO  iotDataVO
-     * @param equipmentSpecificIndex  equipmentSpecificIndex
+     *
+     * @param iotDataVO              iotDataVO
+     * @param equipmentSpecificIndex equipmentSpecificIndex
      */
     private boolean doFoamTankLevel(IotDataVO iotDataVO, EquipmentSpecificIndex equipmentSpecificIndex, Map<String, String> messageBody) {
         boolean alarmFlag = false;
@@ -609,9 +868,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         BigDecimal minValue = getBigDecimal(map.get("minValues"));
         BigDecimal maxValue = getBigDecimal(map.get("maxValues"));
         BigDecimal nowValue = getBigDecimal(iotDataVO.getValue());
-        BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(),iotDataVO.getKey().toLowerCase() , nowValue);
-        if (!ObjectUtils.isEmpty(checkValue)){
-            nowValue =checkValue;
+        BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(), iotDataVO.getKey().toLowerCase(), nowValue);
+        if (!ObjectUtils.isEmpty(checkValue)) {
+            nowValue = checkValue;
         }
         if (nowValue.compareTo(minValue) < 0 || nowValue.compareTo(maxValue) > 0) {
             String body = "";
@@ -625,7 +884,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     equipmentSpecificIndex.getEquipmentSpecificName() + "-" + equipmentSpecificIndex.getLocation(),
                     nowValue,
                     nowValue.compareTo(minValue) < 0 ? "低于最低报警阈值" + minValue : "超过最高报警阈值" + maxValue
-                    );
+            );
             model.setTitle("模拟量提醒");
             model.setBody(bodyMain);
             model.setMsgType("FoamTankOrPipeNetwork");
@@ -651,7 +910,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 //            log.info(String.format("调用平台消息服务成功：%s", JSON.toJSONString(model)));
             alarmFlag = true;
         }
-        return  alarmFlag;
+        return alarmFlag;
     }
 
     /**
@@ -682,9 +941,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 }
                 nowValue = add.divide(new BigDecimal(i), 2, RoundingMode.HALF_UP);
             }
-            BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(),iotDataVO.getKey().toLowerCase() , nowValue);
-            if (!ObjectUtils.isEmpty(checkValue)){
-                nowValue =checkValue;
+            BigDecimal checkValue = checkUnit(equipmentSpecificIndex.getEquipmentSpecificId(), iotDataVO.getKey().toLowerCase(), nowValue);
+            if (!ObjectUtils.isEmpty(checkValue)) {
+                nowValue = checkValue;
             }
             if (nowValue.compareTo(minValue) < 0 || nowValue.compareTo(maxValue) > 0) {
                 String body = "";
@@ -724,42 +983,43 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         return alarmFlag;
     }
 
-    private BigDecimal checkUnit(Long equipmentSpecificId, String indexKey,BigDecimal nowValue){
+    private BigDecimal checkUnit(Long equipmentSpecificId, String indexKey, BigDecimal nowValue) {
         LambdaQueryWrapper<EquipmentSpecificIndex> lambda = new QueryWrapper<EquipmentSpecificIndex>().lambda();
-        lambda.eq(EquipmentSpecificIndex :: getEquipmentSpecificId, equipmentSpecificId);
-        lambda.eq(EquipmentSpecificIndex :: getEquipmentIndexKey, indexKey);
+        lambda.eq(EquipmentSpecificIndex::getEquipmentSpecificId, equipmentSpecificId);
+        lambda.eq(EquipmentSpecificIndex::getEquipmentIndexKey, indexKey);
         EquipmentSpecificIndex equipmentSpecificIndex = equipmentSpecificIndexMapper.selectOne(lambda);
-        if (!ObjectUtils.isEmpty(equipmentSpecificIndex) || !ObjectUtils.isEmpty(equipmentSpecificIndex.getUnit())){
-            if (UnitEnum.MM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.MM.getName().equals(equipmentSpecificIndex.getUnit())){
+        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);
-            }else if (UnitEnum.CM.getKey().equalsIgnoreCase(equipmentSpecificIndex.getUnit()) || UnitEnum.CM.getName().equals(equipmentSpecificIndex.getUnit())){
+                nowValue = nowValue.divide(divide, 2, BigDecimal.ROUND_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, BigDecimal.ROUND_HALF_UP);
             }
         }
         return nowValue;
     }
-    private BigDecimal getBigDecimal( Object value ) {
+
+    private BigDecimal getBigDecimal(Object value) {
         BigDecimal val = null;
-        if( value != null ) {
-            if( value instanceof BigDecimal ) {
+        if (value != null) {
+            if (value instanceof BigDecimal) {
                 val = (BigDecimal) value;
-            } else if( value instanceof String ) {
-                val = new BigDecimal( (String) value );
-            } else if( value instanceof BigInteger) {
-                val = new BigDecimal( (BigInteger) value );
-            } else if( value instanceof Number ) {
+            } else if (value instanceof String) {
+                val = new BigDecimal((String) value);
+            } else if (value instanceof BigInteger) {
+                val = new BigDecimal((BigInteger) value);
+            } else if (value instanceof Number) {
                 val = new BigDecimal(String.valueOf(value));
             } else {
-                throw new ClassCastException("Not possible to coerce ["+value+"] from class "+value.getClass()+" into a BigDecimal.");
+                throw new ClassCastException("Not possible to coerce [" + value + "] from class " + value.getClass() + " into a BigDecimal.");
             }
         }
         return val;
     }
 
 
-    public void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties,TopicEntityVo topicEntity) {
+    public void carRealTimeDate(List<IotDataVO> iotDatalist, List<CarProperty> carProperties, TopicEntityVo topicEntity) {
         List<CarProperty> carIndexsList = new ArrayList<>();
         iotDatalist.forEach(iotDataVO -> {
             // 对指标key为labels的数据处理
@@ -819,7 +1079,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         queryWrapper.eq(EquipmentSpecificAlarmLog::getEquipmentSpecificIndexKey, equipmentSpecificIndexKey);
         queryWrapper.ne(EquipmentSpecificAlarmLog::getStatus, AlarmStatusEnum.HF.getCode());
         List<EquipmentSpecificAlarmLog> logs = equipmentSpecificAlarmLogService.getBaseMapper().selectList(queryWrapper);
-        if(!logs.isEmpty()){
+        if (!logs.isEmpty()) {
             EquipmentSpecificAlarmLog log = logs.get(0);
             EquipmentSpecific specific = equipmentSpecificMapper.selectById(log.getEquipmentSpecificId());
             Date date = new Date();
@@ -990,8 +1250,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         equipmentSpecificAlarmLog.setEquipmentSpecificIndexName(StringUtil.isNotEmpty(equipmentSpecificAlarm.getMessageBody()) ? equipmentSpecificAlarm.getMessageBody() : equipmentSpecificAlarm.getEquipmentSpecificIndexName());
 
         //调整为按照设备归属（可归属公司或者部门)
-        Map<String,Object> mapd= iEquipmentSpecificSerivce.getStationCode(equipmentSpecificAlarm.getEquipmentSpecificId());
-        if(mapd!=null){
+        Map<String, Object> mapd = iEquipmentSpecificSerivce.getStationCode(equipmentSpecificAlarm.getEquipmentSpecificId());
+        if (mapd != null) {
             equipmentSpecificAlarmLog.setStationCode(mapd.get("station_code").toString());
             equipmentSpecificAlarmLog.setStationName(mapd.get("station_name").toString());
         }
@@ -1043,8 +1303,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         equipmentSpecificAlarm.setMessageBody(!ObjectUtils.isEmpty(messageBody) ? messageBody.get("messageBody") : "");
 
         //调整为按照设备归属（可归属公司或者部门
-        Map<String,Object> mapd= iEquipmentSpecificSerivce.getStationCode(equipmentSpecificIndex.getEquipmentSpecificId());
-        if(mapd!=null){
+        Map<String, Object> mapd = iEquipmentSpecificSerivce.getStationCode(equipmentSpecificIndex.getEquipmentSpecificId());
+        if (mapd != null) {
             equipmentSpecificAlarm.setStationCode(mapd.get("station_code").toString());
             equipmentSpecificAlarm.setStationName(mapd.get("station_name").toString());
         }
@@ -1054,7 +1314,6 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 //        equipmentSpecificAlarm.setStationName(stationName);
 
 
-
         equipmentSpecificAlarms.add(equipmentSpecificAlarm);
         return equipmentSpecificAlarms;
     }
@@ -1131,6 +1390,11 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                         || "temperature".equals(x.getKey()) || "ruleTemperature".equals(x.getKey())
                         || "thermometryUnit".equals(x.getKey()) || "alarmRule".equals(x.getKey()))
                 .collect(Collectors.toList());
+
+//        List<IotDataVO> iotDataVOS = iotDatalist.stream()
+//                .filter(x -> Stream.of("alarmLevel", "temperature", "thermometryUnit", "ruleTemperature", "alarmType", "alarmRule")
+//                        .anyMatch(s -> s.equals(x.getKey()))).collect(Collectors.toList());
+
         if (iotDataVOs.size() > 0) {
             Map<String, Object> map = iotDatalist.stream()
                     .collect(Collectors.toMap(IotDataVO::getKey, IotDataVO::getValue));
@@ -1575,15 +1839,15 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 // 1. 获取需要校验的值
                 PressurePumpValueEnum valueEnum = PressurePumpValueEnum.getByCode(pressurePumpEnum.getCompareValue());
                 assert valueEnum != null;
-                EquipmentSpecificIndex data = getPressurePumpDateByType(indexKey,valueEnum, topicEntity, equipmentSpeIndexList, pressurePumpEnum);
+                EquipmentSpecificIndex data = getPressurePumpDateByType(indexKey, valueEnum, topicEntity, equipmentSpeIndexList, pressurePumpEnum);
                 Date newDate = new Date();
                 // 2. 校验
                 if (!ObjectUtils.isEmpty(data.getUpdateDate())) {
                     checkValueByDate(data, newDate, pressurePumpEnum);
-                }else {
+                } else {
                     // 稳压泵漏水告警恢复
                     List<EquipmentSpecificIndex> collect = equipmentSpeIndexList.stream().filter(item -> !ObjectUtils.isEmpty(item.getIotCode()) && item.getIotCode().equals(topicEntity.getIotCode())).collect(Collectors.toList());
-                    if (!ObjectUtils.isEmpty(collect) && !ObjectUtils.isEmpty(collect.get(0)) && !ObjectUtils.isEmpty(collect.get(0).getEquipmentId())){
+                    if (!ObjectUtils.isEmpty(collect) && !ObjectUtils.isEmpty(collect.get(0)) && !ObjectUtils.isEmpty(collect.get(0).getEquipmentId())) {
                         equipmentSpecificAlarmLogService.pressurePumpRestore(collect.get(0).getEquipmentId());
                     }
                 }
@@ -1603,12 +1867,12 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         } else {
             throw new BadRequest("装备物联编码错误，请确认！");
         }
-        String jobName = topicEntity.getIotCode()+"_"+indexKey;
-        String triggerName = PUMP_TRIGGER_NAME+"-"+topicEntity.getIotCode();
+        String jobName = topicEntity.getIotCode() + "_" + indexKey;
+        String triggerName = PUMP_TRIGGER_NAME + "-" + topicEntity.getIotCode();
         switch (valueEnum) {
             case LAST_STOP:
                 List<EquipmentSpecificIndex> lastStop = equipmentSpeIndexList.stream().filter(e ->
-                        StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
+                                StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                         .collect(Collectors.toList());
                 if (!CollectionUtils.isEmpty(lastStop)) {
                     EquipmentSpecificIndex aFalse = getIotDate(equipmentSpecificIndex, lastStop, prefix, suffix, "false");
@@ -1618,12 +1882,12 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             case LAST_START:
                 boolean b = QuartzManager.checkExists(jobName, PUMP_JOB_GROUP_NAME);
                 // 删除这个稳压泵的监听任务
-                if(b) {
-                    QuartzManager.removeJob(jobName,PUMP_JOB_GROUP_NAME,triggerName,PUMP_TRIGGER_GROUP_NAME);
+                if (b) {
+                    QuartzManager.removeJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME);
                 }
 
                 List<EquipmentSpecificIndex> lastStart = equipmentSpeIndexList.stream().filter(e ->
-                        StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
+                                StringUtil.isNotEmpty(e.getValue()) && e.getIotCode().equals(iotCode) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                         .collect(Collectors.toList());
                 if (!CollectionUtils.isEmpty(lastStart)) {
                     EquipmentSpecificIndex aTrue = getIotDate(equipmentSpecificIndex, lastStart, prefix, suffix, "true");
@@ -1633,7 +1897,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 break;
             case LATELY_STOP:
                 List<EquipmentSpecificIndex> latelyStop = equipmentSpeIndexList.stream().filter(e ->
-                        StringUtil.isNotEmpty(e.getValue()) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
+                                StringUtil.isNotEmpty(e.getValue()) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                         .collect(Collectors.toList());
                 if (!CollectionUtils.isEmpty(latelyStop)) {
                     EquipmentSpecificIndex aFalse = getIotDate(equipmentSpecificIndex, latelyStop, prefix, null, "false");
@@ -1643,7 +1907,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 break;
             case LATELY_START:
                 List<EquipmentSpecificIndex> latelyStart = equipmentSpeIndexList.stream().filter(e ->
-                        StringUtil.isNotEmpty(e.getValue()) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
+                                StringUtil.isNotEmpty(e.getValue()) && pressurePumpStart.equals(e.getEquipmentIndexKey())).sorted(Comparator.comparing(EquipmentSpecificIndex::getUpdateDate).reversed())
                         .collect(Collectors.toList());
                 if (!CollectionUtils.isEmpty(latelyStart)) {
                     EquipmentSpecificIndex aTrue = getIotDate(equipmentSpecificIndex, latelyStart, prefix, null, "true");
@@ -1661,7 +1925,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     }
 
 
-    private EquipmentSpecificIndex getIotDate(EquipmentSpecificIndex equipmentSpecificIndex, List<EquipmentSpecificIndex>  listData, String prefix, String suffix, String flag) {
+    private EquipmentSpecificIndex getIotDate(EquipmentSpecificIndex equipmentSpecificIndex, List<EquipmentSpecificIndex> listData, String prefix, String suffix, String flag) {
         ResponseModel start = iotFeign.selectOne(remoteSecurityService.getServerToken().getAppKey(), remoteSecurityService.getServerToken().getProduct(), remoteSecurityService.getServerToken().getToke(), "1", prefix, suffix, flag, pressurePumpStart);
         if (200 == start.getStatus() && !ObjectUtils.isEmpty(start.getResult())) {
             String json1 = JSON.toJSONString(start.getResult());
@@ -1671,7 +1935,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             String startTime = collect.get(0).get("time").substring(0, 19).replace("T", " ");
             Date startDate = DateUtils.dateAddHours(DateUtils.longStr2Date(startTime), +8);
             listData.get(0).setUpdateDate(startDate);
-            BeanUtils.copyProperties(listData.get(0),equipmentSpecificIndex);
+            BeanUtils.copyProperties(listData.get(0), equipmentSpecificIndex);
         }
         return equipmentSpecificIndex;
     }
@@ -1684,7 +1948,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         try {
             long d1 = df.parse(df.format(data.getUpdateDate())).getTime();
             long d2 = df.parse(df.format(newDate)).getTime();
-            diff = (d2-d1)/1000/60;
+            diff = (d2 - d1) / 1000 / 60;
         } catch (Exception e) {
             log.error("时间转换失败" + e.getMessage());
             return;
@@ -1791,19 +2055,19 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     }
 
     private void startTimeCompute(String indexKey, TopicEntityVo topicEntity, PressurePumpEnum pressurePumpEnum) {
-        String jobName = topicEntity.getIotCode()+"_"+indexKey;
+        String jobName = topicEntity.getIotCode() + "_" + indexKey;
 
 
-        String cron ="";
-        String triggerName = PUMP_TRIGGER_NAME+"-"+topicEntity.getIotCode();
+        String cron = "";
+        String triggerName = PUMP_TRIGGER_NAME + "-" + topicEntity.getIotCode();
 
-        if("FHS_PressurePump_Start_ALONE_START_YXSC".equals(pressurePumpEnum.getCode())){
+        if ("FHS_PressurePump_Start_ALONE_START_YXSC".equals(pressurePumpEnum.getCode())) {
 
-         Calendar time= Calendar.getInstance();
-         time.add(Calendar.MINUTE,5);
+            Calendar time = Calendar.getInstance();
+            time.add(Calendar.MINUTE, 5);
             cron = time.get(Calendar.SECOND) + " " + time.get(Calendar.MINUTE) + "/5 * * * ?";
 
-        }else{
+        } else {
             cron = pressurePumpEnum.getLeftValue();
         }
 
@@ -1812,22 +2076,22 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
             LambdaQueryWrapper<EquipmentSpecific> wrapper = new LambdaQueryWrapper<>();
             wrapper.eq(EquipmentSpecific::getIotCode, topicEntity.getIotCode());
             equipmentSpecific = equipmentSpecificMapper.selectOne(wrapper);
-        }catch (Exception e) {
+        } catch (Exception e) {
             log.error("根据iotCod查询失败" + topicEntity.getIotCode());
         }
         boolean b = QuartzManager.checkExists(jobName, PUMP_JOB_GROUP_NAME);
         if (indexKey.equals(pressurePumpStart)) {
             if (b) {
                 // 任务存在  更新时间
-                QuartzManager.modifyJobTime(triggerName,PUMP_TRIGGER_GROUP_NAME,cron);
+                QuartzManager.modifyJobTime(triggerName, PUMP_TRIGGER_GROUP_NAME, cron);
             } else {
-                QuartzManager.removeJob(jobName,PUMP_JOB_GROUP_NAME,triggerName,PUMP_TRIGGER_GROUP_NAME);
+                QuartzManager.removeJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME);
                 // 任务不存在，新增
                 // 传参
                 if (ObjectUtils.isEmpty(equipmentSpecific)) {
                     return;
                 }
-                Map<String,Object> parameter = new HashMap<>(6);
+                Map<String, Object> parameter = new HashMap<>(6);
                 parameter.put("jobName", jobName);
                 parameter.put("triggerName", triggerName);
                 parameter.put("triggerGroupName", PUMP_TRIGGER_GROUP_NAME);
@@ -1838,7 +2102,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 parameter.put("systemctlFeign", systemctlFeign);
                 parameter.put("marqueeDataMapper", marqueeDataMapper);
                 parameter.put("equipmentSpecificAlarmLogService", equipmentSpecificAlarmLogService);
-                QuartzManager.addJob(jobName,PUMP_JOB_GROUP_NAME,triggerName,PUMP_TRIGGER_GROUP_NAME, PumpSendMessage.class,cron,parameter);
+                QuartzManager.addJob(jobName, PUMP_JOB_GROUP_NAME, triggerName, PUMP_TRIGGER_GROUP_NAME, PumpSendMessage.class, cron, parameter);
             }
         }
     }

amos-boot-system-equip/src/main/resources/changelog/wl-3.6.0.xml

@@ -843,4 +843,65 @@
                 esi.emergency_level_describe = ( SELECT emergency_level_describe FROM wl_equipment_index wei WHERE esi.equipment_index_id = wei.id );
         </sql>
     </changeSet>
+
+    <changeSet author="LiuLin" id="1686212667">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <columnExists tableName="wl_equipment_specific_index" columnName="index_address"/>
+            </not>
+        </preConditions>
+        <comment>新增属性字段 index_address</comment>
+        <sql>
+            alter table `wl_equipment_specific_index` add column `index_address` varchar(50) COMMENT '信号的索引键key，用于唯一索引信号';
+        </sql>
+    </changeSet>
+
+    <changeSet author="LiuLin" id="1686212667-1">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <columnExists tableName="wl_equipment_specific_index" columnName="quality"/>
+            </not>
+        </preConditions>
+        <comment>新增属性字段 quality</comment>
+        <sql>
+            alter table `wl_equipment_specific_index` add column `quality` tinyint(4) DEFAULT NULL COMMENT '品质，0为有效，1为无效';
+        </sql>
+    </changeSet>
+
+    <changeSet author="LiuLin" id="1686212667-2">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <columnExists tableName="wl_equipment_specific_index" columnName="data_type"/>
+            </not>
+        </preConditions>
+        <comment>新增属性字段 data_type</comment>
+        <sql>
+            alter table `wl_equipment_specific_index` add column `data_type` varchar(10) DEFAULT NULL COMMENT '测点类型，analog/state';
+        </sql>
+    </changeSet>
+
+    <changeSet author="LiuLin" id="1686212667-3">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <columnExists tableName="wl_equipment_specific_index" columnName="time_stamp"/>
+            </not>
+        </preConditions>
+        <comment>新增属性字段 time_stamp</comment>
+        <sql>
+            alter table `wl_equipment_specific_index` add column `time_stamp` varchar(32) COMMENT '时间';
+        </sql>
+    </changeSet>
+
+    <changeSet author="LiuLin" id="1686564561">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <columnExists tableName="wl_equipment_specific_index" columnName="value_enum"/>
+            </not>
+        </preConditions>
+        <comment>wl_equipment_specific_index add `value_enum`</comment>
+        <sql>
+            ALTER TABLE `wl_equipment_specific_index` ADD COLUMN `value_enum` text NULL COMMENT '指标值枚举' AFTER `value_label`;
+        </sql>
+    </changeSet>
+
 </databaseChangeLog>
\ No newline at end of file

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

@@ -515,4 +515,49 @@
             </if>
         </where>
     </select>
+
+    <!-- 根据信号索引查询装备性能指标 -->
+    <select id="getEquipmentSpeIndexByIndexAddress"
+            resultType="com.yeejoin.equipmanage.common.entity.EquipmentSpecificIndex">
+        SELECT wesi.id      AS id,
+               wei.name_key AS nameKey,
+               wesi.value AS value,
+	           wesi.equipment_specific_id AS equipmentSpecificId,
+	           wesi.equipment_index_id AS equipmentIndexId,
+               wesi.equipment_index_name AS equipmentIndexName,
+	           wesi.equipment_index_key AS equipmentIndexKey,
+	           wesi.value_label AS valueLabel,
+	           wei.type_code AS typeCode,
+	           wei.type_name AS typeName,
+	           wei.name AS indexName,
+	           wei.unit AS indexUnitName,
+	           wes.org_code AS orgCode,
+			   ed.`name` AS equipmentSpecificName,
+			   ed.equipment_name AS equipmentName,
+			   wes.iot_code AS iotCode,
+			   wes.code AS specificCode,
+			   wei.`name` AS equipmentSpecificIndexName,
+			   wei.`value_enum` AS valueEnum,
+			   wei.is_trend AS isTrend,
+			   wes.qr_code AS qrCode,
+			   wesi.update_date AS updateDate,
+			   ed.code AS equipmentCode,
+			   ed.equipment_id AS equipmentId,
+			   ed.id AS equipmentDetailId,
+			   wes.code as equipmentSpecificCode,
+			   wes.system_id as systemId,
+			   wesi.is_alarm as isAlarm,
+			   wesi.emergency_level_color as emergencyLevelColor,
+			   wesi.emergency_level as emergencyLevel,
+			   wesi.emergency_level_describe as emergencyLevelDescribe,
+               wes.biz_org_name AS bizOrgName,
+               wes.biz_org_code AS bizOrgCode
+        FROM wl_equipment_specific_index AS wesi
+        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}
+       AND wei.is_iot = true
+    </select>
 </mapper>
\ No newline at end of file