终端异常告警接口优化

pqs-device/pq-device/pq-device-boot/src/main/java/com/njcn/device/pq/mapper/TopMsgMapper.java

@@ -10,9 +10,9 @@ import org.apache.ibatis.annotations.Param;
 import java.util.List;
 
 public interface TopMsgMapper {
-
    Page<TopMsgPO> getTopMsg(Page<TopMsgPO> page,@Param("lineGrade") String lineGrade, @Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
-   List<TopMsgPO> getTopMsgDetail(@Param("lineGrade") String lineGrade,@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
+
+   List<TopMsgPO> getTopMsgDetail(@Param("lineGrade") String lineGrade,@Param("startTime") DateTime startTime);
 
 
 

pqs-device/pq-device/pq-device-boot/src/main/java/com/njcn/device/pq/mapper/mapping/LineMapper.xml

@@ -865,6 +865,8 @@
         </foreach>
         and
             a.Create_Time between #{startTime} and #{endTime}
+        and
+            b.Run_Flag != 2
         ) t ORDER BY flowProportion DESC
     </select>
 

pqs-device/pq-device/pq-device-boot/src/main/java/com/njcn/device/pq/mapper/mapping/TopMagMapper.xml

@@ -5,12 +5,12 @@
     <select id="getTopMsg" resultType="com.njcn.device.pq.pojo.po.TopMsgPO">
         SELECT
                 ptm.Time_Id AS time,
-                COUNT(ptm.Dev_Id) AS deviceAbnormalNum,
+                IFNULL(COUNT(ptm.Dev_Id),3.14159)AS deviceAbnormalNum,
                 GROUP_CONCAT(sdd.Name) AS lineGrade
         FROM pq_line line
         INNER JOIN pq_line vol ON line.pid=vol.id
         INNER JOIN pq_line device ON vol.pid=device.id
-        INNER JOIN pqs_top_msg ptm ON ptm.Dev_Id = device.Id
+        LEFT JOIN pqs_top_msg ptm ON ptm.Dev_Id = device.Id
         LEFT JOIN pq_line_detail pld ON pld.Id=line.Id
         LEFT JOIN sys_dict_data sdd ON sdd.Id=pld.Line_Grade
         WHERE
@@ -46,7 +46,7 @@
         WHERE
         ptm.State=1
         AND
-        ptm.Time_Id between #{startTime} and #{endTime}
+        ptm.Time_Id = #{startTime}
         <if test="lineGrade!=null and lineGrade!=''">
             AND pld.Line_Grade=#{lineGrade}
         </if>

pqs-device/pq-device/pq-device-boot/src/main/java/com/njcn/device/pq/service/impl/LineServiceImpl.java

@@ -399,8 +399,7 @@ public class LineServiceImpl extends ServiceImpl<LineMapper, Line> implements Li
 
     @Override
     public List<TopMsgPO> dailyDeviceAbnormal(DeviceInfoParam.BusinessParam param) {
-        List<TopMsgPO> topMsgDetails = topMsgMapper.getTopMsgDetail(param.getLineGrade(), DateUtil.beginOfDay(DateUtil.parse(param.getSearchBeginTime())),
+        List<TopMsgPO> topMsgDetails = topMsgMapper.getTopMsgDetail(param.getLineGrade(), DateUtil.beginOfDay(DateUtil.parse(param.getSearchBeginTime())));
-                DateUtil.endOfDay(DateUtil.parse(param.getSearchEndTime())));
         return topMsgDetails;
     }
 

pqs-prepare/prepare-api/src/main/java/com/njcn/prepare/harmonic/pojo/influxdb/po/Communicate.java

@@ -0,0 +1,49 @@
+package com.njcn.prepare.harmonic.pojo.influxdb.po;
+
+import lombok.Data;
+import org.influxdb.annotation.Column;
+import org.influxdb.annotation.Measurement;
+
+import java.time.Instant;
+
+/**
+ * <p>
+ *
+ * </p>
+ *
+ * @author cdf
+ * @since 2022-01-07
+ */
+@Data
+@Measurement(name = "pqs_communicate")
+public class Communicate {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 终端Id
+     */
+    /**
+     * 更新时间
+     */
+    @Column(name = "time")
+    private Instant updateTime;
+
+    @Column(name = "dev_id")
+    private String devId;
+
+
+    /**
+     * 事件类型(0：中断；1：正常；2：退出)
+     */
+    @Column(name = "type")
+    private Integer type;
+
+    /**
+     * 备注
+     */
+    @Column(name = "description")
+    private String remark;
+
+
+}

pqs-prepare/prepare-api/src/main/java/com/njcn/prepare/harmonic/pojo/mysql/po/line/AlarmPO.java

@@ -22,6 +22,7 @@ public class AlarmPO extends BaseEntity {
 
     private String deviceId;
 
+    private LocalDateTime updateTime;
 
     /**
      * 告警描述
@@ -69,9 +70,9 @@ public class AlarmPO extends BaseEntity {
     private Float flowProportion;
 
     /**
-     * 通讯终端次数
+     * 通讯中断次数
      */
-    private Integer comOutNum = 0;
+    private Integer comOutNum;
 
     /**
      * 中断时间段描述

pqs-prepare/prepare-api/src/main/java/com/njcn/prepare/harmonic/pojo/mysql/po/line/TopMsgPO.java

@@ -4,6 +4,8 @@ import com.njcn.db.bo.BaseEntity;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
+import java.util.List;
+
 @Data
 @EqualsAndHashCode(callSuper = true)
 public class TopMsgPO extends BaseEntity {
@@ -26,7 +28,7 @@ public class TopMsgPO extends BaseEntity {
     /**
      * 通信中断时间段描述
      */
-    private String comOutDesc;
+    private List<String> comOutDesc;
 
     /**
      * 数据完整性标识 0.不达标 1.达标

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/mapper/mysql/device/mapping/DeviceAbnormaStatisticsMapper.xml

@@ -80,7 +80,8 @@
                     COUNT(ca.Id) AS alarmCount,
                     GROUP_CONCAT(ca.Remark) AS alarmDesc,
                     IFNULL(d.flow, (select flow from cld_flow_meal where type = 0 and flag = 1)) + ifnull(d1.flow, 0) flowMeal,
-                IFNULL(pmf.Actual_Value,0) statisValue
+                    IFNULL(ANY_VALUE(pmf.Actual_Value),0) statisValue,
+                    ANY_VALUE(device.Update_Time) AS updateTime
             FROM pq_line line
             INNER JOIN pq_line vol ON line.pid=vol.id
             INNER JOIN pq_line device ON vol.pid=device.id

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/service/mysql/Impl/device/DeviceAbnormaStatisticsServiceImpl.java

@@ -2,23 +2,20 @@ package com.njcn.prepare.harmonic.service.mysql.Impl.device;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
-import com.njcn.device.pq.pojo.po.Communicate;
 import com.njcn.influxdb.param.InfluxDBPublicParam;
 import com.njcn.influxdb.utils.InfluxDbUtils;
 import com.njcn.prepare.harmonic.mapper.mysql.device.DeviceAbnormaStatisticsMapper;
+import com.njcn.prepare.harmonic.pojo.influxdb.po.Communicate;
 import com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO;
 import com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmStrategyVO;
-import com.njcn.prepare.harmonic.pojo.mysql.po.line.LinePO;
 import com.njcn.prepare.harmonic.pojo.mysql.po.line.TopMsgPO;
 import com.njcn.prepare.harmonic.pojo.param.DeviceAbnormaStatisticsParam;
 import com.njcn.prepare.harmonic.service.mysql.device.DeviceAbnormalStatisticsService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang.StringUtils;
 import org.influxdb.dto.QueryResult;
 import org.influxdb.impl.InfluxDBResultMapper;
 import org.springframework.stereotype.Service;
-import org.springframework.util.CollectionUtils;
 
 import java.math.BigDecimal;
 import java.util.*;
@@ -37,33 +34,31 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
     private final InfluxDbUtils influxDbUtils;
     @Override
     public boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param) {
+        Map<String, List<TopMsgPO>> comMap = new HashMap<>();Map<String, List<AlarmPO>> devLineMap = new HashMap<>();
         // 获取监测点告警
         List<AlarmPO> lineAlarms = deviceAbnormaStatisticsMapper.getAlarmByLines(
                 DateUtil.beginOfDay(DateUtil.parse(param.getBeginTime())),
                 DateUtil.beginOfDay(DateUtil.parse(param.getEndTime())));
-        Map<String, List<AlarmPO>> lineMap = new HashMap<>();
         if (CollectionUtil.isNotEmpty(lineAlarms)){
             // 根据装置分组
-            lineMap = lineAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+            devLineMap = lineAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
         }
-
         // 获取设备流量
         List<AlarmPO> deviceAlarms = deviceAbnormaStatisticsMapper.getAlarmByDevice(
                 DateUtil.beginOfDay(DateUtil.parse(param.getBeginTime())),
                 DateUtil.beginOfDay(DateUtil.parse(param.getEndTime())));
         List<String> devs = deviceAlarms.stream().map(AlarmPO::getId).collect(Collectors.toList());
-        devs.add("6469e77fda42db12c7ca6620a092f03c");
-        Map<String, List<Communicate>> comMap = new HashMap<>();
         // 获取通信中断信息
-        List<Communicate> communicate = getCommunicate(devs, param.getBeginTime(), param.getEndTime());
+        List<TopMsgPO> comMsgs = getCommunicate(devs, param.getBeginTime(), param.getEndTime());
-        if (CollectionUtil.isNotEmpty(communicate)){
+        if (CollectionUtil.isNotEmpty(comMsgs)){
-            comMap = communicate.stream().collect(Collectors.groupingBy(Communicate::getId));
+           comMap = comMsgs.stream().collect(Collectors.groupingBy(TopMsgPO::getDevId));
         }
+        // 整合监测点告警信息、数据完整性以及监测点等级为设备信息
         for (AlarmPO deviceAlarm : deviceAlarms) {
-            if (!lineMap.containsKey(deviceAlarm.getId())) {
+            if (!devLineMap.containsKey(deviceAlarm.getId())) {
                 continue;
             }
-            AlarmPO lineAlarm = lineMap.get(deviceAlarm.getId()).get(0);
+            AlarmPO lineAlarm = devLineMap.get(deviceAlarm.getId()).get(0);
             deviceAlarm.setAlarmCount(deviceAlarm.getAlarmCount() + lineAlarm.getAlarmCount());
             if (lineAlarm.getDue().compareTo(BigDecimal.ZERO) == 0) {
                 deviceAlarm.setIntegrity(BigDecimal.ZERO);
@@ -73,15 +68,14 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
             if (lineAlarm.getLevel() != null) {
                 deviceAlarm.setLevel(lineAlarm.getLevel());
             }
-            if (CollectionUtil.isNotEmpty(comMap)) {
+            TopMsgPO topMsgPO = comMap.get(deviceAlarm.getId()).get(0);
-                List<Communicate> communicates = comMap.get(deviceAlarm.getId());
+            deviceAlarm.setComOutNum(topMsgPO.getComOutCount());
-                int comOut = communicates.stream().filter(t -> t.getType() == 0).collect(Collectors.toList()).size();
+            deviceAlarm.setComOutDesc(String.join(",",topMsgPO.getComOutDesc()));
-                deviceAlarm.setComOutNum(comOut);
         }
-        }
+        // 比对告警策略并落表
-        Map<Integer, List<AlarmPO>> levelMap = deviceAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getLevel));
         List<AlarmStrategyVO> alarmStrategyVOS = deviceAbnormaStatisticsMapper.selectAlarmStrategy();
         List<TopMsgPO> alarmExceptions = new ArrayList<>();
+        Map<Integer, List<AlarmPO>> levelMap = deviceAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getLevel));
         for (AlarmStrategyVO strategyVO : alarmStrategyVOS) {
             if (!levelMap.containsKey(strategyVO.getAlgoDesc())){
                 continue;
@@ -91,7 +85,7 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
                 TopMsgPO topMsg = new TopMsgPO();
                 topMsg.setDevId(alarmPO.getId());
                 topMsg.setFlowFlag(1);
-                topMsg.setComOutCount(alarmPO.getComOutNum() == 0 ? 0 : alarmPO.getComOutNum());
+                topMsg.setComOutCount(alarmPO.getComOutNum() == null ? 0 : alarmPO.getComOutNum());
                 if (alarmPO.getIntegrity().intValue()<strategyVO.getIntegrityValue()) {
                     topMsg.setIntegrityFlag("0");
                     topMsg.setIntegrityValue(alarmPO.getIntegrity().intValue());
@@ -111,7 +105,7 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
     }
 
 
-    public List<Communicate> getCommunicate(List<String> devs, String startTime, String endTime) {
+    public List<TopMsgPO> getCommunicate(List<String> devs, String startTime, String endTime) {
         //组装sql语句
         StringBuilder stringBuilder = new StringBuilder();
         stringBuilder.append("time >= '").append(DateUtil.beginOfDay(DateUtil.parse(startTime))).append("' and ").append("time <= '").append(DateUtil.endOfDay(DateUtil.parse(endTime))).append("' and ");
@@ -126,7 +120,55 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
         //获取暂降事件
         QueryResult result = influxDbUtils.query(sql);
         InfluxDBResultMapper influxDBResultMapper = new InfluxDBResultMapper();
-        List<Communicate> communicateList = influxDBResultMapper.toPOJO(result, Communicate.class);
+        List<Communicate> communicates = influxDBResultMapper.toPOJO(result, Communicate.class);
-        return communicateList;
+        List<TopMsgPO> comMsgs = new ArrayList<>();
+        if (CollectionUtil.isEmpty(communicates)) {
+            return comMsgs;
+        }
+        Map<String, List<Communicate>> comMap = communicates.stream().collect(Collectors.groupingBy(Communicate::getDevId));
+        for (Map.Entry<String, List<Communicate>> comEntry : comMap.entrySet()) {
+            TopMsgPO topMsgPO = new TopMsgPO(); List<String> comOutDesc = new ArrayList<>();
+            topMsgPO.setDevId(comEntry.getKey());
+            // 根据日期排序
+            List<Communicate> sortedList = comEntry.getValue().stream().sorted(Comparator.comparing(Communicate::getUpdateTime)).collect(Collectors.toList());
+            if (sortedList.size() == 1) {
+                if (sortedList.get(0).getType() == 0) {
+                    comOutDesc.add(sortedList.get(0).getUpdateTime()+"至"+DateUtil.endOfDay(DateUtil.parse(endTime)));
+                } else {
+                    comOutDesc.add((DateUtil.beginOfDay(DateUtil.parse(startTime)))+"至"+sortedList.get(0).getUpdateTime());
+                }
+            } else {
+                switch (sortedList.get(0).getType()) {
+                    case 0:
+                        for (int i = 0; i <sortedList.size() ; i++) {
+                            if (sortedList.get(i).getType()==1) {
+                                String comOutTime = sortedList.get(i-1).getUpdateTime() +"至"+ sortedList.get(i).getUpdateTime();
+                                comOutDesc.add(comOutTime);
+                            }
+                        }
+                        break;
+                    case 1:
+                        comOutDesc.add((DateUtil.beginOfDay(DateUtil.parse(startTime)))+"至"+sortedList.get(0).getUpdateTime());
+                        for (int i = 0; i <sortedList.size();i++) {
+                            if (sortedList.get(i).getType()==0) {
+                                StringBuilder sb = new StringBuilder();
+                                sb.append(sortedList.get(i).getUpdateTime());
+                                sb.append("至");
+                                sb.append(i==sortedList.size()-1?DateUtil.endOfDay(DateUtil.parse(endTime)):sortedList.get(i+1).getUpdateTime());
+                                comOutDesc.add(sb.toString());
+                            }
+                        }
+                        break;
+                    default:
+                        break;
+                }
+            }
+
+            topMsgPO.setComOutDesc(comOutDesc);
+            // 通信中断次数
+            topMsgPO.setComOutCount(sortedList.stream().filter(a -> a.getType() == 0).collect(Collectors.toList()).size());
+            comMsgs.add(topMsgPO);
+        }
+        return comMsgs;
     }
 }