|
|
@@ -3,8 +3,14 @@ package com.zksy.base.service.impl;
|
|
|
import cn.hutool.core.collection.CollUtil;
|
|
|
import cn.hutool.core.util.StrUtil;
|
|
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
|
|
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
|
|
|
+import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
|
|
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
|
|
import com.zksy.base.audio.domain.NoiseInfo;
|
|
|
+
|
|
|
+import java.util.Comparator;
|
|
|
+
|
|
|
import com.zksy.base.audio.mapper.NoiseInfoMapper;
|
|
|
import com.zksy.base.domain.EquipmentBase;
|
|
|
import com.zksy.base.domain.EquipmentType;
|
|
|
@@ -12,18 +18,30 @@ import com.zksy.base.enums.EquipmentTypeEnum;
|
|
|
import com.zksy.base.domain.vo.FlowMonitoringVO;
|
|
|
import com.zksy.base.domain.vo.LeakageMonitoringVO;
|
|
|
import com.zksy.base.domain.vo.PressureMonitoringVO;
|
|
|
+import com.zksy.base.alarm.domain.WarningThreshold;
|
|
|
+import com.zksy.base.alarm.mapper.WarningThresholdMapper;
|
|
|
+import com.zksy.base.domain.EquipmentBase;
|
|
|
+import com.zksy.base.domain.vo.FlowRecordVO;
|
|
|
+import com.zksy.base.domain.vo.HydrantPressureRecordVO;
|
|
|
+import com.zksy.base.domain.vo.LeakageRecordVO;
|
|
|
+import com.zksy.base.domain.vo.SupplyPressureRecordVO;
|
|
|
import com.zksy.base.mapper.EquipmentBaseMapper;
|
|
|
import com.zksy.base.mapper.EquipmentTypeMapper;
|
|
|
import com.zksy.base.pressure.domain.FirefightingPressure;
|
|
|
import com.zksy.base.pressure.mapper.FirefightingPressureMapper;
|
|
|
import com.zksy.base.radar.domain.RadarData;
|
|
|
import com.zksy.base.radar.mapper.RadarDataMapper;
|
|
|
+import com.zksy.base.domain.vo.MonitorOverviewVO;
|
|
|
+import com.zksy.base.domain.vo.TrendPoint;
|
|
|
import com.zksy.base.service.MonitoringService;
|
|
|
import com.zksy.base.telemetry.domain.TelemetryData;
|
|
|
import com.zksy.base.telemetry.mapper.TelemetryDataMapper;
|
|
|
+import com.zksy.base.warning.domain.WarningType;
|
|
|
+import com.zksy.base.warning.mapper.WarningTypeMapper;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.stereotype.Service;
|
|
|
|
|
|
+import java.time.LocalDate;
|
|
|
import java.time.LocalDateTime;
|
|
|
import java.util.*;
|
|
|
import java.util.function.Function;
|
|
|
@@ -35,8 +53,6 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
private static final String RADAR_SOURCE = "雷达流量计";
|
|
|
private static final String TELEMETRY_SOURCE = "遥测终端";
|
|
|
|
|
|
- // 供水监测口径的设备类型主键，取值见 com.zksy.base.enums.EquipmentTypeEnum
|
|
|
-
|
|
|
@Autowired
|
|
|
private EquipmentBaseMapper equipmentBaseMapper;
|
|
|
|
|
|
@@ -55,6 +71,12 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
@Autowired
|
|
|
private NoiseInfoMapper noiseInfoMapper;
|
|
|
|
|
|
+ @Autowired
|
|
|
+ private WarningThresholdMapper warningThresholdMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private WarningTypeMapper warningTypeMapper;
|
|
|
+
|
|
|
@Override
|
|
|
public Page<FlowMonitoringVO> getFlowMonitorPage(long pageNum, long pageSize,
|
|
|
String equipmentName, String equipmentCode,
|
|
|
@@ -190,6 +212,633 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
return pageOf(allRows, pageNum, pageSize);
|
|
|
}
|
|
|
|
|
|
+ // ===================== 原始采集记录列表（按 warning_threshold 判定告警） =====================
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Page<FlowRecordVO> getFlowRecordPage(long pageNum, long pageSize, String deviceCode,
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime, boolean latestOnly) {
|
|
|
+ Page<TelemetryData> page;
|
|
|
+ if (latestOnly) {
|
|
|
+ page = latestPerDevicePage(pageNum, pageSize, telemetryDataMapper, "device_code", deviceCode, startTime, endTime);
|
|
|
+ } else {
|
|
|
+ LambdaQueryWrapper<TelemetryData> qw = new LambdaQueryWrapper<TelemetryData>()
|
|
|
+ .ge(startTime != null, TelemetryData::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, TelemetryData::getCreateTime, endTime)
|
|
|
+ .orderByDesc(TelemetryData::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, TelemetryData::getDeviceCode, deviceCode);
|
|
|
+ page = telemetryDataMapper.selectPage(new Page<>(pageNum, pageSize), qw);
|
|
|
+ }
|
|
|
+ page.getRecords().sort(Comparator.comparing(TelemetryData::getCreateTime).reversed());
|
|
|
+ String warningCode = "WARN-INSTANT-FLOW-SUPPLY";
|
|
|
+ List<String> codes = collectCodes(page.getRecords(), TelemetryData::getDeviceCode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ Page<FlowRecordVO> result = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
|
|
|
+ result.setRecords(page.getRecords().stream().map(r -> {
|
|
|
+ FlowRecordVO vo = new FlowRecordVO();
|
|
|
+ vo.setDeviceCode(r.getDeviceCode());
|
|
|
+ EquipmentBase base = info.get(r.getDeviceCode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getDeviceCode()) : r.getDeviceCode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setMeter1InstantFlow(r.getMeter1InstantFlow());
|
|
|
+ vo.setFlowSpeed(r.getFlowSpeed());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getDeviceCode(), thresholds.get(r.getDeviceCode()),
|
|
|
+ dataType, toDouble(r.getMeter1InstantFlow())));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList()));
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Page<SupplyPressureRecordVO> getSupplyPressureRecordPage(long pageNum, long pageSize, String deviceCode,
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime, boolean latestOnly) {
|
|
|
+ Page<TelemetryData> page;
|
|
|
+ if (latestOnly) {
|
|
|
+ page = latestPerDevicePage(pageNum, pageSize, telemetryDataMapper, "device_code", deviceCode, startTime, endTime);
|
|
|
+ } else {
|
|
|
+ LambdaQueryWrapper<TelemetryData> qw = new LambdaQueryWrapper<TelemetryData>()
|
|
|
+ .ge(startTime != null, TelemetryData::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, TelemetryData::getCreateTime, endTime)
|
|
|
+ .orderByDesc(TelemetryData::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, TelemetryData::getDeviceCode, deviceCode);
|
|
|
+ page = telemetryDataMapper.selectPage(new Page<>(pageNum, pageSize), qw);
|
|
|
+ }
|
|
|
+ page.getRecords().sort(Comparator.comparing(TelemetryData::getCreateTime).reversed());
|
|
|
+ String warningCode = "WARN-PRESSURE";
|
|
|
+ List<String> codes = collectCodes(page.getRecords(), TelemetryData::getDeviceCode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ Page<SupplyPressureRecordVO> result = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
|
|
|
+ result.setRecords(page.getRecords().stream().map(r -> {
|
|
|
+ SupplyPressureRecordVO vo = new SupplyPressureRecordVO();
|
|
|
+ vo.setDeviceCode(r.getDeviceCode());
|
|
|
+ EquipmentBase base = info.get(r.getDeviceCode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getDeviceCode()) : r.getDeviceCode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setPressure(r.getPressure());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getDeviceCode(), thresholds.get(r.getDeviceCode()),
|
|
|
+ dataType, toDouble(r.getPressure())));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList()));
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Page<HydrantPressureRecordVO> getHydrantPressureRecordPage(long pageNum, long pageSize,
|
|
|
+ String telemeteringStation,
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime, boolean latestOnly) {
|
|
|
+ Page<FirefightingPressure> page;
|
|
|
+ if (latestOnly) {
|
|
|
+ page = latestPerDevicePage(pageNum, pageSize, firefightingPressureMapper, "telemetering_station", telemeteringStation, startTime, endTime);
|
|
|
+ } else {
|
|
|
+ LambdaQueryWrapper<FirefightingPressure> qw = new LambdaQueryWrapper<FirefightingPressure>()
|
|
|
+ .ge(startTime != null, FirefightingPressure::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, FirefightingPressure::getCreateTime, endTime)
|
|
|
+ .orderByDesc(FirefightingPressure::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, FirefightingPressure::getTelemeteringStation, telemeteringStation);
|
|
|
+ page = firefightingPressureMapper.selectPage(new Page<>(pageNum, pageSize), qw);
|
|
|
+ }
|
|
|
+ page.getRecords().sort(Comparator.comparing(FirefightingPressure::getCreateTime).reversed());
|
|
|
+ String warningCode = "WARN-FIRE-PRESSURE";
|
|
|
+ List<String> codes = collectCodes(page.getRecords(), FirefightingPressure::getTelemeteringStation);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ Page<HydrantPressureRecordVO> result = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
|
|
|
+ result.setRecords(page.getRecords().stream().map(r -> {
|
|
|
+ HydrantPressureRecordVO vo = new HydrantPressureRecordVO();
|
|
|
+ vo.setTelemeteringStation(r.getTelemeteringStation());
|
|
|
+ EquipmentBase base = info.get(r.getTelemeteringStation());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getTelemeteringStation()) : r.getTelemeteringStation());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setPressureValue(r.getPressureValue());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getTelemeteringStation(), thresholds.get(r.getTelemeteringStation()),
|
|
|
+ dataType, r.getPressureValue()));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList()));
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Page<LeakageRecordVO> getLeakageRecordPage(long pageNum, long pageSize, String encode,
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime, boolean latestOnly) {
|
|
|
+ Page<NoiseInfo> page;
|
|
|
+ if (latestOnly) {
|
|
|
+ page = latestPerDevicePage(pageNum, pageSize, noiseInfoMapper, "encode", encode, startTime, endTime);
|
|
|
+ } else {
|
|
|
+ LambdaQueryWrapper<NoiseInfo> qw = new LambdaQueryWrapper<NoiseInfo>()
|
|
|
+ .ge(startTime != null, NoiseInfo::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, NoiseInfo::getCreateTime, endTime)
|
|
|
+ .orderByDesc(NoiseInfo::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, NoiseInfo::getEncode, encode);
|
|
|
+ page = noiseInfoMapper.selectPage(new Page<>(pageNum, pageSize), qw);
|
|
|
+ }
|
|
|
+ page.getRecords().sort(Comparator.comparing(NoiseInfo::getCreateTime).reversed());
|
|
|
+ String warningCode = "WARN-LEAK";
|
|
|
+ List<String> codes = collectCodes(page.getRecords(), NoiseInfo::getEncode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ Page<LeakageRecordVO> result = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
|
|
|
+ result.setRecords(page.getRecords().stream().map(r -> {
|
|
|
+ LeakageRecordVO vo = new LeakageRecordVO();
|
|
|
+ vo.setEncode(r.getEncode());
|
|
|
+ EquipmentBase base = info.get(r.getEncode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getEncode()) : r.getEncode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setRms(r.getRms());
|
|
|
+ vo.setFilepath(r.getFilepath());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getEncode(), thresholds.get(r.getEncode()),
|
|
|
+ dataType, r.getRms() == null ? null : r.getRms().doubleValue()));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList()));
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ // ===================== 概览（顶部统计指标 + 7 天趋势） =====================
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public MonitorOverviewVO getFlowOverview(LocalDateTime startTime, LocalDateTime endTime, String deviceCode) {
|
|
|
+ LocalDateTime end = endTime != null ? endTime : LocalDateTime.now();
|
|
|
+ LocalDateTime start = startTime != null ? startTime : end.toLocalDate().minusDays(6).atStartOfDay();
|
|
|
+ String valueExpr = "CAST(meter1_instant_flow AS DECIMAL(18,4))";
|
|
|
+ QueryWrapper<TelemetryData> q = new QueryWrapper<TelemetryData>()
|
|
|
+ .select("TO_CHAR(create_time,'YYYY-MM-DD') AS day",
|
|
|
+ "AVG(" + valueExpr + ") AS avg_val",
|
|
|
+ "MAX(" + valueExpr + ") AS max_val",
|
|
|
+ "MIN(" + valueExpr + ") AS min_val",
|
|
|
+ "MAX(create_time) AS latest_time")
|
|
|
+ .ge("create_time", start)
|
|
|
+ .le("create_time", end);
|
|
|
+ if (StrUtil.isNotBlank(deviceCode)) {
|
|
|
+ q.eq("device_code", deviceCode);
|
|
|
+ }
|
|
|
+ q.groupBy("TO_CHAR(create_time,'YYYY-MM-DD')").orderByAsc("day");
|
|
|
+ List<Map<String, Object>> agg = telemetryDataMapper.selectMaps(q);
|
|
|
+ return buildDeviceOverview("WARN-INSTANT-FLOW-SUPPLY", valueExpr, agg, start, end);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public MonitorOverviewVO getSupplyPressureOverview(LocalDateTime startTime, LocalDateTime endTime, String deviceCode) {
|
|
|
+ LocalDateTime end = endTime != null ? endTime : LocalDateTime.now();
|
|
|
+ LocalDateTime start = startTime != null ? startTime : end.toLocalDate().minusDays(6).atStartOfDay();
|
|
|
+ String valueExpr = "CAST(pressure AS DECIMAL(18,4))";
|
|
|
+ QueryWrapper<TelemetryData> q = new QueryWrapper<TelemetryData>()
|
|
|
+ .select("TO_CHAR(create_time,'YYYY-MM-DD') AS day",
|
|
|
+ "AVG(" + valueExpr + ") AS avg_val",
|
|
|
+ "MAX(" + valueExpr + ") AS max_val",
|
|
|
+ "MIN(" + valueExpr + ") AS min_val",
|
|
|
+ "MAX(create_time) AS latest_time")
|
|
|
+ .ge("create_time", start)
|
|
|
+ .le("create_time", end);
|
|
|
+ if (StrUtil.isNotBlank(deviceCode)) {
|
|
|
+ q.eq("device_code", deviceCode);
|
|
|
+ }
|
|
|
+ q.groupBy("TO_CHAR(create_time,'YYYY-MM-DD')").orderByAsc("day");
|
|
|
+ List<Map<String, Object>> agg = telemetryDataMapper.selectMaps(q);
|
|
|
+ return buildDeviceOverview("WARN-PRESSURE", valueExpr, agg, start, end);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public MonitorOverviewVO getHydrantPressureOverview(LocalDateTime startTime, LocalDateTime endTime, String telemeteringStation) {
|
|
|
+ LocalDateTime end = endTime != null ? endTime : LocalDateTime.now();
|
|
|
+ LocalDateTime start = startTime != null ? startTime : end.toLocalDate().minusDays(6).atStartOfDay();
|
|
|
+ String valueExpr = "pressure_value";
|
|
|
+ QueryWrapper<FirefightingPressure> q = new QueryWrapper<FirefightingPressure>()
|
|
|
+ .select("TO_CHAR(create_time,'YYYY-MM-DD') AS day",
|
|
|
+ "AVG(" + valueExpr + ") AS avg_val",
|
|
|
+ "MAX(" + valueExpr + ") AS max_val",
|
|
|
+ "MIN(" + valueExpr + ") AS min_val",
|
|
|
+ "MAX(create_time) AS latest_time")
|
|
|
+ .ge("create_time", start)
|
|
|
+ .le("create_time", end);
|
|
|
+ if (StrUtil.isNotBlank(telemeteringStation)) {
|
|
|
+ q.eq("telemetering_station", telemeteringStation);
|
|
|
+ }
|
|
|
+ q.groupBy("TO_CHAR(create_time,'YYYY-MM-DD')").orderByAsc("day");
|
|
|
+ List<Map<String, Object>> agg = firefightingPressureMapper.selectMaps(q);
|
|
|
+ return buildDeviceOverview("WARN-FIRE-PRESSURE", valueExpr, agg, start, end);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public MonitorOverviewVO getLeakageOverview(LocalDateTime startTime, LocalDateTime endTime, String encode) {
|
|
|
+ LocalDateTime end = endTime != null ? endTime : LocalDateTime.now();
|
|
|
+ LocalDateTime start = startTime != null ? startTime : end.toLocalDate().minusDays(6).atStartOfDay();
|
|
|
+ String valueExpr = "rms";
|
|
|
+ QueryWrapper<NoiseInfo> q = new QueryWrapper<NoiseInfo>()
|
|
|
+ .select("TO_CHAR(create_time,'YYYY-MM-DD') AS day",
|
|
|
+ "AVG(" + valueExpr + ") AS avg_val",
|
|
|
+ "MAX(" + valueExpr + ") AS max_val",
|
|
|
+ "MIN(" + valueExpr + ") AS min_val",
|
|
|
+ "MAX(create_time) AS latest_time")
|
|
|
+ .ge("create_time", start)
|
|
|
+ .le("create_time", end);
|
|
|
+ if (StrUtil.isNotBlank(encode)) {
|
|
|
+ q.eq("encode", encode);
|
|
|
+ }
|
|
|
+ q.groupBy("TO_CHAR(create_time,'YYYY-MM-DD')").orderByAsc("day");
|
|
|
+ List<Map<String, Object>> agg = noiseInfoMapper.selectMaps(q);
|
|
|
+ return buildDeviceOverview("WARN-LEAK", valueExpr, agg, start, end);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 按天聚合，构建最近 7 天整体趋势（每天一个点：均值 / 峰值 / 最低）。
|
|
|
+ * agg 每行：day / avg_val / max_val / min_val
|
|
|
+ */
|
|
|
+ private MonitorOverviewVO buildDeviceOverview(String warningCode, String valueExpr,
|
|
|
+ List<Map<String, Object>> agg, LocalDateTime start, LocalDateTime endTime) {
|
|
|
+ MonitorOverviewVO vo = new MonitorOverviewVO();
|
|
|
+ List<String> days = lastSevenDayLabels(start, endTime);
|
|
|
+ Map<String, Double> dayAvg = new LinkedHashMap<>();
|
|
|
+ Map<String, Double> dayMax = new LinkedHashMap<>();
|
|
|
+ Map<String, Double> dayMin = new LinkedHashMap<>();
|
|
|
+ for (String d : days) {
|
|
|
+ dayAvg.put(d, null);
|
|
|
+ dayMax.put(d, null);
|
|
|
+ dayMin.put(d, null);
|
|
|
+ }
|
|
|
+ double sum = 0;
|
|
|
+ int cnt = 0;
|
|
|
+ double peak = Double.MIN_VALUE;
|
|
|
+ double min = Double.MAX_VALUE;
|
|
|
+ boolean hasAny = false;
|
|
|
+ LocalDateTime latestTime = null;
|
|
|
+ for (Map<String, Object> m : agg) {
|
|
|
+ String day = String.valueOf(m.get("day"));
|
|
|
+ Double a = toBig(m.get("avg_val"));
|
|
|
+ Double pk = toBig(m.get("max_val"));
|
|
|
+ Double mn = toBig(m.get("min_val"));
|
|
|
+ LocalDateTime lt = toLdt(m.get("latest_time"));
|
|
|
+ if (lt != null && (latestTime == null || lt.isAfter(latestTime))) {
|
|
|
+ latestTime = lt;
|
|
|
+ }
|
|
|
+ if (dayAvg.containsKey(day)) {
|
|
|
+ dayAvg.put(day, a);
|
|
|
+ dayMax.put(day, pk);
|
|
|
+ dayMin.put(day, mn);
|
|
|
+ }
|
|
|
+ if (a != null) {
|
|
|
+ sum += a;
|
|
|
+ cnt++;
|
|
|
+ }
|
|
|
+ if (pk != null) {
|
|
|
+ peak = Math.max(peak, pk);
|
|
|
+ hasAny = true;
|
|
|
+ }
|
|
|
+ if (mn != null) {
|
|
|
+ min = Math.min(min, mn);
|
|
|
+ hasAny = true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ List<TrendPoint> trend = new ArrayList<>();
|
|
|
+ for (String d : days) {
|
|
|
+ TrendPoint p = new TrendPoint();
|
|
|
+ p.setTime(d);
|
|
|
+ p.setAvg(dayAvg.get(d));
|
|
|
+ p.setPeak(dayMax.get(d));
|
|
|
+ p.setMin(dayMin.get(d));
|
|
|
+ trend.add(p);
|
|
|
+ }
|
|
|
+ vo.setTrend(trend);
|
|
|
+ vo.setAvgValue(cnt > 0 ? sum / cnt : null);
|
|
|
+ vo.setPeakValue(hasAny ? peak : null);
|
|
|
+ vo.setMinValue(hasAny ? min : null);
|
|
|
+ WarningThreshold threshold = firstThreshold(warningCode);
|
|
|
+ vo.setThresholdMin(threshold == null ? null : threshold.getMinValue());
|
|
|
+ vo.setThresholdMax(threshold == null ? null : threshold.getMaxValue());
|
|
|
+ vo.setAnomalyCount(countAnomaly(warningCode, valueExpr, start, endTime));
|
|
|
+ vo.setLastUpdate(latestTime != null ? latestTime.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) : "");
|
|
|
+ return vo;
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<String> lastSevenDayLabels(LocalDateTime start, LocalDateTime endTime) {
|
|
|
+ LocalDate endDay = (endTime != null ? endTime : LocalDateTime.now()).toLocalDate();
|
|
|
+ LocalDate startDay = (start != null ? start : LocalDateTime.now().minusDays(6)).toLocalDate();
|
|
|
+ List<String> days = new ArrayList<>();
|
|
|
+ for (LocalDate d = startDay; !d.isAfter(endDay); d = d.plusDays(1)) {
|
|
|
+ days.add(d.format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd")));
|
|
|
+ }
|
|
|
+ return days;
|
|
|
+ }
|
|
|
+
|
|
|
+ private long countAnomaly(String warningCode, String valueExpr, LocalDateTime start, LocalDateTime endTime) {
|
|
|
+ WarningThreshold threshold = firstThreshold(warningCode);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ if (threshold == null || dataType == null) {
|
|
|
+ return 0L;
|
|
|
+ }
|
|
|
+ List<String> conditions = new ArrayList<>();
|
|
|
+ if ((dataType == 0 || dataType == 1) && threshold.getMinValue() != null) {
|
|
|
+ conditions.add(valueExpr + " <= " + threshold.getMinValue());
|
|
|
+ }
|
|
|
+ if ((dataType == 0 || dataType == 2) && threshold.getMaxValue() != null) {
|
|
|
+ conditions.add(valueExpr + " >= " + threshold.getMaxValue());
|
|
|
+ }
|
|
|
+ if (conditions.isEmpty()) {
|
|
|
+ return 0L;
|
|
|
+ }
|
|
|
+ String caseSql = "SUM(CASE WHEN " + String.join(" OR ", conditions) + " THEN 1 ELSE 0 END) AS cnt";
|
|
|
+ List<Map<String, Object>> result;
|
|
|
+ if ("WARN-INSTANT-FLOW-SUPPLY".equals(warningCode) || "WARN-PRESSURE".equals(warningCode)) {
|
|
|
+ result = telemetryDataMapper.selectMaps(new QueryWrapper<TelemetryData>().select(caseSql)
|
|
|
+ .ge(start != null, "create_time", start).le(endTime != null, "create_time", endTime));
|
|
|
+ } else if ("WARN-FIRE-PRESSURE".equals(warningCode)) {
|
|
|
+ result = firefightingPressureMapper.selectMaps(new QueryWrapper<FirefightingPressure>().select(caseSql)
|
|
|
+ .ge(start != null, "create_time", start).le(endTime != null, "create_time", endTime));
|
|
|
+ } else {
|
|
|
+ result = noiseInfoMapper.selectMaps(new QueryWrapper<NoiseInfo>().select(caseSql)
|
|
|
+ .ge(start != null, "create_time", start).le(endTime != null, "create_time", endTime));
|
|
|
+ }
|
|
|
+ if (result.isEmpty()) {
|
|
|
+ return 0L;
|
|
|
+ }
|
|
|
+ Object value = result.get(0).get("cnt");
|
|
|
+ return value instanceof Number ? ((Number) value).longValue() : 0L;
|
|
|
+ }
|
|
|
+
|
|
|
+ private WarningThreshold firstThreshold(String warningCode) {
|
|
|
+ return warningThresholdMapper.selectOne(
|
|
|
+ new LambdaQueryWrapper<WarningThreshold>().eq(WarningThreshold::getWarningCode, warningCode));
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ private Double toBig(Object value) {
|
|
|
+ if (value == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ if (value instanceof Number) {
|
|
|
+ return ((Number) value).doubleValue();
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return Double.parseDouble(value.toString());
|
|
|
+ } catch (NumberFormatException e) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // ===================== 原始记录全量（导出用） =====================
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<FlowRecordVO> listFlowRecords(String deviceCode, LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
+ LambdaQueryWrapper<TelemetryData> qw = new LambdaQueryWrapper<TelemetryData>()
|
|
|
+ .ge(startTime != null, TelemetryData::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, TelemetryData::getCreateTime, endTime)
|
|
|
+ .orderByDesc(TelemetryData::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, TelemetryData::getDeviceCode, deviceCode);
|
|
|
+ List<TelemetryData> list = telemetryDataMapper.selectList(qw);
|
|
|
+ return mapFlowRecords(deviceCode, list);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<SupplyPressureRecordVO> listSupplyPressureRecords(String deviceCode, LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
+ LambdaQueryWrapper<TelemetryData> qw = new LambdaQueryWrapper<TelemetryData>()
|
|
|
+ .ge(startTime != null, TelemetryData::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, TelemetryData::getCreateTime, endTime)
|
|
|
+ .orderByDesc(TelemetryData::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, TelemetryData::getDeviceCode, deviceCode);
|
|
|
+ List<TelemetryData> list = telemetryDataMapper.selectList(qw);
|
|
|
+ return mapSupplyPressureRecords(deviceCode, list);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<HydrantPressureRecordVO> listHydrantPressureRecords(String telemeteringStation, LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
+ LambdaQueryWrapper<FirefightingPressure> qw = new LambdaQueryWrapper<FirefightingPressure>()
|
|
|
+ .ge(startTime != null, FirefightingPressure::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, FirefightingPressure::getCreateTime, endTime)
|
|
|
+ .orderByDesc(FirefightingPressure::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, FirefightingPressure::getTelemeteringStation, telemeteringStation);
|
|
|
+ List<FirefightingPressure> list = firefightingPressureMapper.selectList(qw);
|
|
|
+ return mapHydrantPressureRecords(telemeteringStation, list);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public List<LeakageRecordVO> listLeakageRecords(String encode, LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
+ LambdaQueryWrapper<NoiseInfo> qw = new LambdaQueryWrapper<NoiseInfo>()
|
|
|
+ .ge(startTime != null, NoiseInfo::getCreateTime, startTime)
|
|
|
+ .le(endTime != null, NoiseInfo::getCreateTime, endTime)
|
|
|
+ .orderByDesc(NoiseInfo::getCreateTime);
|
|
|
+ applyDeviceFilter(qw, NoiseInfo::getEncode, encode);
|
|
|
+ List<NoiseInfo> list = noiseInfoMapper.selectList(qw);
|
|
|
+ return mapLeakageRecords(encode, list);
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<FlowRecordVO> mapFlowRecords(String deviceCode, List<TelemetryData> list) {
|
|
|
+ String warningCode = "WARN-INSTANT-FLOW-SUPPLY";
|
|
|
+ List<String> codes = collectCodes(list, TelemetryData::getDeviceCode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ return list.stream().map(r -> {
|
|
|
+ FlowRecordVO vo = new FlowRecordVO();
|
|
|
+ vo.setDeviceCode(r.getDeviceCode());
|
|
|
+ EquipmentBase base = info.get(r.getDeviceCode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getDeviceCode()) : r.getDeviceCode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setMeter1InstantFlow(r.getMeter1InstantFlow());
|
|
|
+ vo.setFlowSpeed(r.getFlowSpeed());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getDeviceCode(), thresholds.get(r.getDeviceCode()), dataType, toDouble(r.getMeter1InstantFlow())));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<SupplyPressureRecordVO> mapSupplyPressureRecords(String deviceCode, List<TelemetryData> list) {
|
|
|
+ String warningCode = "WARN-PRESSURE";
|
|
|
+ List<String> codes = collectCodes(list, TelemetryData::getDeviceCode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ return list.stream().map(r -> {
|
|
|
+ SupplyPressureRecordVO vo = new SupplyPressureRecordVO();
|
|
|
+ vo.setDeviceCode(r.getDeviceCode());
|
|
|
+ EquipmentBase base = info.get(r.getDeviceCode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getDeviceCode()) : r.getDeviceCode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setPressure(r.getPressure());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getDeviceCode(), thresholds.get(r.getDeviceCode()), dataType, toDouble(r.getPressure())));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<HydrantPressureRecordVO> mapHydrantPressureRecords(String telemeteringStation, List<FirefightingPressure> list) {
|
|
|
+ String warningCode = "WARN-FIRE-PRESSURE";
|
|
|
+ List<String> codes = collectCodes(list, FirefightingPressure::getTelemeteringStation);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ return list.stream().map(r -> {
|
|
|
+ HydrantPressureRecordVO vo = new HydrantPressureRecordVO();
|
|
|
+ vo.setTelemeteringStation(r.getTelemeteringStation());
|
|
|
+ EquipmentBase base = info.get(r.getTelemeteringStation());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getTelemeteringStation()) : r.getTelemeteringStation());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setPressureValue(r.getPressureValue());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getTelemeteringStation(), thresholds.get(r.getTelemeteringStation()), dataType, r.getPressureValue()));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private List<LeakageRecordVO> mapLeakageRecords(String encode, List<NoiseInfo> list) {
|
|
|
+ String warningCode = "WARN-LEAK";
|
|
|
+ List<String> codes = collectCodes(list, NoiseInfo::getEncode);
|
|
|
+ Map<String, WarningThreshold> thresholds = thresholdMap(warningCode, codes);
|
|
|
+ Map<String, EquipmentBase> info = deviceInfoMap(codes);
|
|
|
+ Integer dataType = resolveDataType(warningCode);
|
|
|
+ return list.stream().map(r -> {
|
|
|
+ LeakageRecordVO vo = new LeakageRecordVO();
|
|
|
+ vo.setEncode(r.getEncode());
|
|
|
+ EquipmentBase base = info.get(r.getEncode());
|
|
|
+ vo.setDeviceName(base != null ? StrUtil.blankToDefault(base.getEquipmentName(), r.getEncode()) : r.getEncode());
|
|
|
+ vo.setDeviceType(base != null ? base.getEquipmentTypeName() : "");
|
|
|
+ vo.setRms(r.getRms());
|
|
|
+ vo.setFilepath(r.getFilepath());
|
|
|
+ vo.setCreateTime(r.getCreateTime());
|
|
|
+ vo.setWarningCode(warningCode);
|
|
|
+ vo.setAlarmStatus(evaluateAlarm(r.getEncode(), thresholds.get(r.getEncode()), dataType, r.getRms() == null ? null : r.getRms().doubleValue()));
|
|
|
+ return vo;
|
|
|
+ }).collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private <T> List<String> collectCodes(List<T> records, Function<T, String> codeGetter) {
|
|
|
+ return records.stream().map(codeGetter).filter(StrUtil::isNotBlank).distinct().collect(Collectors.toList());
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, WarningThreshold> thresholdMap(String warningCode, List<String> deviceCodes) {
|
|
|
+ if (CollUtil.isEmpty(deviceCodes)) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ return warningThresholdMapper.selectList(new LambdaQueryWrapper<WarningThreshold>()
|
|
|
+ .in(WarningThreshold::getDeviceCode, deviceCodes)
|
|
|
+ .eq(WarningThreshold::getWarningCode, warningCode))
|
|
|
+ .stream().collect(Collectors.toMap(WarningThreshold::getDeviceCode, Function.identity(), (left, right) -> left));
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<String, EquipmentBase> deviceInfoMap(List<String> deviceCodes) {
|
|
|
+ if (CollUtil.isEmpty(deviceCodes)) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ List<EquipmentBase> bases = equipmentBaseMapper.selectList(new LambdaQueryWrapper<EquipmentBase>()
|
|
|
+ .in(EquipmentBase::getEquipmentCode, deviceCodes)
|
|
|
+ .select(EquipmentBase::getEquipmentCode, EquipmentBase::getEquipmentName, EquipmentBase::getEquipmentTypeId));
|
|
|
+ if (bases.isEmpty()) {
|
|
|
+ return Collections.emptyMap();
|
|
|
+ }
|
|
|
+ List<Integer> typeIds = bases.stream().map(EquipmentBase::getEquipmentTypeId)
|
|
|
+ .filter(java.util.Objects::nonNull).distinct().collect(Collectors.toList());
|
|
|
+ Map<Integer, String> typeName = typeIds.isEmpty() ? Collections.emptyMap()
|
|
|
+ : equipmentTypeMapper.selectList(new LambdaQueryWrapper<EquipmentType>().in(EquipmentType::getId, typeIds))
|
|
|
+ .stream().collect(Collectors.toMap(EquipmentType::getId, EquipmentType::getTypeName, (l, r) -> l));
|
|
|
+ Map<String, EquipmentBase> map = new HashMap<>(bases.size());
|
|
|
+ for (EquipmentBase b : bases) {
|
|
|
+ String name = typeName.getOrDefault(b.getEquipmentTypeId(),
|
|
|
+ b.getEquipmentTypeId() != null ? String.valueOf(b.getEquipmentTypeId()) : "");
|
|
|
+ b.setEquipmentTypeName(name);
|
|
|
+ map.put(b.getEquipmentCode(), b);
|
|
|
+ }
|
|
|
+ return map;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 设备筛选：支持按设备编码精确匹配，或按设备名称模糊匹配（名称命中后取其编码再 in 查询，避免 SQL 注入）
|
|
|
+ */
|
|
|
+ private <T> void applyDeviceFilter(LambdaQueryWrapper<T> qw, SFunction<T, String> col, String deviceCode) {
|
|
|
+ if (StrUtil.isBlank(deviceCode)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ List<String> nameCodes = equipmentBaseMapper.selectList(new LambdaQueryWrapper<EquipmentBase>()
|
|
|
+ .like(EquipmentBase::getEquipmentName, deviceCode))
|
|
|
+ .stream().map(EquipmentBase::getEquipmentCode).collect(Collectors.toList());
|
|
|
+ List<String> codes = new ArrayList<>();
|
|
|
+ codes.add(deviceCode);
|
|
|
+ codes.addAll(nameCodes);
|
|
|
+ qw.in(col, codes);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 取业务表内最新一条采集时间，用于“最近 7 天”窗口定位（不依赖当前操作系统时间）
|
|
|
+ */
|
|
|
+ private <T> LocalDateTime latestCreateTime(BaseMapper<T> mapper, Class<T> clazz) {
|
|
|
+ List<Map<String, Object>> rows = mapper.selectMaps(new QueryWrapper<T>().select("MAX(create_time) AS mx"));
|
|
|
+ if (rows != null && !rows.isEmpty()) {
|
|
|
+ return toLdt(rows.get(0).get("mx"));
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ private LocalDateTime toLdt(Object v) {
|
|
|
+ if (v == null) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ if (v instanceof LocalDateTime) {
|
|
|
+ return (LocalDateTime) v;
|
|
|
+ }
|
|
|
+ if (v instanceof java.sql.Timestamp) {
|
|
|
+ return ((java.sql.Timestamp) v).toLocalDateTime();
|
|
|
+ }
|
|
|
+ if (v instanceof java.util.Date) {
|
|
|
+ return ((java.util.Date) v).toInstant().atZone(java.time.ZoneId.systemDefault()).toLocalDateTime();
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ private Integer resolveDataType(String warningCode) {
|
|
|
+ WarningType type = warningTypeMapper.selectOne(
|
|
|
+ new LambdaQueryWrapper<WarningType>().eq(WarningType::getWarningCode, warningCode));
|
|
|
+ return type == null ? null : type.getDataType();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 依据 warning_type.data_type 与 warning_threshold 判定告警：0=最小~最大，1=仅最小，2=仅最大，空=不配置阈值
|
|
|
+ */
|
|
|
+ private String evaluateAlarm(String deviceCode, WarningThreshold threshold, Integer dataType, Double value) {
|
|
|
+ if (value == null) {
|
|
|
+ return "未配置";
|
|
|
+ }
|
|
|
+ if (dataType == null) {
|
|
|
+ return "正常";
|
|
|
+ }
|
|
|
+ if (threshold == null) {
|
|
|
+ return "正常";
|
|
|
+ }
|
|
|
+ Double min = threshold.getMinValue();
|
|
|
+ Double max = threshold.getMaxValue();
|
|
|
+ boolean low = (dataType == 0 || dataType == 1) && min != null && value.compareTo(min) <= 0;
|
|
|
+ boolean high = (dataType == 0 || dataType == 2) && max != null && value.compareTo(max) >= 0;
|
|
|
+ if (high) {
|
|
|
+ return "偏高";
|
|
|
+ }
|
|
|
+ if (low) {
|
|
|
+ return "偏低";
|
|
|
+ }
|
|
|
+ return "正常";
|
|
|
+ }
|
|
|
+
|
|
|
+ private Double toDouble(String value) {
|
|
|
+ if (StrUtil.isBlank(value)) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return Double.parseDouble(value.trim());
|
|
|
+ } catch (NumberFormatException e) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
@Override
|
|
|
public List<Map<String, Object>> getFlowTrend(String equipmentId, LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
EquipmentBase device = selectMonitoringDevice(equipmentId, "flow");
|
|
|
@@ -325,8 +974,8 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
}
|
|
|
|
|
|
private Map<String, RadarData> selectLatestRadar(List<EquipmentBase> devices,
|
|
|
- LocalDateTime startTime, LocalDateTime endTime,
|
|
|
- Double flowMin, Double flowMax) {
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime,
|
|
|
+ Double flowMin, Double flowMax) {
|
|
|
List<String> deviceCodes = deviceCodes(devices);
|
|
|
if (deviceCodes.isEmpty()) {
|
|
|
return Collections.emptyMap();
|
|
|
@@ -445,6 +1094,63 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
return page;
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 每个设备仅取最新一条记录的实体分页：先按设备分组求 MAX(create_time)，按最新时间倒序分页，
|
|
|
+ * 再用 (设备, 最新时间) 一次性取回完整记录。latestOnly=false 时退化为原生分页（历史查询使用）。
|
|
|
+ */
|
|
|
+ private <T> Page<T> latestPerDevicePage(long pageNum, long pageSize,
|
|
|
+ BaseMapper<T> mapper,
|
|
|
+ String deviceCol, String deviceKey,
|
|
|
+ LocalDateTime startTime, LocalDateTime endTime) {
|
|
|
+ QueryWrapper<T> g = new QueryWrapper<T>()
|
|
|
+ .select(deviceCol + ", MAX(create_time) AS latest")
|
|
|
+ .ge(startTime != null, "create_time", startTime)
|
|
|
+ .le(endTime != null, "create_time", endTime);
|
|
|
+ if (StrUtil.isNotBlank(deviceKey)) {
|
|
|
+ g.like(deviceCol, deviceKey);
|
|
|
+ }
|
|
|
+ g.groupBy(deviceCol);
|
|
|
+ List<Map<String, Object>> groups = mapper.selectMaps(g);
|
|
|
+
|
|
|
+ groups.sort((a, b) -> {
|
|
|
+ LocalDateTime ta = toLdt(a.get("latest"));
|
|
|
+ LocalDateTime tb = toLdt(b.get("latest"));
|
|
|
+ if (ta == null && tb == null) return 0;
|
|
|
+ if (ta == null) return 1;
|
|
|
+ if (tb == null) return -1;
|
|
|
+ return tb.compareTo(ta);
|
|
|
+ });
|
|
|
+
|
|
|
+ long total = groups.size();
|
|
|
+ long safePageNum = Math.max(pageNum, 1);
|
|
|
+ long safePageSize = Math.max(pageSize, 1);
|
|
|
+ long from = (safePageNum - 1) * safePageSize;
|
|
|
+ long to = Math.min(from + safePageSize, total);
|
|
|
+ List<Map<String, Object>> pagePairs = from < total
|
|
|
+ ? new ArrayList<>(groups.subList((int) from, (int) to))
|
|
|
+ : new ArrayList<>();
|
|
|
+
|
|
|
+ Page<T> result = new Page<>(safePageNum, safePageSize, total);
|
|
|
+ if (pagePairs.isEmpty()) {
|
|
|
+ result.setRecords(new ArrayList<>());
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ QueryWrapper<T> q = new QueryWrapper<T>();
|
|
|
+ for (int i = 0; i < pagePairs.size(); i++) {
|
|
|
+ Map<String, Object> p = pagePairs.get(i);
|
|
|
+ String dev = (String) p.get(deviceCol);
|
|
|
+ LocalDateTime lt = toLdt(p.get("latest"));
|
|
|
+ if (i == 0) {
|
|
|
+ q.eq(deviceCol, dev).eq("create_time", lt);
|
|
|
+ } else {
|
|
|
+ q.or().eq(deviceCol, dev).eq("create_time", lt);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ result.setRecords(mapper.selectList(q));
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
private List<String> deviceCodes(List<EquipmentBase> devices) {
|
|
|
return devices.stream()
|
|
|
.map(EquipmentBase::getEquipmentCode)
|
|
|
@@ -520,7 +1226,9 @@ public class MonitoringServiceImpl implements MonitoringService {
|
|
|
return endTime != null ? null : LocalDateTime.now().minusHours(24);
|
|
|
}
|
|
|
|
|
|
- /** Restrict monitoring pages and trends to the canonical water-supply monitoring types. */
|
|
|
+ /**
|
|
|
+ * Restrict monitoring pages and trends to the canonical water-supply monitoring types.
|
|
|
+ */
|
|
|
private Set<Integer> selectMonitoringTypeIds(String monitorKind) {
|
|
|
Set<Integer> typeIds;
|
|
|
switch (monitorKind) {
|
