package com.battery.device.simulator;
|
|
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;
|
|
/** 实现第一版开发采集器的独立采样与故障注入合同。 */
|
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 DevelopmentCollectorSimulation() { }
|
|
/** 创建带稳定 S 通道和三项固定演示指标的采集器状态。 */
|
public static List<CollectorChannelState> defaultChannels(int channelCount)
|
{
|
List<CollectorChannelState> channels = new ArrayList<>(channelCount);
|
for (int index = 1; index <= channelCount; index++)
|
{
|
channels.add(new CollectorChannelState(channelCode(index), List.of(
|
new CollectorMetricState("temperature", new BigDecimal("25.0")
|
.add(BigDecimal.valueOf(index, 1)), "℃"),
|
new CollectorMetricState("positive_potential", new BigDecimal("3.700"), "V"),
|
new CollectorMetricState("negative_potential", new BigDecimal("-0.100"), "V"))));
|
}
|
return List.copyOf(channels);
|
}
|
|
/** 调整通道数时保留已有通道配置，并为新增通道生成稳定默认配置。 */
|
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);
|
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());
|
}
|
|
/** 更新采样频率和历史下载故障；测量原始历史保持只追加。 */
|
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)
|
{
|
validateMetrics(metrics);
|
int channelIndex = 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());
|
}
|
|
/** 按采集器自身频率推进采样；离线期间不生成记录，恢复后从下一序号继续。 */
|
public static CollectorAdvance advance(CollectorState current, Instant now)
|
{
|
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());
|
|
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)
|
{
|
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());
|
}
|
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);
|
}
|
|
/** 返回独立保存的最新实时快照，不从原始历史反推当前状态。 */
|
public static CollectorRealtimeSnapshot realtime(CollectorState current)
|
{
|
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");
|
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");
|
validateMetrics(channel.metrics());
|
}
|
validateLatestMeasurements(state);
|
validateHistory(state, stateStore);
|
validateFaults(state.historyFaults());
|
}
|
|
private static void validateLatestMeasurements(CollectorState state)
|
{
|
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);
|
}
|
|
private static void validateMetrics(List<CollectorMetricState> metrics)
|
{
|
if (metrics == null || metrics.isEmpty())
|
throw new IllegalArgumentException("Collector channel must contain at least one metric");
|
java.util.HashSet<String> codes = new java.util.HashSet<>();
|
for (CollectorMetricState metric : metrics)
|
{
|
if (metric.metricCode() == null || metric.metricCode().isBlank() || metric.value() == null
|
|| !metric.metricCode().equals(metric.metricCode().trim())
|
|| metric.unit() == null || metric.unit().isBlank()
|
|| !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)
|
{
|
if (value == null || !value.matches("S-\\d{2}"))
|
throw new IllegalArgumentException("Collector channel must use S-01...S-N identity");
|
int index = Integer.parseInt(value.substring(2)) - 1;
|
if (index < 0 || index >= channelCount || !channelCode(index + 1).equals(value))
|
throw new IllegalArgumentException("Unknown collector channel: " + value);
|
return index;
|
}
|
|
private static String channelCode(int index)
|
{
|
return String.format("S-%02d", index);
|
}
|
|
public record CollectorRealtimeSnapshot(String collectorCode, boolean online, int sampleIntervalMs,
|
List<CollectorRealtimeMeasurement> measurements)
|
{
|
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); }
|
}
|
}
|
