2.1 传感器阵列：环境数据采集与输入流

2.1 传感器阵列：环境数据采集与输入流

本节摘要：感知舱的机体是传感器阵列与采集管线。本节按飞行问题 → 舱段设计 → 算法实现 → 联调测试 → 故障排查的五段展开：先看阵列缺员时的事故现场，再给选型矩阵与互补原则，然后实现采样、时间戳、缓冲、重采样的四工序输入流，最后给出时间对齐联调与 timestamp 漂移的排障法。

「传感器融合」这个词在项目书里出现得越来越密，仿佛装满探头就自动有了可靠感知。可真实事故记录讲的另一个故事：某仓储机器人在雨季连续撞上透明卷帘门——单目相机在逆光里看不见玻璃，激光雷达的近红外被水膜吸掉了回波，而整条感知管线上没有任何人意识到"两路情报是同一天气条件的难兄难弟"。这就是本节的飞行问题：阵列不是探头堆，是失效模式的互相补位。上一章状态板上的每个字段，追根溯源都从这里进舱；本节顺着五段节奏，把阵列设计与输入管线一次讲透。

飞行问题：你的阵列缺谁

判断阵列缺员的办法是给每只探头写"病假条"：它在什么条件下会失效，失效时谁替班。常见探头的关键弱点各不相同——单目与双目相机怕暗光、逆光、玻璃；激光雷达怕雨雾吸收与镜面反射；毫米波雷达全天候但角分辨率低，分不清行人与立柱；IMU 与轮式里程计不看外界，只报自身运动，长时间积分会漂移；超声波近距可靠但刷新慢；触觉与力传感器只在接触瞬间说话。除此之外还有一类容易被忽略的软件传感器：外部系统的接口轮询、消息订阅、文件监听——对软件智能体来说，数据库的变更流就是它的雷达回波。

互补原则的检验标准一句话：在最恶劣的那类天气或光照下，仍然至少有两路独立情报在岗。检验方法叫失效演练——人为遮住某路探头，看阵列整体输出降级到什么程度，这是第 7 章仿真测试的常规科目。

舱段设计：阵列的接线图

阵列设计除选探头，还要定三个接口问题。挂载位置：探头装在哪决定视野与遮挡——把相机装在能被自己货箱挡住的位置，参数再好也白搭；供电与带宽：激光雷达动辄每秒百万级点，总线的带宽预算要按"全部探头满帧"算，不是按典型帧算；时钟源：所有探头共用一个单调时钟打时间戳，这是后面时间对齐的物质基础。没有统一时钟的阵列，做出来的融合是把不同时刻的世界硬拼成一张图。

算法实现：四工序输入流

采集管线可以固化成四道工序，顺序不能乱：采样（驱动层把物理量转成数字）、打戳（进入管线的第一时间盖单调时钟时间戳，越早越准）、缓冲（环形队列吸收突发流量，队列满时按策略丢最老的帧）、重采样（把异步到达的各路读数对齐到主循环节拍，插值补齐）。

import time, threading
from collections import deque

class Sensor:
    """探头抽象：所有探头实现同一接口，管线才能一视同仁。"""
    def __init__(self, name, rate_hz):
        self.name = name
        self.rate = rate_hz
        self.ok = True                      # 病假条：False 即请求替班

    def read(self):
        raise NotImplementedError

class RingBuffer:
    """有界缓冲：满时丢最老帧，宁要新数据不要全量历史。"""
    def __init__(self, maxlen):
        self._q = deque(maxlen=maxlen)
        self._lock = threading.Lock()
        self.dropped = 0

    def push(self, frame):
        with self._lock:
            if len(self._q) == self._q.maxlen:
                self.dropped += 1           # 背压留痕，不是静默丢弃
            self._q.append(frame)

    def latest(self, max_age):
        with self._lock:
            for frame in reversed(self._q):  # 从最新往回找
                if time.monotonic() - frame.stamp <= max_age:
                    return frame
        return None                          # 过期即失明，别拿旧图充数

class IngestPipeline:
    def __init__(self, sensors):
        self.buffers = {s.name: RingBuffer(maxlen=32) for s in sensors}
        self.threads = [threading.Thread(
            target=self._pump, args=(s,), daemon=True) for s in sensors]

    def _pump(self, sensor):
        period = 1.0 / sensor.rate
        while True:
            t0 = time.monotonic()
            try:
                raw = sensor.read()
                frame = {"value": raw,
                         "stamp": time.monotonic(),   # 打戳越早越准
                         "sensor": sensor.name}
                self.buffers[sensor.name].push(frame)
            except Exception:
                sensor.ok = False            # 探头报病假，融合层降级
            time.sleep(max(0, period - (time.monotonic() - t0)))

    def aligned_snapshot(self, period=0.05):
        """重采样：把各路最新读数对齐成状态板可用的快照。"""
        snap = {}
        for name, buf in self.buffers.items():
            snap[name] = buf.latest(max_age=2 * period)
        return snap
​

实现里有处处可辨的取舍：环形队列丢帧而不是阻塞，是因为感知管线的下游是实时回路，堵住管线等于全机停摆；latest 找不到新鲜读数时返回空而不是旧值，是把"过期"的判断权交给状态估计段——那里有信念可以兜底，裸旧值只会误导。

探头的健康管理同样值得一段骨架——病假条只有被巡检才算数：

class ArrayHealthCheck:
    """阵列巡检：沉默失效的哨兵，联调与运行共用。"""

    def __init__(self, pipeline, board):
        self.pipeline = pipeline
        self.board = board

    def run(self):
        report = {}
        for name, sensor in self.pipeline.sensors.items():
            buf = self.pipeline.buffers[name]
            report[name] = {
                "ok": sensor.ok,
                "fresh": buf.latest(max_age=1.0) is not None,
                "dropped": buf.dropped,       # 丢帧率反映带宽压力
            }
            if not sensor.ok:
                self.board.write(f"sensor_down:{name}",
                                 True, writer="health")
        return report

    def failure_drill(self, name):
        """失效演练：人为遮住一路探头，看整体降级。"""
        sensor = self.pipeline.sensors[name]
        sensor.ok = False                     # 模拟病假
        snap = self.pipeline.aligned_snapshot()
        alive = [k for k, v in snap.items() if v is not None]
        return {"masked": name, "alive": alive}   # 至少两路在岗才合格
​

联调测试：时间对齐是主科目

阵列的联调核心不是"每只探头都能读数"，而是各路读数在时间上站得住。标准科目：让一个已知速度的目标横穿阵列视野，检验融合快照里"相机看到它"与"激光看到它"的时间差是否在半拍以内；再让探头满帧运行，观察丢帧计数是否稳定。时间对齐失败的症状很有辨识度——融合图上同一物体出现重影，物体速度越快重影越大，这是 timestamp 漂移的指纹。

故障排查：按症状查阵列

读数抖动：先查物理安装（松动、共振），再查供电噪声，最后才是算法滤波——病根在硬件时滤波只是遮丑。某路长期沉默：探头报了病假但没人看——病假条必须接入监控，沉默失效比大声失效危险。重影：查时钟源与打戳时机，驱动层缓存过的读数要回溯原始到达时刻。带宽拥塞：丢帧计数上涨，按探头优先级降帧率，保关键路。每一条处置，最终都要回到那句互补原则：阵列的可靠性是设计出来的，不是调参调出来的。

⚠️ 常见坑：验收时用"典型场景"测阵列。探头缺员的失效恰恰发生在非典型场景——雨季、逆光、满载带宽。按最坏条件验收，按典型条件调优，顺序不能反。

本节要点回顾

• 互补原则：阵列按失效模式互相补位设计，最恶劣条件下至少两路独立情报在岗；失效演练是验收科目。
• 软件传感器：接口轮询与消息订阅也是探头，外部系统就是软件智能体的物理世界。
• 四工序管线：采样、打戳（越早越准）、有界缓冲（丢帧留痕）、重采样对齐；顺序与取舍固定。
• 探针抽象：统一接口加病假条字段，融合层才有依据降级。
• 主联调科目：已知目标横穿测时间对齐，半拍以内合格；重影是 timestamp 漂移的指纹。

阵列送来的是带毛刺的原始流，下一步进调理车间：滤波、剔野值、对坐标，把"垃圾进"改写成"干净出"。