月流量表、数据完整性表切换&&异常统计优化

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/controller/device/DeviceAbnormalController.java

@@ -15,6 +15,8 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.bind.annotation.*;
 
+import java.text.ParseException;
+
 @Slf4j
 @Api(tags = "终端异常")
 @RestController
@@ -28,7 +30,7 @@ public class DeviceAbnormalController extends BaseController {
     @PostMapping("/statistics")
     @ApiOperation("终端异常统计")
     @ApiImplicitParam(name = "param", value = "参数", required = true)
-    public HttpResult<Boolean> dailyDeviceAbnormaStatistics(@RequestBody DeviceAbnormaStatisticsParam param){
+    public HttpResult<Boolean> dailyDeviceAbnormaStatistics(@RequestBody DeviceAbnormaStatisticsParam param) throws ParseException {
         String methodDescribe = getMethodDescribe("dailyDeviceAbnormaStatistics");
         boolean res = deviceAbnormaStatisticsService.dailyDeviceAbnormaStatistics(param);
         if(res){

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

@@ -3,7 +3,6 @@ package com.njcn.prepare.harmonic.mapper.mysql.device;
 import cn.hutool.core.date.DateTime;
 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 org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
@@ -13,11 +12,19 @@ import java.util.List;
 @Mapper
 public interface DeviceAbnormaStatisticsMapper {
 
-    List<LinePO> getLines();
+    /**
+     * 获取所有有等级监测点以及对应设备
+     * @return List<AlarmPO>
+     */
+    List<AlarmPO> getLines();
 
-    List<AlarmPO> getAlarmByDevice(@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
+    List<AlarmPO> getFlowAndOnlineRateByDevice(@Param("list") List<String> list,@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
 
-    List<AlarmPO> getAlarmByLines(@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
+    List<AlarmPO> getIntegrityByLines(@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
+
+    List<AlarmPO> getIntegrityByLineIds(@Param("list") List<String> list,@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
+
+    List<AlarmPO> getAlarmByLines(@Param("monitorIds") List<String> monitorIds,@Param("devIds") List<String> devIds,@Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);
 
     Boolean insertTopMsg(@Param("list") List<TopMsgPO> topMsgPOS);
 

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

@@ -13,6 +13,8 @@
         `Alarm_Desc`,
         `Flow_Flag`,
         `Flow_Value`,
+        `OnlineRate_Flag`,
+        `OnlineRate_Value`,
         `State`
         ) VALUES
         <foreach collection="list" separator="," item="item">
@@ -20,118 +22,138 @@
             now(),
             #{item.devId},
             #{item.comOutCount},
-            #{item.comOutDesc},
+            #{item.comOutDescription},
             #{item.integrityFlag},
             #{item.integrityValue},
             IFNULL(#{item.alarmCount},0),
             #{item.alarmDesc},
             #{item.flowFlag},
             #{item.flowValue},
+            #{item.onlineRateFlag},
+            #{item.onlineRateValue},
             1
             )
         </foreach>
     </insert>
 
-    <select id="getLines" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.LinePO">
-       (SELECT   line.Id AS Id,
-                device.id AS deviceId,
-                COUNT(ca.Id) AS alarmCount,
-				GROUP_CONCAT(ca.Remark) AS alarmDesc
+    <select id="getLines" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
+        SELECT  line.Id AS id,
+                device.Id AS deviceId,
+                sdd.Algo_Describe AS level,
+                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
         LEFT JOIN pq_device pd ON device.id=pd.id
         LEFT JOIN pq_line_detail pld ON pld.id=line.id
-        LEFT JOIN cld_alarm ca ON ca.Line_Id=line.id
+        LEFT JOIN sys_dict_data sdd ON sdd.Id = pld.Line_Grade
         WHERE
-            pd.Dev_Model=1
+                pd.Dev_Model=1
         AND
-            pd.Run_Flag=0
+                pd.Run_Flag=0
         AND
-            pld.Line_Grade is NOT NULL
-        GROUP BY deviceId)
-        UNION ALL
-        (SELECT  device.Id AS Id,
-                device.id AS deviceId,
-				COUNT(ca.Id) AS alarmCount,
-				GROUP_CONCAT(ca.Remark) AS alarmDesc
-        FROM pq_line line
-        INNER JOIN pq_line vol ON line.pid=vol.id
-        INNER JOIN pq_line device ON vol.pid=device.id
-        LEFT JOIN pq_device pd ON device.id=pd.id
-        LEFT JOIN pq_line_detail pld ON pld.id=line.id
-        LEFT JOIN cld_alarm ca ON ca.Line_Id=device.id
-        WHERE
-            pd.Dev_Model=1
-        AND
-            pd.Run_Flag=0
-        AND
-            pld.Line_Grade is NOT NULL
-        GROUP BY deviceId)
+                pld.Line_Grade is NOT NULL
     </select>
 
-    <select id="getAlarmByDevice" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
+    <select id="getIntegrityByLines" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
+       SELECT   line.Id AS Id,
+                device.id AS deviceId,
+                IFNULL(SUM(integrity.due_time),0) AS due,
+				IFNULL(SUM(integrity.real_time),0) AS `real`
+        FROM pq_line line
+        INNER JOIN pq_line vol ON line.pid=vol.id
+        INNER JOIN pq_line device ON vol.pid=device.id
+        LEFT JOIN pq_device pd ON device.id=pd.id
+        LEFT JOIN pq_line_detail pld ON pld.id=line.id
+        LEFT JOIN r_stat_integrity_d integrity ON integrity.line_index = line.id
+        WHERE
+            pd.Dev_Model=1
+        AND
+            pd.Run_Flag=0
+        AND
+            pld.Line_Grade is NOT NULL
+        AND
+            integrity.time_id BETWEEN #{startTime} AND #{endTime}
+        GROUP BY device.Id
+    </select>
+
+    <select id="getIntegrityByLineIds" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
+       SELECT
+            device.id AS deviceId,
+            IFNULL(SUM(integrity.due_time),0) AS due,
+            IFNULL(SUM(integrity.real_time),0) AS `real`
+        FROM r_stat_integrity_d integrity
+        INNER JOIN pq_line line ON integrity.line_index = line.id
+        INNER JOIN pq_line vol ON line.pid=vol.id
+        INNER JOIN pq_line device ON vol.pid=device.id
+        WHERE integrity.line_index IN
+        <foreach collection="list" index="index" item="item" open="(" separator="," close=")">
+            #{item}
+        </foreach>
+        AND
+            integrity.time_id BETWEEN #{startTime} AND #{endTime}
+        GROUP BY device.Id
+    </select>
+
+    <select id="getFlowAndOnlineRateByDevice" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
         SELECT t.*,
-        (t.statisValue)/t.flowMeal flowProportion
+        (t.actualValue)/(t.flowMeal) flowProportion
 		FROM
             (
-            SELECT  device.Id AS Id,
-                    device.id AS deviceId,
-                    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(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
-            LEFT JOIN pq_device pd ON device.id=pd.id
-            LEFT JOIN pq_line_detail pld ON pld.id=line.id
-            LEFT JOIN cld_alarm ca ON ca.Line_Id=device.id
-            LEFT JOIN pqs_month_flow pmf ON pmf.Dev_Id=device.id
-            LEFT JOIN cld_dev_meal c ON device.id = c.line_id
+            SELECT
+                    cmf.Line_Id deviceId,
+                    IFNULL(d.flow, (select flow from cld_flow_meal where type = 0 and flag = 1)) +  IFNULL(d1.flow, 0) flowMeal,
+                    IFNULL(cmf.Actual_Value,0) actualValue,
+                    rsod.online_min/1440 onlineRate
+            FROM  cld_month_flow cmf
+            LEFT JOIN cld_dev_meal c ON cmf.Line_Id = c.Line_Id
             LEFT JOIN cld_flow_meal d ON c.Base_Meal_Id = d.id
             LEFT JOIN cld_flow_meal d1 ON c.Ream_Meal_Id = d1.id
-            WHERE
-                pd.Dev_Model = 1
+            LEFT JOIN r_stat_onlinerate_d rsod ON rsod.dev_index = cmf.Line_Id
+            WHERE cmf.Line_Id IN
+                <foreach collection="list" index="index" item="item" open="(" separator="," close=")">
+                    #{item}
+                </foreach>
             AND
-                pd.Run_Flag = 0
-            AND
-                pld.Line_Grade is NOT NULL
-            AND
-                ca.Occurred_Time between #{startTime} and #{endTime}
-            GROUP BY deviceId
+                cmf.Time_Id BETWEEN #{startTime} AND #{endTime}
             ) t
         ORDER BY flowProportion DESC
     </select>
 
     <select id="getAlarmByLines" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
-       SELECT   GROUP_CONCAT(line.Id) AS Id,
+       (SELECT  GROUP_CONCAT(line.Id) AS Id,
                 device.id AS deviceId,
                 COUNT(ca.Id) AS alarmCount,
-				GROUP_CONCAT(ca.Remark) AS alarmDesc,
-				MIN(sdd.Algo_Describe) AS `level`,
-				IFNULL(SUM(pi.Due),0) AS due,
-				IFNULL(SUM(pi.Real),0) AS `real`
-        FROM pq_line line
+				GROUP_CONCAT(ca.Remark) AS alarmDesc
+        FROM cld_alarm ca
+        INNER JOIN  pq_line line ON ca.Line_Id = line.id
         INNER JOIN pq_line vol ON line.pid=vol.id
         INNER JOIN pq_line device ON vol.pid=device.id
-        LEFT JOIN pq_device pd ON device.id=pd.id
-        LEFT JOIN pq_line_detail pld ON pld.id=line.id
-        LEFT JOIN cld_alarm ca ON ca.Line_Id=line.id
-		LEFT JOIN pqs_integrity pi ON pi.LINE_Id=line.Id
-		LEFT JOIN sys_dict_data sdd ON sdd.Id=pld.Line_Grade
-        WHERE
-            pd.Dev_Model=1
+        WHERE ca.Line_Id IN
+        <foreach collection="monitorIds" index="index" item="item1" open="(" separator="," close=")">
+            #{item1}
+        </foreach>
         AND
-            pd.Run_Flag=0
+            ca.Occurred_Time BETWEEN #{startTime} AND #{endTime}
+        GROUP BY device.Id
+        ORDER BY alarmCount DESC)
+        UNION ALL
+        (SELECT
+                ca.Line_Id AS Id,
+                ca.Line_Id AS deviceId,
+                COUNT(ca.Id) AS alarmCount,
+                GROUP_CONCAT(ca.Remark) AS alarmDesc
+        FROM cld_alarm ca
+        WHERE ca.Line_Id IN
+        <foreach collection="devIds" index="index" item="item2" open="(" separator="," close=")">
+            #{item2}
+        </foreach>
         AND
-            pld.Line_Grade is NOT NULL
-        AND
-            ca.Occurred_Time between #{startTime} and #{endTime}
-        GROUP BY deviceId
-        ORDER BY alarmCount DESC
+        ca.Occurred_Time BETWEEN #{startTime} AND #{endTime}
+        GROUP BY ca.Line_Id
+        ORDER BY alarmCount DESC)
     </select>
+
     <select id="selectAlarmStrategy" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmStrategyVO">
        SELECT
         A.`Id` id,
@@ -143,9 +165,9 @@
         A.`Update_Time` updateTime,
         B.Algo_Describe algoDesc
         FROM
-        pqs_alarm_strategy A,
-        sys_dict_data B
+        pqs_alarm_strategy A
+        LEFT JOIN sys_dict_data B ON A.Id = B.Id
         WHERE A.State = 1
-        AND A.Id = B.Id
     </select>
+
 </mapper>

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

@@ -18,6 +18,10 @@ import org.influxdb.impl.InfluxDBResultMapper;
 import org.springframework.stereotype.Service;
 
 import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.time.*;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -33,79 +37,128 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
 
     private final InfluxDbUtils influxDbUtils;
     @Override
-    public boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param) {
+    public boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param) throws ParseException {
         Map<String, List<TopMsgPO>> comMap = new HashMap<>();Map<String, List<AlarmPO>> devLineMap = new HashMap<>();
-        // 获取监测点告警
+        Map<String, List<AlarmPO>> deviceFlowAndOnlineRateMap  = new HashMap<>();
+        // 获取所有装置以及下属监测点
+        List<AlarmPO> lines = deviceAbnormaStatisticsMapper.getLines();
+        // 根据装置id分组
+        Map<String, List<AlarmPO>> devMap = lines.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+        // 监测点集合
+        List<String> monotorIds = lines.stream().map(AlarmPO::getId).collect(Collectors.toList());
+        // 装置id集合
+        List<String> devIds = new ArrayList<>(devMap.keySet());
+        // 获取监测点数据完整性
+        List<AlarmPO> lineIntegritys = deviceAbnormaStatisticsMapper.getIntegrityByLineIds(
+                monotorIds,
+                DateUtil.beginOfDay(DateUtil.parse(param.getBeginTime())),
+                DateUtil.endOfDay(DateUtil.parse(param.getEndTime())));
+        Map<String, List<AlarmPO>> lineIntegrityMap = lineIntegritys.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
         List<AlarmPO> lineAlarms = deviceAbnormaStatisticsMapper.getAlarmByLines(
+                monotorIds,
+                devIds,
                 DateUtil.beginOfDay(DateUtil.parse(param.getBeginTime())),
                 DateUtil.endOfDay(DateUtil.parse(param.getEndTime())));
-        if (CollectionUtil.isNotEmpty(lineAlarms)){
-            // 根据装置分组
-            devLineMap = lineAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
-        }
+        Map<String, List<AlarmPO>> lineAlarmsMap = lineAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
         // 获取设备流量
-        List<AlarmPO> deviceAlarms = deviceAbnormaStatisticsMapper.getAlarmByDevice(
+        List<AlarmPO> deviceFlowAndOnlineRate = deviceAbnormaStatisticsMapper.getFlowAndOnlineRateByDevice(
+                devIds,
                 DateUtil.beginOfDay(DateUtil.parse(param.getBeginTime())),
                 DateUtil.endOfDay(DateUtil.parse(param.getEndTime())));
-        List<String> devs = deviceAlarms.stream().map(AlarmPO::getId).collect(Collectors.toList());
-        // 获取通信中断信息
-        List<TopMsgPO> comMsgs = getCommunicate(devs, param.getBeginTime(), param.getEndTime());
-        if (CollectionUtil.isNotEmpty(comMsgs)){
-           comMap = comMsgs.stream().collect(Collectors.groupingBy(TopMsgPO::getDevId));
+        if (CollectionUtil.isNotEmpty(deviceFlowAndOnlineRate)) {
+            deviceFlowAndOnlineRateMap = deviceFlowAndOnlineRate.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
         }
-        // 整合监测点告警信息、数据完整性以及监测点等级为设备信息
-        for (AlarmPO deviceAlarm : deviceAlarms) {
-            if (!devLineMap.containsKey(deviceAlarm.getId())) {
-                continue;
-            }
-            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);
-            } else {
-                deviceAlarm.setIntegrity(lineAlarm.getReal().divide(lineAlarm.getDue(), 2, BigDecimal.ROUND_HALF_UP));
-            }
-            if (lineAlarm.getLevel() != null) {
-                deviceAlarm.setLevel(lineAlarm.getLevel());
-            }
-            TopMsgPO topMsgPO = comMap.get(deviceAlarm.getId()).get(0);
-            deviceAlarm.setComOutNum(topMsgPO.getComOutCount());
-            deviceAlarm.setComOutDesc(String.join(",",topMsgPO.getComOutDesc()));
+        // 获取通信中断信息
+       // List<String> list = Arrays.asList("025fa0e4c91f72ad7f1c1bd29026f20a");
+        List<TopMsgPO> comMsgs = getCommunicate(devIds, param.getBeginTime(), param.getEndTime());
+        if (CollectionUtil.isNotEmpty(comMsgs)) {
+           comMap = comMsgs.stream().collect(Collectors.groupingBy(TopMsgPO::getDevId));
         }
         // 比对告警策略并落表
         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())){
+        Map<Integer, List<AlarmStrategyVO>> levMap = alarmStrategyVOS.stream().collect(Collectors.groupingBy(AlarmStrategyVO::getAlgoDesc));
+
+        // 整合装置信息
+        List<TopMsgPO> topMsgPOS = new ArrayList<>();
+        for (Map.Entry<String, List<AlarmPO>> alarmEntry : devMap.entrySet()) {
+            if (!lineIntegrityMap.containsKey(alarmEntry.getKey()) && !lineAlarmsMap.containsKey(alarmEntry.getKey())
+                    && !deviceFlowAndOnlineRateMap.containsKey(alarmEntry.getKey()) && !comMap.containsKey(alarmEntry.getKey())) {
                 continue;
             }
-            List<AlarmPO> alarmPOS = levelMap.get(strategyVO.getAlgoDesc());
-            for (AlarmPO alarmPO : alarmPOS) {
-                TopMsgPO topMsg = new TopMsgPO();
-                topMsg.setDevId(alarmPO.getId());
-                topMsg.setFlowFlag(1);
-                topMsg.setComOutCount(alarmPO.getComOutNum() == null ? 0 : alarmPO.getComOutNum());
-                if (alarmPO.getIntegrity().intValue()<strategyVO.getIntegrityValue()) {
-                    topMsg.setIntegrityFlag("0");
-                    topMsg.setIntegrityValue(alarmPO.getIntegrity().intValue());
-                }
-                if (alarmPO.getAlarmCount()>strategyVO.getWarnValue()){
-                    topMsg.setAlarmCount(alarmPO.getAlarmCount());
-                    topMsg.setAlarmDesc(alarmPO.getAlarmDesc());
-                }
-                if (alarmPO.getStatisValue()>alarmPO.getFlowMeal()) {
-                    topMsg.setFlowFlag(0);
-                    topMsg.setFlowValue(alarmPO.getFlowProportion());
-                }
-                alarmExceptions.add(topMsg);
+            TopMsgPO topMsg = new TopMsgPO();
+            topMsg.setDevId(alarmEntry.getKey());
+            int deviceLevel = alarmEntry.getValue().stream().mapToInt(AlarmPO::getLevel).min().orElse(-1);
+            topMsg.setDeviceLevel(deviceLevel);
+            AlarmStrategyVO alarmStrategyVO = levMap.get(deviceLevel).get(0);
+            // 数据完整性
+            if (lineIntegrityMap.containsKey(alarmEntry.getKey())) {
+                Integer due = lineIntegrityMap.get(alarmEntry.getKey()).get(0).getDue();
+                Integer real = lineIntegrityMap.get(alarmEntry.getKey()).get(0).getReal();
+                float integrity = BigDecimal.valueOf((float) real *100 / due).setScale(2, RoundingMode.HALF_UP).floatValue();
+                topMsg.setIntegrityFlag(integrity < alarmStrategyVO.getIntegrityValue() ? 0 : 1);
+                topMsg.setIntegrityValue(integrity);
+            } else {
+                topMsg.setIntegrityFlag(-1);
             }
+            // 装置告警
+            if (lineAlarmsMap.containsKey(alarmEntry.getKey())) {
+                List<AlarmPO> lineAlarmPOS = lineAlarmsMap.get(alarmEntry.getKey());
+                int alarmCount = lineAlarmPOS.stream().mapToInt(AlarmPO::getAlarmCount).sum();
+                String alarmDesc = String.join(",", lineAlarmPOS.stream().map(AlarmPO::getAlarmDesc).collect(Collectors.toList()));
+                topMsg.setAlarmCountFlag(alarmCount > alarmStrategyVO.getWarnValue() ? 0 : 1);
+                topMsg.setAlarmCount(alarmCount);
+                topMsg.setAlarmDesc(alarmDesc);
+            } else {
+                topMsg.setAlarmCount(0);
+            }
+            // 流量和在线率
+            if (deviceFlowAndOnlineRateMap.containsKey(alarmEntry.getKey())) {
+                AlarmPO alarmPO = deviceFlowAndOnlineRateMap.get(alarmEntry.getKey()).get(0);
+                topMsg.setFlowFlag(alarmPO.getActualValue() > alarmPO.getFlowMeal() ? 0 : 1);
+                topMsg.setOnlineRateFlag(alarmPO.getOnlineRate() *100 < alarmStrategyVO.getOnlineValue() ? 0 : 1);
+                topMsg.setFlowValue(alarmPO.getFlowProportion());
+                topMsg.setOnlineRateValue(alarmPO.getOnlineRate() *100);
+            } else {
+                topMsg.setFlowFlag(-1);
+                topMsg.setOnlineRateFlag(-1);
+            }
+
+            if (comMap.containsKey(alarmEntry.getKey())) {
+                TopMsgPO comOutPO = comMap.get(alarmEntry.getKey()).get(0);
+                topMsg.setOfftimeFlag(comOutPO.getOfftimeValue() > alarmStrategyVO.getOffTimeValue() ? 0 :1);
+                topMsg.setComOutCount(comOutPO.getComOutCount());
+                topMsg.setComOutDescription(String.join(",",comOutPO.getComOutDesc()));
+            } else {
+                topMsg.setOfftimeFlag(0);
+                String updateTime = localDateTimeFormat(alarmEntry.getValue().get(0).getUpdateTime());
+                long time = DateUtil.endOfDay(DateUtil.parse(param.getEndTime())).getTime() -
+                        alarmEntry.getValue().get(0).getUpdateTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
+                topMsg.setOfftimeValue((int)(time/1000/60));
+                topMsg.setComOutCount(1);
+                topMsg.setComOutDescription(updateTime+"至"+DateUtil.endOfDay(DateUtil.parse(param.getEndTime())));
+            }
+            if (topMsg.getFlowFlag() != 1 || topMsg.getIntegrityFlag() != 1 || topMsg.getOfftimeFlag() != 0
+                    || topMsg.getAlarmCountFlag() != 1 || topMsg.getOnlineRateFlag() != 1) {
+                topMsgPOS.add(topMsg);
+            }
+
         }
-        return deviceAbnormaStatisticsMapper.insertTopMsg(alarmExceptions);
+
+        return deviceAbnormaStatisticsMapper.insertTopMsg(topMsgPOS);
+
     }
 
-
-    public List<TopMsgPO> getCommunicate(List<String> devs, String startTime, String endTime) {
+    public String localDateTimeFormat(LocalDateTime localDateTime) {
+        // 通过LocalDateTime.atZone方法，使用系统默认时区ZonId.systemDefault()获取Instant实例
+        ZonedDateTime zonedDateTime = localDateTime.atZone(ZoneId.systemDefault());
+        Instant instant = zonedDateTime.toInstant();
+        // 日期格式化
+        String format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(instant.toEpochMilli());
+        return format;
+    }
+    public List<TopMsgPO> getCommunicate(List<String> devs, String startTime, String endTime) throws ParseException {
+//        startTime ="2023-04-02";
+//        endTime = "2023-04-02";
         //组装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 ");
@@ -125,48 +178,36 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
         if (CollectionUtil.isEmpty(communicates)) {
             return comMsgs;
         }
+        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
         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;
+            Long mills = 0l;
+            if (sortedList.get(0).getType() == 1) {
+                long time = Date.from(sortedList.get(0).getUpdateTime()).getTime() - DateUtil.beginOfDay(DateUtil.parse(startTime)).getTime();
+                mills += time;
+                comOutDesc.add((DateUtil.beginOfDay(DateUtil.parse(startTime)))+"至"+formatter.format(Date.from(sortedList.get(0).getUpdateTime())));
+            }
+            for (int i = 0; i <sortedList.size() ; i++) {
+                if (sortedList.get(i).getType()==0 && i<sortedList.size()-1) {
+                    long time = Date.from(sortedList.get(i+1).getUpdateTime()).getTime() - Date.from(sortedList.get(i).getUpdateTime()).getTime();
+                    mills += time;
+                    String comOutTime =formatter.format(Date.from(sortedList.get(i).getUpdateTime()))+"至"+ formatter.format(Date.from(sortedList.get(i+1).getUpdateTime()));
+                    comOutDesc.add(comOutTime);
+                } else if (sortedList.get(i).getType()==0 && i==sortedList.size()-1) {
+                    long time = DateUtil.endOfDay(DateUtil.parse(endTime)).getTime() - Date.from(sortedList.get(i).getUpdateTime()).getTime();
+                    mills += time;
+                    String comOutTime = formatter.format(Date.from(sortedList.get(i).getUpdateTime())) +"至"+ DateUtil.endOfDay(DateUtil.parse(endTime));
+                    comOutDesc.add(comOutTime);
                 }
             }
-
+            topMsgPO.setOfftimeValue( (int)(mills / 1000 / 60));
             topMsgPO.setComOutDesc(comOutDesc);
             // 通信中断次数
-            topMsgPO.setComOutCount(sortedList.stream().filter(a -> a.getType() == 0).collect(Collectors.toList()).size());
+            topMsgPO.setComOutCount(comOutDesc.size());
             comMsgs.add(topMsgPO);
         }
         return comMsgs;

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

@@ -2,6 +2,8 @@ package com.njcn.prepare.harmonic.service.mysql.device;
 
 import com.njcn.prepare.harmonic.pojo.param.DeviceAbnormaStatisticsParam;
 
+import java.text.ParseException;
+
 public interface DeviceAbnormalStatisticsService {
-    boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param);
+    boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param) throws ParseException;
 }