调整数据中心算法

1.装置在线率算法-日表
2.终端异常通告算法-日表

pqs-device/pq-device/pq-device-api/src/main/java/com/njcn/device/pq/pojo/po/DeviceRunException.java

@@ -14,6 +14,7 @@ import java.io.Serializable;
  */
 @Data
 @Measurement(name = "pqs_top_msg")
+@Deprecated
 public class DeviceRunException implements Serializable {
 
     /**

pqs-device/pq-device/pq-device-api/src/main/java/com/njcn/device/pq/pojo/po/PqsTopMsg.java

@@ -0,0 +1,92 @@
+package com.njcn.device.pq.pojo.po;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.github.jeffreyning.mybatisplus.anno.MppMultiId;
+import com.njcn.db.bo.BaseEntity;
+import lombok.Data;
+
+import java.time.LocalDate;
+
+/**
+ * <p>
+ * 
+ * </p>
+ *
+ * @author xuyang
+ * @since 2023-11-30
+ */
+@Data
+@TableName("pqs_top_msg")
+public class PqsTopMsg {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 日期
+     */
+    @MppMultiId
+    private LocalDate timeId;
+
+    /**
+     * 装置索引（外键）
+     */
+    @MppMultiId
+    private String devId;
+
+    /**
+     * 中断次数
+     */
+    private Integer comoutCount;
+
+    /**
+     * 中断时间段描述
+     */
+    private String comoutDesc;
+
+    /**
+     * 数据完整性标识 0.不达标 1.达标
+     */
+    private Integer integrityFlag;
+
+    /**
+     * 数据完成性百分比  0.3555
+     */
+    private Float integrityValue;
+
+    /**
+     * 告警次数
+     */
+    private Integer alarmCount;
+
+    /**
+     * 告警描述
+     */
+    private String alarmDesc;
+
+    /**
+     * 流量超标标识 0.不达标 1.达标
+     */
+    private Integer flowFlag;
+
+    /**
+     * 流量使用百分比
+     */
+    private Float flowValue;
+
+    /**
+     * 在线率达标标识 0.不达标 1.达标
+     */
+    private Integer onlinerateFlag;
+
+    /**
+     * 在线率百分比
+     */
+    private Float onlinerateValue;
+
+    /**
+     * 状态 0.删除 1.正常
+     */
+    private Integer state;
+
+
+}

pqs-device/pq-device/pq-device-api/src/main/java/com/njcn/device/pq/pojo/po/RStatOnlinerateD.java

@@ -5,6 +5,8 @@ import com.github.jeffreyning.mybatisplus.anno.MppMultiId;
 import com.njcn.db.bo.BaseEntity;
 import java.io.Serializable;
 import java.time.LocalDateTime;
+
+import lombok.Data;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -16,8 +18,7 @@ import lombok.Setter;
  * @author hongawen
  * @since 2023-03-28
  */
-@Getter
+@Data
-@Setter
 @TableName("r_stat_onlinerate_d")
 public class RStatOnlinerateD {
 

pqs-device/pq-device/pq-device-api/src/main/java/com/njcn/device/pq/pojo/po/TopMsg.java

@@ -13,6 +13,7 @@ import java.time.Instant;
  */
 @Data
 @Measurement(name = "pqs_top_msg")
+@Deprecated
 public class TopMsg {
 
     @Column(name = "time")

pqs-harmonic/harmonic-boot/src/main/java/com/njcn/harmonic/service/impl/THDistortionServiceImpl.java

@@ -1,6 +1,7 @@
 package com.njcn.harmonic.service.impl;
 
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
@@ -176,6 +177,24 @@ public class THDistortionServiceImpl implements THDistortionService {
         });
     }
 
+    /**
+     * 计算供电公司畸变率
+     */
+    private void setPowerDistortion(List<THDistortionVO> list) {
+        List<THDistortionVO> result = new ArrayList<>();
+        list.forEach(item->{
+            List<THDistortionVO> children = item.getChildren();
+            children.forEach(item2->{
+                result.addAll(item2.getChildren());
+            });
+            if (CollectionUtil.isNotEmpty(result)){
+                item.setDistortion(roundHalfUp(result.stream().filter(child -> !Objects.equals(child.getDistortion(), 3.14159)).mapToDouble(THDistortionVO::getDistortion).average().orElse(3.14159)));
+            } else {
+                item.setDistortion(3.14159);
+            }
+        });
+    }
+
     /**
      * 组装树层级
      */
@@ -224,8 +243,8 @@ public class THDistortionServiceImpl implements THDistortionService {
             setChildesList(substationList, monitorList);
             setFatherDistortion(substationList);
             powerCompanyList = powerCompanyList.stream().peek(item -> item.setChildren(getChildCategoryList(item, substationList))).collect(Collectors.toList());
-            setFatherDistortion(powerCompanyList);
+//            setFatherDistortion(powerCompanyList);
-
+            setPowerDistortion(powerCompanyList);
         }
         return powerCompanyList;
     }

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/ExecutionCenter.java

@@ -11,10 +11,7 @@ import com.njcn.common.pojo.annotation.OperateInfo;
 import com.njcn.common.pojo.enums.common.LogEnum;
 import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.device.biz.commApi.CommTerminalGeneralClient;
-import com.njcn.device.biz.pojo.dto.DeptGetBusBarDTO;
+import com.njcn.device.biz.pojo.dto.*;
-import com.njcn.device.biz.pojo.dto.DeptGetChildrenDTO;
-import com.njcn.device.biz.pojo.dto.DeptGetChildrenMoreDTO;
-import com.njcn.device.biz.pojo.dto.DeptGetSubStationDTO;
 import com.njcn.device.biz.pojo.param.DeptGetLineParam;
 import com.njcn.prepare.harmonic.enums.PrepareResponseEnum;
 import com.njcn.prepare.harmonic.pojo.bo.BaseParam;
@@ -142,9 +139,48 @@ public class ExecutionCenter extends BaseController {
             liteflowResponse = flowExecutor.execute2Resp("measurement_point", calculatedParam);
             dealResponse(calculatedParam,liteflowResponse,methodDescribe);
         }
-
-
     }
+
+    @OperateInfo(info = LogEnum.BUSINESS_COMMON)
+    @ApiOperation("装置算法执行链")
+    @PostMapping("/deviceExecutor")
+    @Async("asyncExecutor")
+    public void deviceExecutor(@RequestBody BaseParam baseParam) {
+        String methodDescribe = getMethodDescribe("deviceExecutor");
+        //手动判断参数是否合法,
+        CalculatedParam calculatedParam = judgeExecuteParam(baseParam);
+        DeptGetLineParam deptGetLineParam = new DeptGetLineParam();
+        // 设备索引
+        if (CollectionUtils.isEmpty(calculatedParam.getIdList())) {
+            Dept data = deptFeignClient.getRootDept().getData();
+            deptGetLineParam.setDeptId(data.getId());
+            List<DeptGetDeviceDTO> list = commTerminalGeneralClient.deptGetDevice(deptGetLineParam).getData();
+            DeptGetDeviceDTO dto = list.stream().filter(po ->Objects.equals(po.getUnitId(),data.getId())).collect(Collectors.toList()).get(0);
+            List<LineDevGetDTO> devList = dto.getDeviceList();
+            calculatedParam.setIdList(devList.stream().map(LineDevGetDTO::getDevId).distinct().collect(Collectors.toList()));
+        }
+        LiteflowResponse liteflowResponse;
+        if (baseParam.isRepair()) {
+            //补招时，起始日期、截止日期必填
+            DateTime startDate = DateUtil.parse(baseParam.getBeginTime(), DatePattern.NORM_DATE_FORMAT);
+            DateTime endDate = DateUtil.parse(baseParam.getEndTime(), DatePattern.NORM_DATE_FORMAT);
+            long betweenDay = DateUtil.betweenDay(startDate, endDate, true);
+            //递增日期执行算法链
+            for (int i = 0; i < betweenDay; i++) {
+                if (i != 0) {
+                    startDate = DateUtil.offsetDay(startDate, 1);
+                }
+                calculatedParam.setDataDate(DateUtil.format(startDate, DatePattern.NORM_DATE_PATTERN));
+                liteflowResponse = flowExecutor.execute2Resp("device", calculatedParam);
+                dealResponse(calculatedParam,liteflowResponse,methodDescribe);
+            }
+        } else {
+            //非补招
+            liteflowResponse = flowExecutor.execute2Resp("device", calculatedParam);
+            dealResponse(calculatedParam,liteflowResponse,methodDescribe);
+        }
+    }
+
     @OperateInfo(info = LogEnum.BUSINESS_COMMON)
     @ApiOperation("单位监测点算法执行链")
     @PostMapping("/orgPointExecutor")

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/executor/DeviceExecutor.java

@@ -0,0 +1,55 @@
+package com.njcn.prepare.executor;
+
+import com.njcn.prepare.harmonic.service.mysql.device.DeviceAbnormalStatisticsService;
+import com.njcn.prepare.harmonic.service.mysql.line.OnlineRateService;
+import com.yomahub.liteflow.annotation.LiteflowComponent;
+import com.yomahub.liteflow.annotation.LiteflowMethod;
+import com.yomahub.liteflow.core.NodeComponent;
+import com.yomahub.liteflow.enums.LiteFlowMethodEnum;
+import com.yomahub.liteflow.enums.NodeTypeEnum;
+import lombok.RequiredArgsConstructor;
+
+
+/**
+ * @author hongawen
+ * @version 1.0.0
+ * @date 2023年11月06日 15:59
+ */
+@LiteflowComponent
+@RequiredArgsConstructor
+public class DeviceExecutor extends BaseExecutor {
+
+    private final OnlineRateService onlineRateService;
+
+    private final DeviceAbnormalStatisticsService deviceAbnormaStatisticsService;
+
+    /**
+     * 算法名: 暂无-----终端在线率_日表(r_stat_onlinerate_d)
+     *
+     * @author xuyang
+     * @date 2023年11月30日 10:25
+     */
+    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "deviceOnlineRate", nodeType = NodeTypeEnum.COMMON)
+    public boolean deviceOnlineRateAccess(NodeComponent bindCmp) {
+        return isAccess(bindCmp);
+    }
+    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "deviceOnlineRate", nodeType = NodeTypeEnum.COMMON)
+    public void deviceOnlineRateProcess(NodeComponent bindCmp) {
+        onlineRateService.deviceOnlineRate(bindCmp.getRequestData());
+    }
+
+    /**
+     * 算法名: 暂无-----终端异常通告功能-日表(pqs_top_msg)
+     *
+     * @author xuyang
+     * @date 2023年11月30日 13:25
+     */
+    @LiteflowMethod(value = LiteFlowMethodEnum.IS_ACCESS, nodeId = "pqsTopMsg", nodeType = NodeTypeEnum.COMMON)
+    public boolean pqsTopMsgAccess(NodeComponent bindCmp) {
+        return isAccess(bindCmp);
+    }
+    @LiteflowMethod(value = LiteFlowMethodEnum.PROCESS, nodeId = "pqsTopMsg", nodeType = NodeTypeEnum.COMMON)
+    public void pqsTopMsgProcess(NodeComponent bindCmp) {
+        deviceAbnormaStatisticsService.deviceAbnormaStatisticsD(bindCmp.getRequestData());
+    }
+}

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

@@ -1,16 +1,18 @@
 package com.njcn.prepare.harmonic.mapper.mysql.device;
 
 import cn.hutool.core.date.DateTime;
+import com.github.jeffreyning.mybatisplus.base.MppBaseMapper;
+import com.njcn.device.pq.pojo.po.PqsTopMsg;
+import com.njcn.prepare.harmonic.bo.TopMsgPO;
 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.bo.TopMsgPO;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
 @Mapper
-public interface DeviceAbnormaStatisticsMapper {
+public interface DeviceAbnormaStatisticsMapper extends MppBaseMapper<PqsTopMsg>{
 
     /**
      * 获取所有有等级监测点以及对应设备
@@ -18,6 +20,12 @@ public interface DeviceAbnormaStatisticsMapper {
      */
     List<AlarmPO> getLines();
 
+    /**
+     * 获取所有有等级监测点以及对应设备
+     * @return List<AlarmPO>
+     */
+    List<AlarmPO> getLinesById(@Param("list") List<String> list);
+
     List<AlarmPO> getFlowByDevice(@Param("list") List<String> list, @Param("time") String startTime);
 
     List<AlarmPO> getIntegrityByLineIds(@Param("list") List<String> list, @Param("startTime") DateTime startTime, @Param("endTime") DateTime endTime);

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

@@ -55,6 +55,29 @@
                 pld.Line_Grade is NOT NULL
     </select>
 
+    <select id="getLinesById" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
+        SELECT  line.Id AS id,
+                device.Id AS deviceId,
+                sdd.Algo_Describe AS level,
+                DATE_FORMAT(pd.Update_Time, "%Y-%m-%d %H:%i:%s") 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 sys_dict_data sdd ON sdd.Id = pld.Line_Grade
+        WHERE
+                pd.Dev_Model=1
+        AND
+                pd.Run_Flag=0
+        AND
+                pld.Line_Grade is NOT NULL
+        AND pd.id in
+        <foreach collection="list" index="index" item="item" open="(" separator="," close=")">
+            #{item}
+        </foreach>
+    </select>
+
     <select id="getIntegrityByLineIds" resultType="com.njcn.prepare.harmonic.pojo.mysql.po.line.AlarmPO">
        SELECT
             device.id AS deviceId,

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

@@ -1,11 +1,18 @@
 package com.njcn.prepare.harmonic.service.mysql.Impl.device;
 
 import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.date.DateUtil;
+import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
+import com.njcn.common.pojo.enums.common.DataStateEnum;
+import com.njcn.device.pq.pojo.po.PqsTopMsg;
+import com.njcn.device.pq.pojo.po.RStatOnlinerateD;
 import com.njcn.influx.deprecated.InfluxDBPublicParam;
 import com.njcn.influx.pojo.po.PqsCommunicate;
 import com.njcn.influx.utils.InfluxDbUtils;
+import com.njcn.prepare.harmonic.mapper.mysql.day.RStatOnlineRateDMapper;
 import com.njcn.prepare.harmonic.mapper.mysql.device.DeviceAbnormaStatisticsMapper;
+import com.njcn.prepare.harmonic.pojo.bo.CalculatedParam;
 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.bo.TopMsgPO;
@@ -18,6 +25,7 @@ import org.influxdb.impl.InfluxDBResultMapper;
 import org.springframework.stereotype.Service;
 
 import java.text.SimpleDateFormat;
+import java.time.LocalDate;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -27,7 +35,7 @@ import java.util.stream.Collectors;
 @Service
 @RequiredArgsConstructor
 @Slf4j
-public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatisticsService {
+public class DeviceAbnormaStatisticsServiceImpl extends MppServiceImpl<DeviceAbnormaStatisticsMapper, PqsTopMsg> implements DeviceAbnormalStatisticsService {
 
     private final DeviceAbnormaStatisticsMapper deviceAbnormaStatisticsMapper;
 
@@ -145,6 +153,114 @@ public class DeviceAbnormaStatisticsServiceImpl implements DeviceAbnormalStatist
         return topMsgPOS.size()>0 ? deviceAbnormaStatisticsMapper.insertTopMsg(topMsgPOS) : true;
     }
 
+    @Override
+    public void deviceAbnormaStatisticsD(CalculatedParam calculatedParam) {
+        Map<String, List<TopMsgPO>> comMap = new HashMap<>();
+        Map<String, List<AlarmPO>> deviceFlowMap  = new HashMap<>();
+        // 获取所有装置以及下属监测点
+        List<AlarmPO> lines = deviceAbnormaStatisticsMapper.getLinesById(calculatedParam.getIdList());
+        if (CollectionUtil.isNotEmpty(lines)){
+            // 根据装置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<AlarmStrategyVO> alarmStrategyVOS = deviceAbnormaStatisticsMapper.selectAlarmStrategy();
+            Map<Integer, List<AlarmStrategyVO>> levMap = alarmStrategyVOS.stream().collect(Collectors.groupingBy(AlarmStrategyVO::getAlgoDesc));
+            // 获取监测点数据完整性
+            List<AlarmPO> lineIntegritys = deviceAbnormaStatisticsMapper.getIntegrityByLineIds(
+                    monotorIds,
+                    DateUtil.beginOfDay(DateUtil.parse(calculatedParam.getDataDate())),
+                    DateUtil.endOfDay(DateUtil.parse(calculatedParam.getDataDate())));
+            Map<String, List<AlarmPO>> lineIntegrityMap = lineIntegritys.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+            // 告警信息统计
+            List<AlarmPO> lineAlarms = deviceAbnormaStatisticsMapper.getAlarmByLines(
+                    monotorIds,
+                    devIds,
+                    DateUtil.beginOfDay(DateUtil.parse(calculatedParam.getDataDate())),
+                    DateUtil.endOfDay(DateUtil.parse(calculatedParam.getDataDate())));
+            Map<String, List<AlarmPO>> lineAlarmsMap = lineAlarms.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+            // 获取终端在线率
+            List<AlarmPO> deviceOnlineRate = deviceAbnormaStatisticsMapper.getOnlineRateByDevice(
+                    devIds,
+                    DateUtil.beginOfDay(DateUtil.parse(calculatedParam.getDataDate())),
+                    DateUtil.endOfDay(DateUtil.parse(calculatedParam.getDataDate())));
+            Map<String, List<AlarmPO>> deviceOnlineMap = deviceOnlineRate.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+            // 获取设备流量
+            List<AlarmPO> deviceFlow = deviceAbnormaStatisticsMapper.getFlowByDevice(devIds,calculatedParam.getDataDate().substring(0,7));
+            if (CollectionUtil.isNotEmpty(deviceFlow)) {
+                deviceFlowMap = deviceFlow.stream().collect(Collectors.groupingBy(AlarmPO::getDeviceId));
+            }
+            // 获取通信中断信息
+            List<TopMsgPO> comMsgs = getPqsCommunicate(devIds, calculatedParam.getDataDate());
+            if (CollectionUtil.isNotEmpty(comMsgs)) {
+                comMap = comMsgs.stream().collect(Collectors.groupingBy(TopMsgPO::getDevId));
+            }
+            // 整合装置信息
+            List<PqsTopMsg> topMsgPOS = new ArrayList<>();
+            for (Map.Entry<String, List<AlarmPO>> devEntry : devMap.entrySet()) {
+                PqsTopMsg topMsg = new PqsTopMsg();
+                topMsg.setDevId(devEntry.getKey());
+                topMsg.setTimeId(LocalDate.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_FORMATTER));
+                topMsg.setState(DataStateEnum.ENABLE.getCode());
+                // 等级对应的告警策略
+                int deviceLevel = devEntry.getValue().stream().mapToInt(AlarmPO::getLevel).min().orElse(-1);
+                AlarmStrategyVO alarmStrategyVO = levMap.get(deviceLevel).get(0);
+                // 数据完整性
+                if (lineIntegrityMap.containsKey(devEntry.getKey())) {
+                    topMsg.setIntegrityFlag(lineIntegrityMap.get(devEntry.getKey()).get(0).getIntegrity() < alarmStrategyVO.getIntegrityValue() ? 0 : 1);
+                    topMsg.setIntegrityValue(lineIntegrityMap.get(devEntry.getKey()).get(0).getIntegrity());
+                } else {
+                    topMsg.setIntegrityFlag(-1);
+                }
+                // 装置告警
+                if (lineAlarmsMap.containsKey(devEntry.getKey())) {
+                    List<AlarmPO> lineAlarmPOS = lineAlarmsMap.get(devEntry.getKey());
+                    int alarmCount = lineAlarmPOS.stream().mapToInt(AlarmPO::getAlarmCount).sum();
+                    String alarmDesc = String.join(",", lineAlarmPOS.stream().map(AlarmPO::getAlarmDesc).collect(Collectors.toList()));
+                    topMsg.setAlarmCount(alarmCount);
+                    topMsg.setAlarmDesc(alarmDesc);
+                } else {
+                    topMsg.setAlarmCount(0);
+                }
+                // 在线率
+                if (deviceOnlineMap.containsKey(devEntry.getKey())) {
+                    AlarmPO alarmPO = deviceOnlineMap.get(devEntry.getKey()).get(0);
+                    topMsg.setOnlinerateFlag(alarmPO.getOnlineRate() * 100 < alarmStrategyVO.getOnlineValue() ? 0 : 1);
+                    topMsg.setOnlinerateValue(alarmPO.getOnlineRate() *100);
+                } else {
+                    topMsg.setOnlinerateFlag(-1);
+                }
+                // 流量
+                if (deviceFlowMap.containsKey(devEntry.getKey()) && Objects.nonNull(deviceFlowMap.get(devEntry.getKey()).get(0).getFlowProportion())) {
+                    AlarmPO alarmPO = deviceFlowMap.get(devEntry.getKey()).get(0);
+                    topMsg.setFlowFlag(alarmPO.getActualValue() > alarmPO.getFlowMeal() ? 0 : 1);
+                    topMsg.setFlowValue(alarmPO.getFlowProportion());
+                } else {
+                    topMsg.setFlowFlag(-1);
+                }
+
+                if (comMap.containsKey(devEntry.getKey())) {
+                    TopMsgPO comOutPO = comMap.get(devEntry.getKey()).get(0);
+                    List<PqsCommunicate> abnormals = comOutPO.getCommunicates().stream().filter(t -> t.getDuration() > alarmStrategyVO.getOffTimeValue()).collect(Collectors.toList());
+                    topMsg.setComoutCount(abnormals.size());
+                    List<String> abnormalDesc = abnormals.stream().map(PqsCommunicate::getDesc).collect(Collectors.toList());
+                    topMsg.setComoutDesc(String.join(",",abnormalDesc));
+                } else {
+                    topMsg.setComoutCount(0);
+                }
+                if (topMsg.getFlowFlag() ==0 || topMsg.getIntegrityFlag() ==0 || topMsg.getOnlinerateFlag() ==0) {
+                    topMsgPOS.add(topMsg);
+                }
+            }
+            if (CollectionUtil.isNotEmpty(topMsgPOS)){
+                this.saveOrUpdateBatchByMultiId(topMsgPOS);
+            }
+        }
+    }
+
     public List<TopMsgPO> getPqsCommunicate(List<String> devs, String time) {
         // 组装sql语句
         StringBuilder stringBuilder = new StringBuilder();

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/service/mysql/Impl/line/OnlineRateServiceImpl.java

@@ -1,8 +1,11 @@
 package com.njcn.prepare.harmonic.service.mysql.Impl.line;
 
 import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.date.DateUnit;
 import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.date.LocalDateTimeUtil;
 import com.github.jeffreyning.mybatisplus.service.MppServiceImpl;
 import com.njcn.device.pq.api.LineFeignClient;
 import com.njcn.device.pq.pojo.po.RStatOnlinerateD;
@@ -11,6 +14,7 @@ import com.njcn.influx.deprecated.InfluxDBPublicParam;
 import com.njcn.influx.pojo.po.PqsCommunicate;
 import com.njcn.influx.utils.InfluxDbUtils;
 import com.njcn.prepare.harmonic.mapper.mysql.day.RStatOnlineRateDMapper;
+import com.njcn.prepare.harmonic.pojo.bo.CalculatedParam;
 import com.njcn.prepare.harmonic.pojo.dto.OnLineRateDTO;
 import com.njcn.prepare.harmonic.pojo.param.LineParam;
 
@@ -110,6 +114,36 @@ public class OnlineRateServiceImpl extends MppServiceImpl<RStatOnlineRateDMapper
         this.saveOrUpdateBatchByMultiId(list,50);
     }
 
+    @Override
+    public void deviceOnlineRate(CalculatedParam calculatedParam) {
+        List<String> devList = calculatedParam.getIdList();
+        String data = LocalDateTimeUtil.format(
+                LocalDateTimeUtil.beginOfDay(LocalDateTimeUtil.parse(calculatedParam.getDataDate(), DatePattern.NORM_DATE_PATTERN)),
+                DatePattern.NORM_DATETIME_PATTERN
+        );
+        LocalDateTime dateTime = LocalDateTime.parse(data,DatePattern.NORM_DATETIME_FORMATTER);
+        List<PqsCommunicate> outCommunicateData = new ArrayList<>();
+        for (String dev : devList){
+            List<PqsCommunicate> communicateData = getCommunicateData(dev);
+            outCommunicateData.addAll(communicateData);
+        }
+        List<RStatOnlinerateD> list = new ArrayList<>();
+        Date dateOut = DateUtil.parse(calculatedParam.getDataDate());
+        for (PqsCommunicate pqsCommunicate : outCommunicateData){
+            RStatOnlinerateD onlineRateDpo = new RStatOnlinerateD();
+            Date newDate = Date.from(pqsCommunicate.getTime());
+            OnLineRateDTO onLineRate = onLineMinute(newDate,dateOut,pqsCommunicate.getType(),pqsCommunicate.getDevId(),calculatedParam.getDataDate());
+            onlineRateDpo.setTimeId(dateTime);
+            onlineRateDpo.setDevIndex(pqsCommunicate.getDevId());
+            onlineRateDpo.setOnlineMin(onLineRate.getOnLineMinute());
+            onlineRateDpo.setOfflineMin(onLineRate.getOffLineMinute());
+            list.add(onlineRateDpo);
+        }
+        if (CollectionUtil.isNotEmpty(list)){
+            this.saveOrUpdateBatchByMultiId(list,50);
+        }
+    }
+
     /**
      * 获取pqs_communicate数据
      */

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

@@ -1,5 +1,6 @@
 package com.njcn.prepare.harmonic.service.mysql.device;
 
+import com.njcn.prepare.harmonic.pojo.bo.CalculatedParam;
 import com.njcn.prepare.harmonic.pojo.param.DeviceAbnormaStatisticsParam;
 
 import java.text.ParseException;
@@ -7,4 +8,12 @@ import java.text.ParseException;
 public interface DeviceAbnormalStatisticsService {
 
     boolean dailyDeviceAbnormaStatistics(DeviceAbnormaStatisticsParam param);
+
+    /***
+     * 终端异常通告功能-日表
+     * @author xuyang
+     * @date 2023/11/30 13:27
+     * @param calculatedParam 查询条件
+     */
+    void deviceAbnormaStatisticsD(CalculatedParam calculatedParam);
 }

pqs-prepare/prepare-boot/src/main/java/com/njcn/prepare/harmonic/service/mysql/line/OnlineRateService.java

@@ -1,5 +1,6 @@
 package com.njcn.prepare.harmonic.service.mysql.line;
 
+import com.njcn.prepare.harmonic.pojo.bo.CalculatedParam;
 import com.njcn.prepare.harmonic.pojo.param.LineParam;
 
 /**
@@ -12,4 +13,12 @@ public interface OnlineRateService {
     void getOnlineRateData(LineParam lineParam);
 
     void onlineRateData(LineParam lineParam,String time);
+
+    /***
+     * 终端在线率_日表
+     * @author xuyang
+     * @date 2023/11/30 10:27
+     * @param calculatedParam 查询条件
+     */
+    void deviceOnlineRate(CalculatedParam calculatedParam);
 }