From cbf301a41d3ad43a61b472a780f932f3975a5614 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 20:39:17 +0800
Subject: [PATCH] 修复多通道温度覆盖与实时阈值异常漏报
---
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 69 ++++++++++++++++++++--------------
1 files changed, 41 insertions(+), 28 deletions(-)
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
index 71f130d..ed66c11 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
@@ -46,7 +46,7 @@
private final TestWarningExceptionService warningExceptions;
private final Clock clock;
private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
- private final Map<Long, Instant> lastOfflineCheckAt = new java.util.concurrent.ConcurrentHashMap<>();
+ private final Map<Long, Instant> lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>();
@Autowired
public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
@@ -70,13 +70,14 @@
{
for (TTest test : receivingTests()) {
Instant now = clock.instant();
- Instant previous = lastOfflineCheckAt.get(test.getTestId());
- // 离线检查独立于测量采样槽位，并与暂停、停止和删除共享测试事务锁。
+ Instant previous = lastRealtimeCheckAt.get(test.getTestId());
+ // 实时预警独立于持久化槽位，并与暂停、停止和删除共享测试事务锁。
if (previous == null || !now.isBefore(previous.plusSeconds(5))) {
warningExceptions.recordOfflineSources(test, Date.from(now));
- lastOfflineCheckAt.put(test.getTestId(), now);
+ for (TTestBattery battery : batteries(test)) saveRealtime(test, battery);
+ lastRealtimeCheckAt.put(test.getTestId(), now);
}
- for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); }
+ for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery);
}
}
@@ -98,12 +99,24 @@
if (adapter == null) return;
TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId();
- Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return;
CabinetRealtime cabinet = latestCabinet(binding);
if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null;
CollectorRealtime collector = latestCollector(battery);
Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector);
if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return;
+ // 当前成功测量先判定预警；槽位尚未到期或历史行被删除，都不能阻断实时告警。
+ if (cabinet != null) {
+ Date observedAt = Date.from(cabinet.serverReceivedAt());
+ evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(cabinet.voltageV()), "V", observedAt);
+ evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(cabinet.currentA()), "A", observedAt);
+ evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(cabinet.capacityAh()), "Ah", observedAt);
+ evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(cabinet.energyWh()), "Wh", observedAt);
+ }
+ for (CollectorMeasurement measurement : mapped.values().stream().distinct().toList()) {
+ warningExceptions.evaluate(test, "COLLECTOR", battery.getCollectorId() + ":" + measurement.channelCode(),
+ measurement.metricCode(), measurement.value(), measurement.unit(), Date.from(measurement.serverReceivedAt()));
+ }
+ Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return;
Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null);
Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval);
// 删除修订随数据库保留；重启后同一槽位也不能重新出现。
@@ -116,21 +129,6 @@
row.setCollectorSequenceId(mapped.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo).orElse(null)); row.setCollectorDeviceTimeUtc(collectorTime == null ? null : Date.from(collectorTime)); row.setCollectorDeviceTimeRaw(mapped.values().stream().filter(value -> value.deviceTime().equals(collectorTime)).map(CollectorMeasurement::deviceTimeRaw).findFirst().orElse(null));
if (cabinet != null) { row.setCycleNo(cabinet.cycleNo()); row.setStepNo(cabinet.stepNo()); row.setStepType(cabinet.stepType()); row.setStepName(cabinet.stepName()); row.setVoltageV(BigDecimal.valueOf(cabinet.voltageV())); row.setCurrentA(BigDecimal.valueOf(cabinet.currentA())); row.setCapacityAh(BigDecimal.valueOf(cabinet.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(cabinet.energyWh())); }
row.setTemperature(metricValue(mapped, "TEMPERATURE")); row.setPositivePotential(metricValue(mapped, "POSITIVE_POTENTIAL")); row.setNegativePotential(metricValue(mapped, "NEGATIVE_POTENTIAL")); row.setCollectorChannelCode(mapped.values().stream().map(CollectorMeasurement::channelCode).findFirst().orElse(null));
- // 只有本次观测成功取得的值进入阈值判断；离线、请求失败和缺失值在前面被过滤。
- Date cabinetObservedAt = cabinet == null || cabinet.serverReceivedAt() == null ? nowDate(now) : Date.from(cabinet.serverReceivedAt());
- if (cabinet != null)
- {
- evaluateCabinet(test, binding, "VOLTAGE", row.getVoltageV(), "V", cabinetObservedAt);
- evaluateCabinet(test, binding, "CURRENT", row.getCurrentA(), "A", cabinetObservedAt);
- evaluateCabinet(test, binding, "CAPACITY", row.getCapacityAh(), "Ah", cabinetObservedAt);
- evaluateCabinet(test, binding, "ENERGY", row.getEnergyWh(), "Wh", cabinetObservedAt);
- }
- for (CollectorMeasurement measurement : mapped.values())
- {
- if (measurement.value() != null)
- warningExceptions.evaluate(test, "COLLECTOR", String.valueOf(battery.getCollectorId()), measurement.metricCode(), measurement.value(), measurement.unit(),
- measurement.serverReceivedAt() == null ? nowDate(now) : Date.from(measurement.serverReceivedAt()));
- }
Map<String, Object> extension = new LinkedHashMap<>();
for (Map.Entry<String, CollectorMeasurement> e : mapped.entrySet()) if (!isFixedMetric(e.getKey()))
{
@@ -155,8 +153,6 @@
{
if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt);
}
-
- private static Date nowDate(Instant instant) { return Date.from(instant); }
private void saveDeviceAggregates(TTest test, TTestBattery battery)
{
@@ -196,10 +192,26 @@
private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid())); } catch (RuntimeException ex) { return null; } }
private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null) return null; try { return adapter.latestCollector(mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
- private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery b, CollectorRealtime s) { Map<String, CollectorMeasurement> out = new LinkedHashMap<>(); if (s == null || !s.online() || s.measurements() == null || b.getCollectorChannels() == null) return out; for (CollectorMeasurement m : s.measurements()) if (accepted(test, m.deviceTime(), m.serverReceivedAt()) && b.getCollectorChannels().stream().anyMatch(x -> java.util.Objects.equals(x.getChannelCode(), m.channelCode()) && metricMatches(x.getMeasurementCode(), m.metricCode()))) out.put(m.metricCode().toUpperCase(Locale.ROOT), m); return out; }
- private boolean metricMatches(String c, String a) { if (c == null || a == null) return false; c = c.toUpperCase(Locale.ROOT); a = a.toUpperCase(Locale.ROOT); return c.equals(a) || (c.contains("TEMPERATURE") && a.equals("TEMPERATURE")) || (c.contains("POSITIVE_POTENTIAL") && a.equals("POSITIVE_POTENTIAL")) || (c.contains("NEGATIVE_POTENTIAL") && a.equals("NEGATIVE_POTENTIAL")); }
- private static boolean isFixedMetric(String code) { return "TEMPERATURE".equals(code) || "POSITIVE_POTENTIAL".equals(code) || "NEGATIVE_POTENTIAL".equals(code); }
- private static BigDecimal metricValue(Map<String, CollectorMeasurement> values, String code) { return values.get(code) == null ? null : values.get(code).value(); }
+ private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source)
+ {
+ Map<String, CollectorMeasurement> result = new LinkedHashMap<>();
+ if (source == null || !source.online() || battery.getCollectorChannels() == null) return result;
+ for (var binding : battery.getCollectorChannels()) for (var measurement : source.measurements()) {
+ if (accepted(test, measurement.deviceTime(), measurement.serverReceivedAt())
+ && java.util.Objects.equals(binding.getChannelCode(), measurement.channelCode())
+ && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), measurement.metricCode())) {
+ // 绑定角色区分主、辅助测量；不能用设备通用指标名覆盖其他物理通道。
+ result.put(binding.getMeasurementCode().toUpperCase(Locale.ROOT), measurement);
+ }
+ }
+ return result;
+ }
+ private static boolean isFixedMetric(String code) { return List.of("TEMPERATURE", "POSITIVE_POTENTIAL", "NEGATIVE_POTENTIAL", "PRIMARY_TEMPERATURE", "PRIMARY_POSITIVE_POTENTIAL", "PRIMARY_NEGATIVE_POTENTIAL").contains(code); }
+ private static BigDecimal metricValue(Map<String, CollectorMeasurement> values, String code) {
+ var measurement = values.get("PRIMARY_" + code);
+ if (measurement == null) measurement = values.get(code);
+ return measurement == null ? null : measurement.value();
+ }
private static String sourceRecordId(CabinetRealtime cabinet, Map<String, CollectorMeasurement> measurements)
{
List<String> parts = new java.util.ArrayList<>();
@@ -212,7 +224,8 @@
private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }
- private static boolean accepted(TTest test, Instant deviceTime, Instant serverReceivedAt)
+ // 实时提醒与持久化共用归属边界，避免同一旧观测只在首页触发异常。
+ static boolean accepted(TTest test, Instant deviceTime, Instant serverReceivedAt)
{
if (deviceTime != null && test.getStartedAtUtc() != null && deviceTime.isBefore(test.getStartedAtUtc().toInstant())) return false;
return serverReceivedAt != null && (test.getReceptionBoundaryAtUtc() == null || !serverReceivedAt.isBefore(test.getReceptionBoundaryAtUtc().toInstant()));
--
Gitblit v1.7.1