From 99935242ecccd94e76978e2b71cbc46b87d31dc6 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期五, 09 十月 2026 11:17:44 +0800
Subject: [PATCH] 推式来源的工步与循环改为从采样段推导
---
battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml | 9 +
battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java | 21 ++
battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java | 4
battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java | 10 +
battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java | 11 +
battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java | 136 ++++++++++++++
battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java | 264 ++++++++++++++++++++++++++++
battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java | 16 +
ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java | 27 +++
9 files changed, 484 insertions(+), 14 deletions(-)
diff --git a/battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java b/battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java
index c4c6745..5e251f0 100644
--- a/battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java
+++ b/battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java
@@ -41,6 +41,16 @@
@Param("devtype") int devtype, @Param("devid") int devid, @Param("subdevid") int subdevid,
@Param("chlid") int chlid);
+ /**
+ * 某个来源库（厂商侧一次测试）在该通道上的全部采样行，按 (设备时间, 协议序号) 升序。
+ *
+ * <p>推式来源没有设备侧工步层，工步与循环由这些采样行按「来源库 + 循环号 + 工步号」分段推导
+ * （ADR-0067）；读取范围限于一个来源库，也就是一次厂商测试，不随落地表无限增长。
+ */
+ List<TDeviceIngestRecord> selectSamplesOfSource(@Param("adapterCode") String adapterCode,
+ @Param("devtype") int devtype, @Param("devid") int devid, @Param("subdevid") int subdevid,
+ @Param("chlid") int chlid, @Param("sourceRef") String sourceRef);
+
/** 严格晚于 (设备时间, 协议序号) 的采样行，按二元组升序；水位为空表示不取数据。 */
List<TDeviceIngestRecord> selectSamplesSince(@Param("adapterCode") String adapterCode,
@Param("devtype") int devtype, @Param("devid") int devid, @Param("subdevid") int subdevid,
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java b/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
index 4dfdc59..30ff7d0 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
@@ -1,11 +1,13 @@
package com.battery.business.service.impl;
import java.math.BigDecimal;
+import java.math.RoundingMode;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -37,13 +39,15 @@
* 不开入站端口、不在网络上传递厂商库凭据；解释（身份映射、时间换算、幂等键取值）全部在后端完成，
* 所以这里只做投影：落地行的字段由接收端点解析好，本类不出现任何厂商列名。
*
- * <p>三条本形态特有的事实：
+ * <p>四条本形态特有的事实：
* <ul>
* <li><b>来源是否在线由心跳判定</b>：落地表在本地随时可读，因此"能否读到表"不能表示来源可达。
* 取数程序每轮推送一次心跳，心跳缺失即认为来源不可达——发现路径明确失败，
* 已登记设备随之离线、旧快照被撤下（既有路径），页面不会显示伪造的实时值。</li>
* <li><b>最近数据时间只看采样行</b>：没有新采样行就是没有，不按拓扑上报或心跳时间代替（ADR-0063）。</li>
* <li><b>通道工作状态不可观测</b>：落地表里没有设备侧状态，报"无法观测"而不由数据行推断。</li>
+ * <li><b>工步与循环从采样段推导</b>：这类来源的厂商库里没有工步层也没有循环汇总表，工步号与循环号
+ * 只在采样行上，所以按「来源库 + 循环号 + 工步号」分段推导（ADR-0067），推导是无状态重算。</li>
* </ul>
*/
public final class PushIngestAdapter implements DeviceAdapter, TestDataObservationAdapter
@@ -68,6 +72,8 @@
* 只取水位之后的新行，取数更勤不会产生重复记录。厂商写库节奏仍是 60 秒，值不会来得更快。
*/
private static final long PERSISTENCE_INTERVAL_MS = 5000L;
+ /** 厂商写库节奏：工步与循环的重算对齐它，来源的汇总产出不可能比这更快。 */
+ private static final long AGGREGATE_INTERVAL_MS = 60000L;
private static final int SAMPLE_BATCH_LIMIT = 500;
/** 心跳容忍窗口的下限：写库与推送都是 60 秒一轮，窗口至少覆盖三轮。 */
private static final Duration MINIMUM_STALE_AFTER = Duration.ofSeconds(90);
@@ -254,20 +260,37 @@
String unit) { }
/**
- * 22 所没有独立的工步层：工步与循环号只在采样行的 {@code Step_Index} / {@code Cycle_Index} 里，
- * 从采样推导工步是另一项决策（与"工步来自设备工步层"的既有口径冲突），不在本单范围，因此返回空。
+ * 交出该通道已结束的工步。来源没有设备侧工步层，所以按本来源自己的采样行分段推导（ADR-0067）：
+ * 按「来源库 + 循环号 + 工步号」分段，只有最新一段视为进行中、不交出，其余各段各一条工步记录。
*/
@Override
public List<StepRecord> stepRecords(ChannelIdentity channel)
{
- return List.of();
+ Derivation derivation = derive(channel);
+ List<StepRecord> records = new ArrayList<>();
+ for (StepGroup group : derivation.closedSteps()) records.add(stepRecord(derivation.sourceRef(), group));
+ return List.copyOf(records);
}
- /** 同上：循环统计不从采样推导，本单不交出循环记录。 */
+ /**
+ * 交出该通道已结束的循环：来源没有循环汇总行，循环由已结束的工步按方向求和得出（ADR-0066 同一配方）。
+ * 最新一段所属的循环仍在进行，整体不交出——半个循环不是事实。
+ */
@Override
public List<CycleRecord> cycleRecords(ChannelIdentity channel)
{
- return List.of();
+ Derivation derivation = derive(channel);
+ Map<Integer, List<StepGroup>> byCycle = new LinkedHashMap<>();
+ for (StepGroup group : derivation.closedSteps())
+ {
+ // 仍在进行的那一段所属的循环整体排除：它自己的工步行还没写完。
+ if (group.key().cycleNo() == derivation.openCycleNo()) continue;
+ byCycle.computeIfAbsent(group.key().cycleNo(), ignored -> new ArrayList<>()).add(group);
+ }
+ List<CycleRecord> records = new ArrayList<>();
+ for (Map.Entry<Integer, List<StepGroup>> entry : byCycle.entrySet())
+ records.add(cycleRecord(derivation.sourceRef(), entry.getKey(), entry.getValue()));
+ return List.copyOf(records);
}
/** 返回推式来源固定的持久化槽位间隔；采样行按来源自己的落库节奏到达。 */
@@ -275,6 +298,16 @@
public long persistenceIntervalMillis(ChannelIdentity channel)
{
return PERSISTENCE_INTERVAL_MS;
+ }
+
+ /**
+ * 工步与循环的重算跟着来源的写库节奏，不跟实时行的展示节奏：来源最快每 60 秒才产出一行新采样，
+ * 而工步要等下一段出现才算结束，因此重算更勤只会空跑落地表（ADR-0067）。
+ */
+ @Override
+ public long aggregateIntervalMillis(ChannelIdentity channel)
+ {
+ return AGGREGATE_INTERVAL_MS;
}
/** 水位是落地表里该通道已存在的最新一行：设备绝对时间与协议序号。 */
@@ -350,7 +383,8 @@
private CabinetRealtime cabinetRealtime(TDeviceIngestRecord row)
{
- // 充放电容量按电流方向取对应的一组：与直连来源同一口径（22 所的电流符号约定待现场对着上位机确认）。
+ // 充放电容量按电流方向取对应的一组：与直连来源同一口径。电流符号的含义已用该来源的真实推送
+ // 数据核对（恒正即充电、恒负即放电，ADR-0067），不再是待确认的约定。
BigDecimal current = row.getCurrentA() == null ? BigDecimal.ZERO : row.getCurrentA();
boolean charging = current.signum() >= 0;
BigDecimal capacity = charging ? row.getChargeCapacityAh() : row.getDischargeCapacityAh();
@@ -365,6 +399,222 @@
"push:" + row.getRecordKey());
}
+ // ------------------------------------------------------------------ 工步与循环的推导（ADR-0067）
+
+ /** 分段推导的结果：来源库、已结束的各段，以及仍在进行的那一段所属的循环号。 */
+ private record Derivation(String sourceRef, List<StepGroup> closedSteps, int openCycleNo) { }
+
+ /** 一段采样行：同一个来源库里同一个「循环号 + 工步号」的全部行，按 (设备时间, 协议序号) 升序。 */
+ private record StepGroup(StepKey key, List<TDeviceIngestRecord> rows)
+ {
+ TDeviceIngestRecord first() { return rows.get(0); }
+
+ TDeviceIngestRecord last() { return rows.get(rows.size() - 1); }
+ }
+
+ private record StepKey(int cycleNo, int stepNo) { }
+
+ /**
+ * 该段本身的方向。来源没有设备侧工步类型列，方向由该段自己的电流符号判定——依据是用该来源真实推送的
+ * 采样数据核对过的列语义（588 段零反例，见 ADR-0067），不是按数值大小猜方向。
+ *
+ * <p>段内电流正负混用、电流缺失，或恒为零却有容量增长时，方向不唯一：保持未知，明细照原样保留。
+ */
+ private enum StepDirection
+ {
+ CHARGE("charge"), DISCHARGE("discharge"), REST("rest"), UNKNOWN(null);
+
+ private final String label;
+
+ StepDirection(String label) { this.label = label; }
+
+ /** 写进工步记录的类型：来源没给类型时交出的是判定出的方向名称，未知方向留空。 */
+ String label() { return label; }
+
+ /** 该工步自身累计的单项值。搁置段实测两向都不增长，所以它的累计是 0 而不是缺失。 */
+ BigDecimal own(BigDecimal charge, BigDecimal discharge)
+ {
+ return switch (this)
+ {
+ case CHARGE -> charge;
+ case DISCHARGE -> discharge;
+ case REST -> BigDecimal.ZERO;
+ case UNKNOWN -> null;
+ };
+ }
+ }
+
+ /**
+ * 从该通道的采样行推导「已结束的段」。只看该通道**最新的来源库**：来源库就是厂商侧一次测试
+ * （22 所每次测试新建一个库），读取量因此限于一次厂商测试，分段也不会跨测试。
+ *
+ * <p>段是否结束只按一条依据判断：设备是否已经写下**后续工步**的行（与 ADR-0066 同一判据）。
+ * 所以只有最新一段视为进行中；它所属的循环整体不交——半个循环不是事实。
+ */
+ private Derivation derive(ChannelIdentity channel)
+ {
+ TDeviceIngestRecord latest = latestSample(channel);
+ if (latest == null || latest.getSourceRef() == null) return new Derivation(null, List.of(), Integer.MIN_VALUE);
+ List<TDeviceIngestRecord> rows = records.selectSamplesOfSource(config.adapterCode(), latest.getDevtype(),
+ latest.getDevid(), latest.getSubdevid(), latest.getChlid(), latest.getSourceRef());
+ Map<StepKey, List<TDeviceIngestRecord>> grouped = new LinkedHashMap<>();
+ for (TDeviceIngestRecord row : rows == null ? List.<TDeviceIngestRecord>of() : rows)
+ {
+ // 分段依据是设备给出的循环号与工步号；缺任何一个都无法归属到某一段，跳过并留痕。
+ if (row.getCycleNo() == null || row.getStepNo() == null)
+ {
+ log.warn("落地行缺少循环号或工步号，不参与工步推导 ingestId={}", row.getIngestId());
+ continue;
+ }
+ grouped.computeIfAbsent(new StepKey(row.getCycleNo(), row.getStepNo()), ignored -> new ArrayList<>())
+ .add(row);
+ }
+ List<StepGroup> closed = new ArrayList<>();
+ int openCycleNo = Integer.MIN_VALUE;
+ int index = 0;
+ for (Map.Entry<StepKey, List<TDeviceIngestRecord>> entry : grouped.entrySet())
+ {
+ if (++index == grouped.size())
+ {
+ openCycleNo = entry.getKey().cycleNo();
+ break;
+ }
+ closed.add(new StepGroup(entry.getKey(), entry.getValue()));
+ }
+ return new Derivation(latest.getSourceRef(), List.copyOf(closed), openCycleNo);
+ }
+
+ /**
+ * 一条工步记录：容量与能量是**工步累计**（实测段内从未下降，末行即最大值），所以取该段末行的值；
+ * 工步时间取末行的工步内相对时间（秒）。设备没给的项不写零、不伪造中值电压。
+ */
+ private StepRecord stepRecord(String sourceRef, StepGroup group)
+ {
+ TDeviceIngestRecord first = group.first();
+ TDeviceIngestRecord last = group.last();
+ StepDirection direction = direction(group);
+ Map<String, BigDecimal> measurements = new LinkedHashMap<>();
+ putIfPresent(measurements, "stepDuration", last.getRelativeTimeSeconds());
+ putIfPresent(measurements, "startVoltage", first.getVoltageV());
+ // 四组容量能量原样保留设备事实：方向未判定的段也留着明细供核对。
+ putIfPresent(measurements, "chargeCapacity", last.getChargeCapacityAh());
+ putIfPresent(measurements, "dischargeCapacity", last.getDischargeCapacityAh());
+ putIfPresent(measurements, "chargeEnergy", last.getChargeEnergyWh());
+ putIfPresent(measurements, "dischargeEnergy", last.getDischargeEnergyWh());
+ BigDecimal capacity = direction.own(last.getChargeCapacityAh(), last.getDischargeCapacityAh());
+ BigDecimal energy = direction.own(last.getChargeEnergyWh(), last.getDischargeEnergyWh());
+ return new StepRecord(sourceRecordId(sourceRef, "step", group.key().cycleNo() + ":" + group.key().stepNo()),
+ group.key().cycleNo(), group.key().stepNo(), null, direction.label(), deviceTime(first),
+ deviceTime(last), decimal(last.getVoltageV()), decimal(last.getCurrentA()),
+ capacity == null ? null : capacity.doubleValue(), energy == null ? null : energy.doubleValue(),
+ last.getDeviceTimeRaw(), last.getTimezoneId(), receivedAt(last), measurements);
+ }
+
+ /**
+ * 一条循环记录：该循环各已结束工步取自己的累计值一次，按方向归入充电或放电一侧并**求和**
+ * （同一循环可能含多个同方向工步）。效率 = 放电容量 ÷ 充电容量，没有可用充电容量时留空，不写零。
+ * 搁置段两向都不增长、未知方向不按哪一列有值去猜，两者都不进任何一侧。
+ */
+ private CycleRecord cycleRecord(String sourceRef, int cycleNo, List<StepGroup> groups)
+ {
+ BigDecimal chargeCapacity = BigDecimal.ZERO, dischargeCapacity = BigDecimal.ZERO;
+ BigDecimal chargeEnergy = BigDecimal.ZERO, dischargeEnergy = BigDecimal.ZERO;
+ for (StepGroup group : groups)
+ {
+ TDeviceIngestRecord last = group.last();
+ switch (direction(group))
+ {
+ case CHARGE ->
+ {
+ chargeCapacity = chargeCapacity.add(orZero(last.getChargeCapacityAh()));
+ chargeEnergy = chargeEnergy.add(orZero(last.getChargeEnergyWh()));
+ }
+ case DISCHARGE ->
+ {
+ dischargeCapacity = dischargeCapacity.add(orZero(last.getDischargeCapacityAh()));
+ dischargeEnergy = dischargeEnergy.add(orZero(last.getDischargeEnergyWh()));
+ }
+ default -> { }
+ }
+ }
+ TDeviceIngestRecord first = groups.get(0).first();
+ TDeviceIngestRecord last = groups.get(groups.size() - 1).last();
+ return new CycleRecord(sourceRecordId(sourceRef, "cycle", String.valueOf(cycleNo)), cycleNo,
+ deviceTime(first), deviceTime(last), 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.getDeviceTimeRaw(), last.getTimezoneId(), receivedAt(last), CycleRecord.SOURCE_AGGREGATED);
+ }
+
+ /** 段内电流符号一致时给出该方向；正负混用或电流缺失保持未知。 */
+ private static StepDirection direction(StepGroup group)
+ {
+ BigDecimal maximumCharge = group.first().getChargeCapacityAh();
+ BigDecimal maximumDischarge = group.first().getDischargeCapacityAh();
+ boolean positive = false, negative = false;
+ for (TDeviceIngestRecord row : group.rows())
+ {
+ BigDecimal current = row.getCurrentA();
+ // 电流缺失的段判不出方向：不拿同段其它行的符号代替。
+ if (current == null) return StepDirection.UNKNOWN;
+ positive |= current.signum() > 0;
+ negative |= current.signum() < 0;
+ maximumCharge = maximum(maximumCharge, row.getChargeCapacityAh());
+ maximumDischarge = maximum(maximumDischarge, row.getDischargeCapacityAh());
+ }
+ if (positive && negative) return StepDirection.UNKNOWN;
+ if (positive) return StepDirection.CHARGE;
+ if (negative) return StepDirection.DISCHARGE;
+ // 恒为零：搁置。但恒零却出现容量增长时方向未能判定——不按哪一列在涨去猜。
+ boolean accumulated = raised(group.first().getChargeCapacityAh(), maximumCharge)
+ || raised(group.first().getDischargeCapacityAh(), maximumDischarge);
+ return accumulated ? StepDirection.UNKNOWN : StepDirection.REST;
+ }
+
+ /** 来源行身份：来源库 + 工步或循环标识。同一次测试重算多少次都得到同一个身份，重复交出因此幂等。 */
+ private static String sourceRecordId(String sourceRef, String kind, String suffix)
+ {
+ return "push:" + sourceRef + ":" + kind + ":" + suffix;
+ }
+
+ /** 只放设备给了的项：缺列就是没有这一项，不写零代替缺失。 */
+ private static void putIfPresent(Map<String, BigDecimal> target, String key, BigDecimal value)
+ {
+ if (value != null) target.put(key, value);
+ }
+
+ /** 求和时把缺列当没有：缺少的项不参与求和，也不写零占位。 */
+ private static BigDecimal orZero(BigDecimal value)
+ {
+ return value == null ? BigDecimal.ZERO : value;
+ }
+
+ private static BigDecimal maximum(BigDecimal current, BigDecimal candidate)
+ {
+ if (candidate == null) return current;
+ return current == null ? candidate : current.max(candidate);
+ }
+
+ private static boolean raised(BigDecimal first, BigDecimal maximum)
+ {
+ return first != null && maximum != null && maximum.compareTo(first) > 0;
+ }
+
+ private static Instant deviceTime(TDeviceIngestRecord row)
+ {
+ return row.getDeviceTimeUtc() == null ? null : row.getDeviceTimeUtc().toInstant();
+ }
+
+ /**
+ * 接收时间取**我们收到这一行**的时刻，与推式来源的实时路径同一口径，也是接收边界判定的依据；
+ * 直连来源没有这个独立取值，它沿用设备记录时刻（见 KNOWN-GAPS）。
+ */
+ private static Instant receivedAt(TDeviceIngestRecord row)
+ {
+ return row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant();
+ }
+
private RealtimeSnapshot snapshot(ChannelIdentity channel, CabinetRealtime latest)
{
// 读落地表路线下通道工作状态不可观测（ADR-0063）：不按数据行推断，也不沿用上一次状态。
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java b/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java
index 99adf2a..6c1d998 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/RoutingTestDataObservationAdapter.java
@@ -150,6 +150,22 @@
}
/**
+ * 按通道所属来源实例取工步与循环的读取间隔。
+ *
+ * <p>必须在这里转交：不转交就会落到本类的默认实现，从而拿实时持久化间隔去重算汇总，
+ * 来源自己声明的汇总节奏就被忽略了。
+ *
+ * @param channel 检测柜通道身份
+ * @return 汇总记录的读取间隔，单位毫秒
+ * @throws AdapterRegistryException 该通道所属的实例未启用，或实例未提供测试数据观测契约
+ */
+ @Override
+ public long aggregateIntervalMillis(ChannelIdentity channel)
+ {
+ return registry.executeTestData(channel.adapterCode(), adapter -> adapter.aggregateIntervalMillis(channel));
+ }
+
+ /**
* 按通道所属来源实例取设备计划方案，只用于实时展示。
*
* @param channel 检测柜通道身份
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
index 5a77983..aa82ba3 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/TestDataReceptionServiceImpl.java
@@ -340,8 +340,9 @@
if (adapter == null) return; TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return;
String key = test.getTestId() + ":" + battery.getTestBatteryId();
// 接收调度每秒跑一次，而来源按自己的写库节奏产出工步：不按来源节奏限流会每秒重查厂商库。
+ // 汇总用来源声明的汇总间隔：它可能比实时行的间隔长（推式来源的落地表在本地）。
Instant now = clock.instant(); Instant previous = lastAggregateAt.get(key);
- if (previous != null && now.isBefore(previous.plusMillis(persistenceInterval(binding)))) return;
+ if (previous != null && now.isBefore(previous.plusMillis(aggregateInterval(binding)))) return;
lastAggregateAt.put(key, now);
// STEP 2: 读取来源的工步与循环汇总；来源不可达时本轮放弃，下一轮重试。
ChannelIdentity channel = new ChannelIdentity(binding.getAdapterCode(), binding.getDevtype(), binding.getDevid(), binding.getSubdevid(), binding.getChlid());
@@ -410,6 +411,7 @@
return String.join(":", parts);
}
private long persistenceInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.persistenceIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
+ private long aggregateInterval(TTestChannelBinding b) { try { return Math.max(100L, adapter.aggregateIntervalMillis(channelIdentity(b))); } catch (RuntimeException ex) { return 60000L; } }
private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); }
private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); }
private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; }
diff --git a/battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml b/battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml
index c305625..519340d 100644
--- a/battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml
+++ b/battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml
@@ -129,6 +129,15 @@
limit 1
</select>
+ <!-- 一个来源库（厂商侧一次测试）在该通道上的全部采样行，供分段推导工步与循环（ADR-0067）。 -->
+ <select id="selectSamplesOfSource" resultMap="TDeviceIngestRecordResultMap">
+ select <include refid="ingestColumns"/> from t_device_ingest_record
+ where adapter_code = #{adapterCode} and payload_kind = 'SAMPLE'
+ and devtype = #{devtype} and devid = #{devid} and subdevid = #{subdevid} and chlid = #{chlid}
+ and source_ref = #{sourceRef}
+ order by device_time_utc, record_sequence
+ </select>
+
<!-- 只取严格晚于水位的行；同秒多行按协议序号各自成立，不按秒合并。 -->
<select id="selectSamplesSince" resultMap="TDeviceIngestRecordResultMap">
select <include refid="ingestColumns"/> from t_device_ingest_record
diff --git a/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java b/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
index 1d175f7..21b6586 100644
--- a/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
+++ b/battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java
@@ -97,6 +97,17 @@
}
@Override
+ public List<TDeviceIngestRecord> selectSamplesOfSource(String adapterCode, int devtype, int devid, int subdevid,
+ int chlid, String sourceRef)
+ {
+ return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
+ .filter(row -> sourceRef.equals(row.getSourceRef()))
+ .sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
+ .thenComparing(TDeviceIngestRecord::getRecordSequence))
+ .toList();
+ }
+
+ @Override
public List<TDeviceIngestRecord> selectSamplesSince(String adapterCode, int devtype, int devid, int subdevid,
int chlid, Date watermarkTime, long watermarkSequence, int limit)
{
diff --git a/battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java b/battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java
index ec6c1be..14aa901 100644
--- a/battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java
+++ b/battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java
@@ -16,7 +16,9 @@
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
+import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
+import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
@@ -45,6 +47,37 @@
"Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31},
{"record_time":"134349698000000000","Data_Point":422,"Step_Time":600.0,"Step_Index":5,
"Cycle_Index":1,"Voltage":1.2702,"eCurrent":-0.4,"DischargeCapacity":0.05,"DischargeEnergy":0.1}]}
+ """;
+ /**
+ * 两个循环的采样行：循环 1 是充电、搁置、放电三段（都结束了），循环 2 的第一段还在进行。
+ * 容量列是工步累计：充电段从 0.10 涨到 0.12，放电段从 0.05 涨到 0.09。
+ */
+ private static final String TWO_CYCLES = """
+ {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
+ {"record_time":"134349692000000000","Data_Point":420,"Step_Time":540.0,"Step_Index":1,"Cycle_Index":1,
+ "Voltage":1.20,"eCurrent":1.0,"ChargeCapacity":0.10,"ChargeEnergy":0.30},
+ {"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":1,"Cycle_Index":1,
+ "Voltage":1.25,"eCurrent":1.0,"ChargeCapacity":0.12,"ChargeEnergy":0.36},
+ {"record_time":"134349704000000000","Data_Point":422,"Step_Time":1200.0,"Step_Index":2,"Cycle_Index":1,
+ "Voltage":1.25,"eCurrent":0.0,"ChargeCapacity":0.12,"DischargeCapacity":0.0},
+ {"record_time":"134349764000000000","Data_Point":423,"Step_Time":1800.0,"Step_Index":3,"Cycle_Index":1,
+ "Voltage":1.24,"eCurrent":-1.0,"ChargeCapacity":0.12,"DischargeCapacity":0.05,"DischargeEnergy":0.15},
+ {"record_time":"134349824000000000","Data_Point":424,"Step_Time":3600.0,"Step_Index":3,"Cycle_Index":1,
+ "Voltage":1.20,"eCurrent":-1.0,"ChargeCapacity":0.12,"DischargeCapacity":0.09,"DischargeEnergy":0.27},
+ {"record_time":"134349830000000000","Data_Point":425,"Step_Time":60.0,"Step_Index":1,"Cycle_Index":2,
+ "Voltage":1.18,"eCurrent":2.0,"ChargeCapacity":0.01,"ChargeEnergy":0.02}]}
+ """;
+ /** 第一段的电流正负混用：方向判不出来，明细照留但不进循环的任何一侧。 */
+ private static final String MIXED_SIGN = """
+ {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
+ {"record_time":"134349692000000000","Data_Point":420,"Step_Index":1,"Cycle_Index":1,
+ "Voltage":1.20,"eCurrent":1.0,"ChargeCapacity":0.10},
+ {"record_time":"134349698000000000","Data_Point":421,"Step_Index":1,"Cycle_Index":1,
+ "Voltage":1.20,"eCurrent":-1.0,"ChargeCapacity":0.10,"DischargeCapacity":0.02},
+ {"record_time":"134349704000000000","Data_Point":422,"Step_Index":2,"Cycle_Index":1,
+ "Voltage":1.25,"eCurrent":1.0,"ChargeCapacity":0.12,"ChargeEnergy":0.36},
+ {"record_time":"134349830000000000","Data_Point":425,"Step_Index":1,"Cycle_Index":2,
+ "Voltage":1.18,"eCurrent":2.0,"ChargeCapacity":0.01}]}
""";
private final FakeIngestLanding landing = new FakeIngestLanding();
@@ -172,21 +205,118 @@
/** 未声明辅助测量列就不交出测量值：确认之前不按列名推断语义，也不写零占位。 */
@Test
- void collectorAndStepCycleObservationsStayEmptyWithoutDeclaredAuxiliaryMetrics()
+ void collectorObservationsStayEmptyWithoutDeclaredAuxiliaryMetrics()
{
ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
ingest(BATCH);
PushIngestAdapter adapter = adapter(NOW);
assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
- assertTrue(adapter.stepRecords(channel()).isEmpty());
- assertTrue(adapter.cycleRecords(channel()).isEmpty());
// 落地表在本地，取数不产生厂商负载，所以按展示节奏取数；厂商写库节奏仍是 60 秒。
assertEquals(5000L, adapter.persistenceIntervalMillis(channel()));
+ // 工步与循环的重算跟着来源的写库节奏走，不跟实时行的展示节奏：源头最快 60 秒才产出一行。
+ assertEquals(60000L, adapter.aggregateIntervalMillis(channel()));
AdapterCapabilities capabilities = adapter.capabilities();
assertTrue(capabilities.discovery() && capabilities.realtimeObservation() && !capabilities.control());
}
+ /**
+ * 来源没有工步层，工步与循环由采样行分段推导（ADR-0067）：已结束的段各交出一条工步记录，
+ * 取值来自该段末行，方向由该段自己的电流符号判定；最新一段仍在进行，不交出去。
+ */
+ @Test
+ void closedSampleSegmentsBecomeStepRecordsWhileTheNewestSegmentStaysUnpublished()
+ {
+ ingest(BATCH);
+ PushIngestAdapter adapter = adapter(NOW);
+
+ List<StepRecord> steps = adapter.stepRecords(channel());
+
+ // 三行落地行分成两段：(循环 1, 工步 4) 已结束，(循环 1, 工步 5) 还在进行。
+ assertEquals(1, steps.size());
+ StepRecord step = steps.get(0);
+ assertEquals(1, step.cycleNo());
+ assertEquals(4, step.stepNo());
+ // 方向取自该段电流符号：段内两行恒正即充电。
+ assertEquals("charge", step.stepType());
+ // 容量与能量是工步累计，取该段末行的值；工步时间是工步内相对时间（秒）。
+ assertEquals(0.121, step.capacityAh(), 1e-9);
+ assertEquals(0.31, step.energyWh(), 1e-9);
+ assertEquals(600.0, step.measurements().get("stepDuration").doubleValue(), 1e-9);
+ assertEquals(1.26778, step.measurements().get("startVoltage").doubleValue(), 1e-9);
+ // 设备没给的列不写零占位：该段只有充电侧。
+ assertFalse(step.measurements().containsKey("dischargeCapacity"));
+ assertEquals(1.2701, step.endVoltageV(), 1e-9);
+ assertEquals(0.5, step.endCurrentA(), 1e-9);
+ assertEquals(Instant.parse("2026-09-27T07:53:20Z"), step.startedAt());
+ assertEquals(Instant.parse("2026-09-27T08:03:20Z"), step.endedAt());
+ // 接收时间取我们收到该段末行的时刻（与推式实时路径同一口径），设备原始时间照留。
+ assertEquals(NOW, step.serverReceivedAt());
+ assertEquals("134349698000000000", step.deviceTimeRaw());
+ // 来源行身份含来源库与段标识：同一次测试重算多少次都得到同一个身份。
+ assertEquals("push:DB_20260927_155209/Ws26_IV_0:step:1:4", step.sourceRecordId());
+ // 进行中的那一段所属的循环整体不交：半个循环不是事实。
+ assertTrue(adapter.cycleRecords(channel()).isEmpty());
+ }
+
+ /**
+ * 循环由已结束的工步按方向求和（搁置段两向都不增长，贡献 0），效率是放电容量与充电容量之比；
+ * 推导值带来源类别，界面与导出据此标注它是聚合出来的。
+ */
+ @Test
+ void cyclesAggregateClosedStepsByDirectionAndSkipTheRunningCycle()
+ {
+ ingest(TWO_CYCLES);
+ PushIngestAdapter adapter = adapter(NOW);
+
+ List<StepRecord> steps = adapter.stepRecords(channel());
+ List<CycleRecord> cycles = adapter.cycleRecords(channel());
+
+ // 循环 1 的三段都已结束（后面已经出现循环 2 的第一段）；循环 2 的第一段仍在进行。
+ assertEquals(List.of("1:1:charge", "1:2:rest", "1:3:discharge"), steps.stream()
+ .map(step -> step.cycleNo() + ":" + step.stepNo() + ":" + step.stepType()).toList());
+ // 搁置段自身的累计是 0（该段什么都没充放），不是缺失。
+ assertEquals(0.0, steps.get(1).capacityAh(), 1e-9);
+ assertEquals(0.0, steps.get(1).energyWh(), 1e-9);
+ assertEquals(List.of(1), cycles.stream().map(CycleRecord::cycleNo).toList());
+
+ CycleRecord cycle = cycles.get(0);
+ assertEquals(0.12, cycle.chargeCapacityAh(), 1e-9);
+ assertEquals(0.09, cycle.dischargeCapacityAh(), 1e-9);
+ assertEquals(0.36, cycle.chargeEnergyWh(), 1e-9);
+ assertEquals(0.27, cycle.dischargeEnergyWh(), 1e-9);
+ // 效率是百分数：0.09 / 0.12。
+ assertEquals(75.0, cycle.efficiency(), 1e-9);
+ assertEquals(CycleRecord.SOURCE_AGGREGATED, cycle.sourceKind());
+ assertEquals("push:DB_20260927_155209/Ws26_IV_0:cycle:1", cycle.sourceRecordId());
+ }
+
+ /** 方向判不出来时（段内电流正负混用）明细照留、单项容量留空，也不进循环的任何一侧。 */
+ @Test
+ void segmentsWithMixedCurrentSignsStayOutOfTheCycleAggregate()
+ {
+ ingest(MIXED_SIGN);
+ PushIngestAdapter adapter = adapter(NOW);
+
+ List<StepRecord> steps = adapter.stepRecords(channel());
+ List<CycleRecord> cycles = adapter.cycleRecords(channel());
+
+ StepRecord undecided = steps.get(0);
+ assertEquals(1, undecided.cycleNo());
+ assertEquals(1, undecided.stepNo());
+ // 方向不唯一时不猜：类型留空、单项容量与能量留空。
+ assertNull(undecided.stepType());
+ assertNull(undecided.capacityAh());
+ assertNull(undecided.energyWh());
+ // 四组容量能量作为设备事实照原样保留，供核对。
+ assertEquals(0.02, undecided.measurements().get("dischargeCapacity").doubleValue(), 1e-9);
+
+ // 该段不进任何一侧：循环 1 的放电侧算出来是 0，而不是把这一段里的 0.02 算进去。
+ assertEquals(1, cycles.size());
+ assertEquals(0.12, cycles.get(0).chargeCapacityAh(), 1e-9);
+ assertEquals(0.0, cycles.get(0).dischargeCapacityAh(), 1e-9);
+ }
+
/** 环境量与量纲确认后落到配置：列名、指标编码、单位、采集器通道都由声明给出，代码里没有厂商列名。 */
@Test
void declaredAuxiliaryMetricIsProjectedWithItsOwnSourceIdentityValueUnitAndDeviceTime()
diff --git a/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java b/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
index e8d0ea8..22b3918 100644
--- a/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
+++ b/battery-device-api/src/main/java/com/battery/device/api/TestDataObservationAdapter.java
@@ -19,7 +19,11 @@
*/
CollectorRealtime latestCollector(String adapterCode, String collectorCode);
- /** 返回设备直接提供的工步统计，不从实时采样推导。 */
+ /**
+ * 返回该通道的工步统计。来源有工步层时读设备工步层并原样交出；来源没有工步层时由适配器
+ * 从采样段推导（ADR-0067），此时记录里的方向是推导出的方向名称（见 {@link StepRecord}）。
+ * 未结束的段不交出，也不推导来源根本没有的工步层。
+ */
List<StepRecord> stepRecords(ChannelIdentity channel);
/**
@@ -31,6 +35,13 @@
/** 返回适配器定义的实时持久化槽位间隔。 */
long persistenceIntervalMillis(ChannelIdentity channel);
+
+ /**
+ * 返回工步与循环这些汇总记录的读取间隔。默认与实时持久化间隔一致；来源的汇总产出节奏与
+ * 数据行节奏不同的实现覆盖它——推式来源的落地表在本地（实时行按展示节奏取），
+ * 而汇总最快也要等来源写下下一行才可能变化，按展示节奏重算只是空跑。
+ */
+ default long aggregateIntervalMillis(ChannelIdentity channel) { return persistenceIntervalMillis(channel); }
/** 返回设备计划方案；设备未提供时保持为空，不推导计划。 */
default TestPlan testPlan(ChannelIdentity channel) { return null; }
@@ -89,8 +100,12 @@
Instant serverReceivedAt, String metricCode, BigDecimal value, String unit) { }
/**
- * 一条设备工步层记录。{@code stepType} 保留设备给出的原始类型（新威为数字枚举），
- * 方向由调用方按该类型解释，不由适配器按电流符号推断。
+ * 一条设备工步层记录。来源给出工步类型时 {@code stepType} 原样保留（新威为数字枚举），
+ * 方向由调用方按该类型解释，适配器不按电流符号改写它。
+ *
+ * <p>来源**没有**工步类型列时（推式来源的厂商库里既没有工步层也没有循环汇总表，ADR-0067），
+ * 适配器交出的是按已核对列语义判定出的方向名称（{@code charge} / {@code discharge} / {@code rest}），
+ * 判定不出来时留空；此时该工步的其余一切仍是设备采样行上的原文事实。
*
* <p>{@code endVoltageV} 与 {@code endCurrentA} 是该工步结束时的设备读数。
* {@code capacityAh} 与 {@code energyWh} 是该工步自身累计的单项值，只在方向已确认时给出；
diff --git a/ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java b/ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java
index fe58e3f..c1b1d75 100644
--- a/ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java
+++ b/ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java
@@ -207,6 +207,33 @@
assertTrue(latest.getPayloadJson().contains("Voltage"));
}
+ /**
+ * 工步与循环的分段推导读的就是这条查询（ADR-0067）：某一个来源库在该通道上的采样行，
+ * 按 (设备时间, 协议序号) 升序——推导依赖这个次序判断哪一段还在进行；其他来源库、
+ * 其他通道与其他来源的行都不混进来。
+ */
+ @Test
+ void samplesOfOneSourceRefComeBackInWatermarkOrderAndNothingElseDoes()
+ {
+ mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
+ Instant.parse("2026-09-27T07:53:20Z")));
+ mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
+ Instant.parse("2026-09-27T07:52:20Z")));
+ // 换了来源库（厂商侧另一次测试）、另一个通道：都不属于本次推导范围。
+ TDeviceIngestRecord laterTest = sample(26, "DB_20260928_101010/Ws26_IV_0:1", 1L, "{\"Voltage\":1.4}",
+ Instant.parse("2026-09-28T02:10:10Z"));
+ laterTest.setSourceRef("DB_20260928_101010/Ws26_IV_0");
+ mapper.insertIngestRecord(laterTest);
+ mapper.insertIngestRecord(sample(18, "DB_20260923_161840/Ws18_IV_0:5", 5L, "{\"Voltage\":1.1}",
+ Instant.parse("2026-09-23T08:18:40Z")));
+
+ List<TDeviceIngestRecord> rows = mapper.selectSamplesOfSource(ADAPTER, 0, 0, 1, 26,
+ "DB_20260927_155209/Ws26_IV_0");
+
+ assertEquals(List.of(11L, 12L), rows.stream().map(TDeviceIngestRecord::getRecordSequence).toList());
+ assertTrue(mapper.selectSamplesOfSource(ADAPTER, 0, 0, 1, 26, "DB_20261001_000000/Ws26_IV_0").isEmpty());
+ }
+
/** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
public static String hexOf(byte[] value)
{
--
Gitblit v1.7.1