From b78295e8da96c3e5e76befeef9582ce5273dbb74 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期二, 29 九月 2026 13:10:23 +0800
Subject: [PATCH] 按代码规范检查整改实体注释、Service Javadoc 与分层边界
---
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 291 ++++++++++++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 252 insertions(+), 39 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..5a77983 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
@@ -23,6 +23,7 @@
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.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.business.domain.TTest;
@@ -46,7 +47,12 @@
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<>();
+ /** 每个测试电池上次读取工步与循环的时刻；读取节奏对齐来源写库周期，不跟着调度节奏。 */
+ private final Map<String, Instant> lastAggregateAt = new java.util.concurrent.ConcurrentHashMap<>();
+ /** 每个测试电池各自的接收水位；只存在内存中，重启后按来源里的最新一行重新取水位。 */
+ private final Map<String, SampleCursor> sampleCursors = new java.util.concurrent.ConcurrentHashMap<>();
+
@Autowired
public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter,
@@ -64,19 +70,29 @@
TestWarningExceptionService warningExceptions, Clock clock)
{ this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; }
- /** 轮询接收中测试，记录绑定来源离线异常，并保存成功观测的实时槽位及设备汇总。 */
+ /**
+ * 接收调度入口：对全部接收中的测试判定实时超限与来源离线异常，并按各自节奏保存成功观测。
+ *
+ * <p>前置条件：测试处于“检测中 + 接收中”。预警判定按约 5 秒节奏，柜端逐行接收和工步、
+ * 循环汇总按来源写库节奏限流；只有成功观测且可归属本测试的数据才写入事实表，不执行历史补采。</p>
+ *
+ * @throws org.springframework.dao.DataAccessException 任一写入失败时整个接收事务回滚，本轮不保存任何测试的数据
+ */
@Override @Transactional(rollbackFor = Exception.class)
public void captureReceivedData()
{
for (TTest test : receivingTests()) {
Instant now = clock.instant();
- Instant previous = lastOfflineCheckAt.get(test.getTestId());
- // 离线检查独立于测量采样槽位，并与暂停、停止和删除共享测试事务锁。
+ Instant previous = lastRealtimeCheckAt.get(test.getTestId());
+ // STEP 1: 按约 5 秒节奏判定当前来源离线事实与实时超限，预警不受采样入库周期限制。
+ // 实时预警独立于持久化槽位，并与暂停、停止和删除共享测试事务锁。
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); }
+ // STEP 2: 单独按来源写库节奏读取工步与循环汇总，避免每秒重查厂商库。
+ for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery);
}
}
@@ -92,22 +108,186 @@
}
private List<TTestBattery> batteries(TTest test) { return test.getBatteries() == null ? List.of() : test.getBatteries(); }
+ /**
+ * 保存单个测试电池的一次实时观测：来源支持设备数据行时逐行落库，否则回落到轮询槽位。
+ *
+ * <p>前置条件：测试正在接收且绑定通道完整。柜端与采集器分别判断成功，失败来源的字段保持为空。</p>
+ *
+ * @param test 接收中的测试
+ * @param battery 测试内的一个电池号及其通道绑定
+ */
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;
+ // STEP 1: 优先按来源水位逐行接收柜端事实（同秒多行都保留），槽位只留给采集器端。
+ boolean rowBased = persistCabinetSamples(test, battery, binding);
+ // STEP 2: 来源不按设备数据行接收时，落到轮询槽位，保持原来的组合观测语义。
+ saveRealtimeSlot(test, battery, binding, !rowBased);
+ }
+
+ /**
+ * 按来源水位逐行接收柜端事实。水位是"开始接收"或"继续接收"这一刻来源里的最新一行，
+ * 因此水位之前的数据行即使仍在来源里也不进本测试，暂停期间的空档也不会被补回来。
+ *
+ * <p>前置条件：调用方持有该测试电池的接收游标。读取节奏按来源写库周期限流；
+ * 实例缺失、来源不可达或读取失败时明确失败并留痕，不回退到其他实例。</p>
+ *
+ * @param test 接收中的测试
+ * @param battery 测试内的一个电池号
+ * @param binding 该电池号绑定的检测柜通道
+ * @return 该来源是否按设备数据行接收；来源不支持或被拒绝时返回 false，调用方回落到轮询槽位
+ */
+ private boolean persistCabinetSamples(TTest test, TTestBattery battery, TTestChannelBinding binding)
+ {
+ String key = test.getTestId() + ":" + battery.getTestBatteryId();
+ Instant boundary = test.getReceptionBoundaryAtUtc() == null ? null : test.getReceptionBoundaryAtUtc().toInstant();
+ SampleCursor cursor = sampleCursors.computeIfAbsent(key, ignored -> new SampleCursor());
+ synchronized (cursor)
+ {
+ // STEP 1: 开始接收与继续接收都会推进接收边界，边界变化即重新取水位，暂停期间的数据不补采。
+ if (!java.util.Objects.equals(cursor.boundary, boundary))
+ {
+ cursor.boundary = boundary; cursor.watermark = null; cursor.rowBased = null; cursor.sampledAt = null;
+ }
+ if (Boolean.FALSE.equals(cursor.rowBased)) return false;
+ ChannelIdentity channel = channelIdentity(binding);
+ Instant now = clock.instant();
+ long interval = persistenceInterval(binding);
+ // 厂商写库节奏是 60 秒，取数不必更快；展示节奏由控制台轮询单独承担。
+ if (cursor.sampledAt != null && now.isBefore(cursor.sampledAt.plusMillis(interval)))
+ return Boolean.TRUE.equals(cursor.rowBased);
+ try
+ {
+ // STEP 2: 首次接触该边界时只记录水位，不把水位之前的历史行带入本测试。
+ if (cursor.watermark == null)
+ {
+ cursor.sampledAt = now;
+ SampleWatermark watermark = adapter.currentWatermark(channel);
+ if (watermark == null)
+ {
+ // 来源不按设备数据行接收（例如开发模拟器）：柜端事实继续走轮询槽位。
+ cursor.rowBased = false;
+ return false;
+ }
+ cursor.watermark = watermark;
+ cursor.rowBased = true;
+ return true;
+ }
+ // STEP 3: 逐行写入水位之后的设备事实，每行成功后推进水位。
+ List<CabinetRealtime> rows = adapter.samplesSince(channel, cursor.watermark);
+ cursor.sampledAt = now;
+ for (CabinetRealtime row : rows)
+ {
+ insertCabinetSample(test, battery, binding, row);
+ cursor.watermark = new SampleWatermark(row.deviceTime(), row.sequenceId());
+ }
+ return true;
+ }
+ catch (RuntimeException exception)
+ {
+ // 实例缺失、来源不可达或读取失败：明确失败并留痕，下一轮重试；不回退到其他实例。
+ log.warn("柜端逐行接收失败 testId={}, bindingId={}, adapterCode={}", test.getTestId(),
+ binding.getBindingId(), binding.getAdapterCode(), exception);
+ cursor.sampledAt = now;
+ return Boolean.TRUE.equals(cursor.rowBased);
+ }
+ }
+ }
+
+ /**
+ * 把一行设备事实写成一条实时记录；来源行身份保证重复接收幂等、同身份不同内容报冲突。
+ * 归属边界由水位在设备时间里判定（见 {@link #persistCabinetSamples}），
+ * 这里不用服务器取数时间反推设备时间。
+ *
+ * @param test 接收中的测试
+ * @param battery 测试内的一个电池号
+ * @param binding 该电池号绑定的检测柜通道
+ * @param row 来源返回的一行柜端事实；设备时间或服务器接收时间缺失时丢弃不写
+ */
+ private void insertCabinetSample(TTest test, TTestBattery battery, TTestChannelBinding binding, CabinetRealtime row)
+ {
+ if (row.deviceTime() == null || row.serverReceivedAt() == null) return;
+ // STEP 1: 先用本次成功观测判定实时超限，异常不依赖采样入库周期。
+ Date observedAt = Date.from(row.serverReceivedAt());
+ evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(row.voltageV()), "V", observedAt);
+ evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(row.currentA()), "A", observedAt);
+ evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(row.capacityAh()), "Ah", observedAt);
+ evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(row.energyWh()), "Wh", observedAt);
+ // STEP 2: 按来源身份组装记录，槽位留空以允许同秒多行各自成立。
+ String identity = sourceIdentity(binding);
+ String recordId = row.sourceRecordId() == null || row.sourceRecordId().isBlank()
+ ? "cabinet:" + row.sequenceId() : row.sourceRecordId();
+ TTestRealtimeRecord value = new TTestRealtimeRecord();
+ value.setTestId(test.getTestId()); value.setTestBatteryId(battery.getTestBatteryId());
+ value.setBindingId(binding.getBindingId()); value.setAdapterCode(binding.getAdapterCode());
+ // 设备数据行不是系统轮询槽位，槽位留空：同秒多行因此可以各自成立（ADR-0051 的槽位可空机制）。
+ value.setCaptureSlotUtc(null); value.setServerReceivedAtUtc(observedAt);
+ value.setCabinetDevtype(binding.getDevtype()); value.setCabinetDevid(binding.getDevid());
+ value.setCabinetSubdevid(binding.getSubdevid()); value.setCabinetChlid(binding.getChlid());
+ value.setCabinetAdapterTimezoneId(row.timezoneId());
+ value.setCabinetSequenceId(row.sequenceId());
+ value.setCabinetDeviceTimeUtc(Date.from(row.deviceTime()));
+ value.setCabinetDeviceTimeRaw(row.deviceTimeRaw());
+ value.setCycleNo(row.cycleNo()); value.setStepNo(row.stepNo()); value.setStepType(row.stepType());
+ value.setStepName(row.stepName());
+ value.setVoltageV(BigDecimal.valueOf(row.voltageV())); value.setCurrentA(BigDecimal.valueOf(row.currentA()));
+ value.setCapacityAh(BigDecimal.valueOf(row.capacityAh())); value.setEnergyWh(BigDecimal.valueOf(row.energyWh()));
+ Map<String, Object> extension = new LinkedHashMap<>();
+ if (row.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", row.relativeTimeSeconds());
+ value.setExtensionMeasurementsJson(JSON.toJSONString(extension));
+ value.setSourceIdentity(identity); value.setSourceRecordId(recordId);
+ value.setContentSha256(hash(value));
+ // STEP 3: 按来源行身份幂等；同内容重复接收忽略，同身份但内容不同时保留原事实并告警，不覆盖。
+ String existing = mapper.selectRealtimeRecordHashBySourceRecord(battery.getTestBatteryId(), identity, recordId);
+ if (existing != null)
+ {
+ if (!existing.equalsIgnoreCase(HexFormat.of().formatHex(value.getContentSha256())))
+ log.warn("实时记录内容冲突，拒绝覆盖 testId={}, testBatteryId={}, sourceRecordId={}",
+ test.getTestId(), battery.getTestBatteryId(), recordId);
+ return;
+ }
+ mapper.insertRealtimeRecord(value);
+ }
+
+ /**
+ * 向轮询槽位写入一条实时采样记录，柜端与采集器端的成功观测在同一行合并保存。
+ *
+ * <p>前置条件：调用方已判定该来源不按设备数据行接收。槽位按持久化周期对齐，同一槽位只写一行；
+ * 柜端和采集器端分别判断成功，失败来源的身份、时间和指标保持为空。</p>
+ *
+ * @param test 接收中的测试
+ * @param battery 测试内的一个电池号
+ * @param binding 该电池号绑定的检测柜通道
+ * @param includeCabinet 本次是否把柜端事实合并进该槽位
+ */
+ private void saveRealtimeSlot(TTest test, TTestBattery battery, TTestChannelBinding binding, boolean includeCabinet)
+ {
+ // STEP 1: 柜端和采集器端分别判断成功，各取最近一次成功观测；同一槽位只写一行，失败来源的身份、时间和指标保持为空。
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);
+ CabinetRealtime cabinet = includeCabinet ? latestCabinet(binding) : null;
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;
+ // STEP 2: 槽位尚未到期或历史行被删除都不能阻断实时告警，当前成功测量先判定预警。
+ 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()));
+ }
+ // STEP 3: 按持久化周期限流并对齐槽位；删除修订随数据库保留，重启后同一槽位也不能重新出现。
+ 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);
- // 删除修订随数据库保留；重启后同一槽位也不能重新出现。
if (mapper.wasRealtimeSlotDeleted(test.getTestId(), battery.getTestBatteryId(), slot.toEpochMilli())) return;
+ // STEP 4: 组装组合观测行，只写成功来源的字段。
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(cabinet.timezoneId()); }
@@ -116,22 +296,9 @@
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<>();
+ // 工步内相对时长由成功柜端观测提供，不以轮询槽位或服务器时间推算。
+ if (cabinet != null && cabinet.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", cabinet.relativeTimeSeconds());
for (Map.Entry<String, CollectorMeasurement> e : mapped.entrySet()) if (!isFixedMetric(e.getKey()))
{
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);
@@ -139,6 +306,7 @@
row.setExtensionMeasurementsJson(JSON.toJSONString(extension)); row.setSourceIdentity(sourceIdentity(binding));
row.setSourceRecordId(sourceRecordId(cabinet, mapped));
row.setContentSha256(hash(row));
+ // STEP 5: 同一槽位内容不相同报冲突并保留原事实；插入被忽略时再核对一次并发写入。
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);
@@ -156,19 +324,34 @@
if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt);
}
- private static Date nowDate(Instant instant) { return Date.from(instant); }
-
+ /**
+ * 保存单个测试电池的工步与循环汇总事实，来源编号或明确的时间组合保证重复观测幂等。
+ *
+ * <p>前置条件：测试正在接收且绑定通道完整。只接受来源返回的统计记录，不从实时采样推算；
+ * 已按授权物理删除的汇总不会因为重复观测被写回。</p>
+ *
+ * @param test 接收中的测试
+ * @param battery 测试内的一个电池号及其通道绑定
+ */
private void saveDeviceAggregates(TTest test, TTestBattery battery)
{
- // 工步、循环只接受设备直接返回的统计记录，并以来源编号或明确的时间组合执行幂等。
+ // STEP 1: 按来源写库节奏限流，避免接收调度每秒重查厂商库。
+ // 工步、循环只接受来源返回的统计记录，并以来源编号或明确的时间组合执行幂等。
if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
+ String key = test.getTestId() + ":" + battery.getTestBatteryId();
+ // 接收调度每秒跑一次，而来源按自己的写库节奏产出工步：不按来源节奏限流会每秒重查厂商库。
+ Instant now = clock.instant(); Instant previous = lastAggregateAt.get(key);
+ if (previous != null && now.isBefore(previous.plusMillis(persistenceInterval(binding)))) return;
+ lastAggregateAt.put(key, now);
+ // STEP 2: 读取来源的工步与循环汇总；来源不可达时本轮放弃，下一轮重试。
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);
+ String identity = sourceIdentity(binding);
+ // STEP 3: 逐条写入可归属的实际工步，已删除的来源行跳过，内容不一致时保留原事实。
for (StepRecord source : steps == null ? List.<StepRecord>of() : steps)
{
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));
+ 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.setCurrentA(BigDecimal.valueOf(source.endCurrentA())); row.setCapacityAh(decimal(source.capacityAh())); row.setEnergyWh(decimal(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));
@@ -180,10 +363,11 @@
log.warn("工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey());
}
}
+ // STEP 4: 逐条写入可归属的循环汇总，来源类别随记录保存以便区分推导值与设备事实。
for (CycleRecord source : cycles == null ? List.<CycleRecord>of() : cycles)
{
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));
+ 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.setSourceKind(source.sourceKind()); 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(decimal(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)
{
@@ -194,12 +378,29 @@
}
}
- 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 CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(channelIdentity(b)); } catch (RuntimeException ex) { return null; } }
+ private static ChannelIdentity channelIdentity(TTestChannelBinding b) { return new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid()); }
+ private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null || b.getBinding() == null) return null; try { return adapter.latestCollector(b.getBinding().getAdapterCode(), mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } }
+ 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<>();
@@ -208,11 +409,14 @@
.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 long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } 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)
+ /** 设备未提供的单项值保持为空：零是设备事实，缺失不能写成零。 */
+ private static BigDecimal decimal(Double value) { return value == null ? null : BigDecimal.valueOf(value); }
+ // 实时提醒与持久化共用归属边界，避免同一旧观测只在首页触发异常。
+ 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()));
@@ -226,4 +430,13 @@
try { return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content)); }
catch (Exception ex) { throw new IllegalStateException("无法计算测试数据内容摘要", ex); }
}
+
+ /** 单个测试电池的逐行接收状态：接收边界、来源水位、是否按行接收与上次取数时间。 */
+ private static final class SampleCursor
+ {
+ private Instant boundary;
+ private SampleWatermark watermark;
+ private Boolean rowBased;
+ private Instant sampledAt;
+ }
}
--
Gitblit v1.7.1