From 7259e265b9237286996b09da03038ea5a133e1e0 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期六, 05 九月 2026 17:45:07 +0800
Subject: [PATCH] 收敛模拟器为设备事实与实时观测
---
battery-device-simulator/src/main/java/com/battery/device/simulator/DevelopmentCollectorSimulation.java | 374 ++++++----------------------------------------------
1 files changed, 46 insertions(+), 328 deletions(-)
diff --git a/battery-device-simulator/src/main/java/com/battery/device/simulator/DevelopmentCollectorSimulation.java b/battery-device-simulator/src/main/java/com/battery/device/simulator/DevelopmentCollectorSimulation.java
index 0dc231b..b0975f8 100644
--- a/battery-device-simulator/src/main/java/com/battery/device/simulator/DevelopmentCollectorSimulation.java
+++ b/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); }
- }
}
--
Gitblit v1.7.1