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.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 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 long lastDataCacheMillis; private final Map lastDataCache = 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; } /** 工步层记录由本单之后的一张单实现，这里不返回从采样推导出来的工步。 */ @Override public List stepRecords(ChannelIdentity channel) { return List.of(); } /** 循环汇总表不存在，聚合配方须先用真实数据核对，这里不返回推导值。 */ @Override public List cycleRecords(ChannelIdentity channel) { return List.of(); } /** 厂商固定 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 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) { if (value == null) return 0d; // 保留 12 位小数与业务列精度一致，避免在换算环节提前丢失设备原始分辨率。 return value.divide(divisor, 12, RoundingMode.HALF_UP).doubleValue(); } 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 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) { } }