31903 站端设备平台v5>消防告警列表>健康诊断告警,稳压泵单小时“启动”次数计算时，次数少1

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

@@ -23,4 +23,6 @@ public class IotDataVO {
     private String traceId;
 
     private String createdTime;
+
+    private Integer pumNum = 0;
 }

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

@@ -398,18 +398,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         if (eqIotCodeList.size() > 1) {
             log.info("有重复的{}数据！", iotCode);
         }
-        //给 iot服务 推送消息 插数据到 influxdb
-        if (isSendIot) {
-            EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
-            JSONObject messageObj = JSON.parseObject(message);
-            if (!messageObj.containsKey("traceId")) {
-                String traceId = System.currentTimeMillis() + "";
-                messageObj.put("traceId", traceId);
-            }
-            messageObj.put("bizOrgCode", equipmentSpecificVo.getBizOrgCode());
-            String messageTraceId = JSON.toJSONString(messageObj);
-            mqttSendGateway.sendToMqtt("influxdb/" + topic.substring(0, endIndex), messageTraceId);
-        }
+
         EquipmentSpecificVo vo = eqIotCodeList.get(0);
         topicEntity.setType(vo.getType());
         topicEntity.setCode(vo.getCode());
@@ -429,6 +418,10 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 traceId = value.toString();
                 continue;
             }
+            if (iotDataVO.getKey().equalsIgnoreCase(FHS_PressurePump_Start) && "true".equals(iotDataVO.getValue().toString())) {
+                Integer pumNumHour = getPumNumHour(topicEntity);
+                iotDataVO.setPumNum(pumNumHour);
+            }
             iotDatalist.add(iotDataVO);
         }
         if (ObjectUtils.isEmpty(iotDatalist)) {
@@ -436,6 +429,19 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         }
         log.info(String.format("收到mqtt消息：%s", message));
 
+        //给 iot服务 推送消息 插数据到 influxdb
+        if (isSendIot) {
+            EquipmentSpecificVo equipmentSpecificVo = eqIotCodeList.get(0);
+            JSONObject messageObj = JSON.parseObject(message);
+            if (!messageObj.containsKey("traceId")) {
+                String traceIdIot = System.currentTimeMillis() + "";
+                messageObj.put("traceId", traceIdIot);
+            }
+            messageObj.put("bizOrgCode", equipmentSpecificVo.getBizOrgCode());
+            String messageTraceId = JSON.toJSONString(messageObj);
+            mqttSendGateway.sendToMqtt("influxdb/" + topic.substring(0, endIndex), messageTraceId);
+        }
+
         // 发送emq消息转kafka
         JSONObject jsonObject = new JSONObject();
         jsonObject.put("topic", topic);
@@ -1168,7 +1174,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     }
                     //稳压泵启动次数大于15次触发预警
                     if (iotDataVO.getKey().equalsIgnoreCase(FHS_PressurePump_Start) && "true".equals(iotDataVO.getValue().toString())) {
-                        doPressurePumInfo(topicEntity, equipmentSpecificIndex);
+                        doPressurePumInfo(iotDataVO.getPumNum(), equipmentSpecificIndex);
                     }
                     // 遥测数据生成告警事件、日志处理
                     if (iotDataVO.getKey().equalsIgnoreCase(CAFS_FoamTank_FoamTankLevel) ||
@@ -1550,16 +1556,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         }
     }
 
-    private void doPressurePumInfo(TopicEntityVo topicEntity, EquipmentSpecificIndex equipmentSpecificIndex) {
-        // 查询iot该稳压泵的启停次数 一个小时内
-        String endDate = DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN);
-        String startDate = DateUtil.format(DateUtil.offsetHour(new Date(), -1), DatePattern.NORM_DATETIME_PATTERN);
-        String prefix = topicEntity.getIotCode().substring(0, 8);
-        String suffix = topicEntity.getIotCode().substring(8);
-        ResponseModel<Map<String, Integer>> mapResponseModel = iotFeign.queryIotDataNumByIndex(startDate, endDate, prefix, suffix, FHS_PressurePump_Start, "true");
-        if (200 == mapResponseModel.getStatus()) {
-            Map<String, Integer> result = mapResponseModel.getResult();
-            Integer totalNum = result.get("num");
+    private void doPressurePumInfo(Integer totalNum, EquipmentSpecificIndex equipmentSpecificIndex) {
         JSONObject config = this.getRiskTriggerConfig("FIRE_EQUIP_PRESSURE");
         if (Objects.isNull(config)) {
             throw new RuntimeException("获取配置失败！");
@@ -1576,7 +1573,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         TableContentVo tableContentVo2 = new TableContentVo("报警时间", "text", DateUtil.now(), "3");
         TableContentVo tableContentVo3 = new TableContentVo("报警对象", "text", equipmentSpecificIndex.getEquipmentSpecificName(), "4");
         List<TableContentVo> tableContentVos = Arrays.asList(tableContentVo, tableContentVo1, tableContentVo2, tableContentVo3);
-            handlePressureWarning(totalNum.toString(),
+        handlePressureWarning(totalNum + 1,
                 equipmentSpecificIndex,
                 String.valueOf(equipmentSpecificIndex.getEquipmentSpecificId()),
                 "fireIot/data/analysis",
@@ -1584,10 +1581,25 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 extra,
                 "equip",
                 tableContentVos);
+
     }
+
+    private Integer getPumNumHour(TopicEntityVo topicEntity) {
+        // 查询iot该稳压泵的启停次数 一个小时内
+        String endDate = DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN);
+        String startDate = DateUtil.format(DateUtil.offsetHour(new Date(), -1), DatePattern.NORM_DATETIME_PATTERN);
+        String prefix = topicEntity.getIotCode().substring(0, 8);
+        String suffix = topicEntity.getIotCode().substring(8);
+        ResponseModel<Map<String, Integer>> mapResponseModel = iotFeign.queryIotDataNumByIndex(startDate, endDate, prefix, suffix, FHS_PressurePump_Start, "true");
+        if (200 == mapResponseModel.getStatus()) {
+            Map<String, Integer> result = mapResponseModel.getResult();
+            Integer totalNum = result.get("num");
+            return totalNum;
+        }
+        return 0;
     }
 
-    private void handlePressureWarning(String indexValue,
+    private void handlePressureWarning(Integer indexValue,
                                        EquipmentSpecificIndex equipmentSpecificIndex,
                                        String businessId,
                                        String topic,
@@ -1598,7 +1610,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         // 触发预警业务
         BizMessage bizMessage = new BizMessage();
         bizMessage.setIndexKey(indexKey);
-        bizMessage.setIndexValue(indexValue);
+        bizMessage.setIndexValue(indexValue.toString());
         RiskBizInfoVo riskBizInfoVo = fetchData(equipmentSpecificIndex, extra, source, "稳压泵启停频次过高");
         riskBizInfoVo.setWarningObjectCode(businessId);
         riskBizInfoVo.getDynamicDetails().get(0).setTabContent(tableContentVos);