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<Long> 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<Long> 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.<TTestBattery>of() : test.getBatteries())
|
captureCollectorMeasurements(test, battery);
|
}
|
}
|
|
private void captureTest(TTest test)
|
{
|
Date startedAt = test.getStartedAtUtc();
|
for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
|
{
|
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 (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)));
|
}
|
}
|
