From c2dd019c1fddaeae412e2b6f44fb86ff0dc635d4 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期三, 02 九月 2026 22:08:16 +0800
Subject: [PATCH] 实现主动轮询完整实时快照
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java | 87 +++++++++++++++++++++++++++++++++++++++++++
1 files changed, 87 insertions(+), 0 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
index 22f195f..4ba1b52 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
+++ b/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
@@ -4,8 +4,12 @@
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.Date;
+import java.time.Instant;
+import java.util.Map;
+import java.util.LinkedHashMap;
import java.util.HexFormat;
import java.util.List;
+import com.alibaba.fastjson2.JSON;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
@@ -26,6 +30,8 @@
private final TTestMapper mapper;
private final ITTestService tests;
private final BtsDevelopmentSimulator simulator;
+ private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
+ private final Map<Long, Instant> lastRawCaptureAt = new java.util.concurrent.ConcurrentHashMap<>();
public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator)
{
@@ -36,6 +42,87 @@
@Override
@Transactional(rollbackFor = Exception.class)
+ public void captureDevelopmentRealtimeSnapshots()
+ {
+ List<Long> testIds = mapper.selectReceivingTestIds();
+ if (testIds == null) return;
+ Date now = new Date();
+ for (Long testId : testIds)
+ {
+ TTest locked = mapper.selectTestByIdForUpdate(testId);
+ if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
+ TTest test = tests.selectTestById(testId);
+ if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue;
+ // 完整快照与原始事实在同一接收事务内写入；原始事实按发现频率限流，避免 100ms 开发调度重复扫描历史。
+ Instant lastRaw = lastRawCaptureAt.get(testId);
+ if (lastRaw == null || !now.toInstant().isBefore(lastRaw.plusSeconds(5)))
+ {
+ captureTest(test);
+ for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
+ captureCollectorMeasurements(test, battery);
+ lastRawCaptureAt.put(testId, now.toInstant());
+ }
+ for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
+ captureSnapshot(test, battery, now);
+ }
+ }
+
+ private void captureSnapshot(TTest test, TTestBattery battery, Date now)
+ {
+ TTestChannelBinding binding = battery.getBinding();
+ if (binding == null || binding.getDevid() == null || binding.getChlid() == null || battery.getCollectorId() == null) return;
+ final long interval;
+ try { interval = simulator.persistenceIntervalMillis(binding.getDevid(), binding.getChlid()); }
+ catch (IllegalArgumentException ignored) { return; }
+ String key = test.getTestId() + ":" + binding.getBindingId();
+ Instant last = lastSnapshotAt.get(key);
+ if (last != null && now.toInstant().isBefore(last.plusMillis(interval))) return;
+ List<BtsHistoryRecord> records;
+ try { records = simulator.btsRecords(binding.getDevid(), binding.getChlid()); } catch (IllegalArgumentException ignored) { return; }
+ BtsHistoryRecord cabinet = records.stream().max(java.util.Comparator.comparing(BtsHistoryRecord::deviceTime)).orElse(null);
+ CollectorRealtimeSnapshot collector;
+ try { collector = simulator.collectorRealtime(battery.getCollectorId().intValue()); } catch (IllegalArgumentException ignored) { return; }
+ if (cabinet == null || !collector.online()) return;
+ if (test.getStartedAtUtc() != null && cabinet.deviceTime().isBefore(test.getStartedAtUtc().toInstant())) return;
+ if (test.getUpdateTime() != null && cabinet.deviceTime().isBefore(test.getUpdateTime().toInstant())) return;
+ Map<String, CollectorRealtimeMeasurement> metrics = new LinkedHashMap<>();
+ for (CollectorRealtimeMeasurement item : collector.measurements())
+ if (battery.getCollectorChannels() != null && battery.getCollectorChannels().stream().anyMatch(mapping ->
+ java.util.Objects.equals(mapping.getChannelCode(), item.channelCode())
+ && metricMatches(mapping.getMeasurementCode(), item.metricCode())))
+ metrics.put(item.metricCode().toUpperCase(java.util.Locale.ROOT), item);
+ CollectorRealtimeMeasurement temp = metrics.get("TEMPERATURE");
+ CollectorRealtimeMeasurement pos = metrics.get("POSITIVE_POTENTIAL");
+ CollectorRealtimeMeasurement neg = metrics.get("NEGATIVE_POTENTIAL");
+ if (temp == null || pos == null || neg == null) return;
+ if (test.getStartedAtUtc() != null && (temp.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
+ || pos.deviceTime().isBefore(test.getStartedAtUtc().toInstant())
+ || neg.deviceTime().isBefore(test.getStartedAtUtc().toInstant()))) return;
+ if (test.getUpdateTime() != null && (temp.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
+ || pos.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())
+ || neg.serverReceivedAt().isBefore(test.getUpdateTime().toInstant()))) return;
+ Date slot = new Date((now.getTime() / interval) * interval);
+ byte[] hash = snapshotHash(cabinet, temp, pos, neg, slot.toInstant());
+ mapper.insertRealtimeSnapshot(test.getTestId(), binding.getBindingId(), binding.getChannelId(), battery.getCollectorId(), slot,
+ cabinet.sequenceId(), Date.from(cabinet.deviceTime()), cabinet.deviceTime().toString(),
+ temp.sequenceId(), Date.from(temp.deviceTime()), temp.deviceTimeRaw(), now,
+ cabinet.cycleNo(), cabinet.stepNo(), cabinet.stepType(), cabinet.stepName(),
+ BigDecimal.valueOf(cabinet.voltageV()), BigDecimal.valueOf(cabinet.currentA()), BigDecimal.valueOf(cabinet.capacityAh()),
+ BigDecimal.valueOf(cabinet.energyWh()), temp.value(), pos.value(), neg.value(), JSON.toJSONString(metrics), hash);
+ lastSnapshotAt.put(key, now.toInstant());
+ }
+
+ private byte[] snapshotHash(BtsHistoryRecord cabinet, CollectorRealtimeMeasurement temp,
+ CollectorRealtimeMeasurement pos, CollectorRealtimeMeasurement neg, Instant slot)
+ {
+ try { return MessageDigest.getInstance("SHA-256").digest((slot + "|" + cabinet.sequenceId() + "|" + cabinet.deviceTime()
+ + "|" + cabinet.voltageV() + "|" + cabinet.currentA() + "|" + temp.value() + "|" + pos.value() + "|" + neg.value())
+ .getBytes(StandardCharsets.UTF_8)); }
+ catch (Exception exception) { throw new IllegalStateException("无法计算实时快照内容摘要", exception); }
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
public void captureDevelopmentCabinetRecords()
{
List<Long> testIds = mapper.selectReceivingTestIds();
--
Gitblit v1.7.1