package com.ruoyi.system.service.impl; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; import java.util.Date; import java.util.HexFormat; import java.util.List; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import com.ruoyi.device.bts.BtsDevelopmentSimulator; import com.ruoyi.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord; import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement; import com.ruoyi.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot; import com.ruoyi.system.domain.TTest; import com.ruoyi.system.domain.TTestBattery; import com.ruoyi.system.domain.TTestChannelBinding; import com.ruoyi.system.mapper.TTestMapper; import com.ruoyi.system.service.ITTestService; import com.ruoyi.system.service.ITestDataReceptionService; /** 接收时确定测试归属,并以现有采样表追加保存检测柜事实。 */ @Service public class TestDataReceptionServiceImpl implements ITestDataReceptionService { private final TTestMapper mapper; private final ITTestService tests; private final BtsDevelopmentSimulator simulator; public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, BtsDevelopmentSimulator simulator) { this.mapper = mapper; this.tests = tests; this.simulator = simulator; } @Override @Transactional(rollbackFor = Exception.class) public void captureDevelopmentCabinetRecords() { List testIds = mapper.selectReceivingTestIds(); if (testIds == null) return; for (Long testId : testIds) { // 锁住测试行直到本轮采样写入完成,暂停/停止动作只能在本轮接收提交后生效。 TTest locked = mapper.selectTestByIdForUpdate(testId); if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue; TTest test = tests.selectTestById(testId); if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue; captureTest(test); } } @Override @Transactional(rollbackFor = Exception.class) public void captureDevelopmentCollectorMeasurements() { List testIds = mapper.selectReceivingTestIds(); if (testIds == null) return; for (Long testId : testIds) { TTest locked = mapper.selectTestByIdForUpdate(testId); if (locked == null || !"RECEIVING".equals(locked.getReceptionStatus())) continue; TTest test = tests.selectTestById(testId); if (test == null || !"RECEIVING".equals(test.getReceptionStatus())) continue; for (TTestBattery battery : test.getBatteries() == null ? List.of() : test.getBatteries()) captureCollectorMeasurements(test, battery); } } private void captureTest(TTest test) { Date startedAt = test.getStartedAtUtc(); for (TTestBattery battery : test.getBatteries() == null ? List.of() : test.getBatteries()) { TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) continue; List records; try { records = simulator.btsRecords(binding.getDevid(), binding.getChlid()); } catch (IllegalArgumentException ignored) { continue; } for (BtsHistoryRecord record : records) { // 模拟器保留通道历史;只接收本次测试开始后产生的设备时间记录,避免顺序测试串入旧事实。 if (startedAt != null && record.deviceTime().isBefore(startedAt.toInstant())) continue; save(test, binding, record); } } } private void captureCollectorMeasurements(TTest test, TTestBattery battery) { if (battery.getCollectorId() == null || battery.getCollectorChannels() == null || battery.getCollectorChannels().isEmpty()) return; final CollectorRealtimeSnapshot realtime; try { realtime = simulator.collectorRealtime(battery.getCollectorId().intValue()); } catch (IllegalArgumentException ignored) { return; } if (!realtime.online()) return; Date startedAt = test.getStartedAtUtc(); for (CollectorRealtimeMeasurement measurement : realtime.measurements()) { if (startedAt != null && measurement.deviceTime().isBefore(startedAt.toInstant())) continue; // 继续接收只接受恢复动作之后查询到的快照;暂停期间采集器自行产生的历史不回补。 if (test.getUpdateTime() != null && measurement.serverReceivedAt().isBefore(test.getUpdateTime().toInstant())) continue; boolean mapped = battery.getCollectorChannels().stream().anyMatch(mapping -> java.util.Objects.equals(mapping.getChannelCode(), measurement.channelCode()) && metricMatches(mapping.getMeasurementCode(), measurement.metricCode())); if (!mapped) continue; saveCollector(test, battery, measurement); } } private void saveCollector(TTest test, TTestBattery battery, CollectorRealtimeMeasurement measurement) { byte[] hash = hash(measurement); String old = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(), measurement.sequenceId(), measurement.metricCode()); String expected = HexFormat.of().formatHex(hash); if (old != null) { if (!old.equalsIgnoreCase(expected)) mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(), "采集器实时接收内容冲突 collector=" + battery.getCollectorId() + ", channel=" + measurement.channelCode() + ", metric=" + measurement.metricCode() + ", seqid=" + measurement.sequenceId()); return; } Date receivedAt = new Date(); int inserted = mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(), battery.getCollectorId(), measurement.adapterCode(), measurement.channelCode(), measurement.sequenceId(), Date.from(measurement.deviceTime()), measurement.deviceTimeRaw(), receivedAt, measurement.metricCode(), measurement.value(), measurement.unit(), hash); // 唯一键负责并发幂等;内容冲突保留首次事实,不覆盖原始记录。 if (inserted == 0) { String actual = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), measurement.channelCode(), measurement.sequenceId(), measurement.metricCode()); if (actual != null && !actual.equalsIgnoreCase(expected)) mapper.insertSampleConflict(test.getTestId(), battery.getBinding().getChannelId(), "采集器实时并发内容冲突 collector=" + battery.getCollectorId() + ", channel=" + measurement.channelCode() + ", metric=" + measurement.metricCode() + ", seqid=" + measurement.sequenceId()); } } private void 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; } 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()); } } 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))); } }