package com.battery.business.service.impl;
|
|
import java.math.BigDecimal;
|
import java.math.RoundingMode;
|
import java.time.Clock;
|
import java.time.Duration;
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.HashMap;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import com.alibaba.fastjson2.JSON;
|
import com.alibaba.fastjson2.JSONObject;
|
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;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.domain.vo.TDeviceIngestChannel;
|
import com.battery.business.mapper.TDeviceIngestRecordMapper;
|
|
/**
|
* 推式来源的观测实现（存储形态 {@code PUSH_INGEST}，ADR-0065）：读本项目的落地表，
|
* 交出与直连来源完全相同的统一观测。
|
*
|
* <p>这类来源的机器没有可用的网络协议（22 所：TCP 与命名管道都未启用、Windows 账号未获提供），
|
* 只能在它本机跑一个只读取数程序，把厂商原始行推到接收端点。程序本身很蠢——固定查询、原样推出、
|
* 不开入站端口、不在网络上传递厂商库凭据；解释（身份映射、时间换算、幂等键取值）全部在后端完成，
|
* 所以这里只做投影：落地行的字段由接收端点解析好，本类不出现任何厂商列名。
|
*
|
* <p>四条本形态特有的事实：
|
* <ul>
|
* <li><b>来源是否在线由心跳判定</b>：落地表在本地随时可读，因此"能否读到表"不能表示来源可达。
|
* 取数程序每轮推送一次心跳，心跳缺失即认为来源不可达——发现路径明确失败，
|
* 已登记设备随之离线、旧快照被撤下（既有路径），页面不会显示伪造的实时值。</li>
|
* <li><b>最近数据时间只看采样行</b>：没有新采样行就是没有，不按拓扑上报或心跳时间代替（ADR-0063）。</li>
|
* <li><b>通道工作状态不可观测</b>：落地表里没有设备侧状态，报"无法观测"而不由数据行推断。</li>
|
* <li><b>工步与循环从采样段推导</b>：这类来源的厂商库里没有工步层也没有循环汇总表，工步号与循环号
|
* 只在采样行上，所以按「来源库 + 循环号 + 工步号」分段推导（ADR-0067），推导是无状态重算。</li>
|
* </ul>
|
*/
|
public final class PushIngestAdapter implements DeviceAdapter, TestDataObservationAdapter
|
{
|
/** 适配器类型按存储形态划分：这个形态读本项目的落地表（ADR-0065）。 */
|
public static final String ADAPTER_TYPE = "PUSH_INGEST";
|
private static final Logger log = LoggerFactory.getLogger(PushIngestAdapter.class);
|
/**
|
* 身份映射（后端完成）：22 所的存储形态没有设备类型与单元维度，一台电脑就是登记意义上的一个柜，
|
* 通道号就是厂商自己的通道号（表名 {@code Ws<N>_IV_0} 里的 N）。厂商通道号进 {@code chlid}，
|
* {@code subdevid} 固定为 1，使界面通道号显示为「1-26」而不是「0-26」。
|
* 若甲方要求按接口再分成几个柜，那是新增通道→接口映射的新决策，不在本单。
|
*
|
* <p>接收端点写落地行时用的是同一组常量（{@code DeviceIngestServiceImpl.base}），
|
* 使"身份映射只在这个形态里定义一次"成立。
|
*/
|
public static final int DEVICE_TYPE = 0;
|
public static final int DEVICE_ID = 0;
|
public static final int SUB_DEVICE_ID = 1;
|
/**
|
* 落地表在本地，读它不产生厂商负载，所以按展示节奏取数（控制台 5 秒一轮）；
|
* 只取水位之后的新行，取数更勤不会产生重复记录。厂商写库节奏仍是 60 秒，值不会来得更快。
|
*/
|
private static final long PERSISTENCE_INTERVAL_MS = 5000L;
|
/** 厂商写库节奏：工步与循环的重算对齐它，来源的汇总产出不可能比这更快。 */
|
private static final long AGGREGATE_INTERVAL_MS = 60000L;
|
private static final int SAMPLE_BATCH_LIMIT = 500;
|
/** 心跳容忍窗口的下限：写库与推送都是 60 秒一轮，窗口至少覆盖三轮。 */
|
private static final Duration MINIMUM_STALE_AFTER = Duration.ofSeconds(90);
|
|
private final AdapterInstanceConfig config;
|
private final TDeviceIngestRecordMapper records;
|
private final Clock clock;
|
|
public PushIngestAdapter(AdapterInstanceConfig config, TDeviceIngestRecordMapper records)
|
{
|
this(config, records, Clock.systemUTC());
|
}
|
|
public PushIngestAdapter(AdapterInstanceConfig config, TDeviceIngestRecordMapper records, Clock clock)
|
{
|
this.config = config;
|
this.records = records;
|
this.clock = clock;
|
}
|
|
/** 返回本实例登记的适配器编码；推送端点与通道身份都用它定位来源。 */
|
@Override
|
public String adapterCode()
|
{
|
return config.adapterCode();
|
}
|
|
/** 返回本实例登记的能力标记；推式来源同样只观测，不提供控制能力。 */
|
@Override
|
public AdapterCapabilities capabilities()
|
{
|
return config.capabilities();
|
}
|
|
/**
|
* 候选来自取数程序的拓扑上报：每条候选附带落地表里该通道最近的采样行时间。
|
* 不自动判定"在用"，登记哪些通道由操作员决定（ADR-0063）。
|
*
|
* <p>候选的机器地址取实例配置的 {@code host}：这个形态里它表示**取数程序所在那台机器**，
|
* 登记检测柜时会被带进 IP 字段（推送目标地址是取数程序自己的配置项，不在这边）。
|
*
|
* <p>心跳缺失时明确失败（而不是返回空清单）：来源不可达是一个可见的失败事实，
|
* 业务侧据此把已登记设备置为离线并撤下旧快照。
|
*/
|
@Override
|
public List<DetectedChannel> discoverChannels()
|
{
|
requireFreshHeartbeat();
|
List<DetectedChannel> channels = new ArrayList<>();
|
List<TDeviceIngestChannel> reported = records.selectTopologyChannels(config.adapterCode());
|
for (TDeviceIngestChannel channel : reported == null ? List.<TDeviceIngestChannel>of() : reported)
|
{
|
Instant lastDataAt = channel.getLastDataAt() == null ? null : channel.getLastDataAt().toInstant();
|
// 来源自己的通道叫法：22 所软件按 01–30 显示，而它库里的表名比它小 1，
|
// 取数程序按现场核对过的规则补回来（见现场清单）；没报就留空，界面回退到「单元-通道」。
|
String sourceDisplayNo = channel.getVendorChannelNo() == null ? null
|
: String.valueOf(channel.getVendorChannelNo());
|
channels.add(new DetectedChannel(identity(channel.getDevid(), channel.getSubdevid(), channel.getChlid()),
|
config.host(), true, DetectedDeviceMetadata.empty(), lastDataAt, sourceDisplayNo));
|
}
|
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 HashMap<>();
|
for (ChannelIdentity channel : channels)
|
{
|
CabinetRealtime latest = latestCabinet(channel);
|
if (latest == null) continue;
|
result.put(channel, snapshot(channel, latest));
|
}
|
return Map.copyOf(result);
|
}
|
|
/**
|
* 返回该通道在落地表里最近一条采样行投影成的实时观测。
|
*
|
* @param channel 检测柜通道身份
|
* @return 最近一条采样行的实时观测；该通道还没有采样行时为空
|
*/
|
@Override
|
public CabinetRealtime latestCabinet(ChannelIdentity channel)
|
{
|
TDeviceIngestRecord row = latestSample(channel);
|
return row == null ? null : cabinetRealtime(row);
|
}
|
|
/**
|
* 采集器辅助测量按 ADR-0064 由**同一来源实例**提供，所以这里读的是落地表里该来源自己那些通道的原始行。
|
*
|
* <p>哪些列是辅助测量、各叫什么指标、用什么单位、对应哪个采集器通道，全部只在
|
* 来源配置的 {@code auxiliaryMetrics} 里声明（例如
|
* {@code {"vendorChannel":24,"channelCode":"S-24","column":"Tempbox","metricCode":"TEMPERATURE","unit":"℃"}}）。
|
* **没有声明就返回空值**：22 所 IV 表里确实有温箱温度、压力与计数器，是现场唯一见到环境量的来源，
|
* 但列语义与量纲尚未对着现场确认，未确认前不按列名推断语义、也不写出任何值。
|
* 声明之后不必改代码——确认结论落在配置里，不落在推断里。
|
*
|
* <p>来源不可达（心跳缺失或过期）时返回 {@code online=false} 且不带任何测量值：
|
* 离线不伪造测量值，业务侧据此不落任何采集器行。
|
*/
|
@Override
|
public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
|
{
|
List<AuxiliaryMetric> declared = auxiliaryMetrics();
|
if (declared.isEmpty() || !config.adapterCode().equals(adapterCode)) return null;
|
if (!heartbeatFresh())
|
return new CollectorRealtime(collectorCode, false, (int) PERSISTENCE_INTERVAL_MS, List.of(),
|
config.timezoneId());
|
List<CollectorMeasurement> measurements = new ArrayList<>();
|
for (AuxiliaryMetric metric : declared)
|
{
|
TDeviceIngestRecord row = records.selectLatestSample(config.adapterCode(), DEVICE_TYPE, DEVICE_ID,
|
SUB_DEVICE_ID, metric.vendorChannel());
|
if (row == null) continue;
|
BigDecimal value = payloadValue(row.getPayloadJson(), metric.column());
|
// 设备没给这一项就是没给：不写零、不写空值占位，也不阻断其他指标。
|
if (value == null) continue;
|
measurements.add(new CollectorMeasurement(metric.channelCode(),
|
row.getRecordSequence() == null ? 0L : row.getRecordSequence(),
|
row.getDeviceTimeUtc() == null ? null : row.getDeviceTimeUtc().toInstant(),
|
row.getDeviceTimeRaw(),
|
row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant(),
|
metric.metricCode(), value, metric.unit()));
|
}
|
return new CollectorRealtime(collectorCode, true, (int) PERSISTENCE_INTERVAL_MS, measurements,
|
config.timezoneId());
|
}
|
|
/**
|
* 读取来源配置里声明的辅助测量列。配置格式非法时按"没有声明"处理并留一条警告：
|
* 宁可少交出一项测量，也不按猜测解释厂商列。
|
*/
|
private List<AuxiliaryMetric> auxiliaryMetrics()
|
{
|
String declared = config.sourceOption("auxiliaryMetrics");
|
if (declared == null || declared.isBlank()) return List.of();
|
List<AuxiliaryMetric> metrics = new ArrayList<>();
|
try
|
{
|
for (Object item : JSON.parseArray(declared))
|
{
|
if (!(item instanceof JSONObject entry)) continue;
|
Integer vendorChannel = entry.getInteger("vendorChannel");
|
String channelCode = entry.getString("channelCode");
|
String column = entry.getString("column");
|
String metricCode = entry.getString("metricCode");
|
if (vendorChannel == null || channelCode == null || column == null || metricCode == null) continue;
|
metrics.add(new AuxiliaryMetric(vendorChannel, channelCode, column, metricCode,
|
entry.getString("unit")));
|
}
|
}
|
catch (RuntimeException failure)
|
{
|
log.warn("来源 {} 的 auxiliaryMetrics 配置无法解析，按未声明处理", config.adapterCode(), failure);
|
return List.of();
|
}
|
return List.copyOf(metrics);
|
}
|
|
/** 从原始行里按声明取数值：非数值（含 null 与文本）一律当作"设备没给这一项"。 */
|
private static BigDecimal payloadValue(String payloadJson, String column)
|
{
|
if (payloadJson == null || payloadJson.isBlank()) return null;
|
try
|
{
|
Object value = JSON.parseObject(payloadJson).get(column);
|
if (value == null) return null;
|
if (value instanceof BigDecimal decimal) return decimal;
|
if (value instanceof Number number) return new BigDecimal(number.toString());
|
return null;
|
}
|
catch (RuntimeException failure)
|
{
|
return null;
|
}
|
}
|
|
/** 一条声明：来源侧列名到业务指标（含采集器通道与单位）的映射。 */
|
private record AuxiliaryMetric(int vendorChannel, String channelCode, String column, String metricCode,
|
String unit) { }
|
|
/**
|
* 交出该通道已结束的工步。来源没有设备侧工步层，所以按本来源自己的采样行分段推导（ADR-0067）：
|
* 按「来源库 + 循环号 + 工步号」分段，只有最新一段视为进行中、不交出，其余各段各一条工步记录。
|
*/
|
@Override
|
public List<StepRecord> stepRecords(ChannelIdentity channel)
|
{
|
Derivation derivation = derive(channel);
|
List<StepRecord> records = new ArrayList<>();
|
for (StepGroup group : derivation.closedSteps()) records.add(stepRecord(derivation.sourceRef(), group));
|
return List.copyOf(records);
|
}
|
|
/**
|
* 交出该通道已结束的循环：来源没有循环汇总行，循环由已结束的工步按方向求和得出（ADR-0066 同一配方）。
|
* 最新一段所属的循环仍在进行，整体不交出——半个循环不是事实。
|
*/
|
@Override
|
public List<CycleRecord> cycleRecords(ChannelIdentity channel)
|
{
|
Derivation derivation = derive(channel);
|
Map<Integer, List<StepGroup>> byCycle = new LinkedHashMap<>();
|
for (StepGroup group : derivation.closedSteps())
|
{
|
// 仍在进行的那一段所属的循环整体排除：它自己的工步行还没写完。
|
if (group.key().cycleNo() == derivation.openCycleNo()) continue;
|
byCycle.computeIfAbsent(group.key().cycleNo(), ignored -> new ArrayList<>()).add(group);
|
}
|
List<CycleRecord> records = new ArrayList<>();
|
for (Map.Entry<Integer, List<StepGroup>> entry : byCycle.entrySet())
|
records.add(cycleRecord(derivation.sourceRef(), entry.getKey(), entry.getValue()));
|
return List.copyOf(records);
|
}
|
|
/** 返回推式来源固定的持久化槽位间隔；采样行按来源自己的落库节奏到达。 */
|
@Override
|
public long persistenceIntervalMillis(ChannelIdentity channel)
|
{
|
return PERSISTENCE_INTERVAL_MS;
|
}
|
|
/**
|
* 工步与循环的重算跟着来源的写库节奏，不跟实时行的展示节奏：来源最快每 60 秒才产出一行新采样，
|
* 而工步要等下一段出现才算结束，因此重算更勤只会空跑落地表（ADR-0067）。
|
*/
|
@Override
|
public long aggregateIntervalMillis(ChannelIdentity channel)
|
{
|
return AGGREGATE_INTERVAL_MS;
|
}
|
|
/** 水位是落地表里该通道已存在的最新一行：设备绝对时间与协议序号。 */
|
@Override
|
public SampleWatermark currentWatermark(ChannelIdentity channel)
|
{
|
TDeviceIngestRecord row = latestSample(channel);
|
return row == null || row.getDeviceTimeUtc() == null
|
? new SampleWatermark(null, 0L)
|
: new SampleWatermark(row.getDeviceTimeUtc().toInstant(), row.getRecordSequence());
|
}
|
|
/**
|
* 严格晚于水位的落地行。按 {@code (设备时间, 协议序号)} 过滤与排序：同一绝对秒内的多行各自成立，
|
* 不按秒合并。水位为空表示调用方尚未取得水位，返回空列表而不把来源里的历史并入本次接收。
|
*
|
* <p>这里不判心跳：落地行是"我们已经收到并解释过的设备事实"，
|
* 来源离线只影响发现与实时观测，不影响这些已经收到的行。
|
*/
|
@Override
|
public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
|
{
|
if (watermark == null) return List.of();
|
ChannelIdentity key = requireOwnChannel(channel);
|
java.util.Date watermarkTime = watermark.deviceTime() == null ? null : java.util.Date.from(watermark.deviceTime());
|
List<TDeviceIngestRecord> rows = records.selectSamplesSince(config.adapterCode(), key.devtype(), key.devid(),
|
key.subdevid(), key.chlid(), watermarkTime, watermark.sequenceId(), SAMPLE_BATCH_LIMIT);
|
List<CabinetRealtime> samples = new ArrayList<>();
|
for (TDeviceIngestRecord row : rows == null ? List.<TDeviceIngestRecord>of() : rows) samples.add(cabinetRealtime(row));
|
return List.copyOf(samples);
|
}
|
|
// ------------------------------------------------------------------ 投影与判定
|
|
/** 心跳缺失即来源不可达：取数程序停了、或它自己连不上厂商库。 */
|
private void requireFreshHeartbeat()
|
{
|
if (!heartbeatFresh())
|
throw new DeviceAdapterException("SOURCE_OFFLINE",
|
"尚未收到取数程序的推送或推送已超时：请确认 " + config.host()
|
+ " 那台机器上的取数程序已启动，并指向本系统的接收端点", null);
|
}
|
|
/** 来源是否在线：心跳行存在且在容忍窗口内。落地表随时可读，所以"能读到表"不能表示来源可达。 */
|
private boolean heartbeatFresh()
|
{
|
java.util.Date receivedAt = records.selectHeartbeatAt(config.adapterCode());
|
if (receivedAt == null) return false;
|
return Duration.between(receivedAt.toInstant(), clock.instant()).compareTo(staleAfter()) <= 0;
|
}
|
|
/** 容忍窗口按实例配置的轮询周期算：写库与推送都是这个节奏，连丢三轮即认为离线。 */
|
private Duration staleAfter()
|
{
|
Duration window = Duration.ofMillis(config.pollIntervalMs()).multipliedBy(3);
|
return window.compareTo(MINIMUM_STALE_AFTER) < 0 ? MINIMUM_STALE_AFTER : window;
|
}
|
|
private TDeviceIngestRecord latestSample(ChannelIdentity channel)
|
{
|
ChannelIdentity key = requireOwnChannel(channel);
|
return records.selectLatestSample(config.adapterCode(), key.devtype(), key.devid(), key.subdevid(),
|
key.chlid());
|
}
|
|
private ChannelIdentity requireOwnChannel(ChannelIdentity channel)
|
{
|
if (channel == null || !config.adapterCode().equals(channel.adapterCode()))
|
throw new DeviceAdapterException("通道身份不属于本实例："
|
+ (channel == null ? "空" : channel.adapterCode()));
|
return channel;
|
}
|
|
private CabinetRealtime cabinetRealtime(TDeviceIngestRecord row)
|
{
|
// 充放电容量按电流方向取对应的一组：与直连来源同一口径。电流符号的含义已用该来源的真实推送
|
// 数据核对（恒正即充电、恒负即放电，ADR-0067），不再是待确认的约定。
|
BigDecimal current = row.getCurrentA() == null ? BigDecimal.ZERO : row.getCurrentA();
|
boolean charging = current.signum() >= 0;
|
BigDecimal capacity = charging ? row.getChargeCapacityAh() : row.getDischargeCapacityAh();
|
BigDecimal energy = charging ? row.getChargeEnergyWh() : row.getDischargeEnergyWh();
|
// 实时行同样带上方向（类型与名称）：甲方那一列「状态」显示的就是它，实时页的"工步名称"要用。
|
StepDirection direction = directionOf(row);
|
return new CabinetRealtime(row.getRecordSequence() == null ? 0L : row.getRecordSequence(),
|
row.getDeviceTimeUtc() == null ? null : row.getDeviceTimeUtc().toInstant(),
|
current.doubleValue(), decimal(row.getVoltageV()), decimal(capacity), decimal(energy),
|
row.getCycleNo() == null ? 0 : row.getCycleNo(), row.getStepNo() == null ? 0 : row.getStepNo(),
|
// 注意参数次序：CabinetRealtime 是 stepName 在前、stepType 在后。
|
direction.displayName(), direction.label(),
|
row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant(),
|
row.getDeviceTimeRaw(), row.getTimezoneId(),
|
row.getRelativeTimeSeconds() == null ? null : row.getRelativeTimeSeconds().longValue(),
|
"push:" + row.getRecordKey());
|
}
|
|
/**
|
* 单行的方向：这一类来源没有设备侧工步类型，方向只能由电流符号判定（ADR-0067），
|
* 恒正充电、恒负放电、恒为零搁置，电流缺失保持未知。
|
*/
|
private static StepDirection directionOf(TDeviceIngestRecord row)
|
{
|
BigDecimal current = row.getCurrentA();
|
if (current == null) return StepDirection.UNKNOWN;
|
if (current.signum() > 0) return StepDirection.CHARGE;
|
if (current.signum() < 0) return StepDirection.DISCHARGE;
|
return StepDirection.REST;
|
}
|
|
// ------------------------------------------------------------------ 工步与循环的推导（ADR-0067）
|
|
/** 分段推导的结果：来源库、已结束的各段，以及仍在进行的那一段所属的循环号。 */
|
private record Derivation(String sourceRef, List<StepGroup> closedSteps, int openCycleNo) { }
|
|
/** 一段采样行：同一个来源库里同一个「循环号 + 工步号」的全部行，按 (设备时间, 协议序号) 升序。 */
|
private record StepGroup(StepKey key, List<TDeviceIngestRecord> rows)
|
{
|
TDeviceIngestRecord first() { return rows.get(0); }
|
|
TDeviceIngestRecord last() { return rows.get(rows.size() - 1); }
|
}
|
|
private record StepKey(int cycleNo, int stepNo) { }
|
|
/**
|
* 该段本身的方向。来源没有设备侧工步类型列，方向由该段自己的电流符号判定——依据是用该来源真实推送的
|
* 采样数据核对过的列语义（588 段零反例，见 ADR-0067），不是按数值大小猜方向。
|
*
|
* <p>段内电流正负混用、电流缺失，或恒为零却有容量增长时，方向不唯一：保持未知，明细照原样保留。
|
*/
|
private enum StepDirection
|
{
|
CHARGE("charge", "充电"), DISCHARGE("discharge", "放电"), REST("rest", "搁置"), UNKNOWN(null, null);
|
|
private final String label;
|
private final String displayName;
|
|
StepDirection(String label, String displayName) { this.label = label; this.displayName = displayName; }
|
|
/** 写进工步记录的类型：来源没给类型时交出的是判定出的方向名称，未知方向留空。 */
|
String label() { return label; }
|
|
/**
|
* 交出去的工步名称。这类来源的工步方案里只有「步_N」这种标号，没有描述性名称
|
* （2026-10-09 现场核对 .pcj：<Step StepLabel="步_1"> + CtrlTypeName=Rest/Current）；
|
* 而甲方软件那一列「状态」显示的就是充电/放电/搁置——所以用同一套词当工步名称，
|
* 值同样来自电流符号判定，**不是设备给的名称**（现场清单与 ADR-0067 都记了这一条）。
|
*/
|
String displayName() { return displayName; }
|
|
/** 该工步自身累计的单项值。搁置段实测两向都不增长，所以它的累计是 0 而不是缺失。 */
|
BigDecimal own(BigDecimal charge, BigDecimal discharge)
|
{
|
return switch (this)
|
{
|
case CHARGE -> charge;
|
case DISCHARGE -> discharge;
|
case REST -> BigDecimal.ZERO;
|
case UNKNOWN -> null;
|
};
|
}
|
}
|
|
/**
|
* 从该通道的采样行推导「已结束的段」。只看该通道**最新的来源库**：来源库就是厂商侧一次测试
|
* （22 所每次测试新建一个库），读取量因此限于一次厂商测试，分段也不会跨测试。
|
*
|
* <p>段是否结束只按一条依据判断：设备是否已经写下**后续工步**的行（与 ADR-0066 同一判据）。
|
* 所以只有最新一段视为进行中；它所属的循环整体不交——半个循环不是事实。
|
*/
|
private Derivation derive(ChannelIdentity channel)
|
{
|
TDeviceIngestRecord latest = latestSample(channel);
|
if (latest == null || latest.getSourceRef() == null) return new Derivation(null, List.of(), Integer.MIN_VALUE);
|
List<TDeviceIngestRecord> rows = records.selectSamplesOfSource(config.adapterCode(), latest.getDevtype(),
|
latest.getDevid(), latest.getSubdevid(), latest.getChlid(), latest.getSourceRef());
|
Map<StepKey, List<TDeviceIngestRecord>> grouped = new LinkedHashMap<>();
|
for (TDeviceIngestRecord row : rows == null ? List.<TDeviceIngestRecord>of() : rows)
|
{
|
// 分段依据是设备给出的循环号与工步号；缺任何一个都无法归属到某一段，跳过并留痕。
|
if (row.getCycleNo() == null || row.getStepNo() == null)
|
{
|
log.warn("落地行缺少循环号或工步号，不参与工步推导 ingestId={}", row.getIngestId());
|
continue;
|
}
|
grouped.computeIfAbsent(new StepKey(row.getCycleNo(), row.getStepNo()), ignored -> new ArrayList<>())
|
.add(row);
|
}
|
List<StepGroup> closed = new ArrayList<>();
|
int openCycleNo = Integer.MIN_VALUE;
|
int index = 0;
|
for (Map.Entry<StepKey, List<TDeviceIngestRecord>> entry : grouped.entrySet())
|
{
|
if (++index == grouped.size())
|
{
|
openCycleNo = entry.getKey().cycleNo();
|
break;
|
}
|
closed.add(new StepGroup(entry.getKey(), entry.getValue()));
|
}
|
return new Derivation(latest.getSourceRef(), List.copyOf(closed), openCycleNo);
|
}
|
|
/**
|
* 一条工步记录：容量与能量是**工步累计**（实测段内从未下降，末行即最大值），所以取该段末行的值；
|
* 工步时间取末行的工步内相对时间（秒）。设备没给的项不写零、不伪造中值电压。
|
*/
|
private StepRecord stepRecord(String sourceRef, StepGroup group)
|
{
|
TDeviceIngestRecord first = group.first();
|
TDeviceIngestRecord last = group.last();
|
StepDirection direction = direction(group);
|
Map<String, BigDecimal> measurements = new LinkedHashMap<>();
|
putIfPresent(measurements, "stepDuration", last.getRelativeTimeSeconds());
|
putIfPresent(measurements, "startVoltage", first.getVoltageV());
|
// 四组容量能量原样保留设备事实：方向未判定的段也留着明细供核对。
|
putIfPresent(measurements, "chargeCapacity", last.getChargeCapacityAh());
|
putIfPresent(measurements, "dischargeCapacity", last.getDischargeCapacityAh());
|
putIfPresent(measurements, "chargeEnergy", last.getChargeEnergyWh());
|
putIfPresent(measurements, "dischargeEnergy", last.getDischargeEnergyWh());
|
BigDecimal capacity = direction.own(last.getChargeCapacityAh(), last.getDischargeCapacityAh());
|
BigDecimal energy = direction.own(last.getChargeEnergyWh(), last.getDischargeEnergyWh());
|
// 工步名称用判定出的方向（充电/放电/搁置）：该来源的方案里只有「步_N」标号，没有描述性名称，
|
// 而甲方软件自己那一列「状态」显示的就是这三个词。它和方向同源，不是设备给的名称。
|
return new StepRecord(sourceRecordId(sourceRef, "step", group.key().cycleNo() + ":" + group.key().stepNo()),
|
group.key().cycleNo(), group.key().stepNo(), direction.displayName(), direction.label(), deviceTime(first),
|
deviceTime(last), decimal(last.getVoltageV()), decimal(last.getCurrentA()),
|
capacity == null ? null : capacity.doubleValue(), energy == null ? null : energy.doubleValue(),
|
last.getDeviceTimeRaw(), last.getTimezoneId(), receivedAt(last), measurements);
|
}
|
|
/**
|
* 一条循环记录：该循环各已结束工步取自己的累计值一次，按方向归入充电或放电一侧并**求和**
|
* （同一循环可能含多个同方向工步）。效率 = 放电容量 ÷ 充电容量，没有可用充电容量时留空，不写零。
|
* 搁置段两向都不增长、未知方向不按哪一列有值去猜，两者都不进任何一侧。
|
*/
|
private CycleRecord cycleRecord(String sourceRef, int cycleNo, List<StepGroup> groups)
|
{
|
BigDecimal chargeCapacity = BigDecimal.ZERO, dischargeCapacity = BigDecimal.ZERO;
|
BigDecimal chargeEnergy = BigDecimal.ZERO, dischargeEnergy = BigDecimal.ZERO;
|
for (StepGroup group : groups)
|
{
|
TDeviceIngestRecord last = group.last();
|
switch (direction(group))
|
{
|
case CHARGE ->
|
{
|
chargeCapacity = chargeCapacity.add(orZero(last.getChargeCapacityAh()));
|
chargeEnergy = chargeEnergy.add(orZero(last.getChargeEnergyWh()));
|
}
|
case DISCHARGE ->
|
{
|
dischargeCapacity = dischargeCapacity.add(orZero(last.getDischargeCapacityAh()));
|
dischargeEnergy = dischargeEnergy.add(orZero(last.getDischargeEnergyWh()));
|
}
|
default -> { }
|
}
|
}
|
TDeviceIngestRecord first = groups.get(0).first();
|
TDeviceIngestRecord last = groups.get(groups.size() - 1).last();
|
return new CycleRecord(sourceRecordId(sourceRef, "cycle", String.valueOf(cycleNo)), cycleNo,
|
deviceTime(first), deviceTime(last), 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.getDeviceTimeRaw(), last.getTimezoneId(), receivedAt(last), CycleRecord.SOURCE_AGGREGATED);
|
}
|
|
/** 段内电流符号一致时给出该方向；正负混用或电流缺失保持未知。 */
|
private static StepDirection direction(StepGroup group)
|
{
|
BigDecimal maximumCharge = group.first().getChargeCapacityAh();
|
BigDecimal maximumDischarge = group.first().getDischargeCapacityAh();
|
boolean positive = false, negative = false;
|
for (TDeviceIngestRecord row : group.rows())
|
{
|
BigDecimal current = row.getCurrentA();
|
// 电流缺失的段判不出方向：不拿同段其它行的符号代替。
|
if (current == null) return StepDirection.UNKNOWN;
|
positive |= current.signum() > 0;
|
negative |= current.signum() < 0;
|
maximumCharge = maximum(maximumCharge, row.getChargeCapacityAh());
|
maximumDischarge = maximum(maximumDischarge, row.getDischargeCapacityAh());
|
}
|
if (positive && negative) return StepDirection.UNKNOWN;
|
if (positive) return StepDirection.CHARGE;
|
if (negative) return StepDirection.DISCHARGE;
|
// 恒为零：搁置。但恒零却出现容量增长时方向未能判定——不按哪一列在涨去猜。
|
boolean accumulated = raised(group.first().getChargeCapacityAh(), maximumCharge)
|
|| raised(group.first().getDischargeCapacityAh(), maximumDischarge);
|
return accumulated ? StepDirection.UNKNOWN : StepDirection.REST;
|
}
|
|
/** 来源行身份：来源库 + 工步或循环标识。同一次测试重算多少次都得到同一个身份，重复交出因此幂等。 */
|
private static String sourceRecordId(String sourceRef, String kind, String suffix)
|
{
|
return "push:" + sourceRef + ":" + kind + ":" + suffix;
|
}
|
|
/** 只放设备给了的项：缺列就是没有这一项，不写零代替缺失。 */
|
private static void putIfPresent(Map<String, BigDecimal> target, String key, BigDecimal value)
|
{
|
if (value != null) target.put(key, value);
|
}
|
|
/** 求和时把缺列当没有：缺少的项不参与求和，也不写零占位。 */
|
private static BigDecimal orZero(BigDecimal value)
|
{
|
return value == null ? BigDecimal.ZERO : value;
|
}
|
|
private static BigDecimal maximum(BigDecimal current, BigDecimal candidate)
|
{
|
if (candidate == null) return current;
|
return current == null ? candidate : current.max(candidate);
|
}
|
|
private static boolean raised(BigDecimal first, BigDecimal maximum)
|
{
|
return first != null && maximum != null && maximum.compareTo(first) > 0;
|
}
|
|
private static Instant deviceTime(TDeviceIngestRecord row)
|
{
|
return row.getDeviceTimeUtc() == null ? null : row.getDeviceTimeUtc().toInstant();
|
}
|
|
/**
|
* 接收时间取**我们收到这一行**的时刻，与推式来源的实时路径同一口径，也是接收边界判定的依据；
|
* 直连来源没有这个独立取值，它沿用设备记录时刻（见 KNOWN-GAPS）。
|
*/
|
private static Instant receivedAt(TDeviceIngestRecord row)
|
{
|
return row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant();
|
}
|
|
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 ChannelIdentity identity(Integer devid, Integer subdevid, Integer chlid)
|
{
|
return new ChannelIdentity(config.adapterCode(), DEVICE_TYPE,
|
devid == null ? DEVICE_ID : devid, subdevid == null ? SUB_DEVICE_ID : subdevid,
|
chlid == null ? 0 : chlid);
|
}
|
|
private static double decimal(BigDecimal value)
|
{
|
return value == null ? 0d : value.doubleValue();
|
}
|
}
|
