From 92f32acd847ea1700590bc669750f23ef499840e Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 13:39:44 +0800
Subject: [PATCH] 清理旧测试结构并贯通最终业务数据链路
---
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 112 +++++++++++++++++++++----------------------------------
1 files changed, 43 insertions(+), 69 deletions(-)
diff --git a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
similarity index 65%
rename from ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
rename to battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
index 262fa14..32127cf 100644
--- a/ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
@@ -1,4 +1,4 @@
-package com.ruoyi.system.service.impl;
+package com.battery.business.service.impl;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
@@ -24,22 +24,16 @@
import com.battery.device.api.TestDataObservationAdapter.CollectorMeasurement;
import com.battery.device.api.TestDataObservationAdapter.CollectorRealtime;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
-import com.battery.device.model.ChannelIdentity;
-import com.battery.device.simulator.BtsDevelopmentSimulator;
-import com.battery.device.simulator.BtsDevelopmentSimulator.BtsCycleRecord;
-import com.battery.device.simulator.BtsDevelopmentSimulator.BtsStepRecord;
-import com.battery.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord;
-import com.battery.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
-import com.battery.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.TTestCycleRecord;
-import com.ruoyi.system.domain.TTestRealtimeRecord;
-import com.ruoyi.system.domain.TTestStepRecord;
-import com.ruoyi.system.mapper.TTestMapper;
-import com.ruoyi.system.service.ITTestService;
-import com.ruoyi.system.service.ITestDataReceptionService;
+import com.battery.device.api.model.ChannelIdentity;
+import com.battery.business.domain.TTest;
+import com.battery.business.domain.TTestBattery;
+import com.battery.business.domain.TTestChannelBinding;
+import com.battery.business.domain.TTestCycleRecord;
+import com.battery.business.domain.TTestRealtimeRecord;
+import com.battery.business.domain.TTestStepRecord;
+import com.battery.business.mapper.TTestMapper;
+import com.battery.business.service.ITTestService;
+import com.battery.business.service.ITestDataReceptionService;
/** 接收期间将成功观测直接归属测试，并写入实时、工步和循环事实表。 */
@Service
@@ -59,11 +53,11 @@
{ this(mapper, tests, adapter, warningExceptions, Clock.systemUTC()); }
/** 允许测试固定接收时间，生产运行仍使用 UTC 系统时钟。 */
- public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator, Clock clock)
- { this(mapper, tests, new LegacySimulatorAdapter(simulator), new TestWarningExceptionService(mapper), clock); }
+ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter, Clock clock)
+ { this(mapper, tests, adapter, new TestWarningExceptionService(mapper), clock); }
- public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator)
- { this(mapper, tests, new LegacySimulatorAdapter(simulator), new TestWarningExceptionService(mapper), Clock.systemUTC()); }
+ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter)
+ { this(mapper, tests, adapter, new TestWarningExceptionService(mapper), Clock.systemUTC()); }
private TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter,
TestWarningExceptionService warningExceptions, Clock clock)
@@ -71,20 +65,10 @@
/** 轮询接收中测试，只保存本次成功观测的实时槽位及设备直接返回汇总。 */
@Override @Transactional(rollbackFor = Exception.class)
- public void captureDevelopmentRealtimeSnapshots()
+ public void captureReceivedData()
{
for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); }
}
-
- /** 接收检测柜直接返回的实际工步和循环记录，不从实时数据推导。 */
- @Override @Transactional(rollbackFor = Exception.class)
- public void captureDevelopmentCabinetRecords()
- { for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery); }
-
- /** 接收采集器实时指标并并入测试电池轮询槽位，不读取历史补采。 */
- @Override @Transactional(rollbackFor = Exception.class)
- public void captureDevelopmentCollectorMeasurements()
- { for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) saveRealtime(test, battery); }
private List<TTest> receivingTests()
{
@@ -112,12 +96,14 @@
if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return;
Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null);
Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval);
+ // 删除修订随数据库保留；重启后同一槽位也不能重新出现。
+ if (mapper.wasRealtimeSlotDeleted(test.getTestId(), battery.getTestBatteryId(), slot.toEpochMilli())) return;
TTestRealtimeRecord row = new TTestRealtimeRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId());
row.setAdapterCode(binding.getAdapterCode()); row.setCaptureSlotUtc(Date.from(slot)); row.setServerReceivedAtUtc(Date.from(now));
- if (cabinet != null) { row.setCabinetDevtype(binding.getDevtype()); row.setCabinetDevid(binding.getDevid()); row.setCabinetSubdevid(binding.getSubdevid()); row.setCabinetChlid(binding.getChlid()); row.setCabinetAdapterTimezoneId(timezone(binding)); }
- if (collector != null && collector.online() && !mapped.isEmpty()) { row.setCollectorId(battery.getCollectorId()); row.setCollectorAdapterTimezoneId(timezone(binding)); }
- row.setCabinetSequenceId(cabinet == null ? null : cabinet.sequenceId()); row.setCabinetDeviceTimeUtc(cabinet == null ? null : Date.from(cabinet.deviceTime())); row.setCabinetDeviceTimeRaw(cabinet == null ? null : cabinet.deviceTime().toString());
- row.setCollectorSequenceId(mapped.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo).orElse(null)); row.setCollectorDeviceTimeUtc(collectorTime == null ? null : Date.from(collectorTime)); row.setCollectorDeviceTimeRaw(collectorTime == null ? null : collectorTime.toString());
+ if (cabinet != null) { row.setCabinetDevtype(binding.getDevtype()); row.setCabinetDevid(binding.getDevid()); row.setCabinetSubdevid(binding.getSubdevid()); row.setCabinetChlid(binding.getChlid()); row.setCabinetAdapterTimezoneId(cabinet.timezoneId()); }
+ if (collector != null && collector.online() && !mapped.isEmpty()) { row.setCollectorId(battery.getCollectorId()); row.setCollectorAdapterTimezoneId(collector.timezoneId()); }
+ row.setCabinetSequenceId(cabinet == null ? null : cabinet.sequenceId()); row.setCabinetDeviceTimeUtc(cabinet == null ? null : Date.from(cabinet.deviceTime())); row.setCabinetDeviceTimeRaw(cabinet == null ? null : cabinet.deviceTimeRaw());
+ 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));
// 只有本次观测成功取得的值进入阈值判断；离线、请求失败和缺失值在前面被过滤。
@@ -171,28 +157,35 @@
Date receivedAt = Date.from(clock.instant()); String identity = sourceIdentity(binding);
for (StepRecord source : steps == null ? List.<StepRecord>of() : steps)
{
- if (!accepted(test, source.endedAt(), null)) continue;
- TTestStepRecord row = new TTestStepRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setAdapterTimezoneId(timezone(binding)); row.setCycleNo(source.cycleNo()); row.setStepNo(source.stepNo()); row.setStepType(source.stepType()); row.setStepName(source.stepName()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.endedAt().toString()); row.setServerReceivedAtUtc(receivedAt); row.setVoltageV(BigDecimal.valueOf(source.endVoltageV())); row.setCapacityAh(BigDecimal.valueOf(source.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(source.energyWh())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() + ":" + source.stepNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
+ if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
+ TTestStepRecord row = new TTestStepRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setAdapterTimezoneId(source.timezoneId()); row.setCycleNo(source.cycleNo()); row.setStepNo(source.stepNo()); row.setStepType(source.stepType()); row.setStepName(source.stepName()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.deviceTimeRaw()); row.setServerReceivedAtUtc(Date.from(source.serverReceivedAt())); row.setVoltageV(BigDecimal.valueOf(source.endVoltageV())); row.setCapacityAh(BigDecimal.valueOf(source.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(source.energyWh())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() + ":" + source.stepNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
+ // 物理删除依靠已有修订追溯识别来源，重复观测不能把已删除汇总重新写回。
+ row.setExtensionMeasurementsJson(JSON.toJSONString(source.measurements()));
+ row.setContentSha256(hash(row));
+ if (mapper.wasAggregateDeleted(test.getTestId(), "STEP", battery.getTestBatteryId(), identity, row.getDedupeKey())) continue;
if (mapper.insertStepRecord(row) == 0)
{
String actual = mapper.selectStepRecordContentHash(row.getTestBatteryId(), row.getSourceIdentity(), row.getDedupeKey());
- log.warn(actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())) ? "工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}" : "工步记录重复，保留首次事实 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
+ if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
+ log.warn("工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
}
}
for (CycleRecord source : cycles == null ? List.<CycleRecord>of() : cycles)
{
- if (!accepted(test, source.endedAt(), null)) continue;
- TTestCycleRecord row = new TTestCycleRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setAdapterTimezoneId(timezone(binding)); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setCycleNo(source.cycleNo()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.endedAt().toString()); row.setServerReceivedAtUtc(receivedAt); row.setChargeCapacityAh(BigDecimal.valueOf(source.chargeCapacityAh())); row.setDischargeCapacityAh(BigDecimal.valueOf(source.dischargeCapacityAh())); row.setChargeEnergyWh(BigDecimal.valueOf(source.chargeEnergyWh())); row.setDischargeEnergyWh(BigDecimal.valueOf(source.dischargeEnergyWh())); row.setEfficiency(BigDecimal.valueOf(source.efficiency())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
+ if (!accepted(test, source.endedAt(), source.serverReceivedAt())) continue;
+ TTestCycleRecord row = new TTestCycleRecord(); row.setTestId(test.getTestId()); row.setTestBatteryId(battery.getTestBatteryId()); row.setBindingId(binding.getBindingId()); row.setAdapterCode(binding.getAdapterCode()); row.setSourceIdentity(identity); row.setAdapterTimezoneId(source.timezoneId()); row.setSourceRecordId(blankToNull(source.sourceRecordId())); row.setCycleNo(source.cycleNo()); row.setDeviceTimeUtc(Date.from(source.endedAt())); row.setDeviceTimeRaw(source.deviceTimeRaw()); row.setServerReceivedAtUtc(Date.from(source.serverReceivedAt())); row.setChargeCapacityAh(BigDecimal.valueOf(source.chargeCapacityAh())); row.setDischargeCapacityAh(BigDecimal.valueOf(source.dischargeCapacityAh())); row.setChargeEnergyWh(BigDecimal.valueOf(source.chargeEnergyWh())); row.setDischargeEnergyWh(BigDecimal.valueOf(source.dischargeEnergyWh())); row.setEfficiency(BigDecimal.valueOf(source.efficiency())); row.setDedupeKey(row.getSourceRecordId() == null ? "fallback:" + source.endedAt() + ":" + source.cycleNo() : "record:" + row.getSourceRecordId()); row.setContentSha256(hash(row));
+ if (mapper.wasAggregateDeleted(test.getTestId(), "CYCLE", battery.getTestBatteryId(), identity, row.getDedupeKey())) continue;
if (mapper.insertCycleRecord(row) == 0)
{
String actual = mapper.selectCycleRecordContentHash(row.getTestBatteryId(), row.getSourceIdentity(), row.getDedupeKey());
- log.warn(actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())) ? "循环记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}" : "循环记录重复，保留首次事实 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
+ if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
+ log.warn("循环记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
}
}
}
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(b.getCollectorId()); } 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); }
@@ -212,34 +205,15 @@
private 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());
+ return serverReceivedAt != null && (test.getReceptionBoundaryAtUtc() == null || !serverReceivedAt.isBefore(test.getReceptionBoundaryAtUtc().toInstant()));
}
- /** 将开发模拟器的历史扩展映射到业务观测契约，仅供兼容测试构造使用。 */
- private static final class LegacySimulatorAdapter implements TestDataObservationAdapter
+ private byte[] hash(Object value)
{
- private final BtsDevelopmentSimulator simulator;
- private LegacySimulatorAdapter(BtsDevelopmentSimulator simulator) { this.simulator = simulator; }
- @Override public String adapterCode() { return simulator.adapterCode(); }
- @Override public com.battery.device.model.AdapterCapabilities capabilities() { return simulator.capabilities(); }
- @Override public List<com.battery.device.model.DetectedChannel> discoverChannels() { return simulator.discoverChannels(); }
- @Override public Map<ChannelIdentity, com.battery.device.model.DeviceObservation> observe(java.util.Set<ChannelIdentity> channels) { return simulator.observe(channels); }
- @Override public CabinetRealtime latestCabinet(ChannelIdentity channel)
- {
- return simulator.btsRecords(channel.devid(), channel.chlid()).stream()
- .max(java.util.Comparator.comparing(BtsHistoryRecord::deviceTime)).map(r -> new CabinetRealtime(r.sequenceId(), r.deviceTime(), r.currentA(), r.voltageV(), r.capacityAh(), r.energyWh(), r.cycleNo(), r.stepNo(), r.stepName(), r.stepType(), r.serverReceivedAt())).orElse(null);
- }
- @Override public CollectorRealtime latestCollector(long collectorId)
- {
- CollectorRealtimeSnapshot snapshot = simulator.collectorRealtime((int) collectorId);
- return new CollectorRealtime(snapshot.collectorCode(), snapshot.online(), snapshot.sampleIntervalMs(), snapshot.measurements().stream().map(m -> new CollectorMeasurement(m.channelCode(), m.sequenceId(), m.deviceTime(), m.deviceTimeRaw(), m.serverReceivedAt(), m.metricCode(), m.value(), m.unit())).toList());
- }
- @Override public List<StepRecord> stepRecords(ChannelIdentity channel)
- { return simulator.btsSteps(channel.devid(), channel.chlid()).stream().map(s -> new StepRecord(s.recordNo(), s.cycleNo(), s.stepNo(), s.stepName(), s.stepType(), s.startedAt(), s.endedAt(), s.endVoltageV(), s.capacityAh(), s.energyWh())).toList(); }
- @Override public List<CycleRecord> cycleRecords(ChannelIdentity channel)
- { return simulator.btsCycles(channel.devid(), channel.chlid()).stream().map(s -> new CycleRecord(s.recordNo(), s.cycleNo(), s.startedAt(), s.endedAt(), s.chargeCapacityAh(), s.dischargeCapacityAh(), s.chargeEnergyWh(), s.dischargeEnergyWh(), s.efficiency())).toList(); }
- @Override public long persistenceIntervalMillis(ChannelIdentity channel) { return simulator.persistenceIntervalMillis(channel.devid(), channel.chlid()); }
+ // 重复观测比较设备内容，不把服务器轮询时间作为内容变化。
+ Map<String, Object> content = new java.util.TreeMap<>(JSON.parseObject(JSON.toJSONString(value)));
+ content.remove("serverReceivedAtUtc");
+ try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content)); }
+ catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); }
}
-
- private byte[] hash(Object value) { try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONString(value).getBytes(StandardCharsets.UTF_8)); } catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); } }
}
--
Gitblit v1.7.1