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; import java.util.Map; import java.util.Objects; import org.springframework.stereotype.Service; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.lang.Nullable; import com.battery.device.api.TestDataObservationAdapter; import com.battery.device.api.model.ChannelIdentity; import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore; import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate; import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate; import com.battery.device.api.model.Measurement; import com.battery.business.domain.TTest; import com.battery.business.domain.TTestBattery; 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.TTestMapper; /** 将设备最新快照转换为测试视角,按测试接收状态决定是否展示采集参数。 */ @Service public class TTestLiveServiceImpl implements ITestLiveService { private final ITTestService tests; private final DeviceDiscoveryCandidateStore candidates; private final TestDataObservationAdapter adapter; private final TTestMapper mapper; @Autowired public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates, @Nullable TestDataObservationAdapter adapter, TTestMapper mapper) { this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; } public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates, TestDataObservationAdapter adapter) { this(tests, candidates, adapter, null); } @Override public LiveState current(Long testId) { TTest test = tests.selectTestById(testId); List cabinets = candidates.snapshot(Instant.now()).cabinets(); List sources = new ArrayList<>(); boolean receiving = "RUNNING".equals(test.getTestStatus()) && "RECEIVING".equals(test.getReceptionStatus()); for (TTestBattery battery : test.getBatteries() == null ? List.of() : test.getBatteries()) { if (battery.getBinding() != null) sources.add(snapshotFor(cabinetSource(battery, cabinets), receiving)); if (battery.getCollectorId() != null) sources.add(collectorSource(battery, receiving)); } List> exceptions = new ArrayList<>(); java.util.Set offlineSources = new java.util.HashSet<>(); // 实时提醒只反映当前设备和测量事实,异常管理待办是否处理不参与展示判定。 java.util.Set 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> warnings, java.util.Set 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> warnings, java.util.Set 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 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) { if (adapter == null) return null; if (test.getBatteries() == null) return null; for (TTestBattery battery : test.getBatteries()) { if (battery.getBinding() == null || battery.getBinding().getDevid() == null || battery.getBinding().getChlid() == null) continue; try { var binding = battery.getBinding(); var source = adapter.testPlan(new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid())); if (source == null) continue; List rows = source.rows().stream() .map(row -> new TestPlanRow(row.rowType(), row.stepNo(), row.cyclePosition(), row.stepName(), row.durationSeconds(), row.voltageV(), row.executionStatus(), row.cycleCount(), row.recordCount())) .toList(); return new TestPlanSnapshot(source.currentCycleNo(), source.currentStepNo(), source.plannedCycleCount(), rows); } catch (IllegalArgumentException ignored) { /* 来源未提供方案时保持为空。 */ } } return null; } private LiveSource snapshotFor(LiveSource fresh, boolean receiving) { // 暂停或停止只停止数据请求;设备状态仍以本次查询的最新事实为准,采集快照不继续展示。 if (!receiving) return withoutSnapshot(fresh); return fresh; } private LiveSource withoutSnapshot(LiveSource source) { return new LiveSource(source.sourceType(), source.sourceId(), source.batteryNo(), source.deviceStatus(), Map.of()); } private LiveSource cabinetSource(TTestBattery battery, List cabinets) { var binding = battery.getBinding(); for (CabinetCandidate cabinet : cabinets) for (ChannelCandidate channel : cabinet.channels()) if (Objects.equals(channel.identity().adapterCode(), binding.getAdapterCode()) && Objects.equals(channel.identity().devtype(), binding.getDevtype()) && Objects.equals(channel.identity().devid(), binding.getDevid()) && Objects.equals(channel.identity().subdevid(), binding.getSubdevid()) && Objects.equals(channel.identity().chlid(), binding.getChlid())) return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), channel.online() ? "ONLINE" : "OFFLINE", channel.online() ? snapshot(channel) : Map.of()); return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), "OFFLINE", Map.of()); } private Map snapshot(ChannelCandidate channel) { Map value = new LinkedHashMap<>(); 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()); value.put("capacityAh", channel.latestSnapshot().capacityAh()); value.put("energyWh", channel.latestSnapshot().energyWh()); // 实时方案需要设备当前循环和实际工步;这些字段只用于快照展示,不形成历史采集事实。 value.put("cycleNo", channel.latestSnapshot().cycleNo()); value.put("stepNo", channel.latestSnapshot().stepNo()); value.put("stepType", channel.latestSnapshot().stepType()); value.put("totalTimeSeconds", channel.latestSnapshot().totalTimeSeconds()); value.put("measurements", channel.latestSnapshot().measurements().entrySet().stream().map(entry -> { Measurement measurement = entry.getValue(); return Map.of("metricCode", entry.getKey(), "value", measurement.value(), "unit", measurement.unit()); }).toList()); List units = new ArrayList<>(); if (channel.latestSnapshot().currentA() != null) units.add("A"); if (channel.latestSnapshot().voltageV() != null) units.add("V"); if (channel.latestSnapshot().capacityAh() != null) units.add("Ah"); if (channel.latestSnapshot().energyWh() != null) units.add("Wh"); units.addAll(channel.latestSnapshot().measurements().values().stream().map(Measurement::unit).toList()); value.put("units", units.stream().filter(item -> item != null && !item.isBlank()).distinct().toList()); return value; } private LiveSource collectorSource(TTestBattery battery, boolean receiving) { if (adapter == null || mapper == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of()); String collectorCode = mapper.selectCollectorCode(battery.getCollectorId()); // 非接收状态只读设备发现事实,不能先请求测量再隐藏,否则仍会产生新观测。 if (!receiving) { boolean online = candidates.findCollector(collectorCode, Instant.now()) .map(DeviceDiscoveryCandidateStore.CollectorCandidate::online).orElse(false); return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), online ? "ONLINE" : "OFFLINE", Map.of()); } var collector = adapter.latestCollector(collectorCode); if (collector != null) { Map 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()) && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), item.metricCode()))).map(item -> { Map 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()); } return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of()); } }