From 76cd5948e95ddbde9dd4411b466ee75ba20b562e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期四, 03 九月 2026 17:28:29 +0800
Subject: [PATCH] 实现测试预警异常管理
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java | 248 +++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 227 insertions(+), 21 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 025c217..107d501 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,35 +4,130 @@
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 java.util.Objects;
-import java.util.Optional;
+import com.alibaba.fastjson2.JSON;
import org.springframework.stereotype.Service;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import com.ruoyi.device.bts.BtsDevelopmentSimulator;
import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
+import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
+import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestChannelBinding;
-import com.ruoyi.system.domain.TTestReceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITTestService;
import com.ruoyi.system.service.ITestDataReceptionService;
-/** 接收时确定测试归属，并以现有采样表追加保存检测柜事实。 */
+/** 接收时确定测试归属，分别保留检测柜、采集器原始事实，并形成统一实时快照。 */
@Service
public class TestDataReceptionServiceImpl implements ITestDataReceptionService
{
private final TTestMapper mapper;
private final ITTestService tests;
private final BtsDevelopmentSimulator simulator;
+ private final TestWarningExceptionService warningExceptions;
+ 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)
+ @Autowired
+ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator,
+ TestWarningExceptionService warningExceptions)
{
this.mapper = mapper;
this.tests = tests;
this.simulator = simulator;
+ this.warningExceptions = warningExceptions;
+ }
+
+ /** 保留单元测试及旧调用方构造契约。 */
+ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator)
+ { this(mapper, tests, simulator, new TestWarningExceptionService(mapper)); }
+
+ @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
@@ -43,57 +138,145 @@
if (testIds == null) return;
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;
captureTest(test);
}
}
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void captureDevelopmentCollectorMeasurements()
+ {
+ List<Long> testIds = mapper.selectReceivingTestIds();
+ if (testIds == null) return;
+ 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;
+ for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
+ captureCollectorMeasurements(test, battery);
+ }
+ }
+
private void captureTest(TTest test)
{
+ Date startedAt = test.getStartedAtUtc();
for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
{
TTestChannelBinding binding = battery.getBinding();
if (binding == null || binding.getDevid() == null || binding.getChlid() == null) continue;
- Optional<TTestReceptionInterval> open = mapper.selectTestReceptionIntervals(test.getTestId()).stream()
- .filter(item -> Objects.equals(item.getBindingId(), binding.getBindingId())
- && "CABINET_CHANNEL".equals(item.getSourceType())
- && Objects.equals(item.getSourceId(), binding.getChannelId())
- && "OPEN".equals(item.getIntervalStatus()))
- .findFirst();
- if (open.isEmpty()) continue;
- Date start = open.get().getStartBoundaryAtUtc();
List<BtsHistoryRecord> records;
try { records = simulator.btsRecords(binding.getDevid(), binding.getChlid()); }
catch (IllegalArgumentException ignored) { continue; }
for (BtsHistoryRecord record : records)
{
- if (start != null && record.deviceTime().isBefore(start.toInstant())) continue;
- save(test, battery, binding, open.get(), record);
+ // 模拟器保留通道历史；只接收本次测试开始后产生的设备时间记录，避免顺序测试串入旧事实。
+ if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue;
+ if (save(test, binding, record)) {
+ String source = String.join(":", java.util.Objects.toString(binding.getAdapterCode(), ""),
+ java.util.Objects.toString(binding.getDevtype(), ""), java.util.Objects.toString(binding.getDevid(), ""),
+ java.util.Objects.toString(binding.getSubdevid(), ""), java.util.Objects.toString(binding.getChlid(), ""));
+ warningExceptions.evaluate(test, "CABINET", source, "VOLTAGE", BigDecimal.valueOf(record.voltageV()), "V", Date.from(record.deviceTime()));
+ warningExceptions.evaluate(test, "CABINET", source, "CURRENT", BigDecimal.valueOf(record.currentA()), "A", Date.from(record.deviceTime()));
+ warningExceptions.evaluate(test, "CABINET", source, "CAPACITY", BigDecimal.valueOf(record.capacityAh()), "Ah", Date.from(record.deviceTime()));
+ warningExceptions.evaluate(test, "CABINET", source, "ENERGY", BigDecimal.valueOf(record.energyWh()), "Wh", Date.from(record.deviceTime()));
+ }
}
}
}
- private void save(TTest test, TTestBattery battery, TTestChannelBinding binding,
- TTestReceptionInterval interval, BtsHistoryRecord record)
+ private void captureCollectorMeasurements(TTest test, TTestBattery battery)
+ {
+ if (battery.getCollectorId() == null || battery.getCollectorChannels() == null
+ || battery.getCollectorChannels().isEmpty()) return;
+ final CollectorRealtimeSnapshot realtime;
+ try { realtime = simulator.collectorRealtime(battery.getCollectorId().intValue()); }
+ catch (IllegalArgumentException ignored) { return; }
+ if (!realtime.online()) return;
+ Date startedAt = test.getStartedAtUtc();
+ for (CollectorRealtimeMeasurement measurement : realtime.measurements())
+ {
+ if (startedAt != null && measurement.deviceTime().isBefore(startedAt.toInstant())) continue;
+ // 继续接收只接受恢复动作之后查询到的快照；暂停期间采集器自行产生的历史不回补。
+ if (test.getUpdateTime() != null && measurement.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())) continue;
+ boolean mapped = battery.getCollectorChannels().stream().anyMatch(mapping ->
+ java.util.Objects.equals(mapping.getChannelCode(), measurement.channelCode())
+ && metricMatches(mapping.getMeasurementCode(), measurement.metricCode()));
+ if (!mapped) continue;
+ if (saveCollector(test, battery, measurement))
+ warningExceptions.evaluate(test, "COLLECTOR", measurement.adapterCode() + ":" + battery.getCollectorId() + ":" + measurement.channelCode(), measurement.metricCode(), measurement.value(), measurement.unit(), Date.from(measurement.deviceTime()));
+ }
+ }
+
+ private boolean saveCollector(TTest test, TTestBattery battery, CollectorRealtimeMeasurement measurement)
+ {
+ byte[] hash = hash(measurement);
+ String old = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(),
+ measurement.sequenceId(), measurement.metricCode());
+ String expected = HexFormat.of().formatHex(hash);
+ if (old != null)
+ {
+ if (!old.equalsIgnoreCase(expected))
+ mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(),
+ "采集器实时接收内容冲突 collector=" + battery.getCollectorId() + ", channel="
+ + measurement.channelCode() + ", metric=" + measurement.metricCode()
+ + ", seqid=" + measurement.sequenceId());
+ return false;
+ }
+ Date receivedAt = new Date();
+ int inserted = mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(), battery.getCollectorId(),
+ measurement.adapterCode(),
+ measurement.channelCode(), measurement.sequenceId(), Date.from(measurement.deviceTime()),
+ measurement.deviceTimeRaw(), receivedAt, measurement.metricCode(),
+ measurement.value(), measurement.unit(), hash);
+ // 唯一键负责并发幂等；内容冲突保留首次事实，不覆盖原始记录。
+ if (inserted == 0)
+ {
+ String actual = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(),
+ measurement.sequenceId(), measurement.metricCode());
+ if (actual != null && !actual.equalsIgnoreCase(expected))
+ mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(),
+ "采集器实时并发内容冲突 collector=" + battery.getCollectorId() + ", channel="
+ + measurement.channelCode() + ", metric=" + measurement.metricCode()
+ + ", seqid=" + measurement.sequenceId());
+ }
+ return inserted > 0;
+ }
+
+ private boolean save(TTest test, TTestChannelBinding binding,
+ BtsHistoryRecord record)
{
// 模拟器历史 testid 按通道固定；接收事实必须再带测试作用域，避免顺序测试共用全局幂等键。
String btsTestId = "SIM-" + binding.getDevid() + "-" + binding.getChlid() + "-T" + test.getTestId();
byte[] hash = hash(record);
- String old = mapper.selectSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
+ String old = mapper.selectBtsSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
if (old != null)
{
if (!old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
- mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
+ mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
"实时接收内容冲突 testid=" + btsTestId + ", seqid=" + record.sequenceId());
- return;
+ return false;
}
- mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(), interval.getIntervalId(),
+ int inserted = mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(),
btsTestId, record.sequenceId(), Date.from(record.deviceTime()),
record.deviceTime().toString(), new Date(), record.cycleNo(), record.stepNo(), record.stepType(),
record.stepName(), null,
BigDecimal.valueOf(record.voltageV()), BigDecimal.valueOf(record.currentA()),
BigDecimal.valueOf(record.capacityAh()), BigDecimal.valueOf(record.energyWh()), hash);
+ if (inserted == 0)
+ {
+ String actual = mapper.selectBtsSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
+ if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
+ mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
+ "实时接收并发内容冲突 testid=" + btsTestId + ", seqid=" + record.sequenceId());
+ }
+ return inserted > 0;
}
private byte[] hash(BtsHistoryRecord record)
@@ -107,4 +290,27 @@
}
catch (Exception exception) { throw new IllegalStateException("无法计算采样内容摘要", exception); }
}
+
+ private byte[] hash(CollectorRealtimeMeasurement measurement)
+ {
+ try
+ {
+ String value = measurement.adapterCode() + "|" + measurement.collectorCode() + "|"
+ + measurement.channelCode() + "|" + measurement.sequenceId() + "|"
+ + measurement.deviceTimeRaw() + "|" + measurement.metricCode() + "|"
+ + measurement.value() + "|" + measurement.unit();
+ return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8));
+ }
+ catch (Exception exception) { throw new IllegalStateException("无法计算采集器采样内容摘要", exception); }
+ }
+
+ private boolean metricMatches(String configured, String actual)
+ {
+ if (configured == null || actual == null) return false;
+ if (configured.equals(actual)) return true;
+ String normalized = configured.toUpperCase(java.util.Locale.ROOT);
+ return (normalized.contains("TEMPERATURE") && "TEMPERATURE".equals(actual.toUpperCase(java.util.Locale.ROOT)))
+ || (normalized.contains("POSITIVE_POTENTIAL") && "POSITIVE_POTENTIAL".equals(actual.toUpperCase(java.util.Locale.ROOT)))
+ || (normalized.contains("NEGATIVE_POTENTIAL") && "NEGATIVE_POTENTIAL".equals(actual.toUpperCase(java.util.Locale.ROOT)));
+ }
}
--
Gitblit v1.7.1