package com.battery.business.service.impl; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.time.Clock; import java.time.Instant; import java.util.Date; import java.util.HexFormat; import java.util.LinkedHashMap; import java.util.List; 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.SampleWatermark; import com.battery.device.api.TestDataObservationAdapter.StepRecord; 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 public class TestDataReceptionServiceImpl implements ITestDataReceptionService { private static final Logger log = LoggerFactory.getLogger(TestDataReceptionServiceImpl.class); private final TTestMapper mapper; private final ITTestService tests; private final TestDataObservationAdapter adapter; private final TestWarningExceptionService warningExceptions; private final Clock clock; private final Map lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>(); private final Map lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>(); /** 每个测试电池上次读取工步与循环的时刻；读取节奏对齐来源写库周期，不跟着调度节奏。 */ private final Map lastAggregateAt = new java.util.concurrent.ConcurrentHashMap<>(); /** 每个测试电池各自的接收水位；只存在内存中，重启后按来源里的最新一行重新取水位。 */ private final Map sampleCursors = new java.util.concurrent.ConcurrentHashMap<>(); @Autowired public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter, TestWarningExceptionService warningExceptions) { this(mapper, tests, adapter, warningExceptions, Clock.systemUTC()); } /** 允许测试固定接收时间，生产运行仍使用 UTC 系统时钟。 */ public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, TestDataObservationAdapter adapter, Clock clock) { this(mapper, tests, adapter, new TestWarningExceptionService(mapper), clock); } 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) { this.mapper = mapper; this.tests = tests; this.adapter = adapter; this.warningExceptions = warningExceptions; this.clock = clock; } /** 轮询接收中测试，判定实时超限和来源离线异常，并按独立周期保存成功观测。 */ @Override @Transactional(rollbackFor = Exception.class) public void captureReceivedData() { for (TTest test : receivingTests()) { Instant now = clock.instant(); Instant previous = lastRealtimeCheckAt.get(test.getTestId()); // 实时预警独立于持久化槽位，并与暂停、停止和删除共享测试事务锁。 if (previous == null || !now.isBefore(previous.plusSeconds(5))) { warningExceptions.recordOfflineSources(test, Date.from(now)); for (TTestBattery battery : batteries(test)) saveRealtime(test, battery); lastRealtimeCheckAt.put(test.getTestId(), now); } for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery); } } private List receivingTests() { List 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 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; // 来源以设备数据行为接收单位时，柜端事实逐行落库（同秒多行都保留），槽位只留给采集器端。 boolean rowBased = persistCabinetSamples(test, battery, binding); saveRealtimeSlot(test, battery, binding, !rowBased); } /** * 按来源水位逐行接收柜端事实。水位是"开始接收"或"继续接收"这一刻来源里的最新一行， * 因此水位之前的数据行即使仍在来源里也不进本测试，暂停期间的空档也不会被补回来。 * * @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) { // 开始接收与继续接收都会推进接收边界：边界变化即重新取水位，暂停期间的数据不补采。 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 { 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; } List 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}）， * 这里不用服务器取数时间反推设备时间。 */ private void insertCabinetSample(TTest test, TTestBattery battery, TTestChannelBinding binding, CabinetRealtime row) { if (row.deviceTime() == null || row.serverReceivedAt() == null) return; 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); 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 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)); // 重复接收的同内容记录幂等忽略；同身份但内容不同时保留原事实，不覆盖。 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); } private void saveRealtimeSlot(TTest test, TTestBattery battery, TTestChannelBinding binding, boolean includeCabinet) { // 柜端和采集器端分别判断成功；同一槽位只写一行，失败来源的身份、时间和指标保持为空。 Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId(); CabinetRealtime cabinet = includeCabinet ? latestCabinet(binding) : null; if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null; CollectorRealtime collector = latestCollector(battery); Map mapped = mappedCollectorMetrics(test, battery, collector); if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return; // 当前成功测量先判定预警；槽位尚未到期或历史行被删除，都不能阻断实时告警。 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())); } 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; 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()); } 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)); Map extension = new LinkedHashMap<>(); // 工步内相对时长由成功柜端观测提供，不以轮询槽位或服务器时间推算。 if (cabinet != null && cabinet.relativeTimeSeconds() != null) extension.put("relativeTimeSeconds", cabinet.relativeTimeSeconds()); for (Map.Entry e : mapped.entrySet()) if (!isFixedMetric(e.getKey())) { Map 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 evaluateCabinet(TTest test, TTestChannelBinding binding, String metricCode, BigDecimal value, String unit, Date occurredAt) { if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt); } private void saveDeviceAggregates(TTest test, TTestBattery battery) { // 工步、循环只接受来源返回的统计记录，并以来源编号或明确的时间组合执行幂等。 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); ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid()); List steps; List cycles; try { steps = adapter.stepRecords(channel); cycles = adapter.cycleRecords(channel); } catch (RuntimeException ex) { return; } String identity = sourceIdentity(binding); for (StepRecord source : steps == null ? List.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.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)); 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()); if (actual != null && !actual.equalsIgnoreCase(HexFormat.of().formatHex(row.getContentSha256()))) log.warn("工步记录内容冲突，拒绝覆盖 testId={}, dedupeKey={}", test.getTestId(), row.getDedupeKey()); } } for (CycleRecord source : cycles == null ? List.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.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) { String actual = mapper.selectCycleRecordContentHash(row.getTestBatteryId(), row.getSourceIdentity(), 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(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 mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source) { Map 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 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 measurements) { List 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(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 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())); } private byte[] hash(Object value) { // 重复观测比较设备内容，不把服务器轮询时间作为内容变化。 Map 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 static final class SampleCursor { private Instant boundary; private SampleWatermark watermark; private Boolean rowBased; private Instant sampledAt; } }