实现采集器实时扩展测量接收

yeg

2026-09-02 409b5ac5a3b160eacb1fefa6a10bf97b5f6860d7

实现采集器实时扩展测量接收

接入绑定采集器实时快照，保存来源、指标、时间和单位事实，支持冲突审计与实时查询修订。

 ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentSimulatorDiscoveryScheduler.java

@@ -33,5 +33,7 @@
                BtsDevelopmentSimulator.ADAPTER_NAME, BtsDevelopmentSimulator.POLL_INTERVAL_MS));
        // 发现轮询成功后立即把接收窗口内的模拟 record 写入测试事实；页面不再读取模拟器历史。
        reception.captureDevelopmentCabinetRecords();
        // 采集器拥有独立采样频率；只接收本轮实时快照，不读取历史下载接口。
        reception.captureDevelopmentCollectorMeasurements();
    }
}

 ruoyi-admin/src/main/resources/db/migration/V30__collector_sample_source_identity.sql

New file
@@ -0,0 +1,3 @@
ALTER TABLE t_test_collector_sample
    ADD COLUMN adapter_code VARCHAR(64) NOT NULL DEFAULT 'unknown' AFTER collector_id,
    ADD KEY idx_collector_sample_source (adapter_code, collector_id, collector_channel_code, sequence_id);

 ruoyi-admin/src/test/java/com/ruoyi/config/TestPersistedDataQueryContractTest.java

@@ -14,6 +14,7 @@
    {
        String mapper = resource("mapper/system/TTestMapper.xml");
        assertTrue(mapper.contains("id=\"selectPersistedRecords\""));
        assertTrue(mapper.contains("id=\"selectPersistedCollectorRecords\""));
        assertTrue(mapper.contains("id=\"selectPersistedSteps\""));
        assertTrue(mapper.contains("id=\"selectPersistedCycles\""));
        assertTrue(mapper.contains("id=\"selectBtsSampleContentHash\""));

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImplTest.java

@@ -18,6 +18,9 @@
import org.junit.jupiter.api.Test;
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.TTestCollectorChannel;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestChannelBinding;
@@ -71,6 +74,63 @@
                any(BigDecimal.class), any(BigDecimal.class), any(BigDecimal.class), any(byte[].class));
    }

    @Test
    void capturesOnlyMappedCollectorRealtimeMeasurementsAfterReceptionStart()
    {
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        BtsDevelopmentSimulator simulator = mock(BtsDevelopmentSimulator.class);
        TTest test = test(Date.from(Instant.parse("2026-09-02T04:00:00Z")));
        test.getBatteries().get(0).setCollectorId(9L);
        TTestCollectorChannel mapping = new TTestCollectorChannel();
        mapping.setMeasurementCode("temperature"); mapping.setChannelCode("S-01");
        test.getBatteries().get(0).setCollectorChannels(List.of(mapping));
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(tests.selectTestById(1L)).thenReturn(test);
        CollectorRealtimeMeasurement matched = new CollectorRealtimeMeasurement("dev-sim", "collector-9", "S-01",
                3L, Instant.parse("2026-09-02T04:00:01Z"), "2026-09-02 12:00:01.000",
                Instant.parse("2026-09-02T04:00:02Z"), "temperature", new BigDecimal("25.1"), "℃");
        CollectorRealtimeMeasurement ignored = new CollectorRealtimeMeasurement("dev-sim", "collector-9", "S-02",
                3L, matched.deviceTime(), matched.deviceTimeRaw(), matched.serverReceivedAt(), "temperature",
                new BigDecimal("25.2"), "℃");
        when(simulator.collectorRealtime(9)).thenReturn(new CollectorRealtimeSnapshot("collector-9", true, 1000,
                List.of(matched, ignored)));
        when(mapper.selectCollectorSampleContentHash(any(), anyString(), any(), anyString())).thenReturn(null);
        when(mapper.insertCollectorSample(any(), any(), any(), anyString(), anyString(), any(), any(Date.class), anyString(),
                any(Date.class), anyString(), any(BigDecimal.class), anyString(), any(byte[].class))).thenReturn(1);

        new TestDataReceptionServiceImpl(mapper, tests, simulator).captureDevelopmentCollectorMeasurements();

        verify(mapper).insertCollectorSample(eq(1L), eq(2L), eq(9L), eq("dev-sim"), eq("S-01"), eq(3L), any(Date.class),
                eq("2026-09-02 12:00:01.000"), any(Date.class), eq("temperature"), eq(new BigDecimal("25.1")),
                eq("℃"), any(byte[].class));
        verify(mapper, never()).insertCollectorSample(eq(1L), eq(2L), eq(9L), anyString(), eq("S-02"), any(), any(Date.class),
                anyString(), any(Date.class), anyString(), any(BigDecimal.class), anyString(), any(byte[].class));
    }

    @Test
    void recordsCollectorContentConflictWithoutOverwritingOriginal()
    {
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        BtsDevelopmentSimulator simulator = mock(BtsDevelopmentSimulator.class);
        TTest test = test(Date.from(Instant.parse("2026-09-02T04:00:00Z")));
        test.getBatteries().get(0).setCollectorId(9L);
        TTestCollectorChannel mapping = new TTestCollectorChannel(); mapping.setMeasurementCode("temperature"); mapping.setChannelCode("S-01");
        test.getBatteries().get(0).setCollectorChannels(List.of(mapping));
        CollectorRealtimeMeasurement measurement = new CollectorRealtimeMeasurement("dev-sim", "collector-9", "S-01", 3L,
                Instant.parse("2026-09-02T04:00:01Z"), "2026-09-02 12:00:01.000", Instant.now(), "temperature", new BigDecimal("25.1"), "℃");
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L)); when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(tests.selectTestById(1L)).thenReturn(test); when(simulator.collectorRealtime(9)).thenReturn(new CollectorRealtimeSnapshot("collector-9", true, 1000, List.of(measurement)));
        when(mapper.selectCollectorSampleContentHash(9L, "S-01", 3L, "temperature")).thenReturn("different");

        new TestDataReceptionServiceImpl(mapper, tests, simulator).captureDevelopmentCollectorMeasurements();

        verify(mapper).insertSampleConflict(eq(1L), eq(3L), org.mockito.ArgumentMatchers.contains("采集器实时接收内容冲突"));
        verify(mapper, never()).insertCollectorSample(any(), any(), any(), anyString(), anyString(), any(), any(Date.class), anyString(), any(Date.class), anyString(), any(BigDecimal.class), anyString(), any(byte[].class));
    }

    private static TTest test(Date startedAt)
    {
        TTestChannelBinding binding = new TTestChannelBinding();

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java

@@ -54,6 +54,7 @@
    /** 查询当前正在接收的测试，供接收调度绑定测试事实。 */
    List<Long> selectReceivingTestIds();
    List<java.util.Map<String, Object>> selectPersistedRecords(@org.apache.ibatis.annotations.Param("testId") Long testId);
    List<java.util.Map<String, Object>> selectPersistedCollectorRecords(@org.apache.ibatis.annotations.Param("testId") Long testId);
    List<java.util.Map<String, Object>> selectPersistedSteps(@org.apache.ibatis.annotations.Param("testId") Long testId);
    List<java.util.Map<String, Object>> selectPersistedCycles(@org.apache.ibatis.annotations.Param("testId") Long testId);
    List<TTestResult> selectTestResults(TTestResult result);
@@ -106,11 +107,15 @@
            @Param("sourceType") String sourceType);
    int insertCollectorSample(@Param("testId") Long testId, @Param("bindingId") Long bindingId,
            @Param("collectorId") Long collectorId,
            @Param("adapterCode") String adapterCode,
            @Param("channelCode") String channelCode,
            @Param("sequenceId") Long sequenceId, @Param("deviceTimeUtc") Date deviceTimeUtc,
            @Param("deviceTimeRaw") String deviceTimeRaw, @Param("receivedAtUtc") Date receivedAtUtc,
            @Param("metricCode") String metricCode, @Param("metricValue") java.math.BigDecimal metricValue,
            @Param("unit") String unit, @Param("contentSha256") byte[] contentSha256);
    String selectCollectorSampleContentHash(@Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode, @Param("sequenceId") Long sequenceId,
            @Param("metricCode") String metricCode);
    int insertDataRevisionAction(TTestDataRevisionAction action);
    List<TTestDataRevisionAction> selectDataRevisionActions(TTestDataRevisionAction query);
    TTestDataRevisionAction selectDataRevisionActionById(Long actionId);

 ruoyi-system/src/main/java/com/ruoyi/system/service/ITestDataReceptionService.java

@@ -5,4 +5,7 @@
{
    /** 扫描当前接收中的开发检测柜，并把接收窗口内的新记录持久化。 */
    void captureDevelopmentCabinetRecords();

    /** 扫描当前接收中的开发采集器实时快照，并按测试绑定保存扩展测量事实。 */
    void captureDevelopmentCollectorMeasurements();
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestConsoleDataServiceImpl.java

@@ -134,7 +134,7 @@
            List<Map<String, Object>> changes = (List<Map<String, Object>>) (List<?>) JSON.parseArray(action.getChangesSummary(), Map.class);
            Set<String> editable = switch (action.getDataType())
            {
                case "REALTIME" -> Set.of("voltage", "current", "capacity", "energy");
                case "REALTIME" -> Set.of("voltage", "current", "capacity", "energy", "metricValue");
                case "STEP" -> Set.of("stepDuration", "capacity", "energy", "startVoltage", "endVoltage",
                        "chargePeakVoltage", "dischargeValleyVoltage");
                case "CYCLE" -> Set.of("chargeCapacity", "dischargeCapacity", "efficiency", "chargeEnergy",
@@ -215,7 +215,7 @@
    {
        return switch (dataType)
        {
            case "REALTIME" -> Set.of("voltage", "current", "capacity", "energy");
            case "REALTIME" -> Set.of("voltage", "current", "capacity", "energy", "metricValue");
            case "STEP" -> Set.of("stepDuration", "capacity", "energy", "startVoltage", "endVoltage",
                    "chargePeakVoltage", "dischargeValleyVoltage");
            case "CYCLE" -> Set.of("chargeCapacity", "dischargeCapacity", "efficiency", "chargeEnergy",
@@ -227,6 +227,12 @@
    private Long[] parseDeleteIdentity(String value, String dataType)
    {
        String[] parts = value.split(":", -1);
        if ("REALTIME".equals(dataType) && parts.length == 5 && "collector".equals(parts[0])
                && !parts[2].isBlank() && !parts[4].isBlank())
        {
            try { return new Long[] { Long.valueOf(parts[1]), Long.valueOf(parts[3]) }; }
            catch (NumberFormatException exception) { return null; }
        }
        int expectedLength = "STEP".equals(dataType) ? 8 : "CYCLE".equals(dataType) ? 7 : 6;
        int legacyLength = "STEP".equals(dataType) ? 5 : "CYCLE".equals(dataType) ? 4 : 3;
        if (parts.length != expectedLength && parts.length != legacyLength) return null;

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataQueryServiceImpl.java

@@ -28,7 +28,12 @@
    @Override public List<Map<String, Object>> records(Long testId)
    {
        requireTest(testId);
        return applyEffectiveData(testId, "REALTIME", mapper.selectPersistedRecords(testId));
        List<Map<String, Object>> source = new java.util.ArrayList<>();
        List<Map<String, Object>> cabinet = mapper.selectPersistedRecords(testId);
        List<Map<String, Object>> collector = mapper.selectPersistedCollectorRecords(testId);
        if (cabinet != null) source.addAll(cabinet);
        if (collector != null) source.addAll(collector);
        return applyEffectiveData(testId, "REALTIME", source);
    }

    @Override public List<Map<String, Object>> steps(Long testId)
@@ -96,6 +101,9 @@

    private String stableIdentity(String dataType, Map<String, Object> row)
    {
        if ("REALTIME".equals(dataType) && "COLLECTOR".equals(String.valueOf(row.get("sourceType"))))
            return String.join(":", "collector", identityPart(row, "collectorId"),
                    identityPart(row, "channelCode"), value(row, "sequenceId"), identityPart(row, "metricCode"));
        String adapterId = identityPart(row, "adapterId");
        String devtype = identityPart(row, "devtype");
        String deviceId = identityPart(row, "deviceId");

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TestDataReceptionServiceImpl.java

@@ -10,6 +10,8 @@
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;
@@ -49,6 +51,23 @@
        }
    }

    @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();
@@ -65,6 +84,62 @@
                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());
        }
    }

@@ -108,4 +183,27 @@
        }
        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)));
    }
}

 ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml

@@ -334,6 +334,28 @@
        where sample.test_id = #{testId}
        order by sample.device_time_utc desc, sample.seq_id desc, sample.sample_id desc
    </select>
    <select id="selectPersistedCollectorRecords" resultType="map">
        select concat('collector:', sample.collector_id, ':', sample.collector_channel_code, ':',
                      sample.sequence_id, ':', sample.metric_code) id,
               concat('collector:', sample.collector_id, ':', sample.collector_channel_code, ':',
                      sample.sequence_id, ':', sample.metric_code) stableIdentity,
               concat('CR-', sample.collector_id, '-', sample.collector_channel_code, '-',
                      lpad(sample.sequence_id, 6, '0'), '-', sample.metric_code) recordNo,
               'COLLECTOR' sourceType, battery.battery_no batteryNo,
               sample.sequence_id sequenceId, sample.device_time_utc recordedAt,
               sample.device_time_utc absoluteTime, sample.device_time_utc deviceTime,
               sample.server_received_at_utc serverReceivedAt, sample.sample_id sampleId,
               sample.metric_code metricCode,
               sample.metric_value metricValue, sample.unit, sample.collector_id collectorId,
               collector.collector_code collectorCode, sample.collector_channel_code channelCode,
               sample.adapter_code adapterCode
        from t_test_collector_sample sample
        join t_test_channel_binding binding on binding.binding_id = sample.binding_id
        join t_test_battery battery on battery.relation_id = binding.battery_relation_id and battery.del_flag = 0
        join t_collector collector on collector.collector_id = sample.collector_id
        where sample.test_id = #{testId}
        order by sample.device_time_utc desc, sample.sequence_id desc, sample.sample_id desc
    </select>
    <select id="selectPersistedSteps" resultType="map">
        select concat('step:', channel.devid, ':', channel.chlid, ':', sample.cycle_no, ':', sample.step_no) id,
               concat('STEP-', channel.devid, '-', channel.chlid, '-', sample.cycle_no, '-', sample.step_no) recordNo,
@@ -524,11 +546,18 @@
    </insert>
    <insert id="insertCollectorSample">
        insert ignore into t_test_collector_sample
        (test_id, binding_id, collector_id, collector_channel_code, sequence_id, device_time_utc,
        (test_id, binding_id, collector_id, adapter_code, collector_channel_code, sequence_id, device_time_utc,
         device_time_raw, server_received_at_utc, metric_code, metric_value, unit, content_sha256)
        values (#{testId}, #{bindingId}, #{collectorId}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
        values (#{testId}, #{bindingId}, #{collectorId}, #{adapterCode}, #{channelCode}, #{sequenceId}, #{deviceTimeUtc},
                #{deviceTimeRaw}, #{receivedAtUtc}, #{metricCode}, #{metricValue}, #{unit}, #{contentSha256})
    </insert>
    <select id="selectCollectorSampleContentHash" resultType="string">
        select lower(hex(content_sha256))
        from t_test_collector_sample
        where collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and sequence_id = #{sequenceId} and metric_code = #{metricCode}
        limit 1
    </select>
    <insert id="insertSampleConflict">
        insert into t_test_exception
        (test_id, channel_id, exception_type, exception_status, severity,