修复未开始测试继承设备历史

yeg

2026-09-01 d1cb58bebb89c000275d9b092dd644c809296707

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

@@ -11,6 +11,8 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import jakarta.annotation.PreDestroy;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -22,7 +24,11 @@
import com.ruoyi.device.model.Measurement;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITestLiveService.TestPlanRow;
import com.ruoyi.system.service.ITestLiveService.TestPlanSnapshot;
import com.ruoyi.system.service.ITTestService;

/** 将设备最新快照转换为测试视角，并托管 SSE 连接生命周期。 */
@@ -32,43 +38,127 @@
    private final ITTestService tests;
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final TTestMapper testMapper;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; }
            BtsDevelopmentSimulator simulator, TTestMapper testMapper)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.testMapper = testMapper; }

    @Override public LiveState current(Long testId)
    {
        TTest test = tests.selectTestById(testId);
        List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
        List<LiveSource> sources = new ArrayList<>();
        boolean freezeSnapshots = "PAUSED".equals(test.getReceptionStatus())
                || "ENDED".equals(test.getReceptionStatus());
        for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
        {
            if (battery.getBinding() != null) sources.add(cabinetSource(battery, cabinets));
            if (battery.getCollectorId() != null) sources.add(collectorSource(battery));
            if (battery.getBinding() != null)
                sources.add(snapshotFor(testId, cabinetSource(battery, cabinets), freezeSnapshots));
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
        }
        List<Map<String, Object>> exceptions = sources.stream().filter(source -> !"ONLINE".equals(source.deviceStatus())
        List<TTestCollectorExceptionInterval> persisted = testMapper.selectCollectorExceptionIntervals(testId);
        if (persisted == null) persisted = List.of();
        List<Map<String, Object>> exceptions = new ArrayList<>(persisted.stream().map(interval -> {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("sourceType", "COLLECTOR");
            item.put("sourceId", interval.getCollectorId());
            item.put("collectorCode", interval.getCollectorCode());
            item.put("collectorName", interval.getCollectorName());
            item.put("channelCode", interval.getCollectorChannelCode());
            item.put("measurementName", interval.getMeasurementName());
            item.put("exceptionType", interval.getExceptionType());
            item.put("status", interval.getIntervalStatus());
            item.put("openedAt", interval.getOpenedAtUtc());
            item.put("missingFrom", interval.getMissingFromUtc());
            item.put("recoveredAt", interval.getRecoveredAtUtc());
            item.put("missingTo", interval.getMissingToUtc());
            item.put("detail", interval.getDetail());
            return item;
        }).toList());
        exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
                && !"ONLINE".equals(source.deviceStatus())
                && !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
                .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                        "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                .toList();
        return new LiveState(testId, test.getTestStatus(), test.getReceptionStatus(), Instant.now(), sources, exceptions);
                .toList());
        boolean abnormal = persisted.stream().anyMatch(item -> "OPEN".equals(item.getIntervalStatus()));
        String testStatus = abnormal && "RUNNING".equals(test.getTestStatus()) ? "RUNNING_ABNORMAL" : test.getTestStatus();
        List<com.ruoyi.system.domain.TTestIntegrityIssue> integrity = new ArrayList<>();
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedIntegrity = testMapper.selectIntegrityIssues(testId);
        if (persistedIntegrity != null) integrity.addAll(persistedIntegrity);
        List<com.ruoyi.system.domain.TTestIntegrityIssue> persistedExceptions = testMapper.selectTestExceptionFacts(testId);
        if (persistedExceptions != null) integrity.addAll(persistedExceptions);
        TestPlanSnapshot testPlan = detectorTestPlan(test);
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions,
                integrity == null ? List.of() : integrity, testPlan);
    }

    private TestPlanSnapshot detectorTestPlan(TTest test)
    {
        if (test.getBatteries() == null) return null;
        for (TTestBattery battery : test.getBatteries())
        {
            if (battery.getBinding() == null || battery.getBinding().getDevid() == null
                    || battery.getBinding().getChlid() == null) continue;
            try
            {
                var source = "NOT_STARTED".equals(test.getReceptionStatus())
                        ? simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid(), 1)
                        : simulator.btsTest(battery.getBinding().getDevid(), battery.getBinding().getChlid());
                List<TestPlanRow> rows = source.rows().stream()
                        .map(row -> new TestPlanRow(row.rowType(), row.stepNo(), row.cyclePosition(), row.stepName(), row.durationSeconds(),
                                row.voltageV(), row.executionStatus(), row.cycleCount(), row.recordCount()))
                        .toList();
                return new TestPlanSnapshot(source.currentCycleNo(), source.currentStepNo(),
                        source.plannedCycleCount(), rows);
            }
            catch (IllegalArgumentException ignored) { /* 登记设备尚未被开发模拟器提供时保持无方案事实。 */ }
        }
        return null;
    }

    private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean freeze)
    {
        String key = testId + ":" + fresh.sourceType() + ":" + fresh.sourceId() + ":" + fresh.batteryNo();
        if (freeze) return lastSources.getOrDefault(key, fresh);
        lastSources.put(key, fresh);
        return fresh;
    }

    @Override public SseEmitter stream(Long testId)
    {
        SseEmitter emitter = new SseEmitter(0L);
        SseEmitter emitter = createEmitter();
        try { sendAll(emitter, testId); }
        catch (RuntimeException | IOException exception) { emitter.completeWithError(exception); return emitter; }
        catch (IOException exception)
        {
            // 客户端刷新或离开页面会断开连接，不能再交给全局异常处理器写 SSE JSON。
            emitter.complete();
            return emitter;
        }
        catch (RuntimeException exception) { emitter.completeWithError(exception); return emitter; }
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            try { sendAll(emitter, testId); }
            catch (RuntimeException | IOException exception) { emitter.completeWithError(exception); }
            catch (IOException exception)
            {
                // 写出失败表示 SSE 客户端已断开，静默结束连接。
                emitter.complete();
            }
            catch (RuntimeException exception) { emitter.completeWithError(exception); }
        }, 3, 3, TimeUnit.SECONDS);
        emitter.onCompletion(() -> future.cancel(false));
        emitter.onTimeout(() -> future.cancel(false));
        emitter.onError(error -> future.cancel(false));
        return emitter;
    }

    /** 创建实时流发射器，供断连生命周期测试替换。 */
    protected SseEmitter createEmitter()
    {
        return new SseEmitter(0L);
    }

    private void sendAll(SseEmitter emitter, Long testId) throws IOException
@@ -79,6 +169,7 @@
        emitter.send(SseEmitter.event().name("reception-status").data(Map.of("testId", testId, "receptionStatus", state.receptionStatus())));
        emitter.send(SseEmitter.event().name("test-status").data(Map.of("testId", testId, "testStatus", state.testStatus())));
        emitter.send(SseEmitter.event().name("exception").data(state.exceptions()));
        emitter.send(SseEmitter.event().name("integrity").data(state.integrityIssues()));
        emitter.send(SseEmitter.event().name("heartbeat").data(Map.of("at", state.updatedAt())));
    }

@@ -101,6 +192,11 @@
        if (channel.latestSnapshot() == null) return value;
        value.put("source", "检测柜");
        value.put("observedAt", channel.latestSnapshot().receivedAt());
        // 实时方案需要设备当前循环和实际工步；这些字段只用于快照展示，不形成历史采集事实。
        value.put("cycleNo", channel.latestSnapshot().cycleNo());
        value.put("stepNo", channel.latestSnapshot().stepNo());
        value.put("stepType", channel.latestSnapshot().stepType());
        value.put("totalTimeSeconds", channel.latestSnapshot().totalTimeSeconds());
        value.put("measurements", channel.latestSnapshot().measurements().entrySet().stream().map(entry -> {
            Measurement measurement = entry.getValue();
            return Map.of("metricCode", entry.getKey(), "value", measurement.value(), "unit", measurement.unit());