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

yeg

2026-09-05 92f32acd847ea1700590bc669750f23ef499840e

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

@@ -11,15 +11,16 @@
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;
import com.battery.device.simulator.BtsSimulatorStateStore.ChannelState;
import com.battery.device.simulator.BtsSimulatorStateStore.ChannelTombstone;
@@ -43,7 +44,7 @@
import com.battery.device.simulator.DevelopmentCollectorSimulation.CollectorRealtimeSnapshot;

/** 提供由开发后端托管的临时只读 BTS 模拟端点。 */
public final class BtsDevelopmentSimulator implements DeviceObservationAdapter
public final class BtsDevelopmentSimulator implements TestDataObservationAdapter
{
    public static final String HOST = "127.0.0.1";
    public static final int DEFAULT_PORT = 15020;
@@ -495,6 +496,84 @@
        return Math.max(100L, 60000L / Math.max(1, playback.speedMultiplier()));
    }

    @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.deviceTime().toString(), "UTC");
    }

    @Override
    public synchronized CollectorRealtime latestCollector(String collectorCode)
    {
        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(), "Asia/Hong_Kong");
    }

    @Override
    public synchronized List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        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(), 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)
    {
        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(), item.endedAt().toString(), "UTC", receivedAt(channel, item.endedAt()))).toList();
    }

    @Override
    public synchronized long persistenceIntervalMillis(ChannelIdentity channel)
    { 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)
    {
@@ -891,7 +970,8 @@
    {
        return item.stepName() == null || item.stepName().isBlank()
                ? new BtsHistoryRecord(item.sequenceId(), item.deviceTime(), item.currentA(), item.voltageV(),
                        item.capacityAh(), item.energyWh()) : item;
                        item.capacityAh(), item.energyWh(), item.cycleNo(), item.stepNo(), item.stepName(), item.stepType(),
                        item.serverReceivedAt()) : item;
    }

    private String btsKey(DeviceState device, int channelId)
@@ -963,7 +1043,7 @@
            budgetMillis = Math.max(0L, budgetMillis - deltaMillis);
            simulated = simulated.plusSeconds(deltaSeconds);
            BtsHistoryRecord record = new BtsHistoryRecord(fixture.sequenceId(), simulated, fixture.currentA(), fixture.voltageV(),
                    fixture.capacityAh(), fixture.energyWh(), fixture.cycleNo(), fixture.stepNo(), fixture.stepName(), fixture.stepType());
                    fixture.capacityAh(), fixture.energyWh(), fixture.cycleNo(), fixture.stepNo(), fixture.stepName(), fixture.stepType(), now);
            history.add(record);
            appended.add(record);
            lastOffset = fixture.offsetSeconds();
@@ -1003,7 +1083,19 @@

    public record BtsStepRecord(int cycleNo, int stepNo, Integer cyclePosition, String stepName, String stepType, long recordCount,
            Instant startedAt, Instant endedAt, double durationSeconds, double capacityAh, double energyWh,
            double startVoltageV, double endVoltageV, double chargePeakVoltageV, double dischargeValleyVoltageV) { }
            double startVoltageV, double endVoltageV, double chargePeakVoltageV, double dischargeValleyVoltageV,
            String recordNo)
    {
        public BtsStepRecord(int cycleNo, int stepNo, Integer cyclePosition, String stepName, String stepType,
                long recordCount, Instant startedAt, Instant endedAt, double durationSeconds, double capacityAh,
                double energyWh, double startVoltageV, double endVoltageV, double chargePeakVoltageV,
                double dischargeValleyVoltageV)
        {
            this(cycleNo, stepNo, cyclePosition, stepName, stepType, recordCount, startedAt, endedAt,
                    durationSeconds, capacityAh, energyWh, startVoltageV, endVoltageV, chargePeakVoltageV,
                    dischargeValleyVoltageV, null);
        }
    }

    public record BtsCycleRecord(String recordNo, int cycleNo, long recordCount, Instant startedAt, Instant endedAt,
            double chargeCapacityAh, double dischargeCapacityAh, double chargeEnergyWh, double dischargeEnergyWh,
@@ -1030,7 +1122,7 @@
        private double startVoltage; private double endVoltage; private double chargePeak = Double.NEGATIVE_INFINITY; private double dischargeValley = Double.POSITIVE_INFINITY;
        BtsStepAccumulator(BtsHistoryRecord first, int position) { cycle = first.cycleNo(); step = first.stepNo(); this.position = position; name = first.stepName(); type = first.stepType(); }
        void add(BtsHistoryRecord item) { count++; if (first == null) { first = item.deviceTime(); startVoltage = item.voltageV(); } last = item.deviceTime(); endVoltage = item.voltageV(); duration = first == null ? 0 : Math.max(0, java.time.Duration.between(first, last).toMillis() / 1000d); capacity = item.capacityAh(); energy = item.energyWh(); if (item.currentA() >= 0) chargePeak = Math.max(chargePeak, item.voltageV()); else dischargeValley = Math.min(dischargeValley, item.voltageV()); }
        BtsStepRecord finish() { return new BtsStepRecord(cycle, step, position, name, type, count, first, last, duration, capacity, energy, startVoltage, endVoltage, chargePeak == Double.NEGATIVE_INFINITY ? 0 : chargePeak, dischargeValley == Double.POSITIVE_INFINITY ? 0 : dischargeValley); }
        BtsStepRecord finish() { return new BtsStepRecord(cycle, step, position, name, type, count, first, last, duration, capacity, energy, startVoltage, endVoltage, chargePeak == Double.NEGATIVE_INFINITY ? 0 : chargePeak, dischargeValley == Double.POSITIVE_INFINITY ? 0 : dischargeValley, "STEP-" + cycle + "-" + step + "-" + last.toEpochMilli()); }
    }

    private static final class BtsCycleAccumulator