实现 Ticket04：采集器辅助测量与登记入口，拆除开发模拟器

yeg

6 天以前 e38aca8134570513435c003c25dc05f1ae51c9c9

 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;
    }

    /** 容忍窗口按实例配置的轮询周期算：写库与推送都是这个节奏，连丢三轮即认为离线。 */