From 27428b31edee97703b529559146cdd7eef87a7a9 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 11:00:58 +0800
Subject: [PATCH] 完成测试数据接收持久化与查询
---
ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java | 422 +++++++++++++++++++---------------------------------
1 files changed, 156 insertions(+), 266 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 aabbe55..7aa0abf 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
@@ -3,329 +3,219 @@
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
-import java.util.Date;
+import java.time.Clock;
import java.time.Instant;
-import java.util.Map;
-import java.util.LinkedHashMap;
+import java.util.Date;
import java.util.HexFormat;
+import java.util.LinkedHashMap;
import java.util.List;
-import com.alibaba.fastjson2.JSON;
-import org.springframework.stereotype.Service;
+import java.util.Locale;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.lang.Nullable;
+import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
+import com.alibaba.fastjson2.JSON;
+import com.battery.device.api.TestDataObservationAdapter;
+import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
+import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
+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;
-/** 接收时确定测试归属，分别保留检测柜、采集器原始事实，并形成统一实时快照。 */
+/** 接收期间将成功观测直接归属测试，并写入实时、工步和循环事实表。 */
@Service
public class TestDataReceptionServiceImpl implements ITestDataReceptionService
{
+ private static final Logger log = LoggerFactory.getLogger(TestDataReceptionServiceImpl.class);
private final TTestMapper mapper;
private final ITTestService tests;
- private final BtsDevelopmentSimulator simulator;
+ private final TestDataObservationAdapter adapter;
private final TestWarningExceptionService warningExceptions;
+ private final Clock clock;
private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>();
- private final Map<Long, Instant> lastRawCaptureAt = new java.util.concurrent.ConcurrentHashMap<>();
@Autowired
- public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable BtsDevelopmentSimulator simulator,
+ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
TestWarningExceptionService warningExceptions)
- {
- this.mapper = mapper;
- this.tests = tests;
- this.simulator = simulator;
- this.warningExceptions = warningExceptions;
- }
+ { 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, BtsDevelopmentSimulator simulator)
- { this(mapper, tests, simulator, new TestWarningExceptionService(mapper)); }
+ { this(mapper, tests, new LegacySimulatorAdapter(simulator), new TestWarningExceptionService(mapper), Clock.systemUTC()); }
- @Override
- @Transactional(rollbackFor = Exception.class)
+ private TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter,
+ TestWarningExceptionService warningExceptions, Clock clock)
+ { this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; }
+
+ /** 轮询接收中测试，只保存本次成功观测的实时槽位及设备直接返回汇总。 */
+ @Override @Transactional(rollbackFor = Exception.class)
public void captureDevelopmentRealtimeSnapshots()
{
- if (simulator == null) return;
- 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 || !"RUNNING".equals(locked.getTestStatus())
- || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
- TTest test = tests.selectTestById(testId);
- if (test == null || !"RUNNING".equals(test.getTestStatus())
- || !"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);
- }
+ for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); }
}
- 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.getReceptionBoundaryAtUtc() != null && cabinet.serverReceivedAt() != null
- && cabinet.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().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.getReceptionBoundaryAtUtc() != null
- && (temp.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
- || pos.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().toInstant())
- || neg.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().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)
+ /** 接收检测柜直接返回的实际工步和循环记录，不从实时数据推导。 */
+ @Override @Transactional(rollbackFor = Exception.class)
public void captureDevelopmentCabinetRecords()
- {
- if (simulator == null) return;
- List<Long> testIds = mapper.selectReceivingTestIds();
- if (testIds == null) return;
- for (Long testId : testIds)
- {
- // 锁住测试行直到本轮采样写入完成，暂停/停止动作只能在本轮接收提交后生效。
- TTest locked = mapper.selectTestByIdForUpdate(testId);
- if (locked == null || !"RUNNING".equals(locked.getTestStatus())
- || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
- TTest test = tests.selectTestById(testId);
- if (test == null || !"RUNNING".equals(test.getTestStatus())
- || !"RECEIVING".equals(test.getReceptionStatus())) continue;
- captureTest(test);
- }
- }
+ { for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery); }
- @Override
- @Transactional(rollbackFor = Exception.class)
+ /** 接收采集器实时指标并并入测试电池轮询槽位，不读取历史补采。 */
+ @Override @Transactional(rollbackFor = Exception.class)
public void captureDevelopmentCollectorMeasurements()
+ { for (TTest test : receivingTests()) for (TTestBattery battery : batteries(test)) saveRealtime(test, battery); }
+
+ private List<TTest> receivingTests()
{
- if (simulator == null) return;
- List<Long> testIds = mapper.selectReceivingTestIds();
- if (testIds == null) return;
- for (Long testId : testIds)
+ List<Long> ids = mapper.selectReceivingTestIds(); if (ids == null) return List.of();
+ return ids.stream().map(id -> {
+ TTest locked = mapper.selectTestByIdForUpdate(id);
+ if (locked == null || !"RUNNING".equals(locked.getTestStatus()) || !"RECEIVING".equals(locked.getReceptionStatus())) return null;
+ TTest current = tests.selectTestById(id);
+ return current != null && "RUNNING".equals(current.getTestStatus()) && "RECEIVING".equals(current.getReceptionStatus()) ? current : null;
+ }).filter(java.util.Objects::nonNull).toList();
+ }
+ private List<TTestBattery> batteries(TTest test) { return test.getBatteries() == null ? List.of() : test.getBatteries(); }
+
+ private void saveRealtime(TTest test, TTestBattery battery)
+ {
+ 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;
+ Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null);
+ Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval);
+ 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.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));
+ Map<String, Object> extension = new LinkedHashMap<>();
+ for (Map.Entry<String, CollectorMeasurement> e : mapped.entrySet()) if (!isFixedMetric(e.getKey()))
{
- TTest locked = mapper.selectTestByIdForUpdate(testId);
- if (locked == null || !"RUNNING".equals(locked.getTestStatus())
- || !"RECEIVING".equals(locked.getReceptionStatus())) continue;
- TTest test = tests.selectTestById(testId);
- if (test == null || !"RUNNING".equals(test.getTestStatus())
- || !"RECEIVING".equals(test.getReceptionStatus())) continue;
- for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
- captureCollectorMeasurements(test, battery);
+ Map<String, Object> value = new LinkedHashMap<>(); value.put("value", e.getValue().value()); value.put("unit", e.getValue().unit()); value.put("channelCode", e.getValue().channelCode()); extension.put(e.getKey(), value);
}
+ row.setExtensionMeasurementsJson(JSON.toJSONString(extension)); row.setSourceIdentity(sourceIdentity(binding));
+ row.setSourceRecordId(sourceRecordId(cabinet, mapped));
+ row.setContentSha256(hash(row));
+ String old = mapper.selectRealtimeRecordContentHash(test.getTestId(), battery.getTestBatteryId(), row.getCaptureSlotUtc());
+ if (old != null) { if (!old.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256()))) log.warn("实时采样内容冲突，拒绝覆盖 testId={}, testBatteryId={}, captureSlotUtc={}", test.getTestId(), battery.getTestBatteryId(), slot); return; }
+ int inserted = mapper.insertRealtimeRecord(row);
+ if (inserted == 0)
+ {
+ String actual = mapper.selectRealtimeRecordContentHash(test.getTestId(), battery.getTestBatteryId(), row.getCaptureSlotUtc());
+ if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256())))
+ log.warn("实时采样并发内容冲突，拒绝覆盖 testId={}, testBatteryId={}, captureSlotUtc={}", test.getTestId(), battery.getTestBatteryId(), slot);
+ }
+ lastSnapshotAt.put(key, now);
}
- private void captureTest(TTest test)
+ private void saveDeviceAggregates(TTest test, TTestBattery battery)
{
- Date startedAt = test.getStartedAtUtc();
- for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
+ if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
+ ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid());
+ List<StepRecord> steps; List<CycleRecord> cycles; try { steps = adapter.stepRecords(channel); cycles = adapter.cycleRecords(channel); } catch (RuntimeException ex) { return; }
+ Date receivedAt = Date.from(clock.instant()); String identity = sourceIdentity(binding);
+ for (StepRecord source : steps == null ? List.<StepRecord>of() : steps)
{
- TTestChannelBinding binding = battery.getBinding();
- if (binding == null || binding.getDevid() == null || binding.getChlid() == null) continue;
- List<BtsHistoryRecord> records;
- try { records = simulator.btsRecords(binding.getDevid(), binding.getChlid()); }
- catch (IllegalArgumentException ignored) { continue; }
- for (BtsHistoryRecord record : records)
+ 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 (mapper.insertStepRecord(row) == 0)
{
- // 设备时间隔离顺序测试，服务器接收边界排除暂停期间历史；两种时间不可交叉比较。
- if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue;
- if (test.getReceptionBoundaryAtUtc() != null && record.serverReceivedAt() != null
- && record.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().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()));
- }
+ 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());
+ }
+ }
+ 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 (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());
}
}
}
- private void captureCollectorMeasurements(TTest test, TTestBattery battery)
+ 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 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 static String sourceRecordId(CabinetRealtime cabinet, Map<String, CollectorMeasurement> measurements)
{
- 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.getReceptionBoundaryAtUtc() != null
- && measurement.serverReceivedAt().isBefore(test.getReceptionBoundaryAtUtc().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()));
- }
+ List<String> parts = new java.util.ArrayList<>();
+ if (cabinet != null) parts.add("cabinet:" + cabinet.sequenceId());
+ measurements.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo)
+ .ifPresent(sequence -> parts.add("collector:" + sequence));
+ return String.join(":", parts);
+ }
+ private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()))); } catch (RuntimeException ex) { return 60000L; } }
+ 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)
+ {
+ if (deviceTime != null && test.getStartedAtUtc() != null && deviceTime.isBefore(test.getStartedAtUtc().toInstant())) return false;
+ return serverReceivedAt == null || test.getReceptionBoundaryAtUtc() == null || !serverReceivedAt.isBefore(test.getReceptionBoundaryAtUtc().toInstant());
}
- private boolean saveCollector(TTest test, TTestBattery battery, CollectorRealtimeMeasurement measurement)
+ /** 将开发模拟器的历史扩展映射到业务观测契约，仅供兼容测试构造使用。 */
+ private static final class LegacySimulatorAdapter implements TestDataObservationAdapter
{
- 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)
+ 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)
{
- 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;
+ 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);
}
- 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)
+ @Override public CollectorRealtime latestCollector(long collectorId)
{
- 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());
+ 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());
}
- return inserted > 0;
+ @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()); }
}
- 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.selectBtsSampleContentHash(binding.getChannelId(), btsTestId, record.sequenceId());
- if (old != null)
- {
- if (!old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
- mapper.insertSampleConflict(test.getTestId(), binding.getChannelId(),
- "实时接收内容冲突 testid=" + btsTestId + ", seqid=" + record.sequenceId());
- return false;
- }
- 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)
- {
- try
- {
- String value = record.sequenceId() + "|" + record.deviceTime() + "|" + record.currentA() + "|"
- + record.voltageV() + "|" + record.capacityAh() + "|" + record.energyWh() + "|"
- + record.cycleNo() + "|" + record.stepNo() + "|" + record.stepType();
- return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8));
- }
- 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)));
- }
+ 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