feat(alarm): 扩展报警指标并重构遥测报警评估逻辑

- 统一警告编码格式为连字符分隔（下划线改为连字符）
- 新增供水瞬时流量和供水流速两个报警指标枚举
- 雷达报警处理新增水位指标评估
- 重构遥测报警评估，将单一压力评估扩展为压力、瞬时流量、流速三项独立评估
- 提取 evaluateMetric 和 evaluateThreshold 方法，统一汇总后更新设备报警状态
- 调整 radar、telemetry、water-level 三个服务的 netty 端口分配
- 新增 RadarTestFrame 和 TelemetryTestFrame 测试报文构造夹具

pipe-network-service/zksy-system/src/main/java/com/zksy/base/alarm/enums/AlarmMetricSource.java

@@ -8,10 +8,12 @@ import java.util.Arrays;
 @Getter
 @AllArgsConstructor
 public enum AlarmMetricSource {
-    INSTANT_FLOW("WARN_INSTANT_FLOW", "radar_data", "meter1_instant_flow", "timestamp_since", "device_code", null),
-    FLOW_SPEED("WARN_FLOW_SPEED", "radar_data", "flow_speed", "timestamp_since", "device_code", null),
+    INSTANT_FLOW("WARN-INSTANT-FLOW", "radar_data", "meter1_instant_flow", "timestamp_since", "device_code", null),
+    FLOW_SPEED("WARN-FLOW-SPEED", "radar_data", "flow_speed", "timestamp_since", "device_code", null),
     WATER_LEVEL("WARN-WATER-LEVEL", "water_monitor_data", "water_level", "timestamp_since", "device_code", null),
     PRESSURE("WARN-PRESSURE", "telemetry_data", "pressure", "timestamp_since", "device_code", null),
+    SUPPLY_INSTANT_FLOW("WARN-INSTANT-FLOW-SUPPLY", "telemetry_data", "meter1_instant_flow", "timestamp_since", "device_code", null),
+    SUPPLY_FLOW_SPEED("WARN-FLOW-SPEED-SUPPLY", "telemetry_data", "flow_speed", "timestamp_since", "device_code", null),
     SUSPENDED_SOLIDS("WARN_SUSPENDED_SOLIDS", "e_real_time_data", "float_value", "create_time", "device_id", 1),
     COD("WARN_COD", "e_real_time_data", "hum", "create_time", "device_id", 2),
     AMMONIA_NITROGEN("WARN_AMMONIA_NITROGEN", "e_real_time_data", "hum", "create_time", "device_id", 3),

pipe-network-service/zksy-system/src/test/java/com/zksy/base/alarm/mapper/AlarmOperationsMapperTest.java

@@ -74,7 +74,7 @@ class AlarmOperationsMapperTest {
         assertTrue(rendered.get("countByWarningType").contains("wt.warning_name"));
         assertTrue(rendered.get("countByDeviceType").contains("t.type_name"));
         assertTrue(rendered.get("selectDashboardTrend").contains("a.create_time >= ? AND a.create_time < ?"));
-        query.setWarningCode("WARN_INSTANT_FLOW"); query.setAlarmStatus(0);
+        query.setWarningCode("WARN-INSTANT-FLOW"); query.setAlarmStatus(0);
         assertTrue(sql(config,"selectPage",params).contains("a.warning_code = ?"));
         assertTrue(sql(config,"selectPage",params).contains("a.alarm_status = ?"));
         params.put("typeIds",Collections.emptyList()); assertTrue(sql(config,"selectPage",params).contains("1 = 0"));

pipe-network-service/zksy-system/src/test/java/com/zksy/base/alarm/service/AlarmOperationsServiceTest.java

@@ -32,7 +32,7 @@ class AlarmOperationsServiceTest {
         }
         when(dictionaries.selectDictDataList(any())).thenReturn(levels);
         query.setRootTypeId(2);
-        alarm.setId("a1"); alarm.setDeviceCode("D1"); alarm.setWarningCode("WARN_INSTANT_FLOW");
+        alarm.setId("a1"); alarm.setDeviceCode("D1"); alarm.setWarningCode("WARN-INSTANT-FLOW");
         alarm.setCreateTime(LocalDateTime.of(2026,1,10,12,0));
         alarm.setActualValue(BigDecimal.ZERO);
         alarm.setAlarmStatus(1); alarm.setAuditStatus(1); alarm.setAlarmLevel(2); alarm.setWorkOrderId(9L); alarm.setOrderStatus(3);

radar-service/src/main/java/com/zksy/radar/utils/MessageHandler.java

@@ -213,8 +213,9 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
     private void handleRadarAlarm(RadarData resultData) {
         String deviceCode = resultData.getDeviceCode();
         Integer status = alarmStatus(
-                handleAlarm(deviceCode, "WARN_INSTANT_FLOW", "瞬时流量", resultData.getMeter1InstantFlow()),
-                handleAlarm(deviceCode, "WARN_FLOW_SPEED", "流速", resultData.getFlowSpeed()));
+                handleAlarm(deviceCode, "WARN-INSTANT-FLOW", "瞬时流量", resultData.getMeter1InstantFlow()),
+                handleAlarm(deviceCode, "WARN-FLOW-SPEED", "流速", resultData.getFlowSpeed()),
+                handleAlarm(deviceCode, "WARN-WATER-LEVEL", "水位", resultData.getWaterLevel()));
         if (status != null) {
             updateDeviceAlarmStatus(deviceCode, status);
         }

radar-service/src/main/resources/bootstrap.yaml

@@ -32,4 +32,4 @@ spring:
           - dataId: zksy-shared-jdbc.yaml
           - dataId: zksy-shared-log.yaml
 netty:
-  port: 30006
+  port: 30005

radar-service/src/test/java/com/zksy/radar/utils/RadarTestFrame.java

@@ -0,0 +1,54 @@
+package com.zksy.radar.utils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+/** 测试夹具：原有雷达 0x31 报文，仍含流量、流速和水位，不扩展协议。 */
+final class RadarTestFrame {
+    private RadarTestFrame() { }
+
+    static byte[] report(String sourceAddr, float flow, float speed, float waterLevel) throws IOException {
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+        DataOutputStream frame = new DataOutputStream(bytes);
+        frame.write(new byte[]{0x12, 0x34, 0x56});
+        frame.writeShort(0);
+        frame.writeByte(1);
+        frame.writeByte(0x31);
+        frame.writeByte(sourceAddr.length());
+        for (int i = 0; i < sourceAddr.length(); i += 2) {
+            frame.writeByte(Integer.parseInt(sourceAddr.substring(i, i + 2), 16));
+        }
+        frame.writeByte(2);
+        frame.writeByte(1);
+        int innerStart = bytes.size();
+        frame.writeByte(1);
+        frame.writeByte(0x2C);
+        frame.write(new byte[6]);
+        frame.writeByte(1);
+        frame.writeShort(0x003F);
+        frame.writeShort(1200);
+        frame.writeByte(0); // 非最后一包，不产生延迟结束应答。
+        frame.writeByte(1);
+        frame.writeByte(1);
+        frame.writeByte(20);
+        frame.write(new byte[3]);
+        frame.writeInt(800000000);
+        frame.writeFloat(100);
+        frame.writeFloat(flow);
+        frame.writeFloat(speed);
+        frame.writeFloat(waterLevel);
+        frame.writeInt(0);
+        byte[] data = bytes.toByteArray();
+        int crc = ProtocolUtils.calculateCRC16(java.util.Arrays.copyOfRange(data, innerStart, data.length));
+        frame.writeByte(crc & 0xFF);
+        frame.writeByte((crc >>> 8) & 0xFF);
+        byte[] body = bytes.toByteArray();
+        int length = body.length + 1;
+        body[3] = (byte) (length >>> 8);
+        body[4] = (byte) length;
+        byte[] result = java.util.Arrays.copyOf(body, length);
+        result[length - 1] = ProtocolUtils.calculateXorCheck(body);
+        return result;
+    }
+}

telemetry-service/src/main/java/com/zksy/telemetry/utils/MessageHandler.java

@@ -115,7 +115,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
                     }
                 });
                 run(deviceCode, "在线状态同步", () -> deviceOfflineCheckTask.markDeviceOnline(deviceCode));
-                run(deviceCode, "压力报警预警评估", () -> evaluatePressure(resultData, resolveDeviceCode(resultData)));
+                run(deviceCode, "报警预警评估", () -> evaluateTelemetry(resultData));
 
 			} else if (frameType == 0x34) {
 				/*logger.info("收到结束通讯帧，准备发送回应包");
@@ -140,28 +140,40 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
     }
 
     /**
-     * 平升遥测终端的供水压力使用数值阈值判断 WARN-PRESSURE。
-     * 压力值来自本次采集报文，阈值范围来自 warning_threshold 表。
+     * 三项指标独立评估，只汇总一次设备报警状态，避免正常指标覆盖其他指标的报警。
      */
-    private void evaluatePressure(TelemetryData data, String deviceCode) {
+    private void evaluateTelemetry(TelemetryData data) {
+        String deviceCode = resolveDeviceCode(data);
+        Integer status = alarmStatus(
+                evaluateMetric(deviceCode, "WARN-PRESSURE", "供水压力", data.getPressure()),
+                evaluateMetric(deviceCode, "WARN-INSTANT-FLOW-SUPPLY", "供水瞬时流量", data.getMeter1InstantFlow()),
+                evaluateMetric(deviceCode, "WARN-FLOW-SPEED-SUPPLY", "供水流速", data.getFlowSpeed()));
+        if (status != null) {
+            run(deviceCode, "报警状态同步", () -> deviceNotifier.updateAlarmStatus(deviceCode, status));
+        }
+    }
+
+    private Boolean evaluateMetric(String deviceCode, String warningCode, String warningType, String value) {
+        return evaluate(deviceCode, warningCode, () -> evaluateThreshold(deviceCode, warningCode, warningType, value));
+    }
+
+    /** 数值直接使用本次采集字段，范围按设备源地址和各自指标编码查询阈值表。 */
+    private boolean evaluateThreshold(String deviceCode, String warningCode, String warningType, String value) {
         if (warningThresholdService == null || alarmDataService == null || earlyWarningAutoService == null) {
-            throw new IllegalStateException("压力告警依赖未注入");
+            throw new IllegalStateException("阈值告警依赖未注入");
         }
-        if (data.getPressure() == null || data.getPressure().trim().isEmpty()) {
-            throw new IllegalArgumentException("采集报文缺少压力值");
+        if (value == null || value.trim().isEmpty()) {
+            throw new IllegalArgumentException("采集报文缺少指标值: " + warningCode);
         }
-        BigDecimal actual = new BigDecimal(data.getPressure().trim());
-        String warningCode = "WARN-PRESSURE";
+        BigDecimal actual = new BigDecimal(value.trim());
         WarningThreshold threshold = warningThresholdService
                 .getWarningThresholdByDeviceAndCode(deviceCode, warningCode);
         if (threshold == null || (threshold.getMinValue() == null && threshold.getMaxValue() == null)) {
-            log.warn("未配置供水压力阈值，跳过 WARN-PRESSURE 评估: deviceCode={}", deviceCode);
-            return;
+            throw new IllegalStateException("阈值缺失，不能判断设备已恢复正常: " + warningCode);
         }
 
         BigDecimal min = threshold.getMinValue() == null ? null : BigDecimal.valueOf(threshold.getMinValue());
         BigDecimal max = threshold.getMaxValue() == null ? null : BigDecimal.valueOf(threshold.getMaxValue());
-        String warningType = "压力预警";
         boolean over = (min != null && actual.compareTo(min) <= 0)
                 || (max != null && actual.compareTo(max) >= 0);
         if (over) {
@@ -177,7 +189,6 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
                 alarm.setDeviationRatio(AlarmLevelUtil.calculateDeviationRatio(actual, min, max));
                 alarm.setAlarmStatus(0);
                 alarm.setCreateTime(java.time.LocalDateTime.now());
-                alarm.setCreateTime(java.time.LocalDateTime.now());
                 requireSaved(alarmDataService.saveAlarmData(alarm));
             });
             run(deviceCode, warningCode + ":关闭预警", () ->
@@ -192,7 +203,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
                     earlyWarningAutoService.checkAndGenerate(deviceCode, warningType, warningCode,
                             min, max, actual, null));
         }
-        run(deviceCode, "报警状态同步", () -> deviceNotifier.updateAlarmStatus(deviceCode, over ? 1 : 0));
+        return over;
     }
 
     private String resolveDeviceCode(TelemetryData data) {
@@ -202,7 +213,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
         if (data.getDeviceCode() != null && !data.getDeviceCode().trim().isEmpty()) {
             return data.getDeviceCode().trim();
         }
-        throw new IllegalArgumentException("供水压力报文缺少设备源地址");
+        throw new IllegalArgumentException("供水遥测报文缺少设备源地址");
     }
 
     private void sendAlarmSms(String deviceCode, String warningType, BigDecimal actual,
@@ -221,7 +232,7 @@ public class MessageHandler extends ChannelInboundHandlerAdapter {
         params.put("location", "当前值：" + actual + "，正常范围：" + min + "~" + max);
         Map<String, Boolean> result = smsUtil.sendBatchSms(phones, params.toJSONString());
         long successCount = result == null ? 0 : result.values().stream().filter(Boolean.TRUE::equals).count();
-        log.info("一级压力告警短信发送完成: deviceCode={}, success={}/{}", deviceCode, successCount, phones.size());
+        log.info("一级告警短信发送完成: deviceCode={}, warningType={}, success={}/{}", deviceCode, warningType, successCount, phones.size());
     }
 
 

telemetry-service/src/main/resources/bootstrap.yaml

@@ -32,4 +32,4 @@ spring:
           - dataId: zksy-shared-jdbc.yaml
           - dataId: zksy-shared-log.yaml
 netty:
-  port: 30001
+  port: 30006

telemetry-service/src/test/java/com/zksy/telemetry/utils/TelemetryTestFrame.java

@@ -0,0 +1,57 @@
+package com.zksy.telemetry.utils;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+
+/** 现有供水遥测 0x31 报文：使用原有字段顺序、CRC16 和外层异或校验。 */
+final class TelemetryTestFrame {
+    private TelemetryTestFrame() { }
+
+    static byte[] report(String sourceAddr, float pressure, float flow, float speed) throws IOException {
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+        DataOutputStream frame = new DataOutputStream(bytes);
+        frame.write(new byte[]{0x12, 0x34, 0x56});
+        frame.writeShort(0);
+        frame.writeByte(1);
+        frame.writeByte(0x31);
+        frame.writeByte(sourceAddr.length());
+        for (int i = 0; i < sourceAddr.length(); i += 2) {
+            frame.writeByte(Integer.parseInt(sourceAddr.substring(i, i + 2), 16));
+        }
+        frame.writeByte(2);
+        frame.writeByte(1);
+        int innerStart = bytes.size();
+        frame.writeByte(1);
+        frame.writeByte(0x2C);
+        frame.write(new byte[6]);
+        frame.writeByte(1);
+        frame.writeShort(0x00FF);
+        frame.writeShort(1200);
+        frame.writeByte(0); // 非末包，避免测试产生延迟结束应答。
+        frame.writeByte(1);
+        frame.writeByte(1);
+        frame.writeByte(20);
+        frame.write(new byte[3]);
+        frame.writeInt(800000000);
+        frame.writeFloat(100);
+        frame.writeFloat(110);
+        frame.writeFloat(10);
+        frame.writeFloat(flow);
+        frame.writeFloat(speed);
+        frame.writeFloat(pressure);
+        frame.writeFloat(20);
+        frame.writeInt(0);
+        int crc = ProtocolUtils.calculateCRC16(Arrays.copyOfRange(bytes.toByteArray(), innerStart, bytes.size()));
+        frame.writeByte(crc & 0xFF);
+        frame.writeByte((crc >>> 8) & 0xFF);
+        byte[] body = bytes.toByteArray();
+        int length = body.length + 1;
+        body[3] = (byte) (length >>> 8);
+        body[4] = (byte) length;
+        byte[] result = Arrays.copyOf(body, length);
+        result[length - 1] = ProtocolUtils.calculateXorCheck(body);
+        return result;
+    }
+}

water-level-service/src/main/resources/bootstrap.yaml

@@ -32,4 +32,4 @@ spring:
           - dataId: zksy-shared-jdbc.yaml
           - dataId: zksy-shared-log.yaml
 netty:
-  port: 30005
+  port: 30001 #弃用