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<>();

    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)
        {
            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();
        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)));
    }

    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());
    }

    /** 行的序号比水位大就是要的行；序号反而更小说明换了测试，此时按设备时间判定。 */
    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();
    }
}