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}）。
|
*
|
* <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 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<String, CachedLastData> lastDataCache = new ConcurrentHashMap<>();
|
private final Map<String, CachedCapSteps> 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<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;
|
}
|
|
/** 工步层记录与循环统计都读设备的 {@code cap_step} 表；该表不随采样分表滚动。 */
|
@Override
|
public List<StepRecord> stepRecords(ChannelIdentity channel)
|
{
|
CapStepSource source = capStepSource(channel);
|
if (source == null) return List.of();
|
List<StepRecord> records = new ArrayList<>();
|
for (StepGroup group : completedStepGroups(source.rows())) records.add(stepRecord(source.testId(), group));
|
return List.copyOf(records);
|
}
|
|
/**
|
* 循环统计由工步层按循环聚合：设备没有循环汇总行（`cap_plat` 的充放电容量列实测全为 0，
|
* 库内也没有任何按循环汇总的表）。
|
*
|
* <p>配方（行语义见 {@link #completedStepGroups}）：
|
* <ul>
|
* <li>每个已结束工步的累计容量与能量各计一次，重复接收不重复累计。</li>
|
* <li>按方向归入本循环的充电或放电一侧并<b>求和</b>：同一循环可能含多个同方向工步
|
* （该化成协议的前两个循环各有两个充电工步和两个放电工步），求和才是该循环充入/放出的总量。</li>
|
* <li>效率是放电容量除以充电容量；没有可用充电容量时留空，未知比率不写零。</li>
|
* <li>搁置与未知方向的工步不进任何一侧：容量无法归属到方向时，明细留在工步记录里供核对。</li>
|
* </ul>
|
*
|
* <p>不插值、不补齐、不跨循环借值：聚合只用该循环自己的工步行，仍在进行的循环整体不出现。
|
*/
|
@Override
|
public List<CycleRecord> cycleRecords(ChannelIdentity channel)
|
{
|
CapStepSource source = capStepSource(channel);
|
if (source == null) return List.of();
|
List<CapStepRow> rows = source.rows();
|
Map<Integer, List<StepGroup>> 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<CycleRecord> records = new ArrayList<>();
|
for (Map.Entry<Integer, List<StepGroup>> 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)} 过滤与排序，不按秒合并。
|
*
|
* <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 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<CapStepRow> 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<CapStepRow> readCapStepRows(ResultSet rows) throws SQLException
|
{
|
List<CapStepRow> 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;
|
}
|
|
/**
|
* 已结束的工步：按「循环号 + 工步号」分组，每组取设备写的最后一行作为该工步的累计事实。
|
*
|
* <p>实测行语义：同一工步的行数不是常量（多数为进入行加结束行的 2 行，另有 1 行与 3–10 行的少数协议），
|
* 工步的累计容量与能量只出现在结束行上（进入行全为 0，结束行等于该工步的最大值），
|
* 且它是**工步累计**而不是测试累计——同一工步在不同循环的值基本相同，不随测试推进累加。
|
* 因此取最后一行，不按行求和、不取平均。
|
*
|
* <p>工步是否结束只按一条依据判断：设备是否已经写了<b>后续工步</b>的行。只有进入行的工步仍在进行中，
|
* 不落库——否则它结束后的结束行会与已保存的行冲突。代价是设备停止后最后一个未闭合的工步不出现，
|
* 这是"不把未确认的事实当事实"的落地形态，也保证暂停或继续不产生补算。
|
*/
|
private List<StepGroup> completedStepGroups(List<CapStepRow> rows)
|
{
|
Map<StepKey, List<CapStepRow>> groups = new LinkedHashMap<>();
|
for (CapStepRow row : rows)
|
groups.computeIfAbsent(new StepKey(row.cycle(), row.stepId()), ignored -> new ArrayList<>()).add(row);
|
List<StepGroup> completed = new ArrayList<>();
|
int last = groups.size() - 1;
|
int index = 0;
|
for (Map.Entry<StepKey, List<CapStepRow>> 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<String, BigDecimal> 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<StepGroup> 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<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))
|
{
|
// 读取超时有两层：连接串的 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>
|
{
|
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 CachedCapSteps(CapStepSource source, Instant expiresAt) { }
|
|
/** 该通道最新厂商测试在工步层的全部行；{@code testId} 同时用于来源行身份。 */
|
private record CapStepSource(long testId, List<CapStepRow> rows) { }
|
|
/** 工步号在同一个循环内唯一，所以「循环号 + 工步号」就是一条工步事实的身份。 */
|
private record StepKey(int cycle, int stepId) { }
|
|
private record StepGroup(StepKey key, List<CapStepRow> 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) { }
|
}
|
