这类来源的机器没有可用的网络协议（22 所：TCP 与命名管道都未启用、Windows 账号未获提供），
* 只能在它本机跑一个只读取数程序，把厂商原始行推到接收端点。程序本身很蠢——固定查询、原样推出、
* 不开入站端口、不在网络上传递厂商库凭据；解释（身份映射、时间换算、幂等键取值）全部在后端完成，
* 所以这里只做投影：落地行的字段由接收端点解析好，本类不出现任何厂商列名。
*
*
_IV_0} 里的 N）。厂商通道号进 {@code chlid}，
* {@code subdevid} 固定为 1，使界面通道号显示为「1-26」而不是「0-26」。
* 若甲方要求按接口再分成几个柜，那是新增通道→接口映射的新决策，不在本单。
*
* 接收端点写落地行时用的是同一组常量（{@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 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）。
*
*
候选的机器地址取实例配置的 {@code host}：这个形态里它表示**取数程序所在那台机器**，
* 登记检测柜时会被带进 IP 字段（推送目标地址是取数程序自己的配置项，不在这边）。
*
*
心跳缺失时明确失败（而不是返回空清单）：来源不可达是一个可见的失败事实，
* 业务侧据此把已登记设备置为离线并撤下旧快照。
*/
@Override
public List discoverChannels()
{
requireFreshHeartbeat();
List channels = new ArrayList<>();
List reported = records.selectTopologyChannels(config.adapterCode());
for (TDeviceIngestChannel channel : reported == null ? List.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 inquire(Set channels)
{
if (channels == null || channels.isEmpty()) return Map.of();
Map 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 由**同一来源实例**提供，所以这里读的是落地表里该来源自己那些通道的原始行。
*
* 哪些列是辅助测量、各叫什么指标、用什么单位、对应哪个采集器通道，全部只在
* 来源配置的 {@code auxiliaryMetrics} 里声明（例如
* {@code {"vendorChannel":24,"channelCode":"S-24","column":"Tempbox","metricCode":"TEMPERATURE","unit":"℃"}}）。
* **没有声明就返回空值**：22 所 IV 表里确实有温箱温度、压力与计数器，是现场唯一见到环境量的来源，
* 但列语义与量纲尚未对着现场确认，未确认前不按列名推断语义、也不写出任何值。
* 声明之后不必改代码——确认结论落在配置里，不落在推断里。
*
*
来源不可达（心跳缺失或过期）时返回 {@code online=false} 且不带任何测量值：
* 离线不伪造测量值，业务侧据此不落任何采集器行。
*/
@Override
public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
{
List 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 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 auxiliaryMetrics()
{
String declared = config.sourceOption("auxiliaryMetrics");
if (declared == null || declared.isBlank()) return List.of();
List 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) { }
/**
* 22 所没有独立的工步层：工步与循环号只在采样行的 {@code Step_Index} / {@code Cycle_Index} 里，
* 从采样推导工步是另一项决策（与"工步来自设备工步层"的既有口径冲突），不在本单范围，因此返回空。
*/
@Override
public List stepRecords(ChannelIdentity channel)
{
return List.of();
}
/** 同上：循环统计不从采样推导，本单不交出循环记录。 */
@Override
public List cycleRecords(ChannelIdentity channel)
{
return List.of();
}
/** 返回推式来源固定的持久化槽位间隔；采样行按来源自己的落库节奏到达。 */
@Override
public long persistenceIntervalMillis(ChannelIdentity channel)
{
return PERSISTENCE_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 (设备时间, 协议序号)} 过滤与排序：同一绝对秒内的多行各自成立，
* 不按秒合并。水位为空表示调用方尚未取得水位，返回空列表而不把来源里的历史并入本次接收。
*
* 这里不判心跳：落地行是"我们已经收到并解释过的设备事实"，
* 来源离线只影响发现与实时观测，不影响这些已经收到的行。
*/
@Override
public List 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 rows = records.selectSamplesSince(config.adapterCode(), key.devtype(), key.devid(),
key.subdevid(), key.chlid(), watermarkTime, watermark.sequenceId(), SAMPLE_BATCH_LIMIT);
List samples = new ArrayList<>();
for (TDeviceIngestRecord row : rows == null ? List.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)
{
// 充放电容量按电流方向取对应的一组：与直连来源同一口径（22 所的电流符号约定待现场对着上位机确认）。
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();
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(),
null, null, row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant(),
row.getDeviceTimeRaw(), row.getTimezoneId(),
row.getRelativeTimeSeconds() == null ? null : row.getRelativeTimeSeconds().longValue(),
"push:" + row.getRecordKey());
}
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();
}
}