实现 Ticket04：采集器辅助测量与登记入口，拆除开发模拟器

yeg

6 天以前 e38aca8134570513435c003c25dc05f1ae51c9c9

 battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java

@@ -1,6 +1,7 @@
package com.battery.business.service.impl;

import java.time.Instant;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
@@ -17,13 +18,15 @@
import com.battery.device.api.model.Measurement;
import com.battery.business.domain.TTest;
import com.battery.business.domain.TTestBattery;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.service.ITestLiveService;
import com.battery.business.service.ITestLiveService.TestPlanRow;
import com.battery.business.service.ITestLiveService.TestPlanSnapshot;
import com.battery.business.service.ITTestService;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.mapper.TTestMapper;

/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
/** 将设备最新快照转换为测试视角，按测试接收状态决定是否展示采集参数。 */
@Service
public class TTestLiveServiceImpl implements ITestLiveService
{
@@ -31,46 +34,107 @@
    private final DeviceDiscoveryCandidateStore candidates;
    private final TestDataObservationAdapter adapter;
    private final TTestMapper mapper;
    private final TDeviceAdapterMapper adapters;

    @Autowired
    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            @Nullable TestDataObservationAdapter adapter, TTestMapper mapper)
    { this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; }
            @Nullable TestDataObservationAdapter adapter, TTestMapper mapper, TDeviceAdapterMapper adapters)
    { this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; this.adapters = adapters; }

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            TestDataObservationAdapter adapter)
    { this(tests, candidates, adapter, null); }
    { this(tests, candidates, adapter, null, null); }

    /** 返回测试当前设备事实；接收中展示测量和实时提醒，历史异常处理状态不参与判定。 */
    @Override public LiveState current(Long testId)
    {
        TTest test = tests.selectTestById(testId);
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        boolean receiving = "RECEIVING".equals(test.getReceptionStatus());
        boolean receiving = "RUNNING".equals(test.getTestStatus()) && "RECEIVING".equals(test.getReceptionStatus());
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null)
                sources.add(snapshotFor(cabinetSource(battery, cabinets), receiving));
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(collectorSource(battery), receiving));
                sources.add(collectorSource(battery, receiving));
        }
        List<Map<String, Object>> exceptions = new ArrayList<>();
        if (receiving)
            exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
                    && "OFFLINE".equals(source.deviceStatus()))
                    .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                            "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                    .toList());
        // 完成或暂停后只保留历史异常事实，实时控制台不再展示开放预警。
        if (receiving && mapper != null) {
            mapper.selectWarningExceptions(testId, "OPEN").forEach(item -> exceptions.add(Map.of(
                    "exceptionId", item.getExceptionId(), "sourceType", item.getSourceType(),
                    "metricCode", item.getMetricCode(), "value", item.getAbnormalValue(), "unit", item.getUnit(),
                    "status", item.getExceptionStatus(), "message", "测试预警异常")));
        java.util.Set<String> offlineSources = new java.util.HashSet<>();
        // 实时提醒只反映当前设备和测量事实，异常管理待办是否处理不参与展示判定。
        java.util.Set<String> thresholdSources = new java.util.HashSet<>();
        if (receiving) {
            for (LiveSource source : sources) {
                if ("ONLINE".equals(source.deviceStatus())) appendThresholdWarnings(test, source, exceptions, thresholdSources);
                if (!"OFFLINE".equals(source.deviceStatus())) continue;
                String identity = offlineIdentity(test, source);
                if (!offlineSources.add(source.sourceType() + ":" + identity)) continue;
                exceptions.add(Map.of("exceptionType", "DEVICE_OFFLINE", "sourceType", source.sourceType(),
                        "sourceId", source.sourceId(), "sourceIdentity", identity,
                        "status", "OPEN", "message", "接收中数据来源离线，当前无法取得测量数据"));
            }
        }
        String testStatus = test.getTestStatus();
        TestPlanSnapshot testPlan = detectorTestPlan(test);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now().toEpochMilli(), sources, exceptions, testPlan);
    }

    private void appendThresholdWarnings(TTest test, LiveSource source, List<Map<String, Object>> warnings,
            java.util.Set<String> seen)
    {
        var snapshot = source.latestSnapshot();
        if ("CABINET_CHANNEL".equals(source.sourceType())) {
            if (!receivedDuringTest(test, snapshot.get("deviceTime"), snapshot.get("observedAt"))) return;
            String identity = offlineIdentity(test, source);
            for (String[] field : List.of(new String[] { "voltageV", "VOLTAGE", "V" },
                    new String[] { "currentA", "CURRENT", "A" }, new String[] { "capacityAh", "CAPACITY", "Ah" },
                    new String[] { "energyWh", "ENERGY", "Wh" }))
                appendThresholdValue(test, source.sourceType(), identity, field[1], snapshot.get(field[0]), field[2], warnings, seen);
        }
        else if (snapshot.get("measurements") instanceof List<?> measurements) {
            for (Object item : measurements) if (item instanceof Map<?, ?> measurement
                    && receivedDuringTest(test, measurement.get("deviceTime"), measurement.get("serverReceivedAt"))) {
                appendThresholdValue(test, source.sourceType(), source.sourceId() + ":" + measurement.get("channelCode"),
                        String.valueOf(measurement.get("metricCode")), measurement.get("value"),
                        String.valueOf(measurement.get("unit")), warnings, seen);
            }
        }
    }

    private boolean receivedDuringTest(TTest test, Object deviceTime, Object observedAt)
    {
        return TestDataReceptionServiceImpl.accepted(test,
                deviceTime instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null,
                observedAt instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null);
    }

    private void appendThresholdValue(TTest test, String sourceType, String identity, String metricCode,
            Object value, String unit, List<Map<String, Object>> warnings, java.util.Set<String> seen)
    {
        if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())) return;
        BigDecimal measurement = value instanceof BigDecimal decimal ? decimal : new BigDecimal(value.toString());
        for (var rule : TestWarningExceptionService.exceededRules(test, metricCode, measurement)) {
            if (!seen.add(rule.getThresholdId() + ":" + sourceType + ":" + identity + ":" + metricCode)) continue;
            Map<String, Object> warning = new LinkedHashMap<>();
            warning.put("exceptionType", "THRESHOLD"); warning.put("thresholdId", rule.getThresholdId());
            warning.put("sourceType", sourceType); warning.put("sourceIdentity", identity);
            warning.put("metricCode", metricCode); warning.put("value", measurement); warning.put("unit", unit);
            warning.put("minValue", rule.getMinValue()); warning.put("maxValue", rule.getMaxValue());
            warning.put("status", "OPEN"); warning.put("message", "当前测量超出预警阈值");
            warnings.add(warning);
        }
    }

    private String offlineIdentity(TTest test, LiveSource source)
    {
        if ("COLLECTOR".equals(source.sourceType())) return String.valueOf(source.sourceId());
        for (TTestBattery battery : test.getBatteries()) {
            var binding = battery.getBinding();
            if (binding != null && Objects.equals(binding.getChannelId(), source.sourceId()))
                return binding.getAdapterCode() + ":" + binding.getDevtype() + ":" + binding.getDevid()
                        + ":" + binding.getSubdevid() + ":" + binding.getChlid();
        }
        return String.valueOf(source.sourceId());
    }

    private TestPlanSnapshot detectorTestPlan(TTest test)
@@ -130,9 +194,14 @@
    private Map<String, Object> snapshot(ChannelCandidate channel)
    {
        Map<String, Object> value = new LinkedHashMap<>();
        // 通道工作状态是独立设备事实：读库来源下不可观测，界面把它当正常取值展示而不是异常。
        value.put("workStatus", channel.workStatus() == null ? null : channel.workStatus().name());
        // 最近数据时间取来源里最近一次采样行的设备时间，与展示轮询节奏无关。
        value.put("lastDataAt", channel.lastDataAt() == null ? null : channel.lastDataAt().toEpochMilli());
        if (channel.latestSnapshot() == null) return value;
        value.put("source", "检测柜");
        value.put("observedAt", channel.latestSnapshot().receivedAt().toEpochMilli());
        value.put("deviceTime", channel.latestSnapshot().deviceTime() == null ? null : channel.latestSnapshot().deviceTime().toEpochMilli());
        // BTS 核心采样字段是快照的一等事实，不能只依赖动态 measurements 展示。
        value.put("currentA", channel.latestSnapshot().currentA());
        value.put("voltageV", channel.latestSnapshot().voltageV());
@@ -157,34 +226,66 @@
        return value;
    }

    private LiveSource collectorSource(TTestBattery battery)
    /**
     * 采集器来源：测量按 ADR-0064 由同一来源实例作为辅助测量提供，所以它的在线事实就是该来源的可达性
     * （ADR-0063：在线只表示来源可达），不再依赖只有开发模拟器会产出的发现候选。
     *
     * <p>来源不提供辅助测量、或来源可达但这一刻还没有任何辅助测量值时，明确标注为「无辅助数据」：
     * 这是正常取值而不是故障，主测量接收不受影响，也不因此生成离线异常。来源不可达时不标这个标记——
     * 那种情况是来源离线，由既有的离线异常路径表达。
     */
    private LiveSource collectorSource(TTestBattery battery, boolean receiving)
    {
        if (adapter == null || mapper == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
                "OFFLINE", Map.of());
        var collector = adapter.latestCollector(mapper.selectCollectorCode(battery.getCollectorId()));
        if (collector != null)
        String collectorCode = mapper.selectCollectorCode(battery.getCollectorId());
        String adapterCode = battery.getBinding() == null ? null : battery.getBinding().getAdapterCode();
        boolean sourceOnline = sourceOnline(adapterCode);
        // 非接收状态只读来源可达性，不能先请求测量再隐藏，否则仍会产生新观测。
        if (!receiving)
        {
            Map<String, Object> value = new LinkedHashMap<>();
            value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());
            // 在业务响应边界转换毫秒时间，设备时间与服务器接收时间分别保留。
            value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null
                    && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode()))).map(item -> {
                Map<String, Object> measurement = new LinkedHashMap<>();
                measurement.put("metricCode", item.metricCode());
                measurement.put("channelCode", item.channelCode());
                measurement.put("value", item.value());
                measurement.put("unit", item.unit());
                measurement.put("deviceTime", item.deviceTime().toEpochMilli());
                measurement.put("serverReceivedAt", item.serverReceivedAt().toEpochMilli());
                return measurement;
            }).toList());
            value.put("units", collector.measurements().stream().map(item -> item.unit()).distinct().toList());
            collector.measurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
                    .ifPresent(observedAt -> value.put("observedAt", observedAt.toEpochMilli()));
            return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
                    collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());
                    sourceOnline ? "ONLINE" : "OFFLINE", Map.of());
        }
        return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of());
        var collector = adapter.latestCollector(adapterCode, collectorCode);
        Map<String, Object> value = new LinkedHashMap<>();
        value.put("source", "采集器");
        if (collector == null)
        {
            // 该来源根本不提供辅助测量（新威实测无辅助数据；推式来源的列语义确认前同样如此）。
            if (sourceOnline) value.put("auxiliaryDataAvailable", false);
            return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
                    sourceOnline ? "ONLINE" : "OFFLINE", value);
        }
        value.put("sampleIntervalMs", collector.sampleIntervalMs());
        // 在业务响应边界转换毫秒时间，设备时间与服务器接收时间分别保留。
        value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null
                && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode())
                        && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), item.metricCode()))).map(item -> {
            Map<String, Object> measurement = new LinkedHashMap<>();
            measurement.put("metricCode", item.metricCode());
            measurement.put("channelCode", item.channelCode());
            measurement.put("value", item.value());
            measurement.put("unit", item.unit());
            measurement.put("deviceTime", item.deviceTime().toEpochMilli());
            measurement.put("serverReceivedAt", item.serverReceivedAt().toEpochMilli());
            return measurement;
        }).toList());
        value.put("units", collector.measurements().stream().map(item -> item.unit()).distinct().toList());
        collector.measurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
                .ifPresent(observedAt -> value.put("observedAt", observedAt.toEpochMilli()));
        // 来源可达但这一刻还没有任何辅助测量值：同样是"无辅助数据"，按空值展示而不是占位。
        if (collector.online() && collector.measurements().isEmpty()) value.put("auxiliaryDataAvailable", false);
        return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
                collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());
    }

    /** 来源实例是否可达；来源编码缺失或查询不到时按不可达处理，不猜也不回退到其他实例。 */
    private boolean sourceOnline(String adapterCode)
    {
        if (adapters == null || adapterCode == null || adapterCode.isBlank()) return false;
        TDeviceAdapter adapter = adapters.selectDeviceAdapterByCode(adapterCode);
        return adapter != null && "ONLINE".equalsIgnoreCase(adapter.getConnectionStatus());
    }

}