新威直连补上工步层与循环汇总（downloadStepLayer 实机验证后落地）

yeg

4 天以前 6b99cc391064cb38c706979fafcecb9f50ea1587

 battery-device-bts/src/main/java/com/battery/device/bts/NewareBtsAdapter.java

@@ -1,17 +1,21 @@
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;

/**
@@ -26,13 +30,12 @@
 *   <li>{@code inquiredf} → 当前测试号与已落盘行数，后者是逐点接收的增量水位；</li>
 *   <li>{@code download} → 逐点行。它给全精度测量值**和绝对时间**，所以测量值与时间轴取它，
 *       而不取 {@code inquire}（后者是 4 位小数、且没有绝对时间）。</li>
 *   <li>{@code downloadStepLayer} → 工步层；循环汇总由这些工步按循环聚合（见 {@link #cycleRecords}）。</li>
 * </ul>
 *
 * <p><b>不交出的三类事实</b>（都不猜、不用空值冒充已知）：
 * <p><b>不交出的两类事实</b>（都不猜、不用空值冒充已知）：
 *
 * <ul>
 *   <li><b>工步层与循环汇总</b>：协议里读工步层要 {@code downloadStepLayer}，该命令**未验证**，
 *       所以这边为空；它们属于该机厂商库那条路（ADR-0068 §5、未决 1）。</li>
 *   <li><b>辅助通道 / 采集器测量</b>：ADR-0068 已复核不可得（该机 {@code aux_data_*} 全空、
 *       API 的 {@code auxtemp}/{@code auxvol} 为 0），所以 {@link #latestCollector} 返回空值。</li>
 *   <li><b>计划方案</b>：协议里没有读方案的命令（现场实测 {@code inquire_resp} 不带计划），
@@ -49,12 +52,16 @@
    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<>();

    public NewareBtsAdapter(String adapterCode, BtsTransport transport, BtsXmlCodec codec)
    {
@@ -127,18 +134,33 @@
        return null;
    }

    /** 本路线的工步层需要未验证的 {@code downloadStepLayer}，因此不交出工步统计。 */
    /**
     * 返回该通道**已结束**的工步。来源是 {@code downloadStepLayer}（现场实测可用）。
     *
     * <p>这条命令**会把正在进行的那一步也交出来**，而且它的 {@code endseqid} 恰好等于
     * {@code inquiredf} 报的已落盘行数（实测四个测试都是相等）——所以**行数比不出结束没有**，
     * 判据是通道的工作状态：仍在运行、暂停或保护中的测试，最后一步没有结束。
     * 交出未结束的段会让它的结束时间每轮都在变，同一条记录反复以不同内容到达，既留下一条错的结束时间，
     * 又不断触发"内容冲突"告警，所以这里把它连同它所在的循环一起按住，等它结束再交出。
     */
    @Override
    public List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return List.of();
        return stepLayer(channel).steps();
    }

    /** 本路线的循环汇总同上：不交出，也不用实时行推导来源没有的汇总。 */
    /**
     * 返回该通道的循环汇总，由已结束的工步按循环聚合（{@link CycleRecord#SOURCE_AGGREGATED}）。
     *
     * <p>接口没有循环汇总行，聚合口径与库路线一致：只把方向已确认的工步容量能量加进对应一侧，
     * 搁置与未知方向不进任何一侧（按数值猜方向就是编事实）；效率 = 放电容量 / 充电容量 × 100，
     * 没有可用充电容量时留空。含未结束工步的那个循环整轮不交出——那几项合计还在长。
     */
    @Override
    public List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return List.of();
        StepLayer layer = stepLayer(channel);
        return aggregateCycles(layer);
    }

    /** 返回厂商采样周期作为实时持久化槽位间隔。 */
@@ -214,6 +236,111 @@
        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();
        Instant receivedAt = Instant.now();
        List<StepRecord> layers = codec.parseStepLayer(
                transport.exchange(codec.downloadStepLayerRequest(channel)), receivedAt);
        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)
    {
        return codec.parseDataFileState(transport.exchange(codec.inquiredfRequest(channel)));