Merge branch 'develop_dl_3.7.1.1' into develop_dl

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/utils/UnitTransformUtil.java

@@ -33,12 +33,12 @@ public class UnitTransformUtil {
                     if (nowVal.compareTo(new BigDecimal(minValue)) < 0) {
                         map.put("status", "0");
                         map.put("abs", nowVal.divide(new BigDecimal(maxValue),2,BigDecimal.ROUND_HALF_UP).movePointRight(2));
+                    } else if (nowVal.compareTo(new BigDecimal(maxValue)) > 0) {  // 当前值大于最大值
+                        map.put("status", "2");
+                        map.put("abs", 100);
                     } else if (nowVal.compareTo(new BigDecimal(maxValue)) == 0) {
                         map.put("status", "1");
                         map.put("abs", 100);
-                    } else if (nowVal.compareTo(new BigDecimal(0)) == 0) {
-                        map.put("status", "1");
-                        map.put("abs", "--");
                     } else {
                         map.put("status", "1");
                         map.put("abs", nowVal.divide(new BigDecimal(maxValue),2,BigDecimal.ROUND_HALF_UP).movePointRight(2));

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

@@ -135,7 +135,7 @@ public class EquipmentIotMqttReceiveConfig {
         list.add("+/+/perspective"); // 添加交换站事件监听
         list.add(riskMsgCenterEquipTopic);
         list.add(riskMsgCenterPatrolTopic);
-        list.add("+/+/shaoshan"); // 添加换流站韶山监听事件
+        list.add("+/+/eventAlarm"); // 添加换流站韶山监听事件   --- shaoshan 修改为 eventAlarm：事件告警 - 统一
         String[] arr = list.toArray(new String[list.size()]);
         adapter = new MqttPahoMessageDrivenChannelAdapter(clientId + "_inbound", mqttPahoClientFactory(), arr);
         adapter.setCompletionTimeout(completionTimeout);
@@ -161,7 +161,7 @@ public class EquipmentIotMqttReceiveConfig {
                     mqttEventReceiveService.handlerMqttIncrementMessage(topic, msg);
                 }else if (dataType.equals("transmit") && StringUtil.isNotEmpty(msg)){ // 遥信遥测信号
                     mqttReceiveService.handlerMqttRomaMessage(topic,msg);
-                }else if (dataType.equals("shaoshan") && StringUtil.isNotEmpty(msg)){ // 告警信号
+                }else if (dataType.equals("eventAlarm") && 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/controller/StatisticsController.java

 package com.yeejoin.equipmanage.controller;
 
+import com.yeejoin.equipmanage.common.entity.SignalClassify;
 import com.yeejoin.equipmanage.common.vo.EquipCategoryVo;
 import com.yeejoin.equipmanage.service.IStatisticsService;
 import io.swagger.annotations.Api;
@@ -10,6 +11,8 @@ import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
 import org.typroject.tyboot.core.foundation.enumeration.UserType;
 import org.typroject.tyboot.core.restful.doc.TycloudOperation;
+import org.typroject.tyboot.core.restful.utils.ResponseHelper;
+import org.typroject.tyboot.core.restful.utils.ResponseModel;
 
 import javax.servlet.http.HttpServletRequest;
 import java.util.List;
@@ -95,4 +98,11 @@ public class StatisticsController {
 	public List<EquipCategoryVo> equipCategoryVoList(@PathVariable String systemCode) {
 		return iStatisticsService.equipCategoryVoList(systemCode);
 	}
+
+	@RequestMapping(value = "/test", method = RequestMethod.GET)
+	@TycloudOperation(ApiLevel = UserType.AGENCY)
+	@ApiOperation(httpMethod = "GET", value = "根据id查询", notes = "根据id查询")
+	public ResponseModel<String> getStr() {
+		return ResponseHelper.buildResponse(iStatisticsService.getStr());
+	}
 }

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/dto/ReportAlarmDto.java

+package com.yeejoin.equipmanage.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.math.BigDecimal;
+
+/**
+ * @author keyong
+ * @title: ReportAlarmDto
+ * <pre>
+ * @description: TODO
+ * </pre>
+ * @date 2024/5/11 9:56
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class ReportAlarmDto {
+
+    private String equipmentSpecificId;
+
+    private String specificName;
+
+    private String location;
+
+    private String equipmentSpecificIndexName;
+
+    private String alamContent;
+
+    private BigDecimal alarmNum;
+
+    private String createDate;
+
+    private String updateDate;
+
+    private String type;
+}

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

@@ -11,6 +11,7 @@ import com.yeejoin.equipmanage.common.entity.dto.EquipSpecificAlarmDTO;
 import com.yeejoin.equipmanage.common.entity.vo.EquipmentIotAlarmVO;
 import com.yeejoin.equipmanage.common.utils.CommonPageInfoParam;
 import com.yeejoin.equipmanage.common.vo.EquipAlarmLineVo;
+import com.yeejoin.equipmanage.dto.ReportAlarmDto;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 
@@ -235,5 +236,5 @@ public interface EquipmentSpecificAlarmMapper extends BaseMapper<EquipmentSpecif
 
     void clearAlarmData(@Param("equipmentSpecificId") Long equipmentSpecificId);
 
-    List<Map<String, Object>> getAlarmDetailByDate(@Param("bizOrgCode") String bizOrgCode, @Param("startDate") String startDate, @Param("endDate") String endDate, @Param("sysId") Long sysId);
+    List<ReportAlarmDto> getAlarmDetailByDate(@Param("bizOrgCode") String bizOrgCode, @Param("startDate") String startDate, @Param("endDate") String endDate, @Param("sysId") Long sysId);
 }

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

@@ -22,5 +22,6 @@ public interface IStatisticsService {
 	 */
     List<EquipCategoryVo> equipCategoryVoList(String systemCode);
 
+	String getStr();
 
 }

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

@@ -105,8 +105,8 @@ public class AnalysisReportLogServiceImpl extends ServiceImpl<AnalysisReportLogM
                     filter(index -> index.get("code").equals(String.valueOf(f.get("typeCode")))).collect(Collectors.toList());
             if (!ObjectUtils.isEmpty(collect)) {
                 String indicator = String.valueOf(collect.get(0).get("index"));
-                analysisReportMonthMapper.insertSystemMonthData(new ArrayList<>(Arrays.asList(indicator.split(","))), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
-                analysisReportSummaryMapper.insertSystemMonthSummaryData(new ArrayList<>(Arrays.asList(indicator.split(","))), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
+                analysisReportMonthMapper.insertSystemMonthData(new ArrayList<>(), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
+                analysisReportSummaryMapper.insertSystemMonthSummaryData(new ArrayList<>(), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
             }
         });
     }
@@ -136,8 +136,8 @@ public class AnalysisReportLogServiceImpl extends ServiceImpl<AnalysisReportLogM
                     filter(index -> index.get("code").equals(String.valueOf(f.get("typeCode")))).collect(Collectors.toList());
             if (!ObjectUtils.isEmpty(collect)) {
                 String indicator = String.valueOf(collect.get(0).get("index"));
-                analysisReportMonthMapper.insertSystemMonthData(new ArrayList<>(Arrays.asList(indicator.split(","))), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
-                analysisReportSummaryMapper.insertSystemMonthSummaryData(new ArrayList<>(Arrays.asList(indicator.split(","))), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
+                analysisReportMonthMapper.insertSystemMonthData(new ArrayList<>(), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
+                analysisReportSummaryMapper.insertSystemMonthSummaryData(new ArrayList<>(), beginDateStr, endDateStr, Long.valueOf(f.get("id").toString()));
             }
         });
     }

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

@@ -223,7 +223,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     public static List<EquipmentSpecificAlarmLog> upAlarmLogStatus(String iotCode, String equipmentSpecificIndexKey,
                                                                    String traceId, IEquipmentSpecificAlarmLogService equipmentSpecificAlarmLogService,
-                                                                   boolean flag, String batchConfirm) {
+                                                                   boolean flag, boolean batchConfirm, String confirmType) {
 
 //        LambdaQueryWrapper<EquipmentSpecificAlarmLog> queryWrapper = new LambdaQueryWrapper<>();
 //        if (batchConfirm.equalsIgnoreCase(HandleBatchConfirmEnum.SINGLE.getCode())) {
@@ -524,8 +524,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
             indexStateList.add(createIndexStateVo(equipmentSpeIndex));
 
-            // 添加指标报告
-            saveEquipmentAlarmReportDay(equipmentSpeIndex);
+
 
             // 火眼数据构造告警指标逻辑
             equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
@@ -552,6 +551,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                 handlingAlarms(equipmentSpeIndex, alarmFlag);
             }
 
+            // 添加指标报告
+            saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);
+
             // 指标告警处理
             if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
                 equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpeIndex, messageBodyMap));
@@ -625,7 +627,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                             bool = Boolean.FALSE;
                         } else {
                             alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
-                                    equipmentSpecificAlarmLogService, false, ""));
+                                    equipmentSpecificAlarmLogService, false, false, ""));
                             mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                             bool = Boolean.TRUE;
                         }
@@ -791,8 +793,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
                 indexStateList.add(createIndexStateVo(equipmentSpeIndex));
 
-                // 添加指标报告
-                saveEquipmentAlarmReportDay(equipmentSpeIndex);
+
 
                 // 火眼数据构造告警指标逻辑
                 equipmentSpeIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
@@ -819,6 +820,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     handlingAlarms(equipmentSpeIndex, alarmFlag);
                 }
 
+                // 添加指标报告
+                saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);
+
                 // 指标告警处理
                 if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
                     equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpeIndex, messageBodyMap));
@@ -891,7 +895,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                                 bool = Boolean.FALSE;
                             } else {
                                 alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
-                                        equipmentSpecificAlarmLogService, false, ""));
+                                        equipmentSpecificAlarmLogService, false, false, ""));
                                 mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                                 bool = Boolean.TRUE;
                             }
@@ -1075,8 +1079,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                     // 推送数据到组态大屏
                     pushDataToIntegrationPage(equipmentSpecificIndexList);
 
-                    // 添加指标报告
-                    saveEquipmentAlarmReportDay(equipmentSpeIndex);
+
 
                     // 火眼数据构造告警指标逻辑
                     equipmentSpecificIndex = handleTemperatureAlarm(equipmentSpeIndex, iotDatalist);
@@ -1102,6 +1105,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                         handlingAlarms(equipmentSpecificIndex, alarmFlag);
                     }
 
+                    // 添加指标报告
+                    saveEquipmentAlarmReportDay(equipmentSpeIndex, alarmFlag);
+
                     // 指标告警处理
                     if (equipmentSpecificIndex.getIsAlarm() != null && 1 == equipmentSpecificIndex.getIsAlarm() && !equipmentSpecificIndex.getEquipmentIndexKey().equals(pressurePumpStart)) {
                         equipmentSpecificAlarms.addAll(createIndexAlarmRecord(equipmentSpecificIndex, messageBodyMap));
@@ -1208,7 +1214,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
                         bool = Boolean.FALSE;
                     } else {
                         alarmLogs.addAll(upAlarmLogStatus(action.getIotCode(), action.getEquipmentSpecificIndexKey(), action.getTraceId(),
-                                equipmentSpecificAlarmLogService, false, ""));
+                                equipmentSpecificAlarmLogService, false, false, ""));
                         mqttSendGateway.sendToMqtt(TopicEnum.EQYQR.getTopic(), JSONArray.toJSON(action).toString());
                         bool = Boolean.TRUE;
                     }
@@ -1599,7 +1605,7 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         }
     }
 
-    private EquipmentAlarmReportDay addEquipAlarmReportRecord(EquipmentSpecificIndex equipmentSpecificIndex) {
+    private EquipmentAlarmReportDay addEquipAlarmReportRecord(EquipmentSpecificIndex equipmentSpecificIndex, boolean flag) {
         EquipmentAlarmReportDay equipmentAlarmReportDay = new EquipmentAlarmReportDay();
         equipmentAlarmReportDay.setOrgCode(equipmentSpecificIndex.getOrgCode());
         equipmentAlarmReportDay.setReportDate(new Date());
@@ -1611,8 +1617,14 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         equipmentAlarmReportDay.setEquipmentDetailId(equipmentSpecificIndex.getEquipmentDetailId());
         equipmentAlarmReportDay.setEquipmentId(equipmentSpecificIndex.getEquipmentId());
         equipmentAlarmReportDay.setEquipmentCode(equipmentSpecificIndex.getEquipmentCode());
+        if(flag) {
+            equipmentAlarmReportDay
+                    .setIndexTrueNum(1L);
+        } else {
             equipmentAlarmReportDay
                     .setIndexTrueNum(TrueOrFalseEnum.real.value.equals(equipmentSpecificIndex.getValue()) ? 1L : 0L);
+        }
+
         equipmentAlarmReportDay.setAlarmType(equipmentSpecificIndex.getTypeCode());
         equipmentAlarmReportDay.setAlarmTypeName(equipmentSpecificIndex.getTypeName());
         equipmentAlarmReportDay.setIndexId(equipmentSpecificIndex.getEquipmentIndexId());
@@ -2234,9 +2246,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
      *
      * @param equipmentSpecificIndex
      */
-    private void saveEquipmentAlarmReportDay(EquipmentSpecificIndex equipmentSpecificIndex) {
+    private void saveEquipmentAlarmReportDay(EquipmentSpecificIndex equipmentSpecificIndex, boolean flag) {
         SimpleDateFormat sdf = new SimpleDateFormat(DateUtils.DATE_PATTERN);
-        EquipmentAlarmReportDay equipmentAlarmReportDay = addEquipAlarmReportRecord(equipmentSpecificIndex);
+        EquipmentAlarmReportDay equipmentAlarmReportDay = addEquipAlarmReportRecord(equipmentSpecificIndex, flag);
         LambdaQueryWrapper<EquipmentAlarmReportDay> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(EquipmentAlarmReportDay::getReportDate, sdf.format(equipmentAlarmReportDay.getReportDate()))
                 .eq(EquipmentAlarmReportDay::getIndexId, equipmentAlarmReportDay.getIndexId())

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

 package com.yeejoin.equipmanage.service.impl;
 
+import com.yeejoin.equipmanage.utils.ClassToJsonUtil;
+import org.springframework.core.io.Resource;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.yeejoin.amos.feign.privilege.model.AgencyUserModel;
@@ -16,6 +18,7 @@ import com.yeejoin.equipmanage.mapper.StockDetailMapper;
 import com.yeejoin.equipmanage.remote.RemoteSecurityService;
 import com.yeejoin.equipmanage.service.*;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -221,6 +224,16 @@ public class StatisticsServiceImpl implements IStatisticsService {
 		return equipmentSpecificMapper.selectEquipCategoryNumber(fightingSystemEntity.getId());
 	}
 
+	@Value("classpath:/json/commonMessage.json")
+	private Resource commonMessage;
+
+	@Override
+	public String getStr() {
+		String s = "{\"code\":2000,\"id\":1,\"body\":{\"warns\":[{\"systemid\":\"2\",\"pointId\":\"73811\",\"level\":\"3\",\"time\":\"2024-04-08 09:17:56.226\",\"type\":\"遥信变位\",\"deviceId\":\"\",\"content\":\"火灾报警主机_主控楼二层1#蓄电池室可燃气体火警屏蔽状态 消失\"}]}}";
+		String s1 = "{\"data_class\":\"realdata\",\"data_type\":\"alarm\",\"op_type\":\"subscribe_emergency\",\"condition\":{\"station_psr_id\":\"50edcb6c1b8a811030493c80a2014950ed9d4f59e8\",\"station_name\":\"中州换流站\",\"alarm_type\":\"yx_bw\"},\"data\":[{\"psrId\":\"D017020000000000000000999\",\"astId\":\"D017020000000000000000999\",\"equipType\":\"ASTType_0000111\",\"eventType\":\"OtherSignal\",\"alarmSource\":\"OWS\",\"alarmLevel\":\"3\",\"description\":\"2024-03-11 09:06:17::585 S2WCL12A E3.C01软水器再生结束信号 出现\",\"dateTime\":\"2024-03-11 09:06:17.585\"}]}";
+		String s2 = "{\"code\":2000,\"id\":1,\"body\":{\"time_stamp\":\"2024-04-07 16:42:52.616\",\"datatype\":\"analog\",\"name\":\"直流接地极电流IDEE\",\"value\":12.969970703125,\"key\":\"0000YC00010100409C\",\"quality\":0,”code”:1001}}";
+		return ClassToJsonUtil.class2json(s1, commonMessage, "k1");
+	}
 
 
 	private Double getExtinguishingCountByWarehouseId(String categoryCode, Long warehouseId) {

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/utils/ClassToJsonUtil.java

+package com.yeejoin.equipmanage.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.google.gson.JsonObject;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.core.io.Resource;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * @author keyong
+ * @title: ClassToJsonUtil
+ * <pre>
+ * @description: TODO
+ * </pre>
+ * @date 2024/4/11 11:28
+ */
+public class ClassToJsonUtil {
+
+    private static Map<String, Object> map = new HashMap<>();
+
+    public static String class2json(Object obj, Resource commonMessage, String topic) {
+        String json;
+        try {
+            json = IOUtils.toString(commonMessage.getInputStream(), String.valueOf(StandardCharsets.UTF_8));
+        } catch (IOException e) {
+            throw new RuntimeException("获取kafka信息模板失败！");
+        }
+        List<Map> listModel = JSONObject.parseArray(json, Map.class);
+        if (0 < listModel.size()) {
+            List<Map> mapList = listModel.stream().filter(x -> String.valueOf(x.get("kafkaTopic")).equalsIgnoreCase(topic)).collect(Collectors.toList());
+            if (CollectionUtils.isNotEmpty(mapList)) {
+                Map<String, Object> map1 = mapList.get(0);
+                JSONObject object = JSON.parseObject(String.valueOf(obj));
+                analysisJson(object, "");
+                JSONObject res = analyseJson(JSON.toJSONString(map1), map);
+                res.put("kafkaTopic", map1.get("kafkaTopic"));
+                res.put("mqTopic", map1.get("mqTopic"));
+                return JSON.toJSONString(res);
+            }
+        }
+        return null;
+    }
+
+
+    private static JSONObject analysisJson(Object objJson, String flag) {
+        if (objJson instanceof JSONArray) {//如果obj为json数组
+            JSONArray objArray = (JSONArray) objJson;
+            for (int i = 0; i < objArray.size(); i++) {
+                analysisJson(objArray.get(i), flag);
+            }
+        } else if (objJson instanceof JSONObject) {//如果为json对象
+            JSONObject jsonObject = (JSONObject) objJson;
+            Iterator it = jsonObject.keySet().iterator();
+            while (it.hasNext()) {
+                String key = it.next().toString();
+                Object object = jsonObject.get(key);
+                //如果得到的是数组
+                if (object instanceof JSONArray) {
+                    JSONArray objArray = (JSONArray) object;
+                    String path = "";
+                    if (StringUtils.isNotBlank(flag)) {
+                        path = flag + "." + key;
+                    } else {
+                        path = key;
+                    }
+                    analysisJson(objArray, path);
+                } else if (object instanceof JSONObject) {//如果key中是一个json对象
+                    String path = "";
+                    if (StringUtils.isNotBlank(flag)) {
+                        path = flag + "." + key;
+                    } else {
+                        path = key;
+                    }
+                    analysisJson((JSONObject) object, path);
+                } else {//如果key中是其他
+                    String path = "";
+                    if (StringUtils.isNotBlank(flag)) {
+                        path = flag + "." + key;
+                    } else {
+                        path = key;
+                    }
+//                    System.out.println(path+":"+object.toString()+" ");
+                    map.put(path, String.valueOf(object));
+                }
+            }
+        } else {//如果key中是其他
+//            System.out.println(flag+":"+objJson.toString()+" ");
+            map.put(flag, String.valueOf(objJson));
+        }
+        return JSONObject.parseObject(JSON.toJSONString(map));
+    }
+
+    public static JSONObject analyseJson(String jsonData, Map<String, Object> keyMap) {
+        SortedMap<String, Object> map = new TreeMap<>();
+        JSONObject jsonObject = JSON.parseObject(jsonData);
+        for (String key : jsonObject.keySet()) {
+            String resKey = keyMap.get(key) == null ? key : String.valueOf(keyMap.get(key));
+            Object value = jsonObject.get(key);
+            if (value instanceof JSONArray) {
+                JSONArray jsonArray = new JSONArray(new LinkedList<>());
+                JSONArray array = jsonObject.getJSONArray(key);
+                for (int i = 0; i < array.size(); i++) {
+                    Object object = array.get(i);
+                    if (object instanceof String) {
+                        map.put(resKey, array);
+                    } else {
+                        JSONObject sortJson = analyseJson(String.valueOf(object), keyMap);
+                        jsonArray.add(sortJson);
+                        map.put(resKey, jsonArray);
+                    }
+                }
+            } else if (value instanceof JSONObject) {
+                JSONObject sortJson = analyseJson(String.valueOf(value), keyMap);
+                map.put(resKey, sortJson);
+            } else {
+                map.put(resKey, ObjectUtils.isNotEmpty(keyMap.get(value)) ? keyMap.get(value) : "");
+            }
+        }
+        return new JSONObject(map);
+    }
+
+}

amos-boot-system-equip/src/main/resources/application-dev.properties

 spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
-spring.datasource.url = jdbc:mysql://172.16.11.201:3306/dl_business?useUnicode=true&allowMultiQueries=true&characterEncoding=utf-8&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
+spring.datasource.url = jdbc:mysql://172.16.10.241:3306/dl_business_v3.0.1.3?useUnicode=true&allowMultiQueries=true&characterEncoding=utf-8&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=Asia/Shanghai
 spring.datasource.username=root
 spring.datasource.password=Yeejoin@2020
 spring.datasource.type=com.zaxxer.hikari.HikariDataSource
@@ -12,22 +12,24 @@ spring.datasource.hikari.max-lifetime= 1800000
 spring.datasource.hikari.connection-timeout= 60000
 spring.datasource.hikari.connection-test-query= SELECT 1
 
-##########eureka配置信息###############
+##########eureka????###############
 eureka.instance.prefer-ip-address = true
-#服务实例ip地址，若eureka.instance.prefer-ip-address=false,ip-address设置失效，一般情况不用设置。
+#????ip????eureka.instance.prefer-ip-address=false,ip-address??????????????
 #eureka.instance.ip-address= 172.16.3.135
-#eureka服务ip配置，可填写固定ip地址
-eureka.instance.hostname=172.16.11.201
+
 #eureka服务配置的校验账号及密码,配置需和eureka服务后台配置文件中一致
+#eureka??ip????????ip??
+eureka.instance.hostname=172.16.10.241
+#eureka????????????,????eureka???????????
 spring.security.user.name=admin
 spring.security.user.password=a1234560
 eureka.client.serviceUrl.defaultZone=http://${spring.security.user.name}:${spring.security.user.password}@${eureka.instance.hostname}:10001/eureka/
-##########eureka配置信息###############
+##########eureka????###############
 
 
 #redis
 spring.redis.database=1
-spring.redis.host=172.16.11.201
+spring.redis.host=172.16.10.241
 spring.redis.port=6379
 spring.redis.password=yeejoin@2020
 spring.redis.lettuce.pool.max-active=200
@@ -39,10 +41,10 @@ spring.redis.expire.time=30000
 ## emqx
 emqx.clean-session=true
 emqx.client-id=${spring.application.name}-${random.int[1024,65536]}
-emqx.broker=tcp://172.16.11.201:1883
+emqx.broker=tcp://172.16.10.241:1883
 emqx.user-name=admin
 emqx.password=public
-mqtt.scene.host=mqtt://172.16.11.201:8083/mqtt
+mqtt.scene.host=mqtt://172.16.10.241:8083/mqtt
 mqtt.client.product.id=mqtt
 mqtt.topic=topic_mqtt
 spring.mqtt.completionTimeout=3000
@@ -57,7 +59,7 @@ spring.influx.actions=10000
 spring.influx.bufferLimit=20000
 
 
-#系统机器人账号
+#???????
 amos.system.user.user-name=fas_system
 amos.system.user.password=a1234560
 amos.system.user.app-key=studio_normalapp_3056965
@@ -66,36 +68,36 @@ security.productApp=STUDIO_APP_MOBILE
 
 
 
-#向数字站同步设备，巡查点台账信息开关，适用于数字站1.0版本，目前停止同步，默认为false
+#?????????????????????????1.0?????????????false
 systemctl.sync.switch=false
-#数据JCS开关 设备资源数据权限配置，站端部署，库里只有单站数据可不进行开启（false），中心级部署或存在多站数据时需开启（true）数据权限过滤
+#??JCS?? ???????????????????????????????false???????????????????true???????
 systemctl.jcs.switch=true
-#给平台同步告警数据,若不同步影响消防自动化告警消息框数据展示
+#?????????,????????????????????
 systemctl.amos.switch=true
-#告警数据是否发送给App端，发送是通过jpush实现，需要有网情况下。目前电力消防不支持网络使用，默认不进行APP推送
+#?????????App???????jpush??????????????????????????????APP??
 isSendApp=false
 
 
 
-###############向数字化平台3.0进行告警转发配置信息################################
-#告警推送开关，若无推送需求进行关闭（false）
+###############??????3.0??????????################################
+#??????????????????false?
 dcs.send.message=false
-#请求头参数
+#?????
 dcs.client-id=fire_system
 dcs.client.secret=fire_system
 dcs.grant-type=client_credentials
 dcs.scope=all
 dcs.token-key=dcs_token_key
-#华为roma秘钥每个站有区别与集成商索取
+#??roma??????????????
 dcs.x.hw.id=io.cs
 dcs.x.hw.appKey=i!YmnHgJ=Ysz8V+O@ZG237inQKfixkQ.-#-P3IlR8fO%.#F!g%35Z#@tp$6G+52v
-# \u7531\u4E8E\u76EE\u524D\u6709\u4E0D\u6B62\u4E00\u4E2A\u6570\u5B57\u7AD9\u5E73\u53F0\u5BF9\u5E94\u63A5\u53E3\u89C4\u8303\u4E0D\u4E00\u81F4\uFF0C\u6240\u4EE5\u6B64\u5904\u589E\u52A0\u914D\u7F6E\uFF0Ctrue\u8868\u793A\u9002\u7528\u4E8E\u5357\u745E\u79D1\u6280\u7684\u63A5\u53E3\u89C4\u8303
-#平台集成商是否科技接口，(科技为true、继保和许继为false）
+# ?????????????????????????????????true??????????????
+#????????????(???true???????false?
 dcs.NRKJ.use=false
-dcs.source.type=\u9ad8\u7ea7\u5e94\u7528
-#数字化平台3.0权限接口
+dcs.source.type=????
+#?????3.0????
 dcs.url.token=http://172.21.101.123/udaap-ddis/v1/cs/auth-service/oauth/token
-#数字化平台3.0告警存储接口
+#?????3.0??????
 dcs.url.sendalarm=http://172.21.101.101/v1/cs/alarm-service/appalarm/sendalarm
 #################################################################################
 
@@ -103,45 +105,45 @@ dcs.url.sendalarm=http://172.21.101.101/v1/cs/alarm-service/appalarm/sendalarm
 param.system.online.date = 2019-02-12
 
 
-############数字1.0视频接入服务配置，其他版本不支持##################################
-# 视频转码服务开关 hls(关)/flv(开),默认关闭，数字换流站1.0使用时开启
+############??1.0????????????????##################################
+# ???????? hls(?)/flv(?),??????????1.0?????
 window.vedioFormat = hls
-# 航天视频服务地址
+# ????????
 param.htvideo.url=http://192.168.4.174:9001;
-# 南瑞视频转码服务地址
+# ??????????
 param.nrvideo.url=http://198.87.103.158:8001;
-#南瑞视频平台通过视频id获取flv格式视频播放地址
+#??????????id??flv????????
 param.nrflvbyvoideoid.url=http://192.168.4.159:10010/api/media/live
 #####################################################################################
 
-#装备服务在接收到站端iot推送的装备数据后进行influxdb存库
-#1.在装备接口消息处向influxdb/{productKey}/{deviceName} 消息地址推送数据，iot负责存库
-#2.配置文件添加配置项开关，默认为关闭，该功能只使用于中心及系统
-#是否向iot推送消息
+#??????????iot??????????influxdb??
+#1.?????????influxdb/{productKey}/{deviceName} ?????????iot????
+#2.??????????????????????????????
+#???iot????
 isSendIot=false
 
-#站端编码及中文名称，用于数据上送
+#????????????????
 state.code=GW190301
-state.name=\u9526\u5c4f\u6362\u6d41\u7ad9
-#站端编码，用于物联报表查询
+state.name=?????
+#?????????????
 stationCode=LSHLZ1bZAJU645Pgl7
 
-#用于总部系统与站端系统逻辑区分，站端写zd总部默认为空
+#???????????????????zd??????
 system.type=zd
 
-#遥测信号是否向直流中心推送，数字站默认不推送
+#??????????????????????
 is.open.telemetering=false
 
 
 
-# message 接收来自南瑞平台的告警事件描述，如：出现（即为true）
-message.alarm.char=\u51FA\u73B0
+# message ???????????????????????true?
+message.alarm.char=??
 
 
 
-# 电建项目配置
+# ??????
 mileage.parameter=0.5
-# 电建项目配置
+# ??????
 mileage.segmentation.cron=0 0 0 * * ?
 mileage.splitmileage.cron=0 0/1 * * * ?
 mileage.clippingtime=600000
@@ -150,10 +152,10 @@ equip.car.maxTravel=400
 equip.risk.model.topic=""
 #mileage.segmentation.cron= 0 */2 * * * ?
 
-# 机场使用特殊配置iotCode前缀，装备、车辆及导入使用到
-# 机场使用
+# ????????iotCode??????????????
+# ????
 #iot.code.prefix.have.used=20210003,20210004,20210005
-# 电力使用
+# ????
 iot.code.prefix.have.used=
 
 #kafka

amos-boot-system-equip/src/main/resources/application.properties

@@ -46,8 +46,8 @@ iot.vehicle.track=AMOS-API-IOT
 jcs.fegin.name=JCS
 video.fegin.name=VIDEO
 idx.feign.name=AMOS-IDX
-
 mcb.feign.name=MCB-SERVER
+
 #\u9879\u76EE\u521D\u59CB\u5316\u753B\u5E03id
 morphic.projectSeq=1390314016458514433
 #wl_equipment_category \u8868id \uFF08\u53D6\u706B\u707E\u62A5\u8B66\u7CFB\u7EDFid\uFF09
@@ -155,30 +155,34 @@ management.endpoint.health.show-details=always
 logic=false
 
 
-##电力报表定时任务
+##????????
 jobs.month.cron = 0 50 0 1 * ?
 jobs.day.cron = 0 5 0 * * ?
 jobs.week.cron = 0 25 0 ? * 1
 jobs.day.cron.old = 0 25 0 * * ?
+# ?????????
+update.car.qrCode=0 01 00 * * ?
 
-#预案消防炮设备类型编码
+#???????????
 equipment.plan.monitor=92030200,92032200
-#消防泵设备类型编码
+#?????????
 equipment.plan.pump=92010600,92030600,92130400,92140200,92150300
 
-#电力设备报废业务，在设备报废前30日 每日9点执行 系统推送提醒。设备报废后停止消息推送提醒
+#???????????????30? ??9??? ????????????????????
 equipment.scrap.day=30
 equipment.scrap.cron=0 0 9 * * ?
 
+#???????????kks?
+power.station.url=http://139.9.169.123:5024/process/DataInterface
 
-#稳压泵定时向缓存中存昨日启动次数任务
+#??????????????????
 equipment.pressurepump.start.cron=0 0 0 * * ?
 
-#稳压泵启停指标
+#???????
 equipment.pressurepump.start=FHS_PressurePump_Start
-#稳压泵压力指标
+#???????
 equipment.pressurepump.pipepressure=FHS_PipePressureDetector_PipePressure
 
-#液位指标编码
+#??????
 water.level.indexKey=FHS_FirePoolDevice_WaterLevel,FHS_LevelDetector_WaterLevel,FHS_WirelessliquidDetector_WaterLevel,CAFS_FoamTank_FoamTankLevel,CAFS_WaterTank_WaterTankLevel
 

amos-boot-system-equip/src/main/resources/json/commonMessage.json

+[
+  {
+    "kafkaTopic": "k1",
+    "mqTopic": "romaSite/data/transmit",
+    "data": {
+      "dataType": "condition.station_psr_id",
+      "value": "condition.station_psr_id",
+      "timeStamp": "condition.station_psr_id",
+      "quality": "condition.station_psr_id",
+      "scadaId": "condition.station_psr_id",
+      "key": "condition.station_psr_id",
+      "disCreate": "condition.station_psr_id",
+      "name": "condition.station_psr_id"
+    }
+  },
+  {
+    "kafkaTopic": "k2",
+    "mqTopic": "romaSite/data/eventAlarm",
+    "data": {
+      "timeStamp": "body.warns.time",
+      "warns": [
+        {
+          "eventTextL1": "body.warns.systemid",
+          "pointId": "body.warns.pointId",
+          "time": "body.warns.type",
+          "deviceId": "deviceId",
+          "eventstatus": "body.warns.content"
+        }
+      ]
+    }
+  }
+]
\ No newline at end of file

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

@@ -445,6 +445,31 @@
     <select id="selectGasExtinguishing" resultType="java.util.Map">
         SELECT
             wes.id,
+            (SELECT wei.emergency_level FROM  wl_equipment_index wei LEFT JOIN wl_equipment_specific_index wesi ON wei.id = wesi.equipment_index_id WHERE wesi.equipment_specific_id = wes.id AND wesi.equipment_index_key = wes.realtime_iot_index_key) as level,
+            wes.code as code ,
+            wes.name,
+            IFNULL(wes.realtime_iot_index_name, '暂无信号') AS status,
+            CASE
+                WHEN (SELECT is_alarm FROM wl_equipment_specific_index wesi WHERE wesi.equipment_specific_id = wes.id AND wesi.equipment_index_key = wes.realtime_iot_index_key ) = 1 AND wes.realtime_iot_index_value = 'true' THEN 1
+                ELSE 0
+            END AS alarm
+        FROM
+            wl_equipment_specific wes
+                JOIN f_equipment_fire_equipment fire ON wes.id = fire.fire_equipment_id
+        WHERE
+            (
+                wes.equipment_code LIKE concat( '92044300', '%')
+                OR wes.equipment_code LIKE concat( '92040100', '%')
+            )
+            <if test="code != null and  code!='' ">
+                AND fire.equipment_id = (SELECT id  FROM f_equipment WHERE `code` = #{code})
+            </if>
+            AND wes.biz_org_code LIKE CONCAT( #{bizOrgCode}, '%')
+    </select>
+
+    <select id="selectPressureFlow" resultType="java.util.Map">
+        SELECT
+            wes.id,
             '1' as type,
             (SELECT wei.emergency_level FROM  wl_equipment_index wei LEFT JOIN wl_equipment_specific_index wesi ON wei.id = wesi.equipment_index_id WHERE wesi.equipment_specific_id = wes.id AND wesi.equipment_index_key = wes.realtime_iot_index_key) as level,
             wes.code as code ,

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

@@ -1602,18 +1602,19 @@
         DELETE FROM wl_equipment_alarm_report_day WHERE equipment_specific_id = #{equipmentSpecificId};
     </delete>
 
-    <select id="getAlarmDetailByDate" resultType="Map">
+    <select id="getAlarmDetailByDate" resultType="com.yeejoin.equipmanage.dto.ReportAlarmDto">
         SELECT
-            wlesa.equipment_specific_id,
+        wlesa.equipment_specific_id AS equipmentSpecificId,
         wlesa.equipment_specific_name AS specificName,
         wlesa.location AS location,
-            wlesa.equipment_specific_index_name,
+        wlesa.equipment_specific_index_name AS equipmentSpecificIndexName,
         concat(wlesa.equipment_specific_name, wlesa.equipment_specific_index_name) AS alamContent,
-            wlesa.frequency AS alarmNum,
+        IFNULL(SUM(wlesa.frequency), 0) AS alarmNum,
+        wlesa.type AS type,
+        DATE_FORMAT(wlesa.update_date, '%Y-%m-%d %H:%i:%s') AS updateDate,
         DATE_FORMAT(wlesa.create_date, '%Y-%m-%d %H:%i:%s') AS createDate
         FROM
         `wl_equipment_specific_alarm` `wlesa`
-            LEFT JOIN f_fire_fighting_system AS sys ON FIND_IN_SET(sys.id, wlesa.system_ids)
         <where>
             <if test="bizOrgCode != null and bizOrgCode != ''">AND
                 wlesa.biz_org_code like concat (#{bizOrgCode},'%')
@@ -1624,6 +1625,7 @@
                 AND find_in_set(#{sysId}, wlesa.system_ids)
             </if>
         </where>
-            ORDER BY wlesa.create_date DESC
+        GROUP BY wlesa.equipment_specific_id, wlesa.type
+        ORDER BY wlesa.update_date DESC
     </select>
 </mapper>
\ No newline at end of file

amos-boot-utils/amos-boot-utils-message/src/main/java/com/yeejoin/amos/message/kafka/KafkaConsumerService.java

@@ -8,13 +8,11 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.core.io.Resource;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.stereotype.Service;
 import org.typroject.tyboot.component.emq.EmqKeeper;
-
-import org.springframework.core.io.Resource;
-
 import java.io.UnsupportedEncodingException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
@@ -42,6 +40,9 @@ public class KafkaConsumerService {
     @Value("classpath:/json/commonMessage.json")
     private Resource commonMessage;
 
+    @Value("classpath:/json/commonMessage.json")
+    private Resource commonMessage;
+
     /**
      * 批量消费kafka消息
      * Kafka消息转emq
@@ -173,6 +174,77 @@ public class KafkaConsumerService {
             }
         }
     }
+//
+//    /**
+//     * 韶山换流对接Kafka
+//     * @param record record
+//     * @param ack ack
+//     */
+//    @KafkaListener(id = "kafkaConsumer", groupId = "kafkaConsumerGroup", topics = "#{'${queue.kafka.shaoshan.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
+//    public void kafkaConsumer(ConsumerRecord<?, String> record, Acknowledgment ack) {
+//        Optional<?> message = Optional.ofNullable(record.value());
+//        if (message.isPresent()) {
+//            try {
+////                JSONObject messageObj = JSONObject.fromObject(record.value());
+////                JSONObject data = messageObj.getJSONObject("body");
+////                JSONObject object = JSONObject.fromObject(record.value());
+////                String json = ClassToJsonUtil.class2json(object, commonMessage, MQTT_TOPIC_EVENT_ALARM);
+////                emqKeeper.getMqttClient().publish(MQTT_TOPIC_EVENT_ALARM, json.getBytes(StandardCharsets.UTF_8), 0, false);
+//
+//                JSONObject object = JSONObject.fromObject(record.value());
+//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
+//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
+//                ack.acknowledge();
+//            } catch (MqttException e) {
+//                log.error("解析数据失败,{}", e.getMessage());
+//            } catch (UnsupportedEncodingException e) {
+//                e.printStackTrace();
+//            }
+//        }
+//    }
+//
+//    /**
+//     * 事件告警对接Kafka
+//     * @param record record
+//     * @param ack ack
+//     *            groupId = kafkaConsumerGroup
+//     *            该消息的消息格式为
+//     *            {"data_class":"realdata","data_type":"alarm","op_type":"subscribe_emergency","condition":{"station_psr_id":"50edcb6c1b8a811030493c80a2014950ed9d4f59e8","station_name":"中州换流站","alarm_type":"yx_bw"},"data":[{"psrId":"D017020000000000000000999","astId":"D017020000000000000000999","equipType":"ASTType_0000111","eventType":"OtherSignal","alarmSource":"OWS","alarmLevel":"3","description":"2024-03-11 09:06:17::585 S2WCL12A E3.C01软水器再生结束信号 出现","dateTime":"2024-03-11 09:06:17.585"}]}
+//     */
+//
+//    @KafkaListener(id = "kafkaConsumerEventAlarm", groupId = "kafkaConsumerGroupEventAlarm", topics = "#{'${queue.kafka.eventAlarm.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
+//    public void kafkaConsumerEventAlarm(ConsumerRecord<?, String> record, Acknowledgment ack) {
+//        Optional<?> message = Optional.ofNullable(record.value());
+//        if (message.isPresent()) {
+//            try {
+////                JSONObject messageObj = JSONObject.fromObject(record.value());
+////                JSONArray dataArray = messageObj.getJSONArray("data");
+////                JSONArray jsonArray = new JSONArray();
+////                String timestamp = "";
+////                for (Object obj : dataArray) {
+////                    JSONObject finallyObj = new JSONObject();
+////                    com.alibaba.fastjson.JSONObject detail = com.alibaba.fastjson.JSONObject.parseObject(com.alibaba.fastjson.JSONObject.toJSONString(obj));
+////                    finallyObj.put("eventtextL1", detail.get("description"));
+////                    finallyObj.put("pointId", detail.get("astId"));
+////                    finallyObj.put("time", detail.get("dateTime"));
+////                    jsonArray.add(finallyObj);
+////                    timestamp = detail.get("dateTime").toString();
+////                }
+////                JSONObject jsonObjectMessage = new JSONObject();
+////                jsonObjectMessage.put("warns", jsonArray);
+////                jsonObjectMessage.put("timestamp", timestamp);
+//
+//                JSONObject object = JSONObject.fromObject(record.value());
+//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
+//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
+//                ack.acknowledge();
+//            } catch (MqttException e) {
+//                log.error("解析数据失败,{}", e.getMessage());
+//            } catch (UnsupportedEncodingException e) {
+//                e.printStackTrace();
+//            }
+//        }
+//    }
 
 
     ///**

amos-boot-utils/amos-boot-utils-message/src/main/resources/application-dev.properties

@@ -111,3 +111,6 @@ system.zxj=false
 
 queue.kafka.topics=null
 kafka.auto-startup=false
+
+#浜嬩欢鍛婅瀵规帴Kafka涓婚
+queue.kafka.eventAlarm.topics=JKXT2BP-GJ-Topic-site

amos-boot-utils/amos-boot-utils-message/src/main/resources/json/commonMessage.json

 [
   {
-    "kafkaTopic": "k1",
+    "kafkaTopic": "T_DC_MQ_REALDATA__guanggu",
     "mqTopic": "romaSite/data/transmit",
     "data": {
-      "dataType": "condition.station_psr_id",
-      "value": "condition.station_psr_id",
-      "timeStamp": "condition.station_psr_id",
-      "quality": "condition.station_psr_id",
-      "scadaId": "condition.station_psr_id",
-      "key": "condition.station_psr_id",
-      "disCreate": "condition.station_psr_id",
-      "name": "condition.station_psr_id"
+      "dataType": "body.datatype",
+      "value": "body.value",
+      "timeStamp": "body.time_stamp",
+      "quality": "body.quality",
+      "scadaId": "body.key",
+      "key": "body.key",
+      "disCreate": "body.station_psr_id",
+      "name": "body.name"
     }
   },
   {
-    "kafkaTopic": "k2",
+    "kafkaTopic": "T_DC_MQ_ALARM__guanggu",
     "mqTopic": "romaSite/data/eventAlarm",
     "data": {
       "timeStamp": "body.warns.time",