收敛模拟器为设备事实与实时观测

yeg

2026-09-05 7259e265b9237286996b09da03038ea5a133e1e0

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

@@ -2,34 +2,19 @@

import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorChannelState;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorHistoryFaults;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorMetricState;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorRealtimeMeasurement;
import com.battery.device.simulator.BtsSimulatorStateStore.CollectorState;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.simulator.BtsSimulatorStateStore.*;

/** 实现第一版开发采集器的独立采样与故障注入合同。 */
/** 提供请求时的采集器当前值；S 通道仍是待现场验证的开发合同。 */
public final class DevelopmentCollectorSimulation
{
    public static final int DEFAULT_SAMPLE_INTERVAL_MS = 1000;
    public static final int MIN_SAMPLE_INTERVAL_MS = 100;
    public static final int MAX_SAMPLE_INTERVAL_MS = 60000;
    public static final int MAX_HISTORY_PAGE_SIZE = BtsSimulatorStateStore.MAX_HISTORY_BATCH_SIZE;
    private static final int MAX_SAMPLES_PER_ADVANCE = 1000;
    private static final ZoneId DEVICE_ZONE = ZoneId.of("Asia/Hong_Kong");
    private static final DateTimeFormatter DEVICE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
    private static final DateTimeFormatter DEVICE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

    private DevelopmentCollectorSimulation() { }

@@ -48,320 +33,72 @@
        return List.copyOf(channels);
    }

    /** 调整通道数时保留已有通道配置，并为新增通道生成稳定默认配置。 */
    public static CollectorState resize(CollectorState current, String ip, int channelCount, boolean online,
            Instant changedAt)

    /** 调整设备拓扑，保留已有通道的配置。 */
    public static CollectorState resize(CollectorState current, String ip, int channelCount, boolean online, Instant changedAt)
    {
        List<CollectorChannelState> defaults = defaultChannels(channelCount);
        List<CollectorChannelState> channels = new ArrayList<>(channelCount);
        var defaults = defaultChannels(channelCount);
        List<CollectorChannelState> channels = new ArrayList<>();
        for (int index = 0; index < channelCount; index++)
            channels.add(index < current.channels().size() ? current.channels().get(index) : defaults.get(index));
        Instant lastSampledAt = online == current.online() ? current.lastSampledAt() : changedAt;
        List<CollectorRealtimeMeasurement> latestMeasurements = retainConfiguredMeasurements(
                current.latestMeasurements(), channels);
        return copy(current, ip, channelCount, online, current.sampleIntervalMs(), current.nextSequence(),
                lastSampledAt, channels, latestMeasurements, current.historyRecordCount(), current.historyFaults());
        return new CollectorState(current.collectorId(), current.deviceCode(), ip.trim(), current.mac(),
                current.name(), current.manufacturer(), current.model(), channelCount, online,
                current.sampleIntervalMs(), channels, List.of());
    }

    /** 更新采样频率和历史下载故障；测量原始历史保持只追加。 */
    public static CollectorState configureSampling(CollectorState current, int sampleIntervalMs,
            CollectorHistoryFaults historyFaults)
    {
        validateSampleInterval(sampleIntervalMs);
        validateFaults(historyFaults);
        return copy(current, current.ip(), current.channelCount(), current.online(), sampleIntervalMs,
                current.nextSequence(), current.lastSampledAt(), current.channels(), current.latestMeasurements(),
                current.historyRecordCount(), historyFaults);
    }

    /** 替换一个演示通道的指标配置；动态添加项与固定指标共用指标编码合同。 */
    public static CollectorState configureChannel(CollectorState current, String channelCode,
            List<CollectorMetricState> metrics)
    /** 保存当前测量配置，等待下一次实时请求。 */
    public static CollectorState configureChannel(CollectorState current, String channelCode, List<CollectorMetricState> metrics)
    {
        validateMetrics(metrics);
        int channelIndex = parseChannelIndex(channelCode, current.channelCount());
        int index = parseChannelIndex(channelCode, current.channelCount());
        List<CollectorChannelState> channels = new ArrayList<>(current.channels());
        channels.set(channelIndex, new CollectorChannelState(channelCode(channelIndex + 1), metrics));
        List<CollectorRealtimeMeasurement> latestMeasurements = retainConfiguredMeasurements(
                current.latestMeasurements(), channels);
        return copy(current, current.ip(), current.channelCount(), current.online(), current.sampleIntervalMs(),
                current.nextSequence(), current.lastSampledAt(), channels, latestMeasurements,
                current.historyRecordCount(), current.historyFaults());
        channels.set(index, new CollectorChannelState(channelCode, metrics, channels.get(index).workStatus()));
        return copy(current, channels, List.of());
    }

    /** 按采集器自身频率推进采样；离线期间不生成记录，恢复后从下一序号继续。 */
    public static CollectorAdvance advance(CollectorState current, Instant now)
    /** 保存通道工作状态，等待下一次实时请求。 */
    public static CollectorState configureStatus(CollectorState current, String channelCode, DeviceWorkStatus status)
    {
        if (!current.online()) return new CollectorAdvance(current, List.of());
        Instant sampleAt = current.lastSampledAt() == null
                ? now
                : current.lastSampledAt().plusMillis(current.sampleIntervalMs());
        if (sampleAt.isAfter(now)) return new CollectorAdvance(current, List.of());
        int index = parseChannelIndex(channelCode, current.channelCount());
        List<CollectorChannelState> channels = new ArrayList<>(current.channels());
        channels.set(index, new CollectorChannelState(channelCode, channels.get(index).metrics(), status));
        return copy(current, channels, List.of());
    }

        List<CollectorMeasurementRecord> appendedRecords = new ArrayList<>();
        Map<String, CollectorRealtimeMeasurement> latest = new LinkedHashMap<>();
        for (CollectorRealtimeMeasurement measurement : current.latestMeasurements())
            latest.put(measurementKey(measurement.channelCode(), measurement.metricCode()), measurement);
        long sequence = current.nextSequence();
        int generated = 0;
        Instant lastSampledAt = current.lastSampledAt();
        while (!sampleAt.isAfter(now) && generated < MAX_SAMPLES_PER_ADVANCE)
    /** 仅生成本次请求的当前值，不补算上一次请求与本次之间的空档。 */
    public static CollectorState sample(CollectorState current, Instant now, long sequence)
    {
        Instant observedAt = now.truncatedTo(java.time.temporal.ChronoUnit.MILLIS);
        List<CollectorRealtimeMeasurement> measurements = new ArrayList<>();
        for (CollectorChannelState channel : current.channels())
        {
            for (CollectorChannelState channel : current.channels())
            {
                for (CollectorMetricState metric : channel.metrics())
                {
                    CollectorMeasurementRecord record = new CollectorMeasurementRecord("dev-sim", current.deviceCode(),
                            channel.channelCode(), sequence, sampleAt,
                            DEVICE_TIME_FORMATTER.format(sampleAt.atZone(DEVICE_ZONE)), now,
                            metric.metricCode(), metric.value(), metric.unit());
                    appendedRecords.add(record);
                    latest.put(measurementKey(record.channelCode(), record.metricCode()),
                            new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(),
                                    record.channelCode(), record.sequenceId(), record.deviceTime(),
                                    record.deviceTimeRaw(), record.serverReceivedAt(), record.metricCode(),
                                    record.value(), record.unit()));
                }
            }
            lastSampledAt = sampleAt;
            sequence++;
            generated++;
            sampleAt = sampleAt.plusMillis(current.sampleIntervalMs());
            if (channel.workStatus() == DeviceWorkStatus.UNOBSERVABLE) continue;
            for (CollectorMetricState metric : channel.metrics())
                measurements.add(new CollectorRealtimeMeasurement("dev-sim", current.deviceCode(), channel.channelCode(),
                        sequence, observedAt, DEVICE_TIME_FORMATTER.format(observedAt.atZone(DEVICE_ZONE)), observedAt,
                        metric.metricCode(), metric.value(), metric.unit()));
        }
        CollectorState next = copy(current, current.ip(), current.channelCount(), current.online(),
                current.sampleIntervalMs(),
                sequence, lastSampledAt, current.channels(), List.copyOf(latest.values()),
                Math.min(BtsSimulatorStateStore.MAX_HISTORY_RECORDS,
                        current.historyRecordCount() + appendedRecords.size()),
                current.historyFaults());
        return new CollectorAdvance(next, appendedRecords);
        return copy(current, current.channels(), measurements);
    }

    /** 返回独立保存的最新实时快照，不从原始历史反推当前状态。 */
    public static CollectorRealtimeSnapshot realtime(CollectorState current)
    /** 校验恢复后的稳定通道和测量配置。 */
    public static void validate(CollectorState state)
    {
        return new CollectorRealtimeSnapshot(current.deviceCode(), current.online(), current.sampleIntervalMs(),
                current.latestMeasurements());
    }

    /** 对只追加原始模拟历史应用可重复的下载故障视图，再按偏移量分页。 */
    public static CollectorHistoryPage history(CollectorState current, BtsSimulatorStateStore stateStore,
            long offset, int limit)
    {
        if (offset < 0) throw new IllegalArgumentException("offset must not be negative");
        if (limit < 1 || limit > MAX_HISTORY_PAGE_SIZE)
            throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_PAGE_SIZE);
        CollectorHistoryFaults faults = current.historyFaults();
        if (faults.missingSequences().isEmpty() && faults.duplicateSequence() == null
                && faults.conflictingSequence() == null && !faults.outOfOrder())
        {
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    current.deviceCode(), offset, limit, false);
            return new CollectorHistoryPage(current.deviceCode(), offset, batch.nextOffset(), batch.hasMore(),
                    batch.total(), batch.records());
        }
        return faultedHistoryPage(current, stateStore, offset, limit);
    }

    /** 校验 Redis 恢复的采集器采样状态，拒绝损坏状态被静默替换。 */
    public static void validate(CollectorState state, BtsSimulatorStateStore stateStore)
    {
        validateSampleInterval(state.sampleIntervalMs());
        if (state.nextSequence() < 1 || state.channels().size() != state.channelCount())
            throw new IllegalStateException("Invalid collector sampling state in Redis");
        if (state.channels().size() != state.channelCount()) throw new IllegalStateException("Invalid collector topology");
        for (int index = 0; index < state.channels().size(); index++)
        {
            CollectorChannelState channel = state.channels().get(index);
            if (!channelCode(index + 1).equals(channel.channelCode()))
                throw new IllegalStateException("Invalid collector channel identity in Redis");
            var channel = state.channels().get(index);
            if (!channelCode(index + 1).equals(channel.channelCode())) throw new IllegalStateException("Invalid collector channel identity");
            validateMetrics(channel.metrics());
        }
        validateLatestMeasurements(state);
        validateHistory(state, stateStore);
        validateFaults(state.historyFaults());
    }

    private static void validateLatestMeasurements(CollectorState state)
    private static CollectorState copy(CollectorState current, List<CollectorChannelState> channels,
            List<CollectorRealtimeMeasurement> measurements)
    {
        Set<String> configured = configuredMeasurementKeys(state.channels());
        Set<String> identities = new java.util.HashSet<>();
        for (CollectorRealtimeMeasurement measurement : state.latestMeasurements())
        {
            String key = measurementKey(measurement.channelCode(), measurement.metricCode());
            if (!validMeasurement(state, measurement.adapterCode(), measurement.collectorCode(),
                    measurement.channelCode(), measurement.sequenceId(), measurement.deviceTime(),
                    measurement.deviceTimeRaw(), measurement.serverReceivedAt(), measurement.metricCode(),
                    measurement.value(), measurement.unit()) || !configured.contains(key) || !identities.add(key)
                    || measurement.sequenceId() >= state.nextSequence())
                throw new IllegalStateException("Invalid collector realtime state in Redis");
        }
    }

    private static void validateHistory(CollectorState state, BtsSimulatorStateStore stateStore)
    {
        long maximumSequence = 0;
        long previousSequence = 0;
        long recordCount = 0;
        long offset = 0;
        Set<String> identities = new java.util.HashSet<>();
        while (true)
        {
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    state.deviceCode(), offset, MAX_HISTORY_PAGE_SIZE, false);
            if (batch.offset() != offset || batch.records().size() > MAX_HISTORY_PAGE_SIZE)
                throw new IllegalStateException("Invalid collector history page in Redis");
            for (CollectorMeasurementRecord record : batch.records())
            {
                if (record.sequenceId() < previousSequence)
                    throw new IllegalStateException("Invalid collector history state in Redis");
                if (record.sequenceId() != previousSequence)
                {
                    identities.clear();
                    previousSequence = record.sequenceId();
                }
                String identity = record.channelCode() + "\u0000" + record.metricCode();
                if (!validMeasurement(state, record.adapterCode(), record.collectorCode(), record.channelCode(),
                        record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
                        record.metricCode(), record.value(), record.unit()) || !identities.add(identity))
                    throw new IllegalStateException("Invalid collector history state in Redis");
                maximumSequence = Math.max(maximumSequence, record.sequenceId());
                recordCount++;
            }
            if (!batch.hasMore())
            {
                if (batch.total() != recordCount)
                    throw new IllegalStateException("Invalid collector history count in Redis");
                break;
            }
            if (batch.nextOffset() <= offset)
                throw new IllegalStateException("Invalid collector history page in Redis");
            offset = batch.nextOffset();
        }
        if (state.nextSequence() <= maximumSequence)
            throw new IllegalStateException("Invalid collector history sequence in Redis");
        if (state.historyRecordCount() != recordCount)
            throw new IllegalStateException("Invalid collector history count in Redis");
    }

    private static boolean validMeasurement(CollectorState state, String adapterCode, String collectorCode,
            String channelCode, long sequenceId, Instant deviceTime, String deviceTimeRaw,
            Instant serverReceivedAt, String metricCode, BigDecimal value, String unit)
    {
        return "dev-sim".equals(adapterCode) && state.deviceCode().equals(collectorCode)
                && validHistoricalChannelCode(channelCode)
                && sequenceId > 0 && deviceTime != null && deviceTimeRaw != null && !deviceTimeRaw.isBlank()
                && validDeviceTime(deviceTime, deviceTimeRaw) && serverReceivedAt != null
                && !serverReceivedAt.isBefore(deviceTime) && metricCode != null && !metricCode.isBlank()
                && metricCode.equals(metricCode.trim()) && value != null && unit != null && !unit.isBlank()
                && unit.equals(unit.trim());
    }

    private static boolean validHistoricalChannelCode(String channelCode)
    {
        if (channelCode == null || !channelCode.matches("S-\\d{2}")) return false;
        int index = Integer.parseInt(channelCode.substring(2));
        return index >= 1 && channelCode(index).equals(channelCode);
    }

    private static boolean validDeviceTime(Instant deviceTime, String deviceTimeRaw)
    {
        try
        {
            Instant parsed = LocalDateTime.parse(deviceTimeRaw, DEVICE_TIME_FORMATTER)
                    .atZone(DEVICE_ZONE).toInstant();
            return parsed.equals(deviceTime.truncatedTo(ChronoUnit.MILLIS));
        }
        catch (DateTimeParseException exception)
        {
            return false;
        }
    }

    private static List<CollectorRealtimeMeasurement> retainConfiguredMeasurements(
            List<CollectorRealtimeMeasurement> measurements, List<CollectorChannelState> channels)
    {
        Set<String> configured = configuredMeasurementKeys(channels);
        return measurements.stream()
                .filter(item -> configured.contains(measurementKey(item.channelCode(), item.metricCode())))
                .toList();
    }

    private static Set<String> configuredMeasurementKeys(List<CollectorChannelState> channels)
    {
        Set<String> configured = new java.util.HashSet<>();
        for (CollectorChannelState channel : channels)
            for (CollectorMetricState metric : channel.metrics())
                configured.add(measurementKey(channel.channelCode(), metric.metricCode()));
        return configured;
    }

    private static String measurementKey(String channelCode, String metricCode)
    {
        return channelCode + "\u0000" + metricCode;
    }

    private static CollectorHistoryPage faultedHistoryPage(CollectorState current,
            BtsSimulatorStateStore stateStore, long offset, int limit)
    {
        // 故障视图需返回精确 total，因此逐批扫描原始历史；正常下载会直接读取目标页。
        CollectorHistoryFaults faults = current.historyFaults();
        Set<Long> missing = Set.copyOf(faults.missingSequences());
        List<CollectorMeasurementRecord> page = new ArrayList<>(limit);
        long rawOffset = 0;
        long total = 0;
        while (true)
        {
            BtsSimulatorStateStore.CollectorHistoryBatch batch = stateStore.loadCollectorHistoryPage(
                    current.deviceCode(), rawOffset, MAX_HISTORY_PAGE_SIZE, faults.outOfOrder());
            for (CollectorMeasurementRecord record : batch.records())
            {
                if (missing.contains(record.sequenceId())) continue;
                total = appendFaultedRecord(page, record, total, offset, limit);
                if (Long.valueOf(record.sequenceId()).equals(faults.duplicateSequence()))
                    total = appendFaultedRecord(page, record, total, offset, limit);
                if (Long.valueOf(record.sequenceId()).equals(faults.conflictingSequence()))
                    total = appendFaultedRecord(page, conflictingRecord(record), total, offset, limit);
            }
            if (!batch.hasMore()) break;
            if (batch.nextOffset() <= rawOffset)
                throw new IllegalStateException("Invalid collector history page in Redis");
            rawOffset = batch.nextOffset();
        }
        long from = Math.min(offset, total);
        long nextOffset = from + page.size();
        return new CollectorHistoryPage(current.deviceCode(), offset, nextOffset, nextOffset < total, total, page);
    }

    private static long appendFaultedRecord(List<CollectorMeasurementRecord> page,
            CollectorMeasurementRecord record, long index, long offset, int limit)
    {
        if (index >= offset && page.size() < limit) page.add(record);
        return index + 1;
    }

    private static CollectorMeasurementRecord conflictingRecord(CollectorMeasurementRecord record)
    {
        return new CollectorMeasurementRecord(record.adapterCode(), record.collectorCode(), record.channelCode(),
                record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), record.serverReceivedAt(),
                record.metricCode(), record.value().add(new BigDecimal("0.001")), record.unit());
    }

    private static CollectorState copy(CollectorState current, String ip, int channelCount, boolean online,
            int sampleIntervalMs, long nextSequence, Instant lastSampledAt,
            List<CollectorChannelState> channels, List<CollectorRealtimeMeasurement> latestMeasurements,
            long historyRecordCount,
            CollectorHistoryFaults faults)
    {
        return new CollectorState(current.collectorId(), current.deviceCode(), ip.trim(), current.mac(),
                current.name(), current.manufacturer(), current.model(), channelCount, online, sampleIntervalMs,
                nextSequence, lastSampledAt, channels, latestMeasurements, historyRecordCount, faults);
    }

    private static void validateSampleInterval(int sampleIntervalMs)
    {
        if (sampleIntervalMs < MIN_SAMPLE_INTERVAL_MS || sampleIntervalMs > MAX_SAMPLE_INTERVAL_MS)
            throw new IllegalArgumentException("sampleIntervalMs must be between " + MIN_SAMPLE_INTERVAL_MS
                    + " and " + MAX_SAMPLE_INTERVAL_MS);
        return new CollectorState(current.collectorId(), current.deviceCode(), current.ip(), current.mac(),
                current.name(), current.manufacturer(), current.model(), current.channelCount(), current.online(),
                current.sampleIntervalMs(), channels, measurements);
    }

    private static void validateMetrics(List<CollectorMetricState> metrics)
@@ -377,15 +114,6 @@
                    || !metric.unit().equals(metric.unit().trim()) || !codes.add(metric.metricCode()))
                throw new IllegalArgumentException("Collector metric code, value and unit must be valid and unique");
        }
    }

    private static void validateFaults(CollectorHistoryFaults faults)
    {
        if (faults == null) throw new IllegalArgumentException("historyFaults must not be null");
        if (faults.missingSequences().stream().anyMatch(sequence -> sequence == null || sequence < 1)
                || faults.duplicateSequence() != null && faults.duplicateSequence() < 1
                || faults.conflictingSequence() != null && faults.conflictingSequence() < 1)
            throw new IllegalArgumentException("History fault sequences must be positive");
    }

    private static int parseChannelIndex(String value, int channelCount)
@@ -409,14 +137,4 @@
        public CollectorRealtimeSnapshot { measurements = List.copyOf(measurements); }
    }

    public record CollectorAdvance(CollectorState state, List<CollectorMeasurementRecord> appendedRecords)
    {
        public CollectorAdvance { appendedRecords = List.copyOf(appendedRecords); }
    }

    public record CollectorHistoryPage(String collectorCode, long offset, long nextOffset, boolean hasMore,
            long total, List<CollectorMeasurementRecord> records)
    {
        public CollectorHistoryPage { records = List.copyOf(records); }
    }
}