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}）： *

* *

单位换算按实测公式：电压与电流除以 10000（原单位 0.1 mV / 0.1 mA），容量与能量除以 36000000 * （原单位 1e-4 C / 1e-4 J）。工步类型保留设备给出的数字，不按电流符号猜测方向； * 内阻与温度的单位语义仍未确认，因此不对外暴露。 */ 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; 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(), 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 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(), null, 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)) { 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) { } }