| | |
| | | package com.battery.business.runtime; |
| | | |
| | | import java.math.BigDecimal; |
| | | import java.time.Instant; |
| | | import java.util.ArrayList; |
| | | import java.util.List; |
| | | import java.util.Set; |
| | | import java.util.concurrent.TimeUnit; |
| | | import org.springframework.data.redis.core.RedisOperations; |
| | | import org.springframework.data.redis.core.SessionCallback; |
| | | import org.springframework.data.redis.core.RedisTemplate; |
| | | import org.springframework.stereotype.Component; |
| | | import org.springframework.context.annotation.Profile; |
| | |
| | | import com.alibaba.fastjson2.JSONObject; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.BtsDeviceProfile; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.BtsHistoryBatch; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.BtsHistoryRecord; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.BtsPlaybackState; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.ChannelConfiguration; |
| | | import com.battery.device.simulator.BtsSimulatorStateStore.SimulatorPlanStep; |
| | | import com.battery.device.api.model.DeviceWorkStatus; |
| | | |
| | | /** 将开发模拟器完整状态写入 Redis,主状态不设置普通 TTL。 */ |
| | |
| | | { |
| | | static final String STATE_KEY = "dev:bts-simulator:state:v1"; |
| | | static final String RECENT_REQUESTS_KEY = "dev:bts-simulator:recent-requests:v1"; |
| | | static final String HISTORY_KEYS_KEY = "dev:bts-simulator:collector-history-keys:v1"; |
| | | static final String HISTORY_KEY_PREFIX = "dev:bts-simulator:collector-history:v1:"; |
| | | // Ticket21 fixture 变更后切换独立 key 版本,避免把旧的 3 循环演示历史恢复到新时间轴。 |
| | | static final String BTS_HISTORY_KEYS_KEY = "dev:bts-simulator:bts-history-keys:v2"; |
| | | static final String BTS_HISTORY_KEY_PREFIX = "dev:bts-simulator:bts-history:v2:"; |
| | | static final String BTS_PLAYBACK_KEY_PREFIX = "dev:bts-simulator:bts-playback:v2:"; |
| | | private static final long RECENT_REQUESTS_TTL_HOURS = 1; |
| | | |
| | | private final RedisTemplate<Object, Object> redisTemplate; |
| | | private List<CollectorHistoryAppend> pendingEmbeddedHistory = List.of(); |
| | | |
| | | public RedisBtsSimulatorStateStore(RedisTemplate<Object, Object> redisTemplate) |
| | | { |
| | | this.redisTemplate = redisTemplate; |
| | | } |
| | | |
| | | /** 读取模拟器主状态,不存在时原子写入默认拓扑;并发初始化时以 Redis 中先写入的那份为准。 */ |
| | | @Override |
| | | public SimulatorState loadOrInitialize(SimulatorState defaultState) |
| | | { |
| | |
| | | return restoreState(stored); |
| | | } |
| | | |
| | | @Override |
| | | public SimulatorState resetHistoryForDemoStartup(SimulatorState state) |
| | | { |
| | | if (pendingEmbeddedHistory.isEmpty()) return state; |
| | | java.util.Map<String, Integer> embeddedCounts = new java.util.HashMap<>(); |
| | | pendingEmbeddedHistory.forEach(append -> embeddedCounts.put(append.collectorCode(), append.records().size())); |
| | | List<CollectorState> collectors = state.collectors().stream().map(collector -> { |
| | | int embeddedCount = embeddedCounts.getOrDefault(collector.deviceCode(), 0); |
| | | if (embeddedCount == 0) return collector; |
| | | long remaining = Math.max(0L, collector.historyRecordCount() - embeddedCount); |
| | | return new CollectorState(collector.collectorId(), collector.deviceCode(), collector.ip(), collector.mac(), |
| | | collector.name(), collector.manufacturer(), collector.model(), collector.channelCount(), |
| | | collector.online(), collector.sampleIntervalMs(), collector.nextSequence(), collector.lastSampledAt(), |
| | | collector.channels(), collector.latestMeasurements().stream() |
| | | .filter(item -> collector.deviceCode().equals(item.collectorCode())) |
| | | .toList(), remaining, collector.historyFaults()); |
| | | }).toList(); |
| | | // 不删除独立历史 Redis key;这里只阻止旧内嵌数据在本次启动时迁移回来。 |
| | | pendingEmbeddedHistory = List.of(); |
| | | SimulatorState cleaned = new SimulatorState(state.version(), state.devices(), collectors, state.nextDeviceId(), |
| | | state.nextCollectorId(), state.tombstones()); |
| | | save(cleaned); |
| | | return cleaned; |
| | | } |
| | | |
| | | private SimulatorState restoreState(Object stored) |
| | | { |
| | | ParsedSimulatorState parsed = parseState(stored); |
| | | // 旧格式历史先只供恢复校验读取,损坏状态不得被迁移写回覆盖。 |
| | | pendingEmbeddedHistory = parsed.embeddedHistory(); |
| | | return parsed.state(); |
| | | } |
| | | |
| | | @Override |
| | | public void completeValidatedRestore(SimulatorState state) |
| | | { |
| | | if (pendingEmbeddedHistory.isEmpty()) return; |
| | | save(state, pendingEmbeddedHistory); |
| | | pendingEmbeddedHistory = List.of(); |
| | | } |
| | | |
| | | /** 整体覆盖模拟器主状态;该键不设普通 TTL。 */ |
| | | @Override |
| | | public void save(SimulatorState state) |
| | | { |
| | | save(state, List.of()); |
| | | redisTemplate.opsForValue().set(STATE_KEY, JSON.toJSONString(state)); |
| | | } |
| | | |
| | | @Override |
| | | @SuppressWarnings({ "rawtypes", "unchecked" }) |
| | | public void save(SimulatorState state, List<CollectorHistoryAppend> historyAppends) |
| | | { |
| | | List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() |
| | | { |
| | | @Override |
| | | public <K, V> List<Object> execute(RedisOperations<K, V> operations) |
| | | { |
| | | RedisOperations<Object, Object> typed = (RedisOperations) operations; |
| | | typed.multi(); |
| | | for (CollectorHistoryAppend append : historyAppends) |
| | | { |
| | | String historyKey = historyKey(append.collectorCode()); |
| | | if (!append.records().isEmpty()) |
| | | { |
| | | List<Object> records = new ArrayList<>( |
| | | append.records().stream().map(JSON::toJSONString).toList()); |
| | | typed.opsForList().rightPushAll(historyKey, records); |
| | | typed.opsForList().trim(historyKey, -MAX_HISTORY_RECORDS, -1); |
| | | typed.opsForSet().add(HISTORY_KEYS_KEY, historyKey); |
| | | } |
| | | } |
| | | typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state)); |
| | | return typed.exec(); |
| | | } |
| | | }); |
| | | if (results == null) throw new IllegalStateException("BTS simulator state transaction failed in Redis"); |
| | | } |
| | | |
| | | @Override |
| | | public CollectorHistoryBatch loadCollectorHistoryPage(String collectorCode, long offset, int limit, |
| | | boolean reverse) |
| | | { |
| | | if (offset < 0) throw new IllegalArgumentException("offset must not be negative"); |
| | | if (limit < 1 || limit > MAX_HISTORY_BATCH_SIZE) |
| | | throw new IllegalArgumentException("limit must be between 1 and " + MAX_HISTORY_BATCH_SIZE); |
| | | String key = historyKey(collectorCode); |
| | | Long storedSize = redisTemplate.opsForList().size(key); |
| | | long redisCount = storedSize == null ? 0 : storedSize; |
| | | List<CollectorMeasurementRecord> embedded = pendingEmbeddedHistory.stream() |
| | | .filter(append -> append.collectorCode().equals(collectorCode)) |
| | | .findFirst().map(CollectorHistoryAppend::records).orElse(List.of()); |
| | | long total = redisCount + embedded.size(); |
| | | long from = Math.min(offset, total); |
| | | int count = (int) Math.min(limit, total - from); |
| | | List<CollectorMeasurementRecord> records = new ArrayList<>(count); |
| | | if (reverse) |
| | | { |
| | | long lower = total - from - count; |
| | | long upper = total - from; |
| | | long embeddedLower = Math.max(lower, redisCount); |
| | | for (long index = upper - 1; index >= embeddedLower; index--) |
| | | records.add(embedded.get((int) (index - redisCount))); |
| | | long redisUpper = Math.min(upper, redisCount); |
| | | if (lower < redisUpper) |
| | | { |
| | | List<Object> stored = redisTemplate.opsForList().range(key, lower, redisUpper - 1); |
| | | if (stored != null) |
| | | for (int index = stored.size() - 1; index >= 0; index--) |
| | | records.add(collectorMeasurementRecord(stored.get(index))); |
| | | } |
| | | } |
| | | else |
| | | { |
| | | long upper = from + count; |
| | | long redisUpper = Math.min(upper, redisCount); |
| | | if (from < redisUpper) |
| | | { |
| | | List<Object> stored = redisTemplate.opsForList().range(key, from, redisUpper - 1); |
| | | if (stored != null) stored.forEach(item -> records.add(collectorMeasurementRecord(item))); |
| | | } |
| | | long embeddedFrom = Math.max(from, redisCount); |
| | | if (embeddedFrom < upper) |
| | | records.addAll(embedded.subList((int) (embeddedFrom - redisCount), |
| | | (int) (upper - redisCount))); |
| | | } |
| | | if (records.size() != count) |
| | | throw new IllegalStateException("Collector history page changed while reading Redis"); |
| | | long nextOffset = from + records.size(); |
| | | return new CollectorHistoryBatch(offset, nextOffset, nextOffset < total, total, records); |
| | | } |
| | | |
| | | @Override |
| | | public BtsHistoryBatch loadBtsHistoryPage(String historyKey, long offset, int limit) |
| | | { |
| | | if (offset < 0 || limit < 1 || limit > MAX_HISTORY_BATCH_SIZE) |
| | | throw new IllegalArgumentException("invalid BTS history page"); |
| | | String key = BTS_HISTORY_KEY_PREFIX + historyKey; |
| | | Long size = redisTemplate.opsForList().size(key); |
| | | long total = size == null ? 0 : size; |
| | | long from = Math.min(offset, total); |
| | | int count = (int) Math.min(limit, total - from); |
| | | List<Object> values = count == 0 ? List.of() : redisTemplate.opsForList().range(key, from, from + count - 1); |
| | | List<BtsHistoryRecord> records = values == null ? List.of() : values.stream() |
| | | .map(item -> JSON.parseObject(item.toString(), BtsHistoryRecord.class)).toList(); |
| | | long next = from + records.size(); |
| | | return new BtsHistoryBatch(offset, next, next < total, total, records); |
| | | } |
| | | |
| | | @Override |
| | | @SuppressWarnings({ "rawtypes", "unchecked" }) |
| | | public void appendBtsHistory(String historyKey, BtsHistoryRecord record) |
| | | { |
| | | String key = BTS_HISTORY_KEY_PREFIX + historyKey; |
| | | RedisOperations<Object, Object> operations = redisTemplate; |
| | | operations.opsForList().rightPush(key, JSON.toJSONString(record)); |
| | | operations.opsForList().trim(key, -MAX_HISTORY_RECORDS, -1); |
| | | operations.opsForSet().add(BTS_HISTORY_KEYS_KEY, key); |
| | | } |
| | | |
| | | @Override |
| | | public BtsPlaybackState loadBtsPlaybackState(String historyKey) |
| | | { |
| | | Object value = redisTemplate.opsForValue().get(BTS_PLAYBACK_KEY_PREFIX + historyKey); |
| | | return value == null ? null : JSON.parseObject(value.toString(), BtsPlaybackState.class); |
| | | } |
| | | |
| | | @Override |
| | | public void saveBtsPlaybackState(String historyKey, BtsPlaybackState playback) |
| | | { |
| | | redisTemplate.opsForValue().set(BTS_PLAYBACK_KEY_PREFIX + historyKey, JSON.toJSONString(playback)); |
| | | } |
| | | |
| | | @Override |
| | | public void deleteBtsPlaybackState(String historyKey) |
| | | { |
| | | redisTemplate.delete(BTS_PLAYBACK_KEY_PREFIX + historyKey); |
| | | } |
| | | |
| | | @Override |
| | | public void deleteBtsHistory(String historyKey) |
| | | { |
| | | String key = BTS_HISTORY_KEY_PREFIX + historyKey; |
| | | redisTemplate.delete(key); |
| | | redisTemplate.opsForSet().remove(BTS_HISTORY_KEYS_KEY, key); |
| | | } |
| | | |
| | | @Override |
| | | @SuppressWarnings({ "rawtypes", "unchecked" }) |
| | | public void saveBtsPlaybackAndAppendHistory(String historyKey, BtsPlaybackState playback, |
| | | List<BtsHistoryRecord> records) |
| | | { |
| | | List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() |
| | | { |
| | | @Override |
| | | public <K, V> List<Object> execute(RedisOperations<K, V> operations) |
| | | { |
| | | RedisOperations<Object, Object> typed = (RedisOperations) operations; |
| | | typed.multi(); |
| | | String key = BTS_HISTORY_KEY_PREFIX + historyKey; |
| | | if (!records.isEmpty()) |
| | | { |
| | | typed.opsForList().rightPushAll(key, new ArrayList<>(records.stream() |
| | | .map(JSON::toJSONString).toList())); |
| | | typed.opsForList().trim(key, -MAX_HISTORY_RECORDS, -1); |
| | | typed.opsForSet().add(BTS_HISTORY_KEYS_KEY, key); |
| | | } |
| | | typed.opsForValue().set(BTS_PLAYBACK_KEY_PREFIX + historyKey, JSON.toJSONString(playback)); |
| | | return typed.exec(); |
| | | } |
| | | }); |
| | | if (results == null) throw new IllegalStateException("BTS simulator playback transaction failed in Redis"); |
| | | } |
| | | |
| | | @Override |
| | | @SuppressWarnings({ "rawtypes", "unchecked" }) |
| | | public void saveAndDeleteCollectorHistory(SimulatorState state, String collectorCode) |
| | | { |
| | | String historyKey = historyKey(collectorCode); |
| | | List<Object> results = redisTemplate.execute(new SessionCallback<List<Object>>() |
| | | { |
| | | @Override |
| | | public <K, V> List<Object> execute(RedisOperations<K, V> operations) |
| | | { |
| | | RedisOperations<Object, Object> typed = (RedisOperations) operations; |
| | | typed.multi(); |
| | | typed.delete(historyKey); |
| | | typed.opsForSet().remove(HISTORY_KEYS_KEY, historyKey); |
| | | typed.opsForValue().set(STATE_KEY, JSON.toJSONString(state)); |
| | | return typed.exec(); |
| | | } |
| | | }); |
| | | if (results == null) throw new IllegalStateException("BTS simulator collector deletion failed in Redis"); |
| | | } |
| | | |
| | | /** 读取短期保留的最近请求;键不存在时返回空列表。 */ |
| | | @Override |
| | | public List<RecentRequest> loadRecentRequests() |
| | | { |
| | |
| | | return List.copyOf(JSON.parseArray(stored.toString(), RecentRequest.class)); |
| | | } |
| | | |
| | | /** 覆盖保存最近请求,整键按小时级 TTL 自动过期。 */ |
| | | @Override |
| | | public void saveRecentRequests(List<RecentRequest> requests) |
| | | { |
| | |
| | | RECENT_REQUESTS_TTL_HOURS, TimeUnit.HOURS); |
| | | } |
| | | |
| | | /** 人工重置:删除最近请求并写回默认主状态。 */ |
| | | @Override |
| | | public void reset(SimulatorState defaultState) |
| | | { |
| | | pendingEmbeddedHistory = List.of(); |
| | | Set<Object> historyKeys = redisTemplate.opsForSet().members(HISTORY_KEYS_KEY); |
| | | if (historyKeys != null && !historyKeys.isEmpty()) redisTemplate.delete(historyKeys); |
| | | redisTemplate.delete(HISTORY_KEYS_KEY); |
| | | Set<Object> btsHistoryKeys = redisTemplate.opsForSet().members(BTS_HISTORY_KEYS_KEY); |
| | | if (btsHistoryKeys != null && !btsHistoryKeys.isEmpty()) redisTemplate.delete(btsHistoryKeys); |
| | | redisTemplate.delete(BTS_HISTORY_KEYS_KEY); |
| | | // 播放状态按通道独立保存,重置演示时必须一并清理。 |
| | | Set<Object> playbackKeys = redisTemplate.keys(BTS_PLAYBACK_KEY_PREFIX + "*"); |
| | | if (playbackKeys != null && !playbackKeys.isEmpty()) redisTemplate.delete(playbackKeys); |
| | | redisTemplate.delete(RECENT_REQUESTS_KEY); |
| | | save(defaultState); |
| | | } |
| | | |
| | | private ParsedSimulatorState parseState(Object stored) |
| | | private SimulatorState restoreState(Object stored) |
| | | { |
| | | try |
| | | { |
| | |
| | | List<DeviceState> devices = new ArrayList<>(); |
| | | requiredArray(root, "devices").forEach(item -> devices.add(deviceState(item))); |
| | | List<CollectorState> collectors = new ArrayList<>(); |
| | | List<CollectorHistoryAppend> embeddedHistory = new ArrayList<>(); |
| | | requiredArray(root, "collectors").forEach(item -> { |
| | | JSONObject collectorObject = jsonObject(item); |
| | | List<CollectorMeasurementRecord> history = embeddedCollectorHistory(collectorObject); |
| | | CollectorState collector = collectorState(collectorObject, history); |
| | | collectors.add(collector); |
| | | if (!history.isEmpty()) |
| | | embeddedHistory.add(new CollectorHistoryAppend(collector.deviceCode(), history)); |
| | | }); |
| | | requiredArray(root, "collectors").forEach(item -> collectors.add(collectorState(jsonObject(item)))); |
| | | List<ChannelTombstone> tombstones = new ArrayList<>(); |
| | | requiredArray(root, "tombstones").forEach(item -> tombstones.add(channelTombstone(item))); |
| | | SimulatorState state = new SimulatorState(version, devices, collectors, |
| | | requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones); |
| | | return new ParsedSimulatorState(state, embeddedHistory); |
| | | requiredInteger(root, "nextDeviceId"), requiredInteger(root, "nextCollectorId"), tombstones, |
| | | root.getBoolean("softwareAccessible"), root.getLongValue("observationSequence")); |
| | | return state; |
| | | } |
| | | catch (IllegalStateException exception) |
| | | { |
| | |
| | | requiredDecimal(object, "voltageMaxV"), requiredInteger(object, "sampleIntervalMs")); |
| | | } |
| | | |
| | | private List<CollectorMeasurementRecord> embeddedCollectorHistory(JSONObject object) |
| | | private CollectorState collectorState(JSONObject object) |
| | | { |
| | | JSONArray values = object.getJSONArray("history"); |
| | | if (values == null) return List.of(); |
| | | List<CollectorMeasurementRecord> history = new ArrayList<>(values.size()); |
| | | values.forEach(item -> history.add(collectorMeasurementRecord(item))); |
| | | return List.copyOf(history); |
| | | } |
| | | |
| | | private CollectorState collectorState(JSONObject object, List<CollectorMeasurementRecord> embeddedHistory) |
| | | { |
| | | int collectorId = requiredInteger(object, "collectorId"); |
| | | String deviceCode = object.getString("deviceCode"); |
| | | String ip = object.getString("ip"); |
| | | String mac = object.getString("mac"); |
| | | String name = object.getString("name"); |
| | | String manufacturer = object.getString("manufacturer"); |
| | | String model = object.getString("model"); |
| | | int channelCount = requiredInteger(object, "channelCount"); |
| | | boolean online = object.getBooleanValue("online"); |
| | | // Ticket04 状态只包含发现配置;首次恢复时以演示默认值补齐采样状态。 |
| | | if (!object.containsKey("sampleIntervalMs")) |
| | | return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model, |
| | | channelCount, online); |
| | | |
| | | int count = requiredInteger(object, "channelCount"); |
| | | List<CollectorChannelState> channels = new ArrayList<>(); |
| | | requiredArray(object, "channels").forEach(item -> channels.add(collectorChannelState(item))); |
| | | List<CollectorRealtimeMeasurement> latestMeasurements = new ArrayList<>(); |
| | | JSONArray latest = object.getJSONArray("latestMeasurements"); |
| | | if (latest != null) |
| | | latest.forEach(item -> latestMeasurements.add(collectorRealtimeMeasurement(item))); |
| | | else |
| | | latestMeasurements.addAll(latestMeasurementsFromLegacyHistory(embeddedHistory, channels)); |
| | | long historyRecordCount = object.containsKey("historyRecordCount") |
| | | ? requiredLong(object, "historyRecordCount") : embeddedHistory.size(); |
| | | return new CollectorState(collectorId, deviceCode, ip, mac, name, manufacturer, model, |
| | | channelCount, online, requiredInteger(object, "sampleIntervalMs"), |
| | | requiredLong(object, "nextSequence"), optionalInstant(object, "lastSampledAt"), |
| | | channels, latestMeasurements, historyRecordCount, |
| | | collectorHistoryFaults(object.get("historyFaults"))); |
| | | JSONArray configured = object.getJSONArray("channels"); |
| | | if (configured == null) channels.addAll(com.battery.device.simulator.DevelopmentCollectorSimulation.defaultChannels(count)); |
| | | else configured.forEach(item -> channels.add(collectorChannelState(item))); |
| | | // 重启只恢复配置,不把上次响应或已废止的历史作为新观测。 |
| | | return new CollectorState(requiredInteger(object, "collectorId"), object.getString("deviceCode"), |
| | | object.getString("ip"), object.getString("mac"), object.getString("name"), |
| | | object.getString("manufacturer"), object.getString("model"), count, object.getBooleanValue("online"), |
| | | com.battery.device.simulator.DevelopmentCollectorSimulation.DEFAULT_SAMPLE_INTERVAL_MS, channels, List.of()); |
| | | } |
| | | |
| | | private CollectorChannelState collectorChannelState(Object value) |
| | |
| | | JSONObject object = jsonObject(value); |
| | | List<CollectorMetricState> metrics = new ArrayList<>(); |
| | | requiredArray(object, "metrics").forEach(item -> metrics.add(collectorMetricState(item))); |
| | | return new CollectorChannelState(object.getString("channelCode"), metrics); |
| | | return new CollectorChannelState(object.getString("channelCode"), metrics, |
| | | object.getString("workStatus") == null ? DeviceWorkStatus.WORKING |
| | | : DeviceWorkStatus.valueOf(object.getString("workStatus"))); |
| | | } |
| | | |
| | | private CollectorMetricState collectorMetricState(Object value) |
| | |
| | | return new CollectorMetricState(object.getString("metricCode"), measurement, object.getString("unit")); |
| | | } |
| | | |
| | | private CollectorMeasurementRecord collectorMeasurementRecord(Object value) |
| | | { |
| | | JSONObject object = jsonObject(value); |
| | | BigDecimal measurement = object.getBigDecimal("value"); |
| | | if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value"); |
| | | return new CollectorMeasurementRecord(object.getString("adapterCode"), object.getString("collectorCode"), |
| | | object.getString("channelCode"), requiredLong(object, "sequenceId"), |
| | | requiredInstant(object, "deviceTime"), object.getString("deviceTimeRaw"), |
| | | requiredInstant(object, "serverReceivedAt"), object.getString("metricCode"), |
| | | measurement, object.getString("unit")); |
| | | } |
| | | |
| | | private CollectorRealtimeMeasurement collectorRealtimeMeasurement(Object value) |
| | | { |
| | | JSONObject object = jsonObject(value); |
| | | BigDecimal measurement = object.getBigDecimal("value"); |
| | | if (measurement == null) throw new IllegalStateException("Missing BTS simulator state field: value"); |
| | | return new CollectorRealtimeMeasurement(object.getString("adapterCode"), object.getString("collectorCode"), |
| | | object.getString("channelCode"), requiredLong(object, "sequenceId"), |
| | | requiredInstant(object, "deviceTime"), object.getString("deviceTimeRaw"), |
| | | requiredInstant(object, "serverReceivedAt"), object.getString("metricCode"), |
| | | measurement, object.getString("unit")); |
| | | } |
| | | |
| | | /** 兼容 Ticket06 开发期间已写入但尚未拆分快照模型的状态。 */ |
| | | private List<CollectorRealtimeMeasurement> latestMeasurementsFromLegacyHistory( |
| | | List<CollectorMeasurementRecord> history, List<CollectorChannelState> channels) |
| | | { |
| | | java.util.Set<String> configured = new java.util.HashSet<>(); |
| | | channels.forEach(channel -> channel.metrics().forEach(metric -> |
| | | configured.add(channel.channelCode() + "\u0000" + metric.metricCode()))); |
| | | java.util.Map<String, CollectorRealtimeMeasurement> latest = new java.util.LinkedHashMap<>(); |
| | | for (CollectorMeasurementRecord record : history) |
| | | { |
| | | String key = record.channelCode() + "\u0000" + record.metricCode(); |
| | | if (configured.contains(key)) |
| | | latest.put(key, new CollectorRealtimeMeasurement(record.adapterCode(), record.collectorCode(), |
| | | record.channelCode(), record.sequenceId(), record.deviceTime(), record.deviceTimeRaw(), |
| | | record.serverReceivedAt(), record.metricCode(), record.value(), record.unit())); |
| | | } |
| | | return List.copyOf(latest.values()); |
| | | } |
| | | |
| | | private CollectorHistoryFaults collectorHistoryFaults(Object value) |
| | | { |
| | | if (value == null) return CollectorHistoryFaults.none(); |
| | | JSONObject object = jsonObject(value); |
| | | List<Long> missing = new ArrayList<>(); |
| | | JSONArray values = object.getJSONArray("missingSequences"); |
| | | if (values != null) values.forEach(item -> missing.add(Long.valueOf(item.toString()))); |
| | | return new CollectorHistoryFaults(missing, object.getLong("duplicateSequence"), |
| | | object.getLong("conflictingSequence"), object.getBooleanValue("outOfOrder")); |
| | | } |
| | | |
| | | private ChannelState channelState(Object value) |
| | | { |
| | | JSONObject object = jsonObject(value); |
| | | String workStatus = object.getString("workStatus"); |
| | | if (workStatus == null) throw new IllegalStateException("Missing BTS simulator state field: workStatus"); |
| | | JSONObject configuration = object.getJSONObject("configuration"); |
| | | return new ChannelState(requiredInteger(object, "channelId"), object.getBooleanValue("online"), |
| | | DeviceWorkStatus.valueOf(workStatus)); |
| | | DeviceWorkStatus.valueOf(workStatus), channelConfiguration(configuration)); |
| | | } |
| | | |
| | | /** Redis 通用反序列化会把泛型 steps 恢复为 JSONObject,不能让 record 构造器直接强转。 */ |
| | | private ChannelConfiguration channelConfiguration(JSONObject object) |
| | | { |
| | | if (object == null) return null; |
| | | JSONArray configuredSteps = requiredArray(object, "steps"); |
| | | List<SimulatorPlanStep> steps = new ArrayList<>(); |
| | | configuredSteps.forEach(item -> { |
| | | JSONObject step = jsonObject(item); |
| | | steps.add(new SimulatorPlanStep(step.getString("name"), step.getString("stepType"), |
| | | step.getDoubleValue("durationSeconds"), step.getDoubleValue("voltageV"))); |
| | | }); |
| | | return new ChannelConfiguration(requiredDecimal(object, "currentA"), requiredDecimal(object, "voltageV"), |
| | | requiredDecimal(object, "capacityAh"), requiredDecimal(object, "energyWh"), |
| | | requiredInteger(object, "cycleNo"), requiredInteger(object, "stepNo"), |
| | | requiredInteger(object, "plannedCycleCount"), object.getLongValue("totalTimeSeconds"), |
| | | object.getLongValue("relativeTimeSeconds"), steps); |
| | | } |
| | | |
| | | private ChannelTombstone channelTombstone(Object value) |
| | |
| | | requiredInteger(object, "channelId"), requiredInteger(object, "remainingDiscoveries")); |
| | | } |
| | | |
| | | private long requiredLong(JSONObject object, String field) |
| | | { |
| | | Long value = object.getLong(field); |
| | | if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field); |
| | | return value; |
| | | } |
| | | |
| | | private BigDecimal requiredDecimal(JSONObject object, String field) |
| | | { |
| | | BigDecimal value = object.getBigDecimal(field); |
| | |
| | | return value; |
| | | } |
| | | |
| | | private Instant requiredInstant(JSONObject object, String field) |
| | | { |
| | | Instant value = optionalInstant(object, field); |
| | | if (value == null) throw new IllegalStateException("Missing BTS simulator state field: " + field); |
| | | return value; |
| | | } |
| | | |
| | | private Instant optionalInstant(JSONObject object, String field) |
| | | { |
| | | String value = object.getString(field); |
| | | return value == null ? null : Instant.parse(value); |
| | | } |
| | | |
| | | private String historyKey(String collectorCode) |
| | | { |
| | | return HISTORY_KEY_PREFIX + collectorCode; |
| | | } |
| | | |
| | | private record ParsedSimulatorState(SimulatorState state, List<CollectorHistoryAppend> embeddedHistory) |
| | | { |
| | | private ParsedSimulatorState { embeddedHistory = List.copyOf(embeddedHistory); } |
| | | } |
| | | } |