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}）。
|
*
|
* <p>现场确认的取数事实（见 {@code docs/commissioning/现场对接清单.md}）：
|
* <ul>
|
* <li>发现候选来自 {@code channel} 通道登记表，主键 {@code (dev_uid, unit_id, chl_id)}；
|
* 设备号 {@code dev_uid = devtype × 10000 + devid}。登记表不等于"当前在用"，
|
* 所以只把"最近数据时间"作为事实交出，登记哪些候选由操作员决定。</li>
|
* <li>逐点数据在 {@code data_<设备六位>_<年>_<月>_<日>} 分表里，表名由 {@code test} 表的
|
* {@code main_first_table} / {@code main_second_table} 直接给出，不按日期推算。</li>
|
* <li>最新测试按 {@code start_time} 取，不能按 {@code test_id} 排序（实测不单调）。</li>
|
* <li>活跃与否只看有没有新的采样行；{@code end_time} 是否为空与 8.0 的通道水位线列都不作为依据。</li>
|
* <li>厂商固定每 60 秒写一行，实时展示节奏与数据产生节奏天生分离。</li>
|
* <li>辅助表族实测没有数据，采集器辅助测量只能走导入，因此 {@link #latestCollector} 保持为空。</li>
|
* </ul>
|
*
|
* <p>单位换算按实测公式：电压与电流除以 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<String, CachedLastData> 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<DetectedChannel> discoverChannels()
|
{
|
List<ChannelKey> registered = query("select dev_uid, unit_id, chl_id from channel "
|
+ "order by dev_uid, unit_id, chl_id", this::readRegisteredChannels);
|
Map<Long, List<ChannelKey>> byDevice = new LinkedHashMap<>();
|
for (ChannelKey key : registered)
|
byDevice.computeIfAbsent(key.devUid(), ignored -> new ArrayList<>()).add(key);
|
List<DetectedChannel> channels = new ArrayList<>();
|
Instant observedAt = clock.instant();
|
for (Map.Entry<Long, List<ChannelKey>> entry : byDevice.entrySet())
|
{
|
Map<ChannelKey, VendorTest> 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<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
|
{
|
if (channels == null || channels.isEmpty()) return Map.of();
|
Map<ChannelIdentity, RealtimeSnapshot> 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<StepRecord> stepRecords(ChannelIdentity channel)
|
{
|
return List.of();
|
}
|
|
/** 循环汇总表不存在，聚合配方须先用真实数据核对，这里不返回推导值。 */
|
@Override
|
public List<CycleRecord> 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)} 过滤与排序，不按秒合并。
|
*
|
* <p>设备时间为空的水位表示边界时该通道在来源里没有任何行，之后出现的行全部属于本次接收。
|
* 厂商侧测试编号不参与边界判定，只写进来源行身份；接收边界是本系统的接收边界。
|
* 若测试期间分表滚动，两张表都会被读取，避免滚动点整段缺失。
|
*/
|
@Override
|
public List<CabinetRealtime> 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<SampleRow> 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<ChannelKey> readRegisteredChannels(ResultSet rows) throws SQLException
|
{
|
List<ChannelKey> 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<ChannelKey, VendorTest> 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<ChannelKey, VendorTest> 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<VendorTest> 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.<VendorTest>of(),
|
key.devUid(), key.unitId(), key.chlId());
|
return tests.isEmpty() ? null : tests.get(0);
|
}
|
|
/** 测试数据可能跨分表滚动，两张表都要看；优先取第二张表，滚动后它就是最新的那张。 */
|
private SampleRow lastSample(ChannelKey key, VendorTest test)
|
{
|
List<String> tables = new ArrayList<>(test.tables());
|
Collections.reverse(tables);
|
for (String table : tables)
|
{
|
List<SampleRow> 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<SampleRow> readSampleRows(ResultSet rows) throws SQLException
|
{
|
List<SampleRow> 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<String> 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> T query(String sql, ResultSetReader<T> 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>
|
{
|
T read(ResultSet rows) throws SQLException;
|
}
|
|
private record ChannelKey(long devUid, int unitId, int chlId) { }
|
|
private record VendorTest(long testId, List<String> 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) { }
|
}
|
