diff --git a/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarm.java b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarm.java index 0f4d021..e08d3e1 100644 --- a/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarm.java +++ b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarm.java @@ -43,7 +43,7 @@ public class CsAlarm implements Serializable { private String interruptEvent; /** - * 中断事件描述 + * 告警事件描述 */ private String alarmEvent; diff --git a/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarmData.java b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarmData.java new file mode 100644 index 0000000..867fb20 --- /dev/null +++ b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/po/CsAlarmData.java @@ -0,0 +1,61 @@ +package com.njcn.csharmonic.pojo.po; + +import lombok.Data; + +import java.io.Serializable; +import java.util.List; + +/** + * @author xy + */ +@Data +public class CsAlarmData implements Serializable { + + //设备id + private String deviceId; + + // 在线率异常记录 + private OnlineRateAlarm onlineRate; + + // 完整性异常记录 + private IntegrityAlarm integrity; + + /** + * 在线率异常信息 + */ + @Data + public static class OnlineRateAlarm implements Serializable { + // 实际在线率值 + private Double value; + // 阈值 + private Double threshold; + // 是否异常 + private Boolean isAbnormal; + } + + /** + * 完整性异常信息 + */ + @Data + public static class IntegrityAlarm implements Serializable { + // 阈值 + private Double threshold; + // 各监测点的完整性数据 + private List monitorPoints; + + /** + * 监测点完整性信息 + */ + @Data + public static class MonitorPointIntegrity implements Serializable { + // 监测点ID + private String monitorPointId; + // 监测点名称 + private String monitorName; + // 该监测点的完整性值 + private Double value; + // 是否异常 + private Boolean isAbnormal; + } + } +} diff --git a/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/vo/AlarmVO.java b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/vo/AlarmVO.java index 98b5118..65ce37c 100644 --- a/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/vo/AlarmVO.java +++ b/cs-harmonic/cs-harmonic-api/src/main/java/com/njcn/csharmonic/pojo/vo/AlarmVO.java @@ -1,11 +1,11 @@ package com.njcn.csharmonic.pojo.vo; +import com.njcn.csharmonic.pojo.po.CsAlarmData; import io.swagger.annotations.ApiModelProperty; import lombok.Data; import java.io.Serializable; -import java.time.LocalDate; import java.util.ArrayList; import java.util.List; @@ -24,8 +24,11 @@ public class AlarmVO implements Serializable { @ApiModelProperty(value = "告警设备台数") private Integer warnNums; - @ApiModelProperty(value = "通讯中断告警次数") - private Integer interruptCounts; + @ApiModelProperty(value = "在线率是否告警") + private Boolean onlineRateIsWarn; + + @ApiModelProperty(value = "完整性是否告警") + private Boolean integrityIsWarn; @ApiModelProperty(value = "终端告警次数") private Integer warnCounts; @@ -57,11 +60,8 @@ public class AlarmVO implements Serializable { @ApiModelProperty(value = "告警详情") private List warnDetails = new ArrayList<>(); - @ApiModelProperty(value = "中断次数") - private Integer interruptCounts = 0; - - @ApiModelProperty(value = "中断详情") - private List interruptDetails = new ArrayList<>(); + @ApiModelProperty(value = "在线率、完整性详情") + private CsAlarmData dataDetails; @Data public static class WarnDetail implements Serializable { diff --git a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsAlarmServiceImpl.java b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsAlarmServiceImpl.java index 82d2210..4706931 100644 --- a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsAlarmServiceImpl.java +++ b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/CsAlarmServiceImpl.java @@ -19,6 +19,7 @@ import com.njcn.csdevice.pojo.po.CsLedger; import com.njcn.csdevice.pojo.vo.CsLedgerVO; import com.njcn.csharmonic.mapper.CsAlarmMapper; import com.njcn.csharmonic.pojo.po.CsAlarm; +import com.njcn.csharmonic.pojo.po.CsAlarmData; import com.njcn.csharmonic.pojo.po.CsEventPO; import com.njcn.csharmonic.pojo.po.CsEventUserPO; import com.njcn.csharmonic.pojo.vo.AlarmVO; @@ -118,13 +119,12 @@ public class CsAlarmServiceImpl extends ServiceImpl impl alarmVO.setDevIds(matchedDevIds); alarmVO.setIsRead(userEvents.stream().filter(item1->item1.getEventId().equals(alarm.getId())).findFirst().map(CsEventUserPO::getStatus).orElse(1)); - int interruptCounts = 0; int warnCounts = 0; ObjectMapper objectMapper = new ObjectMapper(); - List> resultList; + List resultList; List> resultList2; try { - resultList = objectMapper.readValue(alarm.getInterruptEvent(), new TypeReference>>() {}); + resultList = objectMapper.readValue(alarm.getInterruptEvent(), new TypeReference>() {}); resultList2 = objectMapper.readValue(alarm.getAlarmEvent(), new TypeReference>>() {}); } catch (JsonProcessingException e) { throw new RuntimeException(e); @@ -133,12 +133,17 @@ public class CsAlarmServiceImpl extends ServiceImpl impl for (String matchedDevId : matchedDevIds) { for (int j = 0; j < devIds.length; j++) { if (Objects.equals(matchedDevId, devIds[j])) { - interruptCounts = interruptCounts + (Objects.isNull(resultList.get(j)) ? 0 : resultList.get(j).get(0).split("，").length); warnCounts = warnCounts + (Objects.isNull(resultList2.get(j)) ? 0 : resultList2.get(j).size()); } } } - alarmVO.setInterruptCounts(interruptCounts); + //根据resultList 来查询在线率和完整是否有true的，如果有就设置alarmVO的属性 + resultList.forEach(item->{ + alarmVO.setOnlineRateIsWarn(item.getOnlineRate().getIsAbnormal()); + boolean hasAbnormal = item.getIntegrity().getMonitorPoints().stream() + .anyMatch(CsAlarmData.IntegrityAlarm.MonitorPointIntegrity::getIsAbnormal); + alarmVO.setIntegrityIsWarn(hasAbnormal); + }); alarmVO.setWarnCounts(warnCounts); return alarmVO; }) @@ -147,35 +152,6 @@ public class CsAlarmServiceImpl extends ServiceImpl impl .collect(Collectors.toList()); page1.setRecords(result); } - - -// if (CollectionUtil.isNotEmpty(list)) { -// //获取用户推送事件 -// List userEvents = csEventUserPOService.queryEventListByUserId(RequestUtil.getUserIndex(), list.stream().map(CsAlarm::getId).collect(Collectors.toList())); -// -// List finalDevList = devList; -// List result = list.stream() -// .map(alarm -> { -// AlarmVO alarmVO = new AlarmVO(); -// alarmVO.setEventId(alarm.getId()); -// alarmVO.setDate(alarm.getTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))); -// -// String devListStr = alarm.getDevList(); -// String[] devIds = devListStr.split(","); -// List matchedDevIds = Arrays.stream(devIds) -// .filter(devId -> StrUtil.isNotBlank(devId.trim())) -// .filter(finalDevList::contains) -// .collect(Collectors.toList()); -// alarmVO.setWarnNums(matchedDevIds.size()); -// alarmVO.setDevIds(matchedDevIds); -// alarmVO.setIsRead(userEvents.stream().filter(item1->item1.getEventId().equals(alarm.getId())).findFirst().map(CsEventUserPO::getStatus).orElse(1)); -// return alarmVO; -// }) -// .filter(alarmVO -> alarmVO.getWarnNums() > 0) -// .sorted((a1, a2) -> a2.getDate().compareTo(a1.getDate())) -// .collect(Collectors.toList()); -// page1.setRecords(result); -// } } } return page1; @@ -212,20 +188,16 @@ public class CsAlarmServiceImpl extends ServiceImpl impl if (ObjectUtil.isNotNull(alarm.getInterruptEvent())) { String interruptEvent = alarm.getInterruptEvent(); - List> resultList = new ArrayList<>(); + List resultList = new ArrayList<>(); if (StrUtil.isNotBlank(interruptEvent)) { try { ObjectMapper objectMapper = new ObjectMapper(); - resultList = objectMapper.readValue(interruptEvent, new TypeReference>>() {}); + resultList = objectMapper.readValue(interruptEvent, new TypeReference>() {}); + alarmDetail.setDataDetails(resultList.get(i)); } catch (JsonProcessingException e) { e.printStackTrace(); } } - List interruptDetails = resultList.get(i); - if (CollectionUtil.isNotEmpty(interruptDetails)) { - alarmDetail.setInterruptCounts(interruptDetails.get(0).split("，").length); - alarmDetail.setInterruptDetails(interruptDetails); - } } if (ObjectUtil.isNotNull(alarm.getAlarmEvent())) { String alarmEvent = alarm.getAlarmEvent(); diff --git a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StatisticsDataDataServiceImpl.java b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StatisticsDataDataServiceImpl.java index 7d4fb96..837a5d8 100644 --- a/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StatisticsDataDataServiceImpl.java +++ b/cs-harmonic/cs-harmonic-boot/src/main/java/com/njcn/csharmonic/service/impl/StatisticsDataDataServiceImpl.java @@ -107,8 +107,8 @@ public class StatisticsDataDataServiceImpl implements IStatisticsDataDataService List csLineList = csLineFeignClient.getLinesByDevList(devList).getData(); if (CollectionUtil.isNotEmpty(csLineList)) { Map> devMap = csLineList.stream().collect(Collectors.groupingBy(CsLinePO::getDeviceId)); - List lineList = csLineList.stream().map(CsLinePO::getLineId).collect(Collectors.toList()); - List deviceList = csLineList.stream().map(CsLinePO::getDeviceId).collect(Collectors.toList()); + List lineList = csLineList.stream().map(CsLinePO::getLineId).distinct().collect(Collectors.toList()); + List deviceList = csLineList.stream().map(CsLinePO::getDeviceId).distinct().collect(Collectors.toList()); //获取监测点数据完整性 IcdBzParam param1 = new IcdBzParam(); param1.setLineList(lineList); diff --git a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/controller/task/DataTaskController.java b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/controller/task/DataTaskController.java index bfe17c6..8b767a9 100644 --- a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/controller/task/DataTaskController.java +++ b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/controller/task/DataTaskController.java @@ -46,6 +46,15 @@ public class DataTaskController extends BaseController { return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); } + @OperateInfo(info = LogEnum.BUSINESS_COMMON) + @PostMapping("/runDataAlarm") + @ApiOperation("运行告警(完整性、在线率)") + public HttpResult channelRunDataAlarm(@RequestParam("time") String time){ + String methodDescribe = getMethodDescribe("channelRunDataAlarm"); + dataTaskService.channelRunDataAlarm(time); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, null, methodDescribe); + } + @OperateInfo(info = LogEnum.BUSINESS_COMMON) @PostMapping("/pushAppMsg") @ApiOperation("app消息推送") diff --git a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/IDataTaskService.java b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/IDataTaskService.java index a9e60cf..0d20bbf 100644 --- a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/IDataTaskService.java +++ b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/IDataTaskService.java @@ -23,6 +23,18 @@ public interface IDataTaskService { */ void channelRunAlarm(String time); + /** + 运行告警算法 + 1、每天生成一条记录，查询所有的设备，获取设备告警数据、设备的在线率和完整性，在线率根据设备来判断，完整性根据监测点来判断(多个监测点取最小值)； + 告警判断逻辑：(这个数据是根据一个表配置的，先查询阈值配置表) + a.完整性 < 90(可变动) + b.完整性 > 90(可变动) && 在线率 < 60(可变动) + 2、根据设备获取哪些用户有这个设备的权限，然后将用户id收集起来。(主用户、子用户、管理员id集合)； + 3、将cs_alarm生成的记录，存储到cs_event_user表中，标记为未读 + * @param time + */ + void channelRunDataAlarm(String time); + /** * 推送App消息 * 只推送稳态事件、运行告警数据 diff --git a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/DataTaskServiceImpl.java b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/DataTaskServiceImpl.java index a5942d8..355768f 100644 --- a/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/DataTaskServiceImpl.java +++ b/cs-system/cs-system-boot/src/main/java/com/njcn/cssystem/service/impl/DataTaskServiceImpl.java @@ -8,18 +8,23 @@ import cn.hutool.core.util.ObjectUtil; import cn.hutool.core.util.ReflectUtil; import cn.hutool.core.util.StrUtil; import com.alibaba.nacos.shaded.com.google.gson.Gson; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import com.njcn.access.pojo.dto.NoticeUserDto; import com.njcn.access.utils.SendMessageUtil; -import com.njcn.csdevice.api.CsCommTerminalFeignClient; -import com.njcn.csdevice.api.CsDeviceUserFeignClient; -import com.njcn.csdevice.api.CsLineFeignClient; +import com.njcn.csdevice.api.*; +import com.njcn.csdevice.param.IcdBzParam; import com.njcn.csdevice.pojo.dto.CsLineDTO; import com.njcn.csdevice.pojo.param.UserDevParam; import com.njcn.csdevice.pojo.po.CsDeviceUserPO; +import com.njcn.csdevice.pojo.po.RStatIntegrityD; +import com.njcn.csdevice.pojo.po.RStatOnlineRateD; import com.njcn.csharmonic.api.*; import com.njcn.csharmonic.param.CsEventUserQueryParam; import com.njcn.csharmonic.pojo.param.RStatLimitQueryParam; import com.njcn.csharmonic.pojo.po.*; +import com.njcn.cssystem.pojo.po.CsAlarmSet; +import com.njcn.cssystem.service.ICsAlarmSetService; import com.njcn.cssystem.service.IDataTaskService; import com.njcn.influx.imapper.PqsCommunicateMapper; import com.njcn.influx.pojo.po.PqsCommunicate; @@ -77,6 +82,9 @@ public class DataTaskServiceImpl implements IDataTaskService { private final CsCommTerminalFeignClient csCommTerminalFeignClient; private final CsHarmonicPlanFeignClient csHarmonicPlanFeignClient; private final CsHarmonicPlanLineFeignClient csHarmonicPlanLineFeignClient; + private final ICsAlarmSetService csAlarmSetService; + private final OnlineRateFeignClient onlineRateFeignClient; + private final IntegrityFeignClient integrityClient; @Override @Transactional(rollbackFor = Exception.class) @@ -302,6 +310,135 @@ public class DataTaskServiceImpl implements IDataTaskService { } } + @Override + @Transactional(rollbackFor = Exception.class) + public void channelRunDataAlarm(String time) { + CsAlarm alarm = new CsAlarm(); + List list2 = new ArrayList<>(); + //获取所有监测点 + List allLine = csLineFeignClient.getAllLineDetail().getData(); + if (CollUtil.isNotEmpty(allLine)) { + //过滤物联、在线监测的点 + DictTreeVO portable = dictTreeFeignClient.queryByCode(DicDataEnum.PORTABLE.getCode()).getData(); + if (ObjectUtil.isNotNull(portable)) { + allLine = allLine.stream().filter(item -> !item.getDeviceType().equals(portable.getId())).collect(Collectors.toList()); + } + //设备id集合 + List devList = allLine.stream().map(CsLineDTO::getDeviceId).distinct().collect(Collectors.toList()); + + //获取设备和用户关系 + UserDevParam param = new UserDevParam(); + param.setList(devList); + List userList = csDeviceUserFeignClient.getList(param).getData(); + //根据设备id分组 组是subUserId的集合 + Map> userDevMap = userList.stream().collect(Collectors.groupingBy( + CsDeviceUserPO::getDeviceId, + Collectors.mapping( + CsDeviceUserPO::getSubUserId, + Collectors.toList() + ) + )); + //获取管理员信息 + List adminList = appUserFeignClient.getAdminInfo().getData(); + //获取用户id集合 + List userIdList = adminList.stream().map(User::getId).collect(Collectors.toList()); + + //获取时间 + String date; + if (StringUtils.isNotBlank(time)) { + date = time; + } else { + date = DateUtil.yesterday().toString(NORM_DATE_PATTERN); + } + //获取在线率、完整性阈值 + CsAlarmSet csAlarmSet = csAlarmSetService.getEnabledConfig(); + //获取当日的在线率数据 + IcdBzParam param1 = new IcdBzParam(); + param1.setStartTime(date); + param1.setEndTime(date); + param1.setLineList(devList); + List onlineRateList = onlineRateFeignClient.list(param1).getData(); + //获取当日的完整性数据 + List lineList = allLine.stream().map(CsLineDTO::getLineId).distinct().collect(Collectors.toList()); + param1.setLineList(lineList); + List integrityList = integrityClient.list(param1).getData(); + + // 构建告警数据 + List alarmDataList = buildAlarmData(devList, onlineRateList, integrityList, allLine, csAlarmSet); + Map alarmDataMap = alarmDataList.stream().collect(Collectors.toMap(CsAlarmData::getDeviceId, item -> item)); + + //获取告警数据 + CsEventUserQueryParam queryParam = new CsEventUserQueryParam(); + queryParam.setTarget(devList); + queryParam.setStartTime(date + " 00:00:00"); + queryParam.setEndTime(date + " 23:59:59"); + List alarmList = eventFeignClient.getDevAlarmList(queryParam).getData(); + Map> alarmMap = alarmList.stream().collect(Collectors.groupingBy(CsEventPO::getDeviceId)); + + List l0 = new ArrayList<>(); + List l1 = new ArrayList<>(); + List> l2 = new ArrayList<>(); + //step1:构造告警数据 + String id = IdUtil.fastSimpleUUID(); + alarm.setId(id); + alarm.setTime(LocalDate.parse(date)); + + devList.forEach(item->{ + //写入中断数据 + CsAlarmData data1 = alarmDataMap.get(item); + //写入告警数据 + List data2 = alarmMap.get(item); + List alarmIds = null; + if (CollUtil.isNotEmpty(data2)) { + alarmIds = data2.stream().map(CsEventPO::getId).collect(Collectors.toList()); + } + //只要有一个不为空，就需要统计 + if (Objects.nonNull(data1) || CollUtil.isNotEmpty(alarmIds)) { + // 对中断时间数据进行过滤，只保留间隔超过 6 小时的时间段 + l0.add(item); + l1.add(data1); + l2.add(alarmIds); + } + }); + alarm.setDevList(String.join(",", l0)); + alarm.setInterruptEvent(new Gson().toJson(l1)); + alarm.setAlarmEvent(new Gson().toJson(l2)); + //step2:用户告警通知 + if (CollUtil.isNotEmpty(l0)) { + Set userIds = new HashSet<>(); + l0.forEach(item->{ + List ids = userDevMap.get(item); + if (CollUtil.isNotEmpty(ids)) { + userIds.addAll(ids); + } + }); + //添加管理员用户 + List result = Stream.concat(userIdList.stream(), userIds.stream()) + .distinct() + .collect(Collectors.toList()); + result.forEach(userId->{ + //新增cs_event_user数据 + CsEventUserPO po = new CsEventUserPO(); + po.setUserId(userId); + po.setEventId(id); + po.setStatus(0); + list2.add(po); + }); + } + } + if (ObjectUtil.isNotNull(alarm)) { + //先删除 再录入 + List oldList = csAlarmFeignClient.queryListByTime(time).getData(); + if (CollUtil.isNotEmpty(oldList)) { + List ids = oldList.stream().map(CsAlarm::getId).collect(Collectors.toList()); + csAlarmFeignClient.deleteListByTime(time); + eventUserFeignClient.deleteByIds(ids); + } + csAlarmFeignClient.addList(Collections.singletonList(alarm)); + eventUserFeignClient.addUserEventList(list2); + } + } + @Override public void pushAppMsg(String time) { //查询所有用户-设备关系 @@ -457,6 +594,108 @@ public class DataTaskServiceImpl implements IDataTaskService { } } + /** + * 构建告警数据 + */ + private List buildAlarmData(List devList, + List onlineRateList, + List integrityList, + List allLine, + CsAlarmSet csAlarmSet) { + List result = new ArrayList<>(); + + if (CollUtil.isEmpty(devList)) { + return result; + } + + // 获取在线率阈值和完整性阈值 + double onlineRateThreshold = csAlarmSet.getOnlineRateLimit() != null ? csAlarmSet.getOnlineRateLimit().doubleValue() : 60.0; + double integrityThreshold = csAlarmSet.getIntegrityLimit() != null ? csAlarmSet.getIntegrityLimit().doubleValue() : 90.0; + + // 构建设备ID到监测点列表的映射 + Map> deviceToLinesMap = allLine.stream() + .collect(Collectors.groupingBy(CsLineDTO::getDeviceId)); + + // 构建监测点ID到监测点信息的映射 + Map lineInfoMap = allLine.stream() + .collect(Collectors.toMap(CsLineDTO::getLineId, line -> line, (k1, k2) -> k1)); + + // 将列表转换为Map便于查找（处理null情况） + Map onlineRateMap = CollUtil.isNotEmpty(onlineRateList) + ? onlineRateList.stream().collect(Collectors.toMap(RStatOnlineRateD::getDevIndex, item -> item, (k1, k2) -> k1)) + : new HashMap<>(); + + Map integrityDataMap = CollUtil.isNotEmpty(integrityList) + ? integrityList.stream().collect(Collectors.toMap(RStatIntegrityD::getLineIndex, item -> item, (k1, k2) -> k1)) + : new HashMap<>(); + + // 遍历每个设备 + for (String deviceId : devList) { + // 处理在线率数据 + RStatOnlineRateD onlineRateData = onlineRateMap.get(deviceId); + + // 如果为空，默认值为0 + int onlineMin = onlineRateData != null ? onlineRateData.getOnlineMin() : 0; + double onlineRateValue = Math.round(onlineMin * 100.0 / 1440 * 100.0) / 100.0; + + CsAlarmData.OnlineRateAlarm onlineRateAlarm = new CsAlarmData.OnlineRateAlarm(); + onlineRateAlarm.setValue(onlineRateValue); + onlineRateAlarm.setThreshold(onlineRateThreshold); + // value < threshold 时为 true（低于阈值为异常） + onlineRateAlarm.setIsAbnormal(onlineRateValue < onlineRateThreshold); + + // 处理完整性数据 - 按设备分组 + List deviceLines = deviceToLinesMap.get(deviceId); + boolean hasIntegrityAbnormal = false; + + // 构建完整性数据结构 + CsAlarmData.IntegrityAlarm integrityAlarm = new CsAlarmData.IntegrityAlarm(); + integrityAlarm.setThreshold(integrityThreshold); + + List monitorPoints = new ArrayList<>(); + + if (CollUtil.isNotEmpty(deviceLines)) { + // 为该设备下的每个监测点创建完整性数据 + for (CsLineDTO line : deviceLines) { + String lineId = line.getLineId(); + RStatIntegrityD integrityData = integrityDataMap.get(lineId); + + // 如果为空，默认值为0 + int dueTime = integrityData != null && integrityData.getDueTime() != null ? integrityData.getDueTime() : 0; + int realTime = integrityData != null && integrityData.getRealTime() != null ? integrityData.getRealTime() : 0; + double integrityValue = dueTime > 0 ? Math.round(realTime * 100.0 / dueTime * 100.0) / 100.0 : 0.0; + + CsAlarmData.IntegrityAlarm.MonitorPointIntegrity monitorPoint = new CsAlarmData.IntegrityAlarm.MonitorPointIntegrity(); + monitorPoint.setMonitorPointId(lineId); + monitorPoint.setMonitorName(line.getName()); + monitorPoint.setValue(integrityValue); + // value < threshold 时为 true（低于阈值为异常） + Boolean isAbnormal = integrityValue < integrityThreshold; + monitorPoint.setIsAbnormal(isAbnormal); + + if (isAbnormal) { + hasIntegrityAbnormal = true; + } + + monitorPoints.add(monitorPoint); + } + } + + integrityAlarm.setMonitorPoints(monitorPoints); + + // 只要在线率异常 或 完整性有任意一个监测点异常，就记录该设备的完整数据 + if (onlineRateAlarm.getIsAbnormal() || hasIntegrityAbnormal) { + CsAlarmData alarmData = new CsAlarmData(); + alarmData.setDeviceId(deviceId); + alarmData.setOnlineRate(onlineRateAlarm); + alarmData.setIntegrity(integrityAlarm); + result.add(alarmData); + } + } + + return result; + } + /** * 过滤时间间隔，只保留超过指定小时数的时间段