| | |
| | | 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 com.battery.business.service.ITTestService; |
| | | import com.battery.business.mapper.TTestMapper; |
| | | |
| | | /** 将设备最新快照转换为测试视角,并托管 SSE 连接生命周期。 */ |
| | | /** 将设备最新快照转换为测试视角,按测试接收状态决定是否展示采集参数。 */ |
| | | @Service |
| | | public class TTestLiveServiceImpl implements ITestLiveService |
| | | { |
| | |
| | | TestDataObservationAdapter adapter) |
| | | { this(tests, candidates, adapter, 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) |
| | |
| | | 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()); |
| | |
| | | return value; |
| | | } |
| | | |
| | | private LiveSource collectorSource(TTestBattery battery) |
| | | 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())); |
| | | 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( |
| | | battery.getBinding() == null ? null : battery.getBinding().getAdapterCode(), collectorCode); |
| | | if (collector != null) |
| | | { |
| | | 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 -> { |
| | | && 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()); |