package com.battery.device.neware;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
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;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
/**
* 直连新威上位机 MySQL 库的只读适配器（存储形态 {@code NEWARE_MYSQL}）。
*
* 现场确认的取数事实（见 {@code docs/commissioning/现场对接清单.md}）：
*
* - 发现候选来自 {@code channel} 通道登记表，主键 {@code (dev_uid, unit_id, chl_id)}；
* 设备号 {@code dev_uid = devtype × 10000 + devid}。登记表不等于"当前在用"，
* 所以只把"最近数据时间"作为事实交出，登记哪些候选由操作员决定。
* - 逐点数据在 {@code data_<设备六位>_<年>_<月>_<日>} 分表里，表名由 {@code test} 表的
* {@code main_first_table} / {@code main_second_table} 直接给出，不按日期推算。
* - 最新测试按 {@code start_time} 取，不能按 {@code test_id} 排序（实测不单调）。
* - 活跃与否只看有没有新的采样行；{@code end_time} 是否为空与 8.0 的通道水位线列都不作为依据。
* - 厂商固定每 60 秒写一行，实时展示节奏与数据产生节奏天生分离。
* - 辅助表族实测没有数据，采集器辅助测量只能走导入，因此 {@link #latestCollector} 保持为空。
*
*
* 单位换算按实测公式：电压与电流除以 10000（原单位 0.1 mV / 0.1 mA），容量与能量除以 36000000
* （原单位 1e-4 C / 1e-4 J）。工步类型保留设备给出的数字，不按电流符号猜测方向；工步名称由该类型
* 经 {@link NewareStepDirectionResolver} 解释出的方向名称填写（未确认取值留空，ADR-0067 §4b）；
* 内阻与温度的单位语义仍未确认，因此不对外暴露。
*/
public final class NewareMysqlAdapter implements DeviceAdapter, TestDataObservationAdapter
{
public static final String ADAPTER_TYPE = "NEWARE_MYSQL";
/** 连接口令的默认取值名：本地凭据文件的键、同名环境变量或同名启动参数，见 {@link NewareMysqlAdapterFactory#credential}。 */
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;
/**
* 每条语句的查询截止时间。厂商库的服务端**没有**查询超时，客户端不设就是无限；
* JDBC URL 里的 {@code socketTimeout} 只管单次读，慢速滴流的查询仍可能长期占住连接。
*/
private static final int QUERY_TIMEOUT_SECONDS = 10;
private static final Pattern SAFE_TABLE_NAME = Pattern.compile("[A-Za-z0-9_]{1,64}");
/** 设备时间按适配器时区解释；来源列是秒精度 datetime，小数部分按需接受。 */
private static final DateTimeFormatter DEVICE_TIME = new DateTimeFormatterBuilder()
.appendPattern("yyyy-MM-dd HH:mm:ss")
.optionalStart().appendFraction(ChronoField.NANO_OF_SECOND, 1, 9, true).optionalEnd()
.toFormatter(Locale.ROOT);
/** 水位回写来源时使用秒精度文本，与来源列一致；同秒内的区分由协议序号承担。 */
private static final DateTimeFormatter DEVICE_TIME_TEXT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss",
Locale.ROOT);
private static final BigDecimal VOLTAGE_CURRENT_DIVISOR = BigDecimal.valueOf(10000);
private static final BigDecimal CAPACITY_ENERGY_DIVISOR = BigDecimal.valueOf(36000000);
private final AdapterInstanceConfig config;
private final DataSource dataSource;
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 lastDataCache = new ConcurrentHashMap<>();
private final Map capStepCache = new ConcurrentHashMap<>();
public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource)
{
this(config, dataSource, () -> { }, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
}
public NewareMysqlAdapter(AdapterInstanceConfig config, DataSource dataSource, Runnable connectionReset,
Clock clock, long lastDataCacheMillis)
{
this.config = config;
this.dataSource = dataSource;
this.connectionReset = connectionReset;
this.clock = clock;
this.deviceZone = ZoneId.of(config.timezoneId());
this.lastDataCacheMillis = Math.max(0, lastDataCacheMillis);
}
/** 由工厂使用：把驱动连接包上只读白名单，并在查询失败后作废连接。 */
public static NewareMysqlAdapter create(AdapterInstanceConfig config, SingleConnectionVendorDataSource source)
{
return new NewareMysqlAdapter(config, new ReadOnlyVendorDataSource(source, config.adapterCode()),
source::invalidate, Clock.systemUTC(), SAMPLE_INTERVAL_MS);
}
/** 返回本实例登记的适配器编码。 */
@Override
public String adapterCode()
{
return config.adapterCode();
}
/** 返回实例配置登记的能力标记。 */
@Override
public AdapterCapabilities capabilities()
{
return config.capabilities();
}
/** 发现候选来自厂商库的通道登记表；每条候选附带该通道最近的采样行时间。 */
@Override
public List discoverChannels()
{
List registered = query("select dev_uid, unit_id, chl_id from channel "
+ "order by dev_uid, unit_id, chl_id", this::readRegisteredChannels);
Map> byDevice = new LinkedHashMap<>();
for (ChannelKey key : registered)
byDevice.computeIfAbsent(key.devUid(), ignored -> new ArrayList<>()).add(key);
List channels = new ArrayList<>();
Instant observedAt = clock.instant();
for (Map.Entry> entry : byDevice.entrySet())
{
Map latestTests = latestTests(entry.getKey());
for (ChannelKey key : entry.getValue())
{
VendorTest latest = latestTests.get(key);
Instant lastDataAt = latest == null ? null : cachedLastSampleAt(key, latest, observedAt);
channels.add(new DetectedChannel(identity(key), config.host(), true,
DetectedDeviceMetadata.empty(), lastDataAt));
}
}
return List.copyOf(channels);
}
/** 实时快照取自该通道最新测试的最新一行；没有可用行时不返回该通道，避免沿用过期观测。 */
@Override
public Map inquire(Set channels)
{
if (channels == null || channels.isEmpty()) return Map.of();
Map result = new LinkedHashMap<>();
for (ChannelIdentity channel : channels)
{
CabinetRealtime latest = latestCabinet(channel);
if (latest == null) continue;
result.put(channel, snapshot(channel, latest));
}
return Map.copyOf(result);
}
/** 取该通道最新测试的最新一行采样作为柜端实时观测；没有测试或没有采样行时返回空值。 */
@Override
public CabinetRealtime latestCabinet(ChannelIdentity channel)
{
ChannelKey key = channelKey(channel);
VendorTest latest = latestTest(key);
if (latest == null) return null;
SampleRow row = lastSample(key, latest);
return row == null ? null : cabinetRealtime(row);
}
/** 新威辅助表族实测没有数据，采集器辅助测量只能靠导入（ADR-0064）。 */
@Override
public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
{
return null;
}
/** 工步层记录与循环统计都读设备的 {@code cap_step} 表；该表不随采样分表滚动。 */
@Override
public List stepRecords(ChannelIdentity channel)
{
CapStepSource source = capStepSource(channel);
if (source == null) return List.of();
List records = new ArrayList<>();
for (StepGroup group : completedStepGroups(source.rows())) records.add(stepRecord(source.testId(), group));
return List.copyOf(records);
}
/**
* 循环统计由工步层按循环聚合：设备没有循环汇总行（`cap_plat` 的充放电容量列实测全为 0，
* 库内也没有任何按循环汇总的表）。
*
* 配方（行语义见 {@link #completedStepGroups}）：
*
* - 每个已结束工步的累计容量与能量各计一次，重复接收不重复累计。
* - 按方向归入本循环的充电或放电一侧并求和：同一循环可能含多个同方向工步
* （该化成协议的前两个循环各有两个充电工步和两个放电工步），求和才是该循环充入/放出的总量。
* - 效率是放电容量除以充电容量；没有可用充电容量时留空，未知比率不写零。
* - 搁置与未知方向的工步不进任何一侧：容量无法归属到方向时，明细留在工步记录里供核对。
*
*
* 不插值、不补齐、不跨循环借值：聚合只用该循环自己的工步行，仍在进行的循环整体不出现。
*/
@Override
public List cycleRecords(ChannelIdentity channel)
{
CapStepSource source = capStepSource(channel);
if (source == null) return List.of();
List rows = source.rows();
Map> 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 records = new ArrayList<>();
for (Map.Entry> entry : byCycle.entrySet())
if (entry.getKey() != openCycle) records.add(cycleRecord(source.testId(), entry.getKey(), entry.getValue()));
return List.copyOf(records);
}
/** 厂商固定 60 秒写库；持久化周期与之对齐，不做倍速。 */
@Override
public long persistenceIntervalMillis(ChannelIdentity channel)
{
return SAMPLE_INTERVAL_MS;
}
/** 水位是来源里已存在的最新一行：设备绝对时间与协议序号。 */
@Override
public SampleWatermark currentWatermark(ChannelIdentity channel)
{
ChannelKey key = channelKey(channel);
VendorTest latest = latestTest(key);
if (latest == null) return new SampleWatermark(null, 0L);
SampleRow row = lastSample(key, latest);
return row == null ? new SampleWatermark(null, 0L) : new SampleWatermark(row.deviceTime(), row.sequenceId());
}
/**
* 严格晚于水位的设备数据行。同一绝对秒内的多行各自成立（实测工步切换点会在同一秒写入多行），
* 所以按 {@code (test_atime, seq_id)} 过滤与排序，不按秒合并。
*
* 设备时间为空的水位表示边界时该通道在来源里没有任何行，之后出现的行全部属于本次接收。
* 厂商侧测试编号不参与边界判定，只写进来源行身份；接收边界是本系统的接收边界。
* 若测试期间分表滚动，两张表都会被读取，避免滚动点整段缺失。
*/
@Override
public List samplesSince(ChannelIdentity channel, SampleWatermark watermark)
{
if (watermark == null) return List.of();
ChannelKey key = channelKey(channel);
VendorTest latest = latestTest(key);
if (latest == null) return List.of();
boolean bounded = watermark.deviceTime() != null;
String rawWatermark = bounded ? deviceTimeText(watermark.deviceTime()) : null;
List rows = new ArrayList<>();
for (String table : latest.tables())
{
String predicate = bounded
? "and (test_atime > ? or (test_atime = ? and seq_id > ?)) "
: "";
Object[] parameters = bounded
? new Object[] { key.unitId(), key.chlId(), rawWatermark, rawWatermark, watermark.sequenceId() }
: new Object[] { key.unitId(), key.chlId() };
rows.addAll(query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, test_cur, "
+ "test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
+ " where unit_id = ? and chl_id = ? " + predicate
+ "order by test_atime, seq_id limit " + SAMPLE_BATCH_LIMIT,
this::readSampleRows, parameters));
}
rows.sort(Comparator.comparing(SampleRow::deviceTime).thenComparingLong(SampleRow::sequenceId));
return rows.stream().map(this::cabinetRealtime).toList();
}
// ------------------------------------------------------------------ 来源读取
private List readRegisteredChannels(ResultSet rows) throws SQLException
{
List channels = new ArrayList<>();
while (rows.next())
channels.add(new ChannelKey(rows.getLong("dev_uid"), rows.getInt("unit_id"), rows.getInt("chl_id")));
return channels;
}
/** 按起始时间取每台设备每个通道的最新测试；test_id 实测不单调，不能用来排序。 */
private Map latestTests(long devUid)
{
return query("select unit_id, chl_id, test_id, main_first_table, main_second_table from test "
+ "where dev_uid = ? and start_time = (select max(start_time) from test newer "
+ "where newer.dev_uid = test.dev_uid and newer.unit_id = test.unit_id and newer.chl_id = test.chl_id) "
+ "order by unit_id, chl_id, test_id", rows -> {
Map result = new LinkedHashMap<>();
while (rows.next())
{
ChannelKey key = new ChannelKey(devUid, rows.getInt("unit_id"), rows.getInt("chl_id"));
result.putIfAbsent(key, vendorTest(rows.getLong("test_id"),
rows.getString("main_first_table"), rows.getString("main_second_table")));
}
return result;
}, devUid);
}
private VendorTest latestTest(ChannelKey key)
{
List tests = query("select test_id, main_first_table, main_second_table from test "
+ "where dev_uid = ? and unit_id = ? and chl_id = ? order by start_time desc limit 1",
rows -> rows.next() ? List.of(vendorTest(rows.getLong("test_id"), rows.getString("main_first_table"),
rows.getString("main_second_table"))) : List.of(),
key.devUid(), key.unitId(), key.chlId());
return tests.isEmpty() ? null : tests.get(0);
}
/** 测试数据可能跨分表滚动，两张表都要看；优先取第二张表，滚动后它就是最新的那张。 */
private SampleRow lastSample(ChannelKey key, VendorTest test)
{
List tables = new ArrayList<>(test.tables());
Collections.reverse(tables);
for (String table : tables)
{
List rows = query("select test_id, seq_id, cycle, step_id, step_type, test_time, test_vol, "
+ "test_cur, test_capchg, test_engchg, test_capdchg, test_engdchg, test_atime from " + quote(table)
+ " where unit_id = ? and chl_id = ? and test_id = ? order by seq_id desc limit 1",
this::readSampleRows, key.unitId(), key.chlId(), test.testId());
if (!rows.isEmpty()) return rows.get(0);
}
return null;
}
/** 最近数据时间来自采样行本身：没有新采样行就是没有，不看结束时间也不看通道水位线列。 */
private Instant cachedLastSampleAt(ChannelKey key, VendorTest test, Instant now)
{
String cacheKey = key.devUid() + ":" + key.unitId() + ":" + key.chlId() + ":" + test.testId();
CachedLastData cached = lastDataCache.get(cacheKey);
if (cached != null && now.isBefore(cached.expiresAt())) return cached.lastDataAt();
SampleRow row = lastSample(key, test);
Instant lastDataAt = row == null ? null : row.deviceTime();
if (lastDataCacheMillis > 0)
lastDataCache.put(cacheKey, new CachedLastData(lastDataAt, now.plusMillis(lastDataCacheMillis)));
return lastDataAt;
}
private List readSampleRows(ResultSet rows) throws SQLException
{
List result = new ArrayList<>();
while (rows.next())
{
String raw = rows.getString("test_atime");
LocalDateTime deviceTime = parseDeviceTime(raw);
result.add(new SampleRow(rows.getLong("test_id"), 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"), raw, deviceTime.atZone(deviceZone).toInstant()));
}
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 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 readCapStepRows(ResultSet rows) throws SQLException
{
List 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;
}
/**
* 已结束的工步：按「循环号 + 工步号」分组，每组取设备写的最后一行作为该工步的累计事实。
*
* 实测行语义：同一工步的行数不是常量（多数为进入行加结束行的 2 行，另有 1 行与 3–10 行的少数协议），
* 工步的累计容量与能量只出现在结束行上（进入行全为 0，结束行等于该工步的最大值），
* 且它是**工步累计**而不是测试累计——同一工步在不同循环的值基本相同，不随测试推进累加。
* 因此取最后一行，不按行求和、不取平均。
*
*
工步是否结束只按一条依据判断：设备是否已经写了后续工步的行。只有进入行的工步仍在进行中，
* 不落库——否则它结束后的结束行会与已保存的行冲突。代价是设备停止后最后一个未闭合的工步不出现，
* 这是"不把未确认的事实当事实"的落地形态，也保证暂停或继续不产生补算。
*/
private List completedStepGroups(List rows)
{
Map> groups = new LinkedHashMap<>();
for (CapStepRow row : rows)
groups.computeIfAbsent(new StepKey(row.cycle(), row.stepId()), ignored -> new ArrayList<>()).add(row);
List completed = new ArrayList<>();
int last = groups.size() - 1;
int index = 0;
for (Map.Entry> 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 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(), direction.displayName(), 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 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)
{
// 充放电容量按电流方向取对应的一组：设备界面对正在放电的通道显示放电容量。
boolean charging = row.current().signum() >= 0;
BigDecimal capacity = charging ? row.capacityCharge() : row.capacityDischarge();
BigDecimal energy = charging ? row.energyCharge() : row.energyDischarge();
return new CabinetRealtime(row.sequenceId(), row.deviceTime(), scaled(row.current(), VOLTAGE_CURRENT_DIVISOR),
scaled(row.voltage(), VOLTAGE_CURRENT_DIVISOR), scaled(capacity, CAPACITY_ENERGY_DIVISOR),
scaled(energy, CAPACITY_ENERGY_DIVISOR), row.cycle(), row.stepId(),
// 工步名称填设备工步类型解释出的方向名称；类型未确认时留空，不用默认名冒充已知。
directions.resolve(row.stepType()).displayName(), row.stepType(),
clock.instant(), row.deviceTimeRaw(), deviceZone.getId(),
row.testTimeMillis() == null ? null : row.testTimeMillis() / 1000,
"neware:" + row.testId() + ":" + row.sequenceId());
}
private RealtimeSnapshot snapshot(ChannelIdentity channel, CabinetRealtime latest)
{
// 读库路线下通道工作状态不可观测（ADR-0063）：不按数据行推断，也不沿用上一次状态。
return new RealtimeSnapshot(channel, DeviceWorkStatus.UNOBSERVABLE, latest.cycleNo(), latest.stepNo(),
latest.stepType(), null, BigDecimal.valueOf(latest.currentA()), BigDecimal.valueOf(latest.voltageV()),
BigDecimal.valueOf(latest.capacityAh()), BigDecimal.valueOf(latest.energyWh()), null,
latest.relativeTimeSeconds(), null, latest.serverReceivedAt(), Map.of(),
DeviceConnectionStatus.ONLINE, latest.deviceTime(), List.of(), null);
}
private static double scaled(BigDecimal value, BigDecimal divisor)
{
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)
{
// 设备号编码 devtype × 10000 + devid；unit_id 是该柜内的单元，作为 subdevid 保留。
return new ChannelIdentity(config.adapterCode(), (int) (key.devUid() / 10000),
(int) (key.devUid() % 10000), key.unitId(), key.chlId());
}
private ChannelKey channelKey(ChannelIdentity channel)
{
if (!config.adapterCode().equals(channel.adapterCode()))
throw new DeviceAdapterException("通道身份不属于本实例：" + channel.adapterCode());
return new ChannelKey(channel.devtype() * 10000L + channel.devid(), channel.subdevid(), channel.chlid());
}
private static VendorTest vendorTest(long testId, String firstTable, String secondTable)
{
List tables = new ArrayList<>(2);
if (isSafeTable(firstTable)) tables.add(firstTable.trim());
if (isSafeTable(secondTable) && !tables.contains(secondTable.trim())) tables.add(secondTable.trim());
return new VendorTest(testId, tables);
}
private static boolean isSafeTable(String table)
{
return table != null && SAFE_TABLE_NAME.matcher(table.trim()).matches();
}
private static String quote(String table)
{
// 表名来自厂商库自身的 test 记录，必须先按白名单校验再拼接，避免把库内容当 SQL。
if (!isSafeTable(table)) throw new DeviceAdapterException("来源表名不合法：" + table);
return "`" + table.trim() + "`";
}
private LocalDateTime parseDeviceTime(String raw)
{
if (raw == null || raw.isBlank())
throw new DeviceAdapterException("PARSE_FAILED", "设备时间为空，无法解释为设备事实", null);
String normalized = raw.trim().replace('T', ' ');
if (normalized.endsWith("Z")) normalized = normalized.substring(0, normalized.length() - 1).trim();
try
{
return LocalDateTime.parse(normalized, DEVICE_TIME);
}
catch (RuntimeException exception)
{
throw new DeviceAdapterException("PARSE_FAILED", "设备时间无法解释：" + raw, exception);
}
}
private String deviceTimeText(Instant instant)
{
return instant.atZone(deviceZone).toLocalDateTime().format(DEVICE_TIME_TEXT);
}
// ------------------------------------------------------------------ JDBC
/**
* 厂商数据源交出的是长期复用的一条连接，这里只关闭语句与结果集，不关闭连接本身；
* 查询失败时作废连接，下一轮取数重新建立。
*/
private T query(String sql, ResultSetReader reader, Object... parameters)
{
try
{
Connection connection = dataSource.getConnection();
try (PreparedStatement statement = connection.prepareStatement(sql))
{
// 读取超时有两层：连接串的 socketTimeout 管单次读，这里给整条语句一个截止时间。
statement.setQueryTimeout(QUERY_TIMEOUT_SECONDS);
for (int index = 0; index < parameters.length; index++)
statement.setObject(index + 1, parameters[index]);
try (ResultSet rows = statement.executeQuery())
{
return reader.read(rows);
}
}
}
catch (NonQueryStatementException exception)
{
throw exception;
}
catch (SQLException exception)
{
connectionReset.run();
throw new DeviceAdapterException("CONNECTION_FAILED",
"厂商库读取失败：" + config.host() + ":" + config.port() + "，原因：" + exception.getMessage(),
exception);
}
}
private interface ResultSetReader
{
T read(ResultSet rows) throws SQLException;
}
private record ChannelKey(long devUid, int unitId, int chlId) { }
private record VendorTest(long testId, List tables) { }
private record CachedLastData(Instant lastDataAt, Instant expiresAt) { }
private record CachedCapSteps(CapStepSource source, Instant expiresAt) { }
/** 该通道最新厂商测试在工步层的全部行；{@code testId} 同时用于来源行身份。 */
private record CapStepSource(long testId, List rows) { }
/** 工步号在同一个循环内唯一，所以「循环号 + 工步号」就是一条工步事实的身份。 */
private record StepKey(int cycle, int stepId) { }
private record StepGroup(StepKey key, List 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,
String deviceTimeRaw, Instant deviceTime) { }
}