package com.battery.business.service.impl;
|
|
import java.time.Instant;
|
import java.math.BigDecimal;
|
import java.util.ArrayList;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Objects;
|
import org.springframework.stereotype.Service;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.lang.Nullable;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
|
import com.battery.device.api.model.Measurement;
|
import com.battery.business.domain.TTest;
|
import com.battery.business.domain.TTestBattery;
|
import com.battery.business.service.ITestLiveService;
|
import com.battery.business.service.ITestLiveService.TestPlanRow;
|
import com.battery.business.service.ITestLiveService.TestPlanSnapshot;
|
import com.battery.business.service.ITTestService;
|
import com.battery.business.mapper.TTestMapper;
|
|
/** 将设备最新快照转换为测试视角，按测试接收状态决定是否展示采集参数。 */
|
@Service
|
public class TTestLiveServiceImpl implements ITestLiveService
|
{
|
private final ITTestService tests;
|
private final DeviceDiscoveryCandidateStore candidates;
|
private final TestDataObservationAdapter adapter;
|
private final TTestMapper mapper;
|
|
@Autowired
|
public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
|
@Nullable TestDataObservationAdapter adapter, TTestMapper mapper)
|
{ this.tests = tests; this.candidates = candidates; this.adapter = adapter; this.mapper = mapper; }
|
|
public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
|
TestDataObservationAdapter adapter)
|
{ this(tests, candidates, adapter, null); }
|
|
/** 返回测试当前设备事实；接收中展示测量和实时提醒，历史异常处理状态不参与判定。 */
|
@Override public LiveState current(Long testId)
|
{
|
TTest test = tests.selectTestById(testId);
|
List<CabinetCandidate> cabinets = candidates.snapshot(Instant.now()).cabinets();
|
List<LiveSource> sources = new ArrayList<>();
|
boolean receiving = "RUNNING".equals(test.getTestStatus()) && "RECEIVING".equals(test.getReceptionStatus());
|
for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
|
{
|
if (battery.getBinding() != null)
|
sources.add(snapshotFor(cabinetSource(battery, cabinets), receiving));
|
if (battery.getCollectorId() != null)
|
sources.add(collectorSource(battery, receiving));
|
}
|
List<Map<String, Object>> exceptions = new ArrayList<>();
|
java.util.Set<String> offlineSources = new java.util.HashSet<>();
|
// 实时提醒只反映当前设备和测量事实，异常管理待办是否处理不参与展示判定。
|
java.util.Set<String> thresholdSources = new java.util.HashSet<>();
|
if (receiving) {
|
for (LiveSource source : sources) {
|
if ("ONLINE".equals(source.deviceStatus())) appendThresholdWarnings(test, source, exceptions, thresholdSources);
|
if (!"OFFLINE".equals(source.deviceStatus())) continue;
|
String identity = offlineIdentity(test, source);
|
if (!offlineSources.add(source.sourceType() + ":" + identity)) continue;
|
exceptions.add(Map.of("exceptionType", "DEVICE_OFFLINE", "sourceType", source.sourceType(),
|
"sourceId", source.sourceId(), "sourceIdentity", identity,
|
"status", "OPEN", "message", "接收中数据来源离线，当前无法取得测量数据"));
|
}
|
}
|
String testStatus = test.getTestStatus();
|
TestPlanSnapshot testPlan = detectorTestPlan(test);
|
return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now().toEpochMilli(), sources, exceptions, testPlan);
|
}
|
|
private void appendThresholdWarnings(TTest test, LiveSource source, List<Map<String, Object>> warnings,
|
java.util.Set<String> seen)
|
{
|
var snapshot = source.latestSnapshot();
|
if ("CABINET_CHANNEL".equals(source.sourceType())) {
|
if (!receivedDuringTest(test, snapshot.get("deviceTime"), snapshot.get("observedAt"))) return;
|
String identity = offlineIdentity(test, source);
|
for (String[] field : List.of(new String[] { "voltageV", "VOLTAGE", "V" },
|
new String[] { "currentA", "CURRENT", "A" }, new String[] { "capacityAh", "CAPACITY", "Ah" },
|
new String[] { "energyWh", "ENERGY", "Wh" }))
|
appendThresholdValue(test, source.sourceType(), identity, field[1], snapshot.get(field[0]), field[2], warnings, seen);
|
}
|
else if (snapshot.get("measurements") instanceof List<?> measurements) {
|
for (Object item : measurements) if (item instanceof Map<?, ?> measurement
|
&& receivedDuringTest(test, measurement.get("deviceTime"), measurement.get("serverReceivedAt"))) {
|
appendThresholdValue(test, source.sourceType(), source.sourceId() + ":" + measurement.get("channelCode"),
|
String.valueOf(measurement.get("metricCode")), measurement.get("value"),
|
String.valueOf(measurement.get("unit")), warnings, seen);
|
}
|
}
|
}
|
|
private boolean receivedDuringTest(TTest test, Object deviceTime, Object observedAt)
|
{
|
return TestDataReceptionServiceImpl.accepted(test,
|
deviceTime instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null,
|
observedAt instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null);
|
}
|
|
private void appendThresholdValue(TTest test, String sourceType, String identity, String metricCode,
|
Object value, String unit, List<Map<String, Object>> warnings, java.util.Set<String> seen)
|
{
|
if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())) return;
|
BigDecimal measurement = value instanceof BigDecimal decimal ? decimal : new BigDecimal(value.toString());
|
for (var rule : TestWarningExceptionService.exceededRules(test, metricCode, measurement)) {
|
if (!seen.add(rule.getThresholdId() + ":" + sourceType + ":" + identity + ":" + metricCode)) continue;
|
Map<String, Object> warning = new LinkedHashMap<>();
|
warning.put("exceptionType", "THRESHOLD"); warning.put("thresholdId", rule.getThresholdId());
|
warning.put("sourceType", sourceType); warning.put("sourceIdentity", identity);
|
warning.put("metricCode", metricCode); warning.put("value", measurement); warning.put("unit", unit);
|
warning.put("minValue", rule.getMinValue()); warning.put("maxValue", rule.getMaxValue());
|
warning.put("status", "OPEN"); warning.put("message", "当前测量超出预警阈值");
|
warnings.add(warning);
|
}
|
}
|
|
private String offlineIdentity(TTest test, LiveSource source)
|
{
|
if ("COLLECTOR".equals(source.sourceType())) return String.valueOf(source.sourceId());
|
for (TTestBattery battery : test.getBatteries()) {
|
var binding = battery.getBinding();
|
if (binding != null && Objects.equals(binding.getChannelId(), source.sourceId()))
|
return binding.getAdapterCode() + ":" + binding.getDevtype() + ":" + binding.getDevid()
|
+ ":" + binding.getSubdevid() + ":" + binding.getChlid();
|
}
|
return String.valueOf(source.sourceId());
|
}
|
|
private TestPlanSnapshot detectorTestPlan(TTest test)
|
{
|
if (adapter == null) return null;
|
if (test.getBatteries() == null) return null;
|
for (TTestBattery battery : test.getBatteries())
|
{
|
if (battery.getBinding() == null || battery.getBinding().getDevid() == null
|
|| battery.getBinding().getChlid() == null) continue;
|
try
|
{
|
var binding = battery.getBinding();
|
var source = adapter.testPlan(new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(),
|
binding.getDevid(), binding.getSubdevid(), binding.getChlid()));
|
if (source == null) continue;
|
List<TestPlanRow> rows = source.rows().stream()
|
.map(row -> new TestPlanRow(row.rowType(), row.stepNo(), row.cyclePosition(), row.stepName(), row.durationSeconds(),
|
row.voltageV(), row.executionStatus(), row.cycleCount(), row.recordCount()))
|
.toList();
|
return new TestPlanSnapshot(source.currentCycleNo(), source.currentStepNo(),
|
source.plannedCycleCount(), rows);
|
}
|
catch (IllegalArgumentException ignored) { /* 来源未提供方案时保持为空。 */ }
|
}
|
return null;
|
}
|
|
private LiveSource snapshotFor(LiveSource fresh, boolean receiving)
|
{
|
// 暂停或停止只停止数据请求；设备状态仍以本次查询的最新事实为准，采集快照不继续展示。
|
if (!receiving)
|
return withoutSnapshot(fresh);
|
return fresh;
|
}
|
|
private LiveSource withoutSnapshot(LiveSource source)
|
{
|
return new LiveSource(source.sourceType(), source.sourceId(), source.batteryNo(), source.deviceStatus(), Map.of());
|
}
|
|
private LiveSource cabinetSource(TTestBattery battery, List<CabinetCandidate> cabinets)
|
{
|
var binding = battery.getBinding();
|
for (CabinetCandidate cabinet : cabinets) for (ChannelCandidate channel : cabinet.channels())
|
if (Objects.equals(channel.identity().adapterCode(), binding.getAdapterCode())
|
&& Objects.equals(channel.identity().devtype(), binding.getDevtype())
|
&& Objects.equals(channel.identity().devid(), binding.getDevid())
|
&& Objects.equals(channel.identity().subdevid(), binding.getSubdevid())
|
&& Objects.equals(channel.identity().chlid(), binding.getChlid()))
|
return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(),
|
channel.online() ? "ONLINE" : "OFFLINE",
|
channel.online() ? snapshot(channel) : Map.of());
|
return new LiveSource("CABINET_CHANNEL", binding.getChannelId(), battery.getBatteryNo(), "OFFLINE", Map.of());
|
}
|
|
private Map<String, Object> snapshot(ChannelCandidate channel)
|
{
|
Map<String, Object> value = new LinkedHashMap<>();
|
if (channel.latestSnapshot() == null) return value;
|
value.put("source", "检测柜");
|
value.put("observedAt", channel.latestSnapshot().receivedAt().toEpochMilli());
|
value.put("deviceTime", channel.latestSnapshot().deviceTime() == null ? null : channel.latestSnapshot().deviceTime().toEpochMilli());
|
// BTS 核心采样字段是快照的一等事实，不能只依赖动态 measurements 展示。
|
value.put("currentA", channel.latestSnapshot().currentA());
|
value.put("voltageV", channel.latestSnapshot().voltageV());
|
value.put("capacityAh", channel.latestSnapshot().capacityAh());
|
value.put("energyWh", channel.latestSnapshot().energyWh());
|
// 实时方案需要设备当前循环和实际工步；这些字段只用于快照展示，不形成历史采集事实。
|
value.put("cycleNo", channel.latestSnapshot().cycleNo());
|
value.put("stepNo", channel.latestSnapshot().stepNo());
|
value.put("stepType", channel.latestSnapshot().stepType());
|
value.put("totalTimeSeconds", channel.latestSnapshot().totalTimeSeconds());
|
value.put("measurements", channel.latestSnapshot().measurements().entrySet().stream().map(entry -> {
|
Measurement measurement = entry.getValue();
|
return Map.of("metricCode", entry.getKey(), "value", measurement.value(), "unit", measurement.unit());
|
}).toList());
|
List<String> units = new ArrayList<>();
|
if (channel.latestSnapshot().currentA() != null) units.add("A");
|
if (channel.latestSnapshot().voltageV() != null) units.add("V");
|
if (channel.latestSnapshot().capacityAh() != null) units.add("Ah");
|
if (channel.latestSnapshot().energyWh() != null) units.add("Wh");
|
units.addAll(channel.latestSnapshot().measurements().values().stream().map(Measurement::unit).toList());
|
value.put("units", units.stream().filter(item -> item != null && !item.isBlank()).distinct().toList());
|
return value;
|
}
|
|
private LiveSource collectorSource(TTestBattery battery, boolean receiving)
|
{
|
if (adapter == null || mapper == null) return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
|
"OFFLINE", Map.of());
|
String collectorCode = mapper.selectCollectorCode(battery.getCollectorId());
|
// 非接收状态只读设备发现事实，不能先请求测量再隐藏，否则仍会产生新观测。
|
if (!receiving)
|
{
|
boolean online = candidates.findCollector(collectorCode, Instant.now())
|
.map(DeviceDiscoveryCandidateStore.CollectorCandidate::online).orElse(false);
|
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
|
online ? "ONLINE" : "OFFLINE", Map.of());
|
}
|
var collector = adapter.latestCollector(collectorCode);
|
if (collector != null)
|
{
|
Map<String, Object> value = new LinkedHashMap<>();
|
value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs());
|
// 在业务响应边界转换毫秒时间，设备时间与服务器接收时间分别保留。
|
value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null
|
&& battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode())
|
&& TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), item.metricCode()))).map(item -> {
|
Map<String, Object> measurement = new LinkedHashMap<>();
|
measurement.put("metricCode", item.metricCode());
|
measurement.put("channelCode", item.channelCode());
|
measurement.put("value", item.value());
|
measurement.put("unit", item.unit());
|
measurement.put("deviceTime", item.deviceTime().toEpochMilli());
|
measurement.put("serverReceivedAt", item.serverReceivedAt().toEpochMilli());
|
return measurement;
|
}).toList());
|
value.put("units", collector.measurements().stream().map(item -> item.unit()).distinct().toList());
|
collector.measurements().stream().map(item -> item.serverReceivedAt()).max(Instant::compareTo)
|
.ifPresent(observedAt -> value.put("observedAt", observedAt.toEpochMilli()));
|
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(),
|
collector.online() ? "ONLINE" : "OFFLINE", collector.online() ? value : Map.of());
|
}
|
return new LiveSource("COLLECTOR", battery.getCollectorId(), battery.getBatteryNo(), "OFFLINE", Map.of());
|
}
|
|
}
|
