清理旧测试结构并贯通最终业务数据链路

yeg

2026-09-05 92f32acd847ea1700590bc669750f23ef499840e

 battery-device-simulator/src/main/java/com/battery/device/simulator/BtsDevelopmentSimulator.java

@@ -11,14 +11,14 @@
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamConstants;
import javax.xml.stream.XMLStreamReader;
import com.battery.device.model.DeviceWorkStatus;
import com.battery.device.model.DeviceConnectionStatus;
import com.battery.device.model.DeviceIdentity;
import com.battery.device.model.DeviceObservation;
import com.battery.device.model.DetectedChannel;
import com.battery.device.model.DetectedDeviceMetadata;
import com.battery.device.model.ChannelIdentity;
import com.battery.device.model.AdapterCapabilities;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceIdentity;
import com.battery.device.api.model.DeviceObservation;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.DeviceObservationAdapter;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.simulator.BtsSimulatorStateStore;
@@ -497,44 +497,82 @@
    }

    @Override
    public synchronized TestPlan testPlan(ChannelIdentity channel)
    {
        if (!canObserve(channel)) return null;
        var plan = btsTest(channel.devid(), channel.chlid());
        return new TestPlan(plan.currentCycleNo(), plan.currentStepNo(), plan.plannedCycleCount(),
                plan.rows().stream().map(row -> new PlanRow(row.rowType(), row.stepNo(), row.cyclePosition(),
                        row.stepName(), row.durationSeconds(), row.voltageV(), row.executionStatus(),
                        row.cycleCount(), row.recordCount())).toList());
    }

    @Override
    public synchronized CabinetRealtime latestCabinet(ChannelIdentity channel)
    {
        if (!canObserve(channel)) return null;
        BtsHistoryRecord record = btsRecords(channel.devid(), channel.chlid()).stream()
                .max(java.util.Comparator.comparing(BtsHistoryRecord::deviceTime)).orElse(null);
        return record == null ? null : new CabinetRealtime(record.sequenceId(), record.deviceTime(), record.currentA(),
                record.voltageV(), record.capacityAh(), record.energyWh(), record.cycleNo(), record.stepNo(),
                record.stepName(), record.stepType(), record.serverReceivedAt());
                record.stepName(), record.stepType(), record.serverReceivedAt(), record.deviceTime().toString(), "UTC");
    }

    @Override
    public synchronized CollectorRealtime latestCollector(long collectorId)
    public synchronized CollectorRealtime latestCollector(String collectorCode)
    {
        CollectorRealtimeSnapshot snapshot = collectorRealtime((int) collectorId);
        var collector = snapshot().collectors().stream().filter(item -> item.deviceCode().equals(collectorCode))
                .findFirst().orElse(null);
        if (collector == null) return null;
        CollectorRealtimeSnapshot snapshot = collectorRealtime(collector.collectorId());
        return new CollectorRealtime(snapshot.collectorCode(), snapshot.online(), snapshot.sampleIntervalMs(),
                snapshot.measurements().stream().map(item -> new CollectorMeasurement(item.channelCode(), item.sequenceId(),
                        item.deviceTime(), item.deviceTimeRaw(), item.serverReceivedAt(), item.metricCode(), item.value(),
                        item.unit())).toList());
                        item.unit())).toList(), "Asia/Hong_Kong");
    }

    @Override
    public synchronized List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return btsSteps(channel.devid(), channel.chlid()).stream().map(item -> new StepRecord(item.recordNo(),
        if (!canObserve(channel)) return List.of();
        var playback = playbackState(btsKey(requireDevice(channel.devid()), channel.chlid()), Instant.now());
        // 仅发布已离开的工步；手动跳转也可结束工步，不用记录数作为完整性门禁。
        return btsSteps(channel.devid(), channel.chlid()).stream().filter(item -> playback.naturallyFinished()
                || item.cycleNo() != playback.cycleNo() || item.stepNo() != playback.stepNo()).map(item -> new StepRecord(item.recordNo(),
                item.cycleNo(), item.stepNo(), item.stepName(), item.stepType(), item.startedAt(), item.endedAt(),
                item.endVoltageV(), item.capacityAh(), item.energyWh())).toList();
                item.endVoltageV(), item.capacityAh(), item.energyWh(), item.endedAt().toString(), "UTC", receivedAt(channel, item.endedAt()),
                java.util.Map.of("stepDuration", java.math.BigDecimal.valueOf(item.durationSeconds()),
                        "startVoltage", java.math.BigDecimal.valueOf(item.startVoltageV()),
                        "chargePeakVoltage", java.math.BigDecimal.valueOf(item.chargePeakVoltageV()),
                        "dischargeValleyVoltage", java.math.BigDecimal.valueOf(item.dischargeValleyVoltageV())))).toList();
    }

    @Override
    public synchronized List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return btsCycles(channel.devid(), channel.chlid()).stream().map(item -> new CycleRecord(item.recordNo(),
        if (!canObserve(channel)) return List.of();
        var playback = playbackState(btsKey(requireDevice(channel.devid()), channel.chlid()), Instant.now());
        return btsCycles(channel.devid(), channel.chlid()).stream().filter(item -> playback.naturallyFinished()
                || item.cycleNo() != playback.cycleNo()).map(item -> new CycleRecord(item.recordNo(),
                item.cycleNo(), item.startedAt(), item.endedAt(), item.chargeCapacityAh(), item.dischargeCapacityAh(),
                item.chargeEnergyWh(), item.dischargeEnergyWh(), item.efficiency())).toList();
                item.chargeEnergyWh(), item.dischargeEnergyWh(), item.efficiency(), item.endedAt().toString(), "UTC", receivedAt(channel, item.endedAt()))).toList();
    }

    @Override
    public synchronized long persistenceIntervalMillis(ChannelIdentity channel)
    { return persistenceIntervalMillis(channel.devid(), channel.chlid()); }
    { return canObserve(channel) ? persistenceIntervalMillis(channel.devid(), channel.chlid()) : 60000L; }

    private boolean canObserve(ChannelIdentity channel)
    {
        // 全身份与在线事实共同限定来源，避免同编号的其他适配器被写入模拟数据。
        return channel != null && observe(java.util.Set.of(channel)).get(channel).connectionStatus() == DeviceConnectionStatus.ONLINE;
    }

    private Instant receivedAt(ChannelIdentity channel, Instant endedAt)
    {
        return btsRecords(channel.devid(), channel.chlid()).stream().filter(row -> row.deviceTime().equals(endedAt))
                .map(BtsHistoryRecord::serverReceivedAt).filter(java.util.Objects::nonNull).max(Instant::compareTo).orElse(null);
    }

    /** 按实际出现的循环和工步汇总 record，不补齐缺失行。 */
    public synchronized List<BtsStepRecord> btsSteps(int deviceId, int channelId)