From b78295e8da96c3e5e76befeef9582ce5273dbb74 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期二, 29 九月 2026 13:10:23 +0800
Subject: [PATCH] 按代码规范检查整改实体注释、Service Javadoc 与分层边界
---
battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java | 113 +++++++++++++++++++++++++++++++++++++++++++++++---------
1 files changed, 95 insertions(+), 18 deletions(-)
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
index ceedeb7..f9a7270 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TTestLiveServiceImpl.java
+++ b/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;
@@ -23,7 +24,7 @@
import com.battery.business.service.ITTestService;
import com.battery.business.mapper.TTestMapper;
-/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
+/** 将设备最新快照转换为测试视角，按测试接收状态决定是否展示采集参数。 */
@Service
public class TTestLiveServiceImpl implements ITestLiveService
{
@@ -41,36 +42,96 @@
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)
@@ -130,9 +191,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,18 +223,29 @@
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());
--
Gitblit v1.7.1