fix(cockpit): 按采集源读取设备最新指标并保留节点时间

修复驾驶舱仅按行业读通用表导致环境等设备有源数据却返回空指标的问题。
按设备类型的采集表白名单复用现有查询；环境各节点独立取最新记录，缺值不补零。
保留入库与报告时间来源，历史趋势不扩展。
隔离离线回归30项通过；未构建、部署、重启或运行浏览器验收。

pipe-network-service/zksy-admin/src/test/java/com/zksy/web/controller/cockpit/BusinessCockpitCollectedMetricsTest.java

@@ -0,0 +1,258 @@
+package com.zksy.web.controller.cockpit;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.zksy.base.domain.EquipmentBase;
+import com.zksy.base.domain.EquipmentType;
+import com.zksy.base.environment.domain.ERealTimeData;
+import com.zksy.base.environment.service.ERealTimeDataService;
+import com.zksy.base.perception.mapper.PerceptionMapper;
+import com.zksy.base.gas.service.GasMonitorDataService;
+import com.zksy.base.service.EquipmentBaseService;
+import com.zksy.base.service.EquipmentTypeService;
+import com.zksy.base.water.service.WaterMonitorDataService;
+import com.zksy.cockpit.type.*;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.math.BigDecimal;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/** Public current-metrics seam; only database-facing dependencies are mocked. */
+class BusinessCockpitCollectedMetricsTest {
+    private final EquipmentBaseService devices = mock(EquipmentBaseService.class);
+    private final EquipmentTypeService types = mock(EquipmentTypeService.class);
+    private final GasMonitorDataService gas = mock(GasMonitorDataService.class);
+    private final WaterMonitorDataService water = mock(WaterMonitorDataService.class);
+    private final ERealTimeDataService environment = mock(ERealTimeDataService.class);
+    private final PerceptionMapper perception = mock(PerceptionMapper.class);
+    private final BusinessCockpitTypeMetricsAdapter adapter = new BusinessCockpitTypeMetricsAdapter(
+            devices, types, gas, water, new CollectedMetricsReader(environment, perception));
+    private final CockpitTypeMetricsService service = new CockpitTypeMetricsService(adapter,
+            Clock.fixed(Instant.parse("2026-10-30T04:00:00Z"), ZoneOffset.UTC));
+
+    @Test
+    void environmentUsesLatestRecordForEachNodeAndKeepsOldNullAndZero() {
+        device("21119740", "e_real_time_data");
+        when(environment.getLatestByDeviceNode(21119740, 1)).thenReturn(node(1, "0", 999f, 21f, "2026-10-30T11:59:50"));
+        // node 2 has not reported recently; a recent global 50-row window would lose this factor.
+        when(environment.getLatestByDeviceNode(21119740, 2)).thenReturn(node(2, "0.00", 17.3f, 22.2f, "2026-10-23T17:04:29"));
+        when(environment.getLatestByDeviceNode(21119740, 3)).thenReturn(node(3, "999", null, null, "2026-10-30T11:59:55"));
+        when(environment.getLatestByDeviceNode(21119740, 4)).thenReturn(node(4, "999", 420f, 23f, "2026-10-30T11:59:56"));
+        when(environment.getLatestByDeviceNode(21119740, 5)).thenReturn(node(5, "999", 7.2f, 24f, "2026-10-30T11:59:57"));
+
+        CockpitTypeMetricsResponse response = service.loadCurrent("test", " 21119740 ");
+        assertEquals("21119740", response.getDeviceCode());
+        assertEquals("STALE", response.getState());
+        assertEquals(10, response.getMetrics().size());
+        assertMetric(response, "ENV_SUSPENDED_SOLIDS", "0", "mg/L", "CURRENT");
+        assertMetric(response, "ENV_COD", "17.3", "mg/L", "STALE");
+        assertMetric(response, "ENV_AMMONIA_NITROGEN", null, "mg/L", "CURRENT");
+        assertMetric(response, "ENV_CONDUCTIVITY", "420.0", "μS/cm", "CURRENT");
+        assertMetric(response, "ENV_PH", "7.2", "", "CURRENT");
+        assertMetric(response, "ENV_NODE_2_TEMPERATURE", "22.2", "℃", "STALE");
+        assertEquals("2026-10-23T17:04:29+08:00", metric(response, "ENV_COD").getSampledAt());
+        assertTrue(response.getMetrics().stream().allMatch(m -> "CREATED_AT".equals(m.getTimeSource())));
+        assertTrue(response.getMetrics().stream().allMatch(m -> "UNKNOWN".equals(m.getStatus())));
+        for (int node = 1; node <= 5; node++) verify(environment).getLatestByDeviceNode(21119740, node);
+        verifyNoMoreInteractions(environment);
+        verifyNoInteractions(gas, water, perception);
+    }
+
+    @ParameterizedTest
+    @NullAndEmptySource
+    @ValueSource(strings = {"untrusted_table", "gas_monitor_data; DROP TABLE equipment_base", "app_user.gas_monitor_data"})
+    void unknownCollectionSourceNeverFallsBackToIndustryTable(String source) {
+        device("21119740", source);
+        assertEquals("EMPTY", service.loadCurrent("test", "21119740").getState());
+        verifyNoInteractions(gas, water, environment, perception);
+    }
+
+    @ParameterizedTest
+    @CsvSource({
+            "firefighting_pressure,telemetering_station,pressure_value,FIREFIGHTING_PRESSURE,MPa",
+            "jg_device_data,imei_card_number,tilt_angle,MANHOLE_TILT_ANGLE,°",
+            "radar_data,device_code,meter1_instant_flow,RADAR_INSTANT_FLOW,m³/h",
+            "telemetry_data,device_code,pressure,TELEMETRY_PRESSURE,MPa",
+            "noise_info,encode,rms,NOISE_RMS,''"
+    })
+    void readsOnlyWhitelistedActualSourceAndDevice(String table, String idColumn, String field, String metricCode, String unit) {
+        device("REAL-CODE-7", table);
+        Map<String, Object> row = new HashMap<>();
+        row.put(idColumn, "REAL-CODE-7");
+        row.put(field, "0.00");
+        row.put("create_time", java.sql.Timestamp.valueOf("2026-10-30 11:59:55"));
+        when(perception.selectLatestRecords(table, idColumn, "REAL-CODE-7", 1)).thenReturn(Collections.singletonList(row));
+
+        CockpitTypeMetricsResponse response = service.loadCurrent("test", "REAL-CODE-7");
+        assertEquals("CURRENT", response.getState());
+        assertMetric(response, metricCode, "0.00", unit, "CURRENT");
+        assertEquals("UNKNOWN", metric(response, metricCode).getStatus());
+        assertEquals("CREATED_AT", metric(response, metricCode).getTimeSource());
+        verify(perception).selectLatestRecords(table, idColumn, "REAL-CODE-7", 1);
+        verifyNoMoreInteractions(perception);
+        verifyNoInteractions(gas, water, environment);
+    }
+
+    @ParameterizedTest
+    @CsvSource({
+            "firefighting_pressure,telemetering_station,observed_time,FIREFIGHTING_PRESSURE",
+            "radar_data,device_code,timestamp_since,RADAR_INSTANT_FLOW",
+            "telemetry_data,device_code,timestamp_since,TELEMETRY_PRESSURE",
+            "noise_info,encode,ts,NOISE_RMS"
+    })
+    void retainsOldDeviceReportedTimeEvenWhenReceiptIsRecent(String table, String idColumn, String timeColumn, String metricCode) {
+        device("REAL-CODE-7", table);
+        Map<String, Object> row = new HashMap<>();
+        row.put(timeColumn, "2026-10-23 17:04:29");
+        row.put("create_time", LocalDateTime.parse("2026-10-30T11:59:55"));
+        when(perception.selectLatestRecords(table, idColumn, "REAL-CODE-7", 1)).thenReturn(Collections.singletonList(row));
+        CockpitTypeMetricsResponse response = service.loadCurrent("test", "REAL-CODE-7");
+        assertEquals("STALE", response.getState());
+        CockpitMetric metric = metric(response, metricCode);
+        assertEquals("2026-10-23T17:04:29+08:00", metric.getSampledAt());
+        assertEquals("REPORTED_AT", metric.getTimeSource());
+        assertNull(metric.getValue());
+    }
+
+    @Test
+    void latestWideRecordKeepsNullAndUnparseableFactorsInsteadOfLookingBack() {
+        device("REAL-CODE-7", "telemetry_data");
+        Map<String, Object> row = new HashMap<>();
+        row.put("pressure", null);
+        row.put("meter1_instant_flow", "not-a-number");
+        row.put("flow_speed", "0");
+        row.put("temperature", "21.5");
+        row.put("create_time", LocalDateTime.parse("2026-10-30T11:59:55"));
+        when(perception.selectLatestRecords("telemetry_data", "device_code", "REAL-CODE-7", 1))
+                .thenReturn(Collections.singletonList(row));
+        CockpitTypeMetricsResponse response = service.loadCurrent("test", "REAL-CODE-7");
+        assertMetric(response, "TELEMETRY_PRESSURE", null, "MPa", "CURRENT");
+        assertMetric(response, "TELEMETRY_INSTANT_FLOW", null, "m³/h", "CURRENT");
+        assertMetric(response, "TELEMETRY_FLOW_SPEED", "0", "m/s", "CURRENT");
+        assertMetric(response, "TELEMETRY_TEMPERATURE", "21.5", "℃", "CURRENT");
+        verify(perception).selectLatestRecords("telemetry_data", "device_code", "REAL-CODE-7", 1);
+        verifyNoMoreInteractions(perception);
+    }
+
+    @Test
+    void onlyExplicitManholeAlarmBitsProduceAlarmOrNormalStatus() {
+        device("REAL-CODE-7", "jg_device_data");
+        Map<String, Object> row = new HashMap<>();
+        row.put("water_infiltration_alarm_status", "1");
+        row.put("water_level_alarm_status", "0");
+        row.put("tilt_angle", "0");
+        row.put("battery_level", "85");
+        row.put("temperature_value", "23");
+        row.put("create_time", LocalDateTime.parse("2026-10-30T11:59:55"));
+        when(perception.selectLatestRecords("jg_device_data", "imei_card_number", "REAL-CODE-7", 1))
+                .thenReturn(Collections.singletonList(row));
+        CockpitTypeMetricsResponse response = service.loadCurrent("test", "REAL-CODE-7");
+        assertEquals("ALARM", metric(response, "MANHOLE_WATER_INFILTRATION_ALARM").getStatus());
+        assertEquals("NORMAL", metric(response, "MANHOLE_WATER_LEVEL_ALARM").getStatus());
+        assertEquals("UNKNOWN", metric(response, "MANHOLE_TILT_ANGLE").getStatus());
+        assertMetric(response, "MANHOLE_BATTERY", "85", "%", "CURRENT");
+        assertMetric(response, "MANHOLE_TEMPERATURE", "23", "℃", "CURRENT");
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"CN-ENV-21119740", "021119740", "21119740x", "9999999999999"})
+    void doesNotCoerceAnotherDeviceCodeIntoAnEnvironmentId(String code) {
+        device(code, "e_real_time_data");
+        assertEquals("EMPTY", service.loadCurrent("test", code).getState());
+        verifyNoInteractions(environment, perception, gas, water);
+    }
+
+    @Test
+    void noCollectedRowsRemainEmptyWithoutSynthesizingMetrics() {
+        device("21119740", "e_real_time_data");
+        assertEquals("EMPTY", service.loadCurrent("test", "21119740").getState());
+        for (int node = 1; node <= 5; node++) verify(environment).getLatestByDeviceNode(21119740, node);
+        device("REAL-CODE-7", "radar_data");
+        assertEquals("EMPTY", service.loadCurrent("test", "REAL-CODE-7").getState());
+    }
+
+    @Test
+    void timestampWithExplicitOffsetKeepsItsInstantInBusinessZone() {
+        device("REAL-CODE-7", "radar_data");
+        Map<String, Object> row = new HashMap<>();
+        row.put("timestamp_since", "2026-10-30T03:59:55Z");
+        when(perception.selectLatestRecords("radar_data", "device_code", "REAL-CODE-7", 1))
+                .thenReturn(Collections.singletonList(row));
+        CockpitMetric metric = metric(service.loadCurrent("test", "REAL-CODE-7"), "RADAR_INSTANT_FLOW");
+        assertEquals("2026-10-30T11:59:55+08:00", metric.getSampledAt());
+        assertEquals("CURRENT", metric.getFreshness());
+        row.put("timestamp_since", java.time.OffsetDateTime.parse("2026-10-30T03:59:55Z"));
+        assertEquals("2026-10-30T11:59:55+08:00",
+                metric(service.loadCurrent("test", "REAL-CODE-7"), "RADAR_INSTANT_FLOW").getSampledAt());
+    }
+
+    @Test
+    void missingOrUnspecifiedEpochTimeFailsClosedInsteadOfClaimingCurrent() {
+        device("REAL-CODE-7", "radar_data");
+        Map<String, Object> row = new HashMap<>();
+        row.put("meter1_instant_flow", "3.5");
+        when(perception.selectLatestRecords("radar_data", "device_code", "REAL-CODE-7", 1))
+                .thenReturn(Collections.singletonList(row));
+        assertThrows(com.zksy.cockpit.foundation.CockpitReadException.class,
+                () -> service.loadCurrent("test", "REAL-CODE-7"));
+        row.put("timestamp_since", 1793332795000L);
+        assertThrows(com.zksy.cockpit.foundation.CockpitReadException.class,
+                () -> service.loadCurrent("test", "REAL-CODE-7"));
+    }
+
+    private void device(String code, String source) {
+        EquipmentBase device = new EquipmentBase();
+        device.setEquipmentId("database-primary-key-must-not-be-used");
+        device.setEquipmentCode(code);
+        device.setEquipmentTypeId(15);
+        EquipmentType root = new EquipmentType();
+        root.setId(3);
+        root.setTypeName("燃气");
+        root.setParentId(0);
+        EquipmentType type = new EquipmentType();
+        type.setId(15);
+        type.setParentId(3);
+        type.setTypeName("实际采集类型");
+        type.setOrigCollTable(source);
+        when(devices.list(any(LambdaQueryWrapper.class))).thenReturn(Collections.singletonList(device));
+        when(types.list()).thenReturn(Arrays.asList(root, type));
+    }
+
+    private static ERealTimeData node(int node, String solids, Float hum, Float tem, String at) {
+        ERealTimeData row = new ERealTimeData();
+        row.setDeviceId(21119740);
+        row.setNodeId(node);
+        row.setFloatValue(solids);
+        row.setHum(hum);
+        row.setTem(tem);
+        row.setCreateTime(LocalDateTime.parse(at));
+        return row;
+    }
+
+    private static CockpitMetric metric(CockpitTypeMetricsResponse response, String code) {
+        return response.getMetrics().stream().filter(m -> code.equals(m.getMetricCode())).findFirst()
+                .orElseThrow(() -> new AssertionError("missing metric: " + code));
+    }
+
+    private static void assertMetric(CockpitTypeMetricsResponse response, String code, String value, String unit, String freshness) {
+        CockpitMetric metric = metric(response, code);
+        if (value == null) assertNull(metric.getValue());
+        else assertEquals(0, new BigDecimal(value).compareTo(metric.getValue()), code);
+        assertEquals(unit, metric.getUnit());
+        assertEquals(freshness, metric.getFreshness());
+    }
+}

pipe-network-service/zksy-admin/src/test/java/com/zksy/web/controller/cockpit/BusinessCockpitTypeMetricsAdapterTest.java

@@ -10,6 +10,9 @@ import com.zksy.base.service.EquipmentBaseService;
 import com.zksy.base.service.EquipmentTypeService;
 import com.zksy.base.water.domain.WaterMonitorData;
 import com.zksy.base.water.service.WaterMonitorDataService;
+import com.zksy.cockpit.type.CollectedMetricsReader;
+import com.zksy.base.environment.service.ERealTimeDataService;
+import com.zksy.base.perception.mapper.PerceptionMapper;
 import com.zksy.cockpit.type.BusinessCockpitTypeMetricsAdapter;
 import com.zksy.cockpit.type.CockpitMetricSnapshot;
 import com.zksy.cockpit.type.CockpitTypeMetricsSnapshot;
@@ -47,12 +50,15 @@ class BusinessCockpitTypeMetricsAdapterTest {
         gasMonitorDataService = mock(GasMonitorDataService.class);
         waterMonitorDataService = mock(WaterMonitorDataService.class);
         adapter = new BusinessCockpitTypeMetricsAdapter(
-                equipmentBaseService, equipmentTypeService, gasMonitorDataService, waterMonitorDataService);
+                equipmentBaseService, equipmentTypeService, gasMonitorDataService, waterMonitorDataService,
+                new CollectedMetricsReader(mock(ERealTimeDataService.class), mock(PerceptionMapper.class)));
 
         EquipmentType gasRoot = type(3, "燃气", 0);
         EquipmentType gasSensor = type(11, "燃气浓度计", 3);
+        gasSensor.setOrigCollTable("gas_monitor_data");
         EquipmentType waterRoot = type(1, "供水", 0);
         EquipmentType waterMeter = type(5, "供水流量计", 1);
+        waterMeter.setOrigCollTable("water_monitor_data");
         when(equipmentTypeService.list()).thenReturn(
                 Arrays.asList(gasRoot, gasSensor, waterRoot, waterMeter));
     }
@@ -78,6 +84,7 @@ class BusinessCockpitTypeMetricsAdapterTest {
         assertEquals("%LEL", concentration.getUnit());
         assertEquals("ALARM", concentration.getStatus());
         assertEquals("2026-09-02T09:00:00+08:00", concentration.getSampledAt());
+        assertEquals("REPORTED_AT", concentration.getTimeSource());
     }
 
     @Test
@@ -145,6 +152,7 @@ class BusinessCockpitTypeMetricsAdapterTest {
         assertEquals(new BigDecimal("3.2"), level.getValue());
         assertEquals("UNKNOWN", level.getStatus());
         assertEquals("2026-09-02T10:00:00+08:00", level.getSampledAt());
+        assertEquals("CREATED_AT", level.getTimeSource());
         assertEquals(new BigDecimal("12.6"), snapshot.getMetrics().get(1).getValue());
     }
 

pipe-network-service/zksy-system/src/main/java/com/zksy/cockpit/type/BusinessCockpitTypeMetricsAdapter.java

@@ -32,19 +32,13 @@ import java.util.stream.Collectors;
 /**
  * 类型指标真实数据适配器（BT-5）。
  *
- * <p>指标采样来源按设备行业分流：
- * <ul>
- *   <li>燃气（{@code GAS}）：{@code gas_monitor_data}，现网惯例以
- *       {@code mac_address} 存设备编码（见 {@code GasDashboardServiceImpl}），
- *       指标为气体浓度/%LEL、电池电量/%、信号强度/dBm；报警位取自帧内 {@code alarm_info}；</li>
- *   <li>供水/排水（{@code WATER_SUPPLY}、{@code DRAINAGE}）：{@code water_monitor_data}，
- *       {@code device_code} 关联设备编码，指标为水位、电源电压、信号质量；
- *       水位/电压为字符串字段，可解析成数值才上报，解析失败按缺失处理；</li>
- *   <li>其他行业或类型不可推导：不臆造数据源，返回空指标（服务层按 EMPTY 处理）。</li>
- * </ul>
+ * <p>当前指标只按设备自身类型的 {@code orig_coll_table} 选择采集源：
+ * 气体、水位保留专用取数；环境节点和其他白名单采集表由 {@link CollectedMetricsReader} 读取。
+ * 未知/未配置源返回空指标，不按所属行业猜表。设备主键不作为采集编码。
  *
- * <p><b>不臆造</b>：水位帧无报警位字段 → 状态为 {@code UNKNOWN}；气体帧有报警位 → {@code ALARM}/{@code NORMAL}；
- * 无报告时间回退入库时间；不补造趋势与阈值判定。
+ * <p>最新整行里的缺值保持 null；只有真实报警位才判定报警，其他指标状态为 UNKNOWN。
+ * {@code sampledAt} 为兼容字段，实际时间来源通过 {@code timeSource} 区分报告与入库。
+ * 历史接口保持原有行业、分页、时间范围和指标契约，不在本次扩展。
  */
 @Component
 @Primary
@@ -57,16 +51,19 @@ public class BusinessCockpitTypeMetricsAdapter implements CockpitTypeMetricsAdap
     private final EquipmentTypeService equipmentTypeService;
     private final GasMonitorDataService gasMonitorDataService;
     private final WaterMonitorDataService waterMonitorDataService;
+    private final CollectedMetricsReader collectedMetricsReader;
 
     public BusinessCockpitTypeMetricsAdapter(
             EquipmentBaseService equipmentBaseService,
             EquipmentTypeService equipmentTypeService,
             GasMonitorDataService gasMonitorDataService,
-            WaterMonitorDataService waterMonitorDataService) {
+            WaterMonitorDataService waterMonitorDataService,
+            CollectedMetricsReader collectedMetricsReader) {
         this.equipmentBaseService = equipmentBaseService;
         this.equipmentTypeService = equipmentTypeService;
         this.gasMonitorDataService = gasMonitorDataService;
         this.waterMonitorDataService = waterMonitorDataService;
+        this.collectedMetricsReader = collectedMetricsReader;
     }
 
     @Override
@@ -79,14 +76,17 @@ public class BusinessCockpitTypeMetricsAdapter implements CockpitTypeMetricsAdap
         Map<Integer, EquipmentType> types = equipmentTypes();
         String typeCode = typeCodeOf(device, types);
         String typeName = typeNameOf(device, types);
-        String industry = industryOf(device, types);
-        if ("GAS".equals(industry)) {
+        EquipmentType type = types.get(device.getEquipmentTypeId());
+        String source = type == null || type.getOrigCollTable() == null
+                ? null : type.getOrigCollTable().trim();
+        if ("gas_monitor_data".equals(source)) {
             return gasCurrent(device, code, typeCode, typeName);
         }
-        if ("WATER_SUPPLY".equals(industry) || "DRAINAGE".equals(industry)) {
+        if ("water_monitor_data".equals(source)) {
             return waterCurrent(device, code, typeCode, typeName);
         }
-        return new CockpitTypeMetricsSnapshot(code, typeCode, typeName, Collections.emptyList());
+        return new CockpitTypeMetricsSnapshot(code, typeCode, typeName,
+                collectedMetricsReader.loadCurrent(source, code));
     }
 
     @Override
@@ -124,13 +124,14 @@ public class BusinessCockpitTypeMetricsAdapter implements CockpitTypeMetricsAdap
         }
         String sampledAt = offsetTime(sampledAt(latest.getReportTime(), latest.getCreateTime()));
         String status = gasStatus(latest);
+        String timeSource = latest.getReportTime() == null ? "CREATED_AT" : "REPORTED_AT";
         List<CockpitMetricSnapshot> metrics = new ArrayList<>();
-        addMetric(metrics, "GAS_CONCENTRATION", "气体浓度", latest.getGasConcentration(), "%LEL", status, sampledAt);
-        addMetric(metrics, "GAS_BATTERY", "电池电量", latest.getBattery(), "%", status, sampledAt);
+        addMetric(metrics, "GAS_CONCENTRATION", "气体浓度", latest.getGasConcentration(), "%LEL", status, sampledAt, timeSource);
+        addMetric(metrics, "GAS_BATTERY", "电池电量", latest.getBattery(), "%", status, sampledAt, timeSource);
         addMetric(metrics, "GAS_SIGNAL_STRENGTH", "信号强度",
                 latest.getSignalStrength() == null
                         ? null : BigDecimal.valueOf(latest.getSignalStrength().doubleValue()),
-                "dBm", status, sampledAt);
+                "dBm", status, sampledAt, timeSource);
         return new CockpitTypeMetricsSnapshot(code, typeCode, typeName, metrics);
     }
 
@@ -202,10 +203,11 @@ public class BusinessCockpitTypeMetricsAdapter implements CockpitTypeMetricsAdap
             return new CockpitTypeMetricsSnapshot(code, typeCode, typeName, Collections.emptyList());
         }
         String sampledAt = offsetTime(sampledAt(latest.getTimestampSince(), latest.getCreateTime()));
+        String timeSource = latest.getTimestampSince() == null ? "CREATED_AT" : "REPORTED_AT";
         List<CockpitMetricSnapshot> metrics = new ArrayList<>();
-        addMetric(metrics, "WATER_LEVEL", "水位", decimal(latest.getWaterLevel()), null, "UNKNOWN", sampledAt);
-        addMetric(metrics, "WATER_POWER_VOLTAGE", "电源电压", decimal(latest.getPowerVoltage()), "V", "UNKNOWN", sampledAt);
-        addMetric(metrics, "WATER_SIGNAL_QUALITY", "信号质量", decimal(latest.getSignalQuality()), null, "UNKNOWN", sampledAt);
+        addMetric(metrics, "WATER_LEVEL", "水位", decimal(latest.getWaterLevel()), null, "UNKNOWN", sampledAt, timeSource);
+        addMetric(metrics, "WATER_POWER_VOLTAGE", "电源电压", decimal(latest.getPowerVoltage()), "V", "UNKNOWN", sampledAt, timeSource);
+        addMetric(metrics, "WATER_SIGNAL_QUALITY", "信号质量", decimal(latest.getSignalQuality()), null, "UNKNOWN", sampledAt, timeSource);
         return new CockpitTypeMetricsSnapshot(code, typeCode, typeName, metrics);
     }
 
@@ -307,8 +309,8 @@ public class BusinessCockpitTypeMetricsAdapter implements CockpitTypeMetricsAdap
 
     private static void addMetric(
             List<CockpitMetricSnapshot> metrics, String code, String name,
-            BigDecimal value, String unit, String status, String sampledAt) {
-        metrics.add(new CockpitMetricSnapshot(code, name, value, unit, status, sampledAt, null));
+            BigDecimal value, String unit, String status, String sampledAt, String timeSource) {
+        metrics.add(new CockpitMetricSnapshot(code, name, value, unit, status, sampledAt, null, timeSource));
     }
 
     private static BigDecimal decimal(String value) {

pipe-network-service/zksy-system/src/main/java/com/zksy/cockpit/type/CockpitMetric.java

@@ -11,6 +11,8 @@ public class CockpitMetric {
     private final String status;
     private final String sampledAt;
     private final String freshness;
+    /** sampledAt 的实际来源；CREATED_AT 不代表采样时间。 */
+    private final String timeSource;
     private final List<CockpitMetricPoint> points;
 
     public CockpitMetric(
@@ -22,6 +24,11 @@ public class CockpitMetric {
             String sampledAt,
             String freshness,
             List<CockpitMetricPoint> points) {
+        this(metricCode, metricName, value, unit, status, sampledAt, freshness, points, null);
+    }
+
+    public CockpitMetric(String metricCode, String metricName, BigDecimal value, String unit,
+            String status, String sampledAt, String freshness, List<CockpitMetricPoint> points, String timeSource) {
         this.metricCode = metricCode;
         this.metricName = metricName;
         this.value = value;
@@ -30,6 +37,7 @@ public class CockpitMetric {
         this.sampledAt = sampledAt;
         this.freshness = freshness;
         this.points = points;
+        this.timeSource = timeSource;
     }
 
     public String getMetricCode() { return metricCode; }
@@ -39,5 +47,6 @@ public class CockpitMetric {
     public String getStatus() { return status; }
     public String getSampledAt() { return sampledAt; }
     public String getFreshness() { return freshness; }
+    public String getTimeSource() { return timeSource; }
     public List<CockpitMetricPoint> getPoints() { return points; }
 }

pipe-network-service/zksy-system/src/main/java/com/zksy/cockpit/type/CockpitMetricSnapshot.java

@@ -10,6 +10,8 @@ public class CockpitMetricSnapshot {
     private final String unit;
     private final String status;
     private final String sampledAt;
+    /** CREATED_AT = 入库时间；REPORTED_AT = 设备报告时间。null 保持历史契约。 */
+    private final String timeSource;
     private final List<PointSnapshot> points;
 
     public CockpitMetricSnapshot(
@@ -20,6 +22,11 @@ public class CockpitMetricSnapshot {
             String status,
             String sampledAt,
             List<PointSnapshot> points) {
+        this(metricCode, metricName, value, unit, status, sampledAt, points, null);
+    }
+
+    public CockpitMetricSnapshot(String metricCode, String metricName, BigDecimal value, String unit,
+            String status, String sampledAt, List<PointSnapshot> points, String timeSource) {
         this.metricCode = metricCode;
         this.metricName = metricName;
         this.value = value;
@@ -27,6 +34,7 @@ public class CockpitMetricSnapshot {
         this.status = status;
         this.sampledAt = sampledAt;
         this.points = points;
+        this.timeSource = timeSource;
     }
 
     public String getMetricCode() { return metricCode; }
@@ -35,6 +43,7 @@ public class CockpitMetricSnapshot {
     public String getUnit() { return unit; }
     public String getStatus() { return status; }
     public String getSampledAt() { return sampledAt; }
+    public String getTimeSource() { return timeSource; }
     public List<PointSnapshot> getPoints() { return points; }
 
     public static class PointSnapshot {

pipe-network-service/zksy-system/src/main/java/com/zksy/cockpit/type/CockpitTypeMetricsService.java

@@ -55,7 +55,8 @@ public class CockpitTypeMetricsService {
                         source.getStatus(),
                         source.getSampledAt(),
                         freshness,
-                        null));
+                        null,
+                        source.getTimeSource()));
             }
         }
 

pipe-network-service/zksy-system/src/main/java/com/zksy/cockpit/type/CollectedMetricsReader.java

@@ -0,0 +1,168 @@
+package com.zksy.cockpit.type;
+
+import com.zksy.base.environment.domain.ERealTimeData;
+import com.zksy.base.environment.service.ERealTimeDataService;
+import com.zksy.base.perception.mapper.PerceptionMapper;
+import com.zksy.cockpit.foundation.CockpitReadException;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeParseException;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/** 只读原始采集源；节点分别取最新整行，缺值不回退旧行、不补零。 */
+@Component
+public class CollectedMetricsReader {
+    private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
+    private final ERealTimeDataService environment;
+    private final PerceptionMapper perception;
+
+    public CollectedMetricsReader(ERealTimeDataService environment, PerceptionMapper perception) {
+        this.environment = environment;
+        this.perception = perception;
+    }
+
+    public List<CockpitMetricSnapshot> loadCurrent(String table, String deviceCode) {
+        // 表名、标识列均为固定白名单；未知类型绝不回退行业表。
+        switch (table == null ? "" : table) {
+            case "e_real_time_data":
+                return environmentCurrent(deviceCode);
+            case "firefighting_pressure":
+                return wideCurrent("firefighting_pressure", "telemetering_station", deviceCode, "observed_time",
+                        field("pressure_value", "FIREFIGHTING_PRESSURE", "压力", "MPa"));
+            case "jg_device_data":
+                return wideCurrent("jg_device_data", "imei_card_number", deviceCode, null,
+                        field("tilt_angle", "MANHOLE_TILT_ANGLE", "倾斜角度", "°"),
+                        field("water_infiltration_alarm_status", "MANHOLE_WATER_INFILTRATION_ALARM", "水浸报警状态", ""),
+                        field("water_level_alarm_status", "MANHOLE_WATER_LEVEL_ALARM", "水位报警状态", ""),
+                        field("temperature_value", "MANHOLE_TEMPERATURE", "温度", "℃"),
+                        field("battery_level", "MANHOLE_BATTERY", "电池电量", "%"));
+            case "radar_data":
+                return wideCurrent("radar_data", "device_code", deviceCode, "timestamp_since",
+                        // 累计量不能沿用感知页的 m³/h 标注；单位未确认时留空。
+                        field("meter1_net_total", "RADAR_NET_TOTAL", "累计流量", ""),
+                        field("meter1_instant_flow", "RADAR_INSTANT_FLOW", "瞬时流量", "m³/h"),
+                        field("flow_speed", "RADAR_FLOW_SPEED", "流速", "m/s"),
+                        field("water_level", "RADAR_WATER_LEVEL", "水位", "m"));
+            case "telemetry_data":
+                return wideCurrent("telemetry_data", "device_code", deviceCode, "timestamp_since",
+                        field("meter1_instant_flow", "TELEMETRY_INSTANT_FLOW", "瞬时流量", "m³/h"),
+                        field("flow_speed", "TELEMETRY_FLOW_SPEED", "流速", "m/s"),
+                        field("pressure", "TELEMETRY_PRESSURE", "压力", "MPa"),
+                        field("temperature", "TELEMETRY_TEMPERATURE", "温度", "℃"));
+            case "noise_info":
+                return wideCurrent("noise_info", "encode", deviceCode, "ts",
+                        field("rms", "NOISE_RMS", "特征值", ""),
+                        field("gain", "NOISE_GAIN", "增益值", ""),
+                        field("center_frequency", "NOISE_CENTER_FREQUENCY", "中心频率", ""));
+            default:
+                return Collections.emptyList();
+        }
+    }
+
+    private List<CockpitMetricSnapshot> environmentCurrent(String deviceCode) {
+        Integer deviceId;
+        try {
+            deviceId = Integer.valueOf(deviceCode);
+        } catch (NumberFormatException exception) {
+            return Collections.emptyList();
+        }
+        // device_id 是整数；不能截取业务编码或把另一种文本编码强行映射到同一设备。
+        if (!deviceId.toString().equals(deviceCode)) {
+            return Collections.emptyList();
+        }
+        String[] codes = {"SUSPENDED_SOLIDS", "COD", "AMMONIA_NITROGEN", "CONDUCTIVITY", "PH"};
+        String[] names = {"悬浮物", "COD", "氨氮", "电导率", "pH"};
+        String[] units = {"mg/L", "mg/L", "mg/L", "μS/cm", ""};
+        List<CockpitMetricSnapshot> metrics = new ArrayList<>();
+        for (int node = 1; node <= 5; node++) {
+            ERealTimeData row = environment.getLatestByDeviceNode(deviceId, node);
+            if (row == null) {
+                continue;
+            }
+            String at = offsetTime(row.getCreateTime());
+            metrics.add(new CockpitMetricSnapshot("ENV_" + codes[node - 1], names[node - 1],
+                    decimal(node == 1 ? row.getFloatValue() : row.getHum()), units[node - 1],
+                    "UNKNOWN", at, null, "CREATED_AT"));
+            metrics.add(new CockpitMetricSnapshot("ENV_NODE_" + node + "_TEMPERATURE",
+                    names[node - 1] + "节点温度", decimal(row.getTem()), "℃",
+                    "UNKNOWN", at, null, "CREATED_AT"));
+        }
+        return metrics;
+    }
+
+    private List<CockpitMetricSnapshot> wideCurrent(String table, String idColumn, String deviceCode,
+            String reportTimeColumn, String[]... fields) {
+        List<Map<String, Object>> rows = perception.selectLatestRecords(table, idColumn, deviceCode, 1);
+        if (rows == null || rows.isEmpty() || rows.get(0) == null) {
+            return Collections.emptyList();
+        }
+        Map<String, Object> row = rows.get(0);
+        Object reportedAt = reportTimeColumn == null ? null : row.get(reportTimeColumn);
+        String at = recordTime(reportedAt == null ? row.get("create_time") : reportedAt);
+        String timeSource = reportedAt == null ? "CREATED_AT" : "REPORTED_AT";
+        List<CockpitMetricSnapshot> metrics = new ArrayList<>();
+        for (String[] field : fields) {
+            BigDecimal value = decimal(row.get(field[0]));
+            String status = "UNKNOWN";
+            if (field[0].endsWith("_alarm_status") && value != null) {
+                status = BigDecimal.ZERO.compareTo(value) == 0 ? "NORMAL"
+                        : BigDecimal.ONE.compareTo(value) == 0 ? "ALARM" : "UNKNOWN";
+            }
+            metrics.add(new CockpitMetricSnapshot(field[1], field[2], value, field[3], status,
+                    at, null, timeSource));
+        }
+        return metrics;
+    }
+
+    private static String[] field(String column, String code, String name, String unit) {
+        return new String[]{column, code, name, unit};
+    }
+
+    private static String recordTime(Object value) {
+        if (value instanceof java.sql.Timestamp) {
+            return offsetTime(((java.sql.Timestamp) value).toLocalDateTime());
+        }
+        if (value instanceof LocalDateTime) {
+            return offsetTime((LocalDateTime) value);
+        }
+        if (value == null) {
+            return null;
+        }
+        String text = value.toString().trim().replace(' ', 'T');
+        try {
+            return DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(
+                    OffsetDateTime.parse(text).atZoneSameInstant(BUSINESS_ZONE));
+        } catch (DateTimeParseException noOffset) {
+            try {
+                return offsetTime(LocalDateTime.parse(text));
+            } catch (DateTimeParseException invalid) {
+                // 不猜测数字的秒/毫秒基准，不用当前时间掩盖未知时间。
+                throw new CockpitReadException("invalid collected metric timestamp");
+            }
+        }
+    }
+
+    private static String offsetTime(LocalDateTime value) {
+        return value == null ? null : DateTimeFormatter.ISO_OFFSET_DATE_TIME
+                .format(value.atZone(BUSINESS_ZONE).toOffsetDateTime());
+    }
+
+    private static BigDecimal decimal(Object value) {
+        if (value == null) {
+            return null;
+        }
+        try {
+            return new BigDecimal(value.toString().trim());
+        } catch (NumberFormatException exception) {
+            return null;
+        }
+    }
+}