package com.battery.device.bts;
|
|
import java.math.BigDecimal;
|
import java.math.RoundingMode;
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.concurrent.ConcurrentHashMap;
|
import com.battery.device.api.DeviceAdapter;
|
import com.battery.device.api.StepDirectionResolver;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.device.api.model.AdapterCapabilities;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.model.DetectedChannel;
|
import com.battery.device.api.model.DeviceWorkStatus;
|
import com.battery.device.api.model.RealtimeSnapshot;
|
|
/**
|
* 新威 BTS 对外接口的只读适配器：走远程命名管道，一台上位机一个实例（ADR-0068）。
|
*
|
* <p>它同时实现观测契约与测试数据观测契约，使这条路线直接接上既有的接收与展示路径，
|
* 不需要在业务层为新威另开一条路。各命令的分工是现场实测定下来的：
|
*
|
* <ul>
|
* <li>{@code getdevinfo} → 通道登记事实（发现）；</li>
|
* <li>{@code inquire} → 实时值与**通道工作状态**（读库路线拿不到工作状态，这是本路线的主要收获）；</li>
|
* <li>{@code inquiredf} → 当前测试号与已落盘行数，后者是逐点接收的增量水位；</li>
|
* <li>{@code download} → 逐点行。它给全精度测量值**和绝对时间**，所以测量值与时间轴取它，
|
* 而不取 {@code inquire}（后者是 4 位小数、且没有绝对时间）。</li>
|
* <li>{@code downloadStepLayer} → 工步层；循环汇总由这些工步按循环聚合（见 {@link #cycleRecords}）。</li>
|
* </ul>
|
*
|
* <p><b>不交出的两类事实</b>（都不猜、不用空值冒充已知）：
|
*
|
* <ul>
|
* <li><b>辅助通道 / 采集器测量</b>：ADR-0068 已复核不可得（该机 {@code aux_data_*} 全空、
|
* API 的 {@code auxtemp}/{@code auxvol} 为 0），所以 {@link #latestCollector} 返回空值。</li>
|
* <li><b>计划方案</b>：协议里没有读方案的命令（现场实测 {@code inquire_resp} 不带计划），
|
* 方案来自方案文件与厂商库表，因此这里不提供。</li>
|
* </ul>
|
*
|
* <p><b>空闲通道的读数不进本适配器的持久化路径</b>：实测空闲通道的 {@code inquire} 电压是悬空负值
|
* （≈ −2.4 / −3.4 V），而该通道库内从无负值。逐点行（{@code download}）只在测试进行中有意义，
|
* 所以按行接收天然只覆盖测试期；实时值本身仍可用于展示。
|
*/
|
public final class NewareBtsAdapter implements DeviceAdapter, TestDataObservationAdapter
|
{
|
/** 该存储形态的适配器类型标识，登记实例时用它匹配本工厂。 */
|
public static final String ADAPTER_TYPE = "NEWARE_BTS";
|
/** 厂商采样周期实测为 60 秒；实时持久化槽位按它对齐，取数不必更快。 */
|
public static final long SAMPLE_INTERVAL_MS = 60_000L;
|
/** 工步层读取结果的复用窗口：只为让工步与循环来自同一次读取，不做缓存加速。 */
|
private static final long STEP_LAYER_CACHE_NANOS = 5_000_000_000L;
|
|
private final String adapterCode;
|
private final BtsTransport transport;
|
private final BtsXmlCodec codec;
|
private final StepDirectionResolver directions = new BtsStepDirectionResolver();
|
private final AdapterCapabilities capabilities;
|
/** 每通道最近一次工步层读取；见 {@link #stepLayer}。 */
|
private final Map<String, StepLayer> stepLayers = new ConcurrentHashMap<>();
|
/**
|
* 该客户端有没有数据文件命令（{@code inquiredf} / {@code download} / {@code downloadStepLayer}）。
|
* 未知为 null；探测一次后固定下来——见 {@link #supportsDataFileCommands}。
|
*/
|
private volatile Boolean dataFileCommandsSupported;
|
|
public NewareBtsAdapter(String adapterCode, BtsTransport transport, BtsXmlCodec codec)
|
{
|
this(adapterCode, transport, codec, new AdapterCapabilities(true, true, true, false));
|
}
|
|
public NewareBtsAdapter(String adapterCode, BtsTransport transport, BtsXmlCodec codec,
|
AdapterCapabilities capabilities)
|
{
|
this.adapterCode = adapterCode;
|
this.transport = transport;
|
this.codec = codec;
|
this.capabilities = capabilities;
|
}
|
|
/** 返回本实例登记的适配器编码。 */
|
@Override
|
public String adapterCode()
|
{
|
return adapterCode;
|
}
|
|
/** 返回本实例登记的能力；本路线可发现、可观测实时值、可下载历史逐点，且始终不控制。 */
|
@Override
|
public AdapterCapabilities capabilities()
|
{
|
return capabilities;
|
}
|
|
/** 通过 {@code getdevinfo} 读取通道登记事实，作为发现候选返回。 */
|
@Override
|
public List<DetectedChannel> discoverChannels()
|
{
|
return codec.parseDeviceInfo(transport.exchange(codec.getDeviceInfoRequest()));
|
}
|
|
/** 按给定通道集合请求一次实时状态；通道为空时不发请求。 */
|
@Override
|
public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
|
{
|
if (channels == null || channels.isEmpty()) return Map.of();
|
// 带上请求集合：实测服务端会改写应答里的 devtype，按请求的身份归集才拿得到自己要的通道。
|
return codec.parseSnapshots(transport.exchange(codec.inquireRequest(channels, false)), channels,
|
Instant.now());
|
}
|
|
/**
|
* 返回该通道最近一次实时观测。
|
*
|
* <p>{@code inquire} 不带绝对时间与协议序号，所以这里的时间口径是**服务器接收时间**，
|
* 序号为 0；需要设备时间与真实序号的场合取 {@link #samplesSince}。
|
*/
|
@Override
|
public CabinetRealtime latestCabinet(ChannelIdentity channel)
|
{
|
RealtimeSnapshot snapshot = inquire(Set.of(channel)).get(channel);
|
if (snapshot == null) return null;
|
return new CabinetRealtime(0L, snapshot.deviceTime(), value(snapshot.currentA()), value(snapshot.voltageV()),
|
value(snapshot.capacityAh()), value(snapshot.energyWh()),
|
snapshot.cycleNo() == null ? 0 : snapshot.cycleNo(),
|
snapshot.stepNo() == null ? 0 : snapshot.stepNo(),
|
directions.resolve(snapshot.stepType()).displayName(), snapshot.stepType(), snapshot.receivedAt(),
|
null, codec.timezoneId(), snapshot.relativeTimeSeconds(), null);
|
}
|
|
/** 本路线的辅助测量不可得（ADR-0068 已复核），返回空值而不是零值。 */
|
@Override
|
public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
|
{
|
return null;
|
}
|
|
/**
|
* 返回该通道**已结束**的工步。来源是 {@code downloadStepLayer}（现场实测可用）。
|
*
|
* <p>这条命令**会把正在进行的那一步也交出来**，而且它的 {@code endseqid} 恰好等于
|
* {@code inquiredf} 报的已落盘行数（实测四个测试都是相等）——所以**行数比不出结束没有**，
|
* 判据是通道的工作状态：仍在运行、暂停或保护中的测试，最后一步没有结束。
|
* 交出未结束的段会让它的结束时间每轮都在变，同一条记录反复以不同内容到达，既留下一条错的结束时间，
|
* 又不断触发"内容冲突"告警，所以这里把它连同它所在的循环一起按住，等它结束再交出。
|
*/
|
@Override
|
public List<StepRecord> stepRecords(ChannelIdentity channel)
|
{
|
return stepLayer(channel).steps();
|
}
|
|
/**
|
* 返回该通道的循环汇总，由已结束的工步按循环聚合（{@link CycleRecord#SOURCE_AGGREGATED}）。
|
*
|
* <p>接口没有循环汇总行，聚合口径与库路线一致：只把方向已确认的工步容量能量加进对应一侧，
|
* 搁置与未知方向不进任何一侧（按数值猜方向就是编事实）；效率 = 放电容量 / 充电容量 × 100，
|
* 没有可用充电容量时留空。含未结束工步的那个循环整轮不交出——那几项合计还在长。
|
*/
|
@Override
|
public List<CycleRecord> cycleRecords(ChannelIdentity channel)
|
{
|
StepLayer layer = stepLayer(channel);
|
return aggregateCycles(layer);
|
}
|
|
/** 返回厂商采样周期作为实时持久化槽位间隔。 */
|
@Override
|
public long persistenceIntervalMillis(ChannelIdentity channel)
|
{
|
return SAMPLE_INTERVAL_MS;
|
}
|
|
/**
|
* 读取该通道当前的接收水位：{@code inquiredf} 给测试号与已落盘行数，
|
* 再用 {@code download} 取最后一行拿到它的设备时间。
|
*
|
* <p>行数就是最后一行的 {@code seqid}（实测一致），所以水位是"行数那一行"。
|
* 该通道当前没有行时水位为 {@code (空时间, 0)}：契约规定空时间表示"这一刻来源里没有任何可用行，
|
* 之后出现的行全部属于本次接收"。
|
*
|
* <p>客户端不提供 {@code inquiredf} 时（实测 7.6.0.1475 无此命令）返回空值，
|
* 表示本来源不按设备数据行接收，调用方回落到轮询槽位——这不是失败，是该构建的真实能力。
|
*/
|
@Override
|
public SampleWatermark currentWatermark(ChannelIdentity channel)
|
{
|
BtsXmlCodec.DataFileState state;
|
try
|
{
|
state = dataFileState(channel);
|
}
|
catch (BtsProtocolException unsupported)
|
{
|
// 客户端不提供 inquiredf（实测 7.6.0.1475）：本来源不按设备数据行接收，逐点回落到轮询槽位。
|
return null;
|
}
|
catch (BtsTransportException unanswered)
|
{
|
// 探测说支持、发出去却不作答（已登记不再试）：同样回落到轮询槽位，不把来源不可达说成"不按行接收"。
|
return null;
|
}
|
if (state.count() < 1) return new SampleWatermark(null, 0L);
|
List<CabinetRealtime> tail = download(channel, state.testId(), state.count(), 1);
|
if (tail.isEmpty()) return new SampleWatermark(null, 0L);
|
CabinetRealtime last = tail.get(0);
|
return new SampleWatermark(last.deviceTime(), last.sequenceId());
|
}
|
|
/**
|
* 返回严格晚于水位的逐点行，按 {@code startpos} 分页取到已落盘行数为止。
|
*
|
* <p>{@code seqid} 在每个测试内从 1 重编：水位比现有行数还大说明该通道换了测试，
|
* 此时从第 1 行重读，并按设备时间把旧测试的行过掉（新测试的行时间必然晚于旧水位）。
|
*
|
* @param channel 检测柜通道身份
|
* @param watermark 上次取得的水位；为空表示尚未取水位，返回空列表而不并入设备历史
|
*/
|
@Override
|
public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
|
{
|
if (watermark == null) return List.of();
|
BtsXmlCodec.DataFileState state = dataFileState(channel);
|
if (state.count() < 1) return List.of();
|
long next = watermark.sequenceId() >= state.count() ? 1L : watermark.sequenceId() + 1L;
|
List<CabinetRealtime> rows = new ArrayList<>();
|
while (next <= state.count())
|
{
|
int page = (int) Math.min(BtsXmlCodec.MAX_DOWNLOAD_PAGE, state.count() - next + 1);
|
List<CabinetRealtime> fetched = download(channel, state.testId(), next, page);
|
if (fetched.isEmpty()) break;
|
rows.addAll(fetched);
|
if (fetched.size() < page) break;
|
next = fetched.get(fetched.size() - 1).sequenceId() + 1L;
|
}
|
return rows.stream().filter(row -> isAfter(watermark, row)).toList();
|
}
|
|
/** 关闭底层传输连接。 */
|
@Override
|
public void close()
|
{
|
transport.close();
|
}
|
|
/**
|
* 取该通道的工步层，并按"最后一步行不行"切出已结束的部分。
|
*
|
* <p>结果在 {@link #STEP_LAYER_CACHE_NANOS} 内复用：工步与循环必须来自**同一次读取**，
|
* 否则一个工步边界落在两次读取之间时，循环的合计会包含工步列表里没有的那一步，
|
* 使循环记录的内容每轮都不同（同一条循环反复以不同内容到达）。
|
*/
|
private StepLayer stepLayer(ChannelIdentity channel)
|
{
|
String key = channel.vendorKey();
|
StepLayer cached = stepLayers.get(key);
|
if (cached != null && System.nanoTime() - cached.fetchedAtNanos() < STEP_LAYER_CACHE_NANOS) return cached;
|
StepLayer fetched = fetchStepLayer(channel);
|
stepLayers.put(key, fetched);
|
return fetched;
|
}
|
|
private StepLayer fetchStepLayer(ChannelIdentity channel)
|
{
|
long fetchedAtNanos = System.nanoTime();
|
// 客户端没有这条命令时一条都不发：见 supportsDataFileCommands。
|
if (!supportsDataFileCommands(channel)) return new StepLayer(List.of(), null, fetchedAtNanos);
|
Instant receivedAt = Instant.now();
|
List<StepRecord> layers;
|
try
|
{
|
layers = codec.parseStepLayer(transport.exchange(codec.downloadStepLayerRequest(channel)), receivedAt);
|
}
|
catch (BtsTransportException unanswered)
|
{
|
rememberUnanswered(unanswered);
|
return new StepLayer(List.of(), null, fetchedAtNanos);
|
}
|
if (layers.isEmpty()) return new StepLayer(List.of(), null, fetchedAtNanos);
|
// 通道仍在动（运行、暂停、保护中）时最后一步没结束：它和它所在的循环都先按住。
|
if (!isStillRunning(channel)) return new StepLayer(layers, null, fetchedAtNanos);
|
StepRecord running = layers.get(layers.size() - 1);
|
return new StepLayer(layers.subList(0, layers.size() - 1), running.cycleNo(), fetchedAtNanos);
|
}
|
|
/**
|
* 通道工作状态是否说明测试仍在进行。
|
*
|
* <p>只有读到"测试进行中"的状态才按住最后一步；其余状态（含读不到该通道）按已结束处理。
|
* **读取失败不吞**：{@code inquire} 抛错时让它传出去，调用方本来就有"来源读不到就本轮放弃、
|
* 下一轮重试"的约定，这样一次传输抖动不会变成"把一个还在跑的工步交出去"。
|
*/
|
private boolean isStillRunning(ChannelIdentity channel)
|
{
|
RealtimeSnapshot snapshot = inquire(Set.of(channel)).get(channel);
|
if (snapshot == null) return false;
|
DeviceWorkStatus status = snapshot.workStatus();
|
return status == DeviceWorkStatus.WORKING || status == DeviceWorkStatus.PAUSED
|
|| status == DeviceWorkStatus.PROTECTED || status == DeviceWorkStatus.SYNCHRONIZING;
|
}
|
|
/** 把已结束的工步按循环聚合；含未结束工步的那个循环不交出。 */
|
private List<CycleRecord> aggregateCycles(StepLayer layer)
|
{
|
Map<Integer, List<StepRecord>> byCycle = new LinkedHashMap<>();
|
for (StepRecord step : layer.steps())
|
{
|
if (layer.unfinishedCycleNo() != null && layer.unfinishedCycleNo() == step.cycleNo()) continue;
|
byCycle.computeIfAbsent(step.cycleNo(), ignored -> new ArrayList<>()).add(step);
|
}
|
List<CycleRecord> cycles = new ArrayList<>(byCycle.size());
|
byCycle.forEach((cycleNo, group) -> {
|
BigDecimal chargeCapacity = BigDecimal.ZERO, dischargeCapacity = BigDecimal.ZERO;
|
BigDecimal chargeEnergy = BigDecimal.ZERO, dischargeEnergy = BigDecimal.ZERO;
|
for (StepRecord step : group)
|
{
|
if (step.capacityAh() == null || step.energyWh() == null) continue;
|
BigDecimal capacity = BigDecimal.valueOf(step.capacityAh());
|
BigDecimal energy = BigDecimal.valueOf(step.energyWh());
|
switch (directions.resolve(step.stepType()))
|
{
|
case CHARGE ->
|
{
|
chargeCapacity = chargeCapacity.add(capacity);
|
chargeEnergy = chargeEnergy.add(energy);
|
}
|
case DISCHARGE ->
|
{
|
dischargeCapacity = dischargeCapacity.add(capacity);
|
dischargeEnergy = dischargeEnergy.add(energy);
|
}
|
// 搁置与未知方向不进任何一侧：把容量算进某一侧就是凭数值猜方向。
|
default -> { }
|
}
|
}
|
StepRecord first = group.get(0), last = group.get(group.size() - 1);
|
cycles.add(new CycleRecord("neware:" + testIdOf(first) + ":cycle:" + cycleNo, cycleNo,
|
first.startedAt(), last.endedAt(), chargeCapacity.doubleValue(), dischargeCapacity.doubleValue(),
|
chargeEnergy.doubleValue(), dischargeEnergy.doubleValue(),
|
chargeCapacity.signum() == 0 ? null : dischargeCapacity.multiply(BigDecimal.valueOf(100))
|
.divide(chargeCapacity, 4, RoundingMode.HALF_UP).doubleValue(),
|
last.deviceTimeRaw(), codec.timezoneId(), last.serverReceivedAt(), CycleRecord.SOURCE_AGGREGATED));
|
});
|
return List.copyOf(cycles);
|
}
|
|
/** 从工步的来源行身份里取回测试号（形如 {@code neware:<测试号>:step:<序号>}）。 */
|
private static String testIdOf(StepRecord step)
|
{
|
String identity = step.sourceRecordId();
|
if (identity == null) return "unknown";
|
String[] parts = identity.split(":");
|
return parts.length >= 2 ? parts[1] : "unknown";
|
}
|
|
/**
|
* 一次工步层读取的结果：已结束的工步、被按住的未结束工步所属循环（没有则为空）。
|
*/
|
private record StepLayer(List<StepRecord> steps, Integer unfinishedCycleNo, long fetchedAtNanos) { }
|
|
private BtsXmlCodec.DataFileState dataFileState(ChannelIdentity channel)
|
{
|
if (!supportsDataFileCommands(channel))
|
throw new BtsProtocolException(BtsProtocolException.Reason.MISSING_FIELD,
|
"该客户端不提供 inquiredf");
|
try
|
{
|
return codec.parseDataFileState(transport.exchange(codec.inquiredfRequest(channel)));
|
}
|
catch (BtsTransportException unanswered)
|
{
|
rememberUnanswered(unanswered);
|
throw unanswered;
|
}
|
}
|
|
private List<CabinetRealtime> download(ChannelIdentity channel, long testId, long startPosition, int count)
|
{
|
return codec.parseDownload(transport.exchange(codec.downloadRequest(channel, testId, startPosition, count)),
|
Instant.now());
|
}
|
|
/**
|
* 这台客户端有没有数据文件命令（{@code inquiredf}/{@code download}/{@code downloadStepLayer}）。
|
*
|
* <p>为什么必须先问一句再发：**对不支持这些命令的客户端发出去，代价不只是超时**。2026-10-12 实测
|
* （客户端 7.6.0.1475）：发一条 {@code inquiredf} 之后，该接口的 {@code inquire} 连续 8 分钟无应答
|
* （{@code getdevinfo} 仍正常）——那等于干扰了那台上位机的对外接口。所以不支持的客户端
|
* **一条都不发**，逐点回落到轮询槽位、工步层为空。
|
*
|
* <p>探测依据是 {@code inquire} 应答的字段：7.6 的应答不带 {@code cycle_id}，8.0 的带；
|
* 实测三个构建（7.6.0.1475 / 8.0.0.484 / 8.0.0.530）里"有没有 {@code cycle_id}"与
|
* "有没有这三条数据文件命令"同向。探测本身读不到应答时**抛出**（不吞）：那说明来源不可达，
|
* 按契约要抛异常，不能被当成"该来源不按数据行接收"。
|
*/
|
private boolean supportsDataFileCommands(ChannelIdentity channel)
|
{
|
Boolean known = dataFileCommandsSupported;
|
if (known != null) return known;
|
RealtimeSnapshot snapshot = inquire(Set.of(channel)).get(channel);
|
boolean supported = snapshot != null && snapshot.cycleNo() != null;
|
dataFileCommandsSupported = supported;
|
return supported;
|
}
|
|
/** 探测说支持、真发出去却不作答：记下来不再反复试——反复发一条无应答的命令会持续干扰那台接口。 */
|
private void rememberUnanswered(BtsTransportException failure)
|
{
|
if (failure.reason() == BtsTransportException.Reason.TIMEOUT) dataFileCommandsSupported = false;
|
}
|
|
/** 行的序号比水位大就是要的行；序号反而更小说明换了测试，此时按设备时间判定。 */
|
private static boolean isAfter(SampleWatermark watermark, CabinetRealtime row)
|
{
|
if (row.sequenceId() > watermark.sequenceId()) return true;
|
return watermark.deviceTime() != null && row.deviceTime() != null
|
&& row.deviceTime().isAfter(watermark.deviceTime());
|
}
|
|
private static double value(BigDecimal source)
|
{
|
return source == null ? 0d : source.doubleValue();
|
}
|
}
|
