实现新威 7.6 的工步与循环数据取数

yeg

2026-09-28 9c93934b70ccfca081102944a47f3baafad3c218

 battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapter.java

@@ -26,6 +26,7 @@
import javax.sql.DataSource;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.DeviceAdapterException;
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;
@@ -63,6 +64,8 @@
    public static final String DEFAULT_PASSWORD_ENV = "BATTERY_VENDOR_DB_PASSWORD";
    /** 厂商实测写库周期；它同时是持久化周期，展示节奏不随之加快。 */
    public static final long SAMPLE_INTERVAL_MS = 60000L;
    /** 工步层的共用读取窗口：足以覆盖一轮接收里工步与循环的连续调用，又不会压住新出现的工步行。 */
    private static final long CAP_STEP_CACHE_MS = 5000L;
    private static final int SAMPLE_BATCH_LIMIT = 500;
    private static final Pattern SAFE_TABLE_NAME = Pattern.compile("[A-Za-z0-9_]{1,64}");
    /** 设备时间按适配器时区解释；来源列是秒精度 datetime，小数部分按需接受。 */
@@ -81,9 +84,12 @@
    private final Runnable connectionReset;
    private final Clock clock;
    private final ZoneId deviceZone;
    /** 工步类型到方向的映射：正向聚合循环容量时需要它区分充、放、搁置与未知。 */
    private final StepDirectionResolver directions = new NewareStepDirectionResolver();
    /** 最近数据时间最快也要一分钟才可能变化，缓存使发现清单不必每轮重读全部通道。 */
    private final long lastDataCacheMillis;
    private final Map<String, CachedLastData> lastDataCache = new ConcurrentHashMap<>();
    private final Map<String, CachedCapSteps> capStepCache = new ConcurrentHashMap<>();

    public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource)
    {
@@ -177,18 +183,47 @@
        return null;
    }

    /** 工步层记录由本单之后的一张单实现，这里不返回从采样推导出来的工步。 */
    /** 工步层记录与循环统计都读设备的 {@code cap_step} 表；该表不随采样分表滚动。 */
    @Override
    public List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return List.of();
        CapStepSource source = capStepSource(channel);
        if (source == null) return List.of();
        List<StepRecord> records = new ArrayList<>();
        for (StepGroup group : completedStepGroups(source.rows())) records.add(stepRecord(source.testId(), group));
        return List.copyOf(records);
    }

    /** 循环汇总表不存在，聚合配方须先用真实数据核对，这里不返回推导值。 */
    /**
     * 循环统计由工步层按循环聚合：设备没有循环汇总行（`cap_plat` 的充放电容量列实测全为 0，
     * 库内也没有任何按循环汇总的表）。
     *
     * <p>配方（行语义见 {@link #completedStepGroups}）：
     * <ul>
     *   <li>每个已结束工步的累计容量与能量各计一次，重复接收不重复累计。</li>
     *   <li>按方向归入本循环的充电或放电一侧并<b>求和</b>：同一循环可能含多个同方向工步
     *       （该化成协议的前两个循环各有两个充电工步和两个放电工步），求和才是该循环充入/放出的总量。</li>
     *   <li>效率是放电容量除以充电容量；没有可用充电容量时留空，未知比率不写零。</li>
     *   <li>搁置与未知方向的工步不进任何一侧：容量无法归属到方向时，明细留在工步记录里供核对。</li>
     * </ul>
     *
     * <p>不插值、不补齐、不跨循环借值：聚合只用该循环自己的工步行，仍在进行的循环整体不出现。
     */
    @Override
    public List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return List.of();
        CapStepSource source = capStepSource(channel);
        if (source == null) return List.of();
        List<CapStepRow> rows = source.rows();
        Map<Integer, List<StepGroup>> byCycle = new LinkedHashMap<>();
        for (StepGroup group : completedStepGroups(rows))
            byCycle.computeIfAbsent(group.key().cycle(), ignored -> new ArrayList<>()).add(group);
        // 最后一行所属的循环仍在进行：它的工步行还没写完，聚合出来的是半个循环。
        int openCycle = rows.isEmpty() ? Integer.MIN_VALUE : rows.get(rows.size() - 1).cycle();
        List<CycleRecord> records = new ArrayList<>();
        for (Map.Entry<Integer, List<StepGroup>> entry : byCycle.entrySet())
            if (entry.getKey() != openCycle) records.add(cycleRecord(source.testId(), entry.getKey(), entry.getValue()));
        return List.copyOf(records);
    }

    /** 厂商固定 60 秒写库；持久化周期与之对齐，不做倍速。 */
@@ -328,6 +363,154 @@
        return result;
    }

    // ------------------------------------------------------------------ 工步层与循环聚合

    /** 读该通道最新测试在工步层的全部行；工步号与循环号都取自设备，不按采样分表推算。 */
    private CapStepSource capStepSource(ChannelIdentity channel)
    {
        ChannelKey key = channelKey(channel);
        String cacheKey = key.devUid() + ":" + key.unitId() + ":" + key.chlId();
        Instant now = clock.instant();
        CachedCapSteps cached = capStepCache.get(cacheKey);
        // 工步与循环在一次接收里被连续请求，同一次读取供两条路径共用，不重复查厂商库。
        if (cached != null && now.isBefore(cached.expiresAt())) return cached.source();
        VendorTest latest = latestTest(key);
        if (latest == null) return null;
        List<CapStepRow> rows = query("select seq_id, cycle, step_id, step_type, test_time, test_vol, test_cur, "
                + "test_capchg, test_engchg, test_capdchg, test_engdchg, middle_vol, test_atime from cap_step "
                + "where test_id = ? and dev_uid = ? and unit_id = ? and chl_id = ? order by seq_id",
                this::readCapStepRows, latest.testId(), key.devUid(), key.unitId(), key.chlId());
        CapStepSource source = new CapStepSource(latest.testId(), rows);
        capStepCache.put(cacheKey, new CachedCapSteps(source, now.plusMillis(CAP_STEP_CACHE_MS)));
        return source;
    }

    private List<CapStepRow> readCapStepRows(ResultSet rows) throws SQLException
    {
        List<CapStepRow> result = new ArrayList<>();
        while (rows.next())
        {
            String raw = rows.getString("test_atime");
            LocalDateTime deviceTime = parseDeviceTime(raw);
            result.add(new CapStepRow(rows.getLong("seq_id"), rows.getInt("cycle"), rows.getInt("step_id"),
                    rows.getString("step_type"), rows.getLong("test_time"), rows.getBigDecimal("test_vol"),
                    rows.getBigDecimal("test_cur"), rows.getBigDecimal("test_capchg"),
                    rows.getBigDecimal("test_engchg"), rows.getBigDecimal("test_capdchg"),
                    rows.getBigDecimal("test_engdchg"), rows.getBigDecimal("middle_vol"), raw,
                    deviceTime.atZone(deviceZone).toInstant()));
        }
        return result;
    }

    /**
     * 已结束的工步：按「循环号 + 工步号」分组，每组取设备写的最后一行作为该工步的累计事实。
     *
     * <p>实测行语义：同一工步的行数不是常量（多数为进入行加结束行的 2 行，另有 1 行与 3–10 行的少数协议），
     * 工步的累计容量与能量只出现在结束行上（进入行全为 0，结束行等于该工步的最大值），
     * 且它是**工步累计**而不是测试累计——同一工步在不同循环的值基本相同，不随测试推进累加。
     * 因此取最后一行，不按行求和、不取平均。
     *
     * <p>工步是否结束只按一条依据判断：设备是否已经写了<b>后续工步</b>的行。只有进入行的工步仍在进行中，
     * 不落库——否则它结束后的结束行会与已保存的行冲突。代价是设备停止后最后一个未闭合的工步不出现，
     * 这是"不把未确认的事实当事实"的落地形态，也保证暂停或继续不产生补算。
     */
    private List<StepGroup> completedStepGroups(List<CapStepRow> rows)
    {
        Map<StepKey, List<CapStepRow>> groups = new LinkedHashMap<>();
        for (CapStepRow row : rows)
            groups.computeIfAbsent(new StepKey(row.cycle(), row.stepId()), ignored -> new ArrayList<>()).add(row);
        List<StepGroup> completed = new ArrayList<>();
        int last = groups.size() - 1;
        int index = 0;
        for (Map.Entry<StepKey, List<CapStepRow>> entry : groups.entrySet())
            if (index++ < last) completed.add(new StepGroup(entry.getKey(), entry.getValue()));
        return completed;
    }

    /**
     * 一条工步记录。工步时间取设备给的 {@code test_time}（毫秒 → 秒），四组容量能量与中值电压
     * 放进扩展测量；单项容量只在方向已确认时给出。
     */
    private StepRecord stepRecord(long testId, StepGroup group)
    {
        CapStepRow first = group.rows().get(0);
        CapStepRow last = group.last();
        StepDirectionResolver.Direction direction = directions.resolve(last.stepType());
        Map<String, BigDecimal> measurements = new LinkedHashMap<>();
        measurements.put("stepDuration", BigDecimal.valueOf(last.testTimeMillis()).movePointLeft(3));
        measurements.put("startVoltage", scaledValue(first.voltage(), VOLTAGE_CURRENT_DIVISOR));
        // 中值电压按 ADR-0054 的口径改由设备工步层字段承载，不用峰值或谷值替代；设备给零就保留零。
        BigDecimal medianVoltage = scaledValue(last.middleVoltage(), VOLTAGE_CURRENT_DIVISOR);
        if (direction == StepDirectionResolver.Direction.CHARGE) measurements.put("chargeMedianVoltage", medianVoltage);
        if (direction == StepDirectionResolver.Direction.DISCHARGE) measurements.put("dischargeMedianVoltage", medianVoltage);
        // 四组容量能量原样保留设备事实：未知方向的工步也留着明细供核对。
        measurements.put("chargeCapacity", scaledValue(last.capacityCharge(), CAPACITY_ENERGY_DIVISOR));
        measurements.put("dischargeCapacity", scaledValue(last.capacityDischarge(), CAPACITY_ENERGY_DIVISOR));
        measurements.put("chargeEnergy", scaledValue(last.energyCharge(), CAPACITY_ENERGY_DIVISOR));
        measurements.put("dischargeEnergy", scaledValue(last.energyDischarge(), CAPACITY_ENERGY_DIVISOR));
        // 单项容量与能量只在方向已确认时给值：未知方向留空，不按数值大小猜它属于哪一组。
        BigDecimal capacity = null, energy = null;
        if (direction == StepDirectionResolver.Direction.CHARGE)
        {
            capacity = last.capacityCharge(); energy = last.energyCharge();
        }
        else if (direction == StepDirectionResolver.Direction.DISCHARGE)
        {
            capacity = last.capacityDischarge(); energy = last.energyDischarge();
        }
        else if (direction == StepDirectionResolver.Direction.REST)
        {
            // 搁置工步设备两向都为 0，取充电侧的 0 承载，使单项列与充放电工步同列可比。
            capacity = last.capacityCharge(); energy = last.energyCharge();
        }
        // 服务器接收时间取设备写结束行的时刻：接收边界据它在暂停与继续之间留白，不把暂停期的工步补回来。
        Double capacityAh = capacity == null ? null : scaled(capacity, CAPACITY_ENERGY_DIVISOR);
        Double energyWh = energy == null ? null : scaled(energy, CAPACITY_ENERGY_DIVISOR);
        return new StepRecord(sourceRecordId(testId, "step", group.key().cycle() + ":" + group.key().stepId()),
                group.key().cycle(), group.key().stepId(), null, last.stepType(), first.deviceTime(),
                last.deviceTime(), scaled(last.voltage(), VOLTAGE_CURRENT_DIVISOR),
                scaled(last.current(), VOLTAGE_CURRENT_DIVISOR), capacityAh, energyWh,
                last.deviceTimeRaw(), deviceZone.getId(), last.deviceTime(), measurements);
    }

    private CycleRecord cycleRecord(long testId, int cycleNo, List<StepGroup> groups)
    {
        BigDecimal chargeCapacity = BigDecimal.ZERO, dischargeCapacity = BigDecimal.ZERO;
        BigDecimal chargeEnergy = BigDecimal.ZERO, dischargeEnergy = BigDecimal.ZERO;
        for (StepGroup group : groups)
        {
            CapStepRow last = group.last();
            switch (directions.resolve(last.stepType()))
            {
                case CHARGE ->
                {
                    chargeCapacity = chargeCapacity.add(scaledValue(last.capacityCharge(), CAPACITY_ENERGY_DIVISOR));
                    chargeEnergy = chargeEnergy.add(scaledValue(last.energyCharge(), CAPACITY_ENERGY_DIVISOR));
                }
                case DISCHARGE ->
                {
                    dischargeCapacity = dischargeCapacity.add(scaledValue(last.capacityDischarge(), CAPACITY_ENERGY_DIVISOR));
                    dischargeEnergy = dischargeEnergy.add(scaledValue(last.energyDischarge(), CAPACITY_ENERGY_DIVISOR));
                }
                // 搁置与未知方向不进任何一侧：方向未确认时把容量算进某一侧就是凭数值猜方向。
                default -> { }
            }
        }
        CapStepRow last = groups.get(groups.size() - 1).last();
        return new CycleRecord(sourceRecordId(testId, "cycle", String.valueOf(cycleNo)), cycleNo,
                groups.get(0).rows().get(0).deviceTime(), last.deviceTime(), 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(), deviceZone.getId(), last.deviceTime(), CycleRecord.SOURCE_AGGREGATED);
    }

    /** 来源行身份：厂商测试编号 + 工步或循环标识；重复接收据此幂等。 */
    private static String sourceRecordId(long testId, String kind, String suffix)
    {
        return "neware:" + testId + ":" + kind + ":" + suffix;
    }

    private CabinetRealtime cabinetRealtime(SampleRow row)
    {
        // 充放电容量按电流方向取对应的一组：设备界面对正在放电的通道显示放电容量。
@@ -354,9 +537,14 @@

    private static double scaled(BigDecimal value, BigDecimal divisor)
    {
        if (value == null) return 0d;
        // 保留 12 位小数与业务列精度一致，避免在换算环节提前丢失设备原始分辨率。
        return value.divide(divisor, 12, RoundingMode.HALF_UP).doubleValue();
        return scaledValue(value, divisor).doubleValue();
    }

    /** 保留 12 位小数与业务列精度一致，避免在换算环节提前丢失设备原始分辨率。 */
    private static BigDecimal scaledValue(BigDecimal value, BigDecimal divisor)
    {
        if (value == null) return BigDecimal.ZERO;
        return value.divide(divisor, 12, RoundingMode.HALF_UP);
    }

    private ChannelIdentity identity(ChannelKey key)
@@ -459,6 +647,24 @@

    private record CachedLastData(Instant lastDataAt, Instant expiresAt) { }

    private record CachedCapSteps(CapStepSource source, Instant expiresAt) { }

    /** 该通道最新厂商测试在工步层的全部行；{@code testId} 同时用于来源行身份。 */
    private record CapStepSource(long testId, List<CapStepRow> rows) { }

    /** 工步号在同一个循环内唯一，所以「循环号 + 工步号」就是一条工步事实的身份。 */
    private record StepKey(int cycle, int stepId) { }

    private record StepGroup(StepKey key, List<CapStepRow> rows)
    {
        CapStepRow last() { return rows.get(rows.size() - 1); }
    }

    private record CapStepRow(long sequenceId, int cycle, int stepId, String stepType, long testTimeMillis,
            BigDecimal voltage, BigDecimal current, BigDecimal capacityCharge, BigDecimal energyCharge,
            BigDecimal capacityDischarge, BigDecimal energyDischarge, BigDecimal middleVoltage,
            String deviceTimeRaw, Instant deviceTime) { }

    private record SampleRow(long testId, long sequenceId, int cycle, int stepId, String stepType,
            Long testTimeMillis, BigDecimal voltage, BigDecimal current, BigDecimal capacityCharge,
            BigDecimal energyCharge, BigDecimal capacityDischarge, BigDecimal energyDischarge,