From e38aca8134570513435c003c25dc05f1ae51c9c9 Mon Sep 17 00:00:00 2001
From: yeg <1347747024@qq.com>
Date: 星期四, 08 十月 2026 13:56:28 +0800
Subject: [PATCH] 实现 Ticket04：采集器辅助测量与登记入口，拆除开发模拟器
---
battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java | 113 +++++++++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 102 insertions(+), 11 deletions(-)
diff --git a/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java b/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
index f988911..4dfdc59 100644
--- a/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
+++ b/battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java
@@ -9,6 +9,10 @@
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;
@@ -46,6 +50,7 @@
{
/** 适配器类型按存储形态划分：这个形态读本项目的落地表（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}，
@@ -155,14 +160,98 @@
}
/**
- * 22 所的 IV 表里有温箱温度、压力与计数器，是现场唯一见到环境量的来源，
- * 但类目语义与量纲尚未确认。按 ADR-0064 逐来源确认可得性，确认之前不交出测量值。
+ * 采集器辅助测量按 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)
{
- return null;
+ 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) { }
/**
* 22 所没有独立的工步层：工步与循环号只在采样行的 {@code Step_Index} / {@code Cycle_Index} 里，
@@ -223,16 +312,18 @@
/** 心跳缺失即来源不可达：取数程序停了、或它自己连不上厂商库。 */
private void requireFreshHeartbeat()
{
- java.util.Date receivedAt = records.selectHeartbeatAt(config.adapterCode());
- if (receivedAt == null)
+ if (!heartbeatFresh())
throw new DeviceAdapterException("SOURCE_OFFLINE",
- "尚未收到取数程序的推送：请确认 " + config.host()
+ "尚未收到取数程序的推送或推送已超时：请确认 " + config.host()
+ " 那台机器上的取数程序已启动，并指向本系统的接收端点", null);
- Duration staleAfter = staleAfter();
- Duration silent = Duration.between(receivedAt.toInstant(), clock.instant());
- if (silent.compareTo(staleAfter) > 0)
- throw new DeviceAdapterException("SOURCE_OFFLINE", "取数程序已停止推送：最近一次推送在 "
- + silent.toSeconds() + " 秒前，超过容忍窗口 " + staleAfter.toSeconds() + " 秒", 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;
}
/** 容忍窗口按实例配置的轮询周期算：写库与推送都是这个节奏，连丢三轮即认为离线。 */
--
Gitblit v1.7.1