多读头融合：从"谁在场"到"在哪里"

# 多读头融合：从"谁在场"到"在哪里"

承接上篇：我们把盘点快照变成了业务事件，通过 MQTT 推给了 WMS。

但现实仓库不是"一个读头对一个门口"。一个货架区可能有 4 个 KLM9700，天线交叉覆盖。同一个标签，左边读头说"在 A 区"，右边读头说"在 B 区"——WMS 听谁的？

这篇讲多读头融合：从"哪个标签在场"到"这个标签在哪里"。

0. 先看一段真实的混乱

这是 4 个读头同时跑 10 秒的事件流（截段）：

<pre><code>[15:30:45.100] reader-001/tag_enter: EPC=E2003412012A1B2C3D4, zone=shelf-a [15:30:45.150] reader-002/tag_enter: EPC=E2003412012A1B2C3D4, zone=shelf-b [15:30:45.200] reader-001/tag_leave: EPC=E2003412012A1B2C3D4, zone=shelf-a [15:30:45.300] reader-003/tag_enter: EPC=E2003412012A1B2C3D4, zone=aisle-1 [15:30:45.500] reader-002/tag_leave: EPC=E2003412012A1B2C3D4, zone=shelf-b [15:30:46.000] reader-003/tag_leave: EPC=E2003412012A1B2C3D4, zone=aisle-1</code></pre>

看出问题了吗？

<strong>重叠覆盖：</strong> 同一个标签，reader-001 和 reader-002 都报 enter，间隔只有 50ms。是标签真的同时在两个区，还是天线覆盖重叠？

<strong>快速切换：</strong> 100ms 后 reader-001 报 leave，300ms 后 reader-003 报 enter。是标签被移动了，还是信号漂移？

<strong>缺失上下文：</strong> 4 个读头各自为政，没有全局视图。WMS 收到 6 个事件，但不知道"这个标签现在到底在哪"。

我们需要一个<strong>融合层</strong>：把多个读头的原始事件，融合成一个全局一致的"位置状态"。

1. 空间模型：从物理到逻辑

人话版

读头的天线覆盖是物理的、模糊的、重叠的。但业务需要的是逻辑的、明确的、互斥的。

"shelf-a" 和 "shelf-b" 不是两个天线，而是两个<strong>逻辑区域</strong>。一个标签不可能同时在两个逻辑区域。

区域定义

<pre><code class="lang-yaml"># 空间区域模型 # 逻辑区域（业务可见） vs 读头覆盖（物理层） zones: # 逻辑区域 - id: "shelf-a" type: "storage" description: "A 货架区" readers: ["reader-001", "reader-002"] # 哪些读头覆盖这个区域 priority: 1 # 冲突时的优先级 - id: "shelf-b" type: "storage" description: "B 货架区" readers: ["reader-002", "reader-003"] priority: 1 - id: "aisle-1" type: "transit" description: "1 号通道" readers: ["reader-003", "reader-004"] priority: 2 # 通道优先级低于货架 - id: "inbound-zone" type: "gate" description: "入库口" readers: ["reader-004"] priority: 3 # 门口优先级最高 reader_coverage: # 每个读头的覆盖范围（可能跨多个逻辑区域） reader-001: primary_zone: "shelf-a" bleed_zones: ["aisle-1"] # 信号溢出区域 confidence_map: shelf-a: 0.9 aisle-1: 0.3 reader-002: primary_zone: "shelf-b" bleed_zones: ["shelf-a", "aisle-1"] confidence_map: shelf-b: 0.85 shelf-a: 0.4 aisle-1: 0.35 reader-003: primary_zone: "aisle-1" bleed_zones: ["shelf-b"] confidence_map: aisle-1: 0.88 shelf-b: 0.3 reader-004: primary_zone: "inbound-zone" bleed_zones: ["aisle-1"] confidence_map: inbound-zone: 0.95 aisle-1: 0.25</code></pre>

一个坑：

<strong>primary_zone 和 bleed_zones 必须现场标定。</strong> 不能靠"理论上天线覆盖范围"。实际部署时，在目标区域放 20 个标签跑 10 分钟，统计每个读头对每个区域的检出率，就是 confidence_map。

2. 融合算法：从多源到单点

人话版

4 个读头同时报"看到这个标签"，融合层的任务是决定"它最可能在哪"。

不是"投票最多的赢"，而是"置信度最高的赢"。reader-001 在 shelf-a 的置信度是 0.9，reader-002 在 shelf-b 的置信度是 0.85——如果两个读头同时报 enter，标签更可能在 shelf-a。

融合逻辑

<pre><code class="lang-python">from dataclasses import dataclass from typing import Dict, List, Optional import time @dataclass class ReaderEvent: epc: str reader_id: str event_type: str # "enter" | "leave" ts: float confidence: float @dataclass class LocationState: epc: str zone: str confidence: float last_update: float source_readers: List[str] class MultiReaderFusion: def __init__(self, zone_config: dict, fusion_window: float = 0.5): self.zone_config = zone_config self.fusion_window = fusion_window # 融合窗口（秒） # 每个读头的覆盖配置 self.reader_coverage = zone_config["reader_coverage"] # 每个逻辑区域的优先级 self.zone_priority = {z["id"]: z["priority"] for z in zone_config["zones"]} # 当前状态：epc -> LocationState self.location_state = {} # 事件缓冲：用于融合窗口内的批量处理 self.event_buffer = [] def feed(self, event: ReaderEvent) -> Optional[LocationState]: """输入一个读头事件，输出融合后的位置状态（如果有变化）""" self.event_buffer.append(event) # 按融合窗口批量处理 now = event.ts self.event_buffer = [ e for e in self.event_buffer if now - e.ts <= self.fusion_window ] # 只处理当前事件触发的融合 return self._fuse(event.epc, now) def _fuse(self, epc: str, now: float) -> Optional[LocationState]: """对单个 EPC 执行融合""" # 收集融合窗口内所有相关事件 related_events = [ e for e in self.event_buffer if e.epc == epc and now - e.ts <= self.fusion_window ] if not related_events: return None # 计算每个区域的综合置信度 zone_scores: Dict[str, float] = {} zone_readers: Dict[str, List[str]] = {} for event in related_events: if event.event_type != "enter": continue reader_id = event.reader_id coverage = self.reader_coverage.get(reader_id, {}) # 遍历这个读头覆盖的所有区域 for zone, base_conf in coverage.get("confidence_map", {}).items(): # 综合置信度 = 读头报告置信度 × 区域覆盖置信度 score = event.confidence * base_conf if zone not in zone_scores: zone_scores[zone] = 0.0 zone_readers[zone] = [] zone_scores[zone] += score if reader_id not in zone_readers[zone]: zone_readers[zone].append(reader_id) if not zone_scores: # 所有事件都是 leave，标签离开所有区域 if epc in self.location_state: del self.location_state[epc] return None # 选置信度最高的区域 best_zone = max(zone_scores.keys(), key=lambda z: (zone_scores[z], self.zone_priority.get(z, 0))) best_score = zone_scores[best_zone] # 检查是否有变化 current = self.location_state.get(epc) if current and current.zone == best_zone and current.confidence >= best_score * 0.9: # 位置没变，置信度没显著下降，不更新 return None # 更新状态 new_state = LocationState( epc=epc, zone=best_zone, confidence=best_score, last_update=now, source_readers=zone_readers[best_zone] ) self.location_state[epc] = new_state return new_state</code></pre>

两个反直觉点：

<strong>不是"投票最多的赢"。</strong> 如果 reader-001 和 reader-002 都报 enter，但 reader-001 在 shelf-a 的置信度是 0.9，reader-002 在 shelf-b 的置信度是 0.4，那么 shelf-a 的综合得分是 0.9，shelf-b 是 0.4——即使两个读头都报了这个标签，标签更可能在 shelf-a。

<strong>融合窗口不能太短。</strong> 0.5 秒是经验值。如果设成 0.1 秒，网络抖动导致的事件延迟会让你错过融合机会；如果设成 2 秒，标签快速移动时你会把"经过通道"误判为"停留在货架"。

3. 位置事件：从融合到业务

人话版

融合层输出的是"这个标签现在在 shelf-a，置信度 0.85"。但业务系统需要的是"这个标签从 aisle-1 移动到了 shelf-a"。

位置事件不是"在哪"，而是"从哪到哪"。

位置事件状态机

<pre><code class="lang-yaml"># 位置事件状态机 # 输入：融合后的位置状态流 (epc, zone, confidence) # 输出：位置事件流 (move/enter_zone/leave_zone) location_state_machine: name: "LocationTracker" states: - UNKNOWN: "位置未知（初始状态，或长时间未检出）" - IN_ZONE: "在某个逻辑区域内" - IN_TRANSIT: "在区域间移动（置信度下降，或多次切换）" transitions: - from: UNKNOWN to: IN_ZONE trigger: "fusion confidence >= 0.7" action: "emit ENTER_ZONE event" - from: IN_ZONE to: IN_ZONE trigger: "zone changed && new_confidence >= 0.7" action: "emit MOVE event (from_zone -> to_zone)" - from: IN_ZONE to: IN_TRANSIT trigger: "confidence < 0.5 持续 2s，或多个区域得分接近" action: "emit LEAVE_ZONE event" - from: IN_TRANSIT to: IN_ZONE trigger: "new zone confidence >= 0.7" action: "emit ENTER_ZONE event" - from: IN_TRANSIT to: UNKNOWN trigger: "所有区域 confidence < 0.3 持续 5s" action: "emit LEAVE_ZONE event" parameters: enter_threshold: 0.7 transit_threshold: 0.5 unknown_threshold: 0.3 transit_duration: 2.0 # 秒 unknown_duration: 5.0 # 秒</code></pre>

实现

<pre><code class="lang-python">from enum import Enum from dataclasses import dataclass from typing import Optional class LocationState(Enum): UNKNOWN = "unknown" IN_ZONE = "in_zone" IN_TRANSIT = "in_transit" @dataclass class LocationEvent: epc: str event_type: str # "enter_zone" | "leave_zone" | "move" from_zone: Optional[str] to_zone: Optional[str] confidence: float ts: float class LocationTracker: def __init__(self, enter_threshold=0.7, transit_threshold=0.5, unknown_threshold=0.3, transit_duration=2.0, unknown_duration=5.0): self.enter_threshold = enter_threshold self.transit_threshold = transit_threshold self.unknown_threshold = unknown_threshold self.transit_duration = transit_duration self.unknown_duration = unknown_duration # epc -> {state, zone, last_update, leave_timer} self.tracker = {} def feed(self, fusion_result) -> Optional[LocationEvent]: """输入融合结果，输出位置事件""" epc = fusion_result.epc zone = fusion_result.zone confidence = fusion_result.confidence ts = fusion_result.last_update state = self.tracker.setdefault(epc, { "state": LocationState.UNKNOWN, "zone": None, "last_update": 0, "leave_timer": None, }) current = state["state"] current_zone = state["zone"] # 状态跃迁 if current == LocationState.UNKNOWN: if confidence >= self.enter_threshold: state["state"] = LocationState.IN_ZONE state["zone"] = zone state["last_update"] = ts return LocationEvent(epc, "enter_zone", None, zone, confidence, ts) elif current == LocationState.IN_ZONE: if confidence >= self.enter_threshold: if zone != current_zone: # 区域切换 state["zone"] = zone state["last_update"] = ts return LocationEvent(epc, "move", current_zone, zone, confidence, ts) else: # 同一区域，更新置信度 state["last_update"] = ts return None elif confidence < self.transit_threshold: state["state"] = LocationState.IN_TRANSIT state["leave_timer"] = ts return None else: # 置信度在 [transit, enter) 之间，保持 state["last_update"] = ts return None elif current == LocationState.IN_TRANSIT: if confidence >= self.enter_threshold: state["state"] = LocationState.IN_ZONE state["zone"] = zone state["leave_timer"] = None state["last_update"] = ts return LocationEvent(epc, "enter_zone", current_zone, zone, confidence, ts) elif confidence < self.unknown_threshold: if ts - state["leave_timer"] >= self.unknown_duration: state["state"] = LocationState.UNKNOWN state["zone"] = None state["leave_timer"] = None return LocationEvent(epc, "leave_zone", current_zone, None, confidence, ts) else: # 置信度在 [unknown, transit) 之间，继续等待 if ts - state["leave_timer"] >= self.transit_duration: # 超过 transit 时间，确认离开 state["state"] = LocationState.UNKNOWN state["zone"] = None state["leave_timer"] = None return LocationEvent(epc, "leave_zone", current_zone, None, confidence, ts) return None</code></pre>

一个坑：

<strong>move 事件必须原子化。</strong> 不能先发 leave_zone 再发 enter_zone，因为中间有个"未知"状态。业务系统收到 leave_zone 会认为标签"消失了"，然后 enter_zone 又"出现了"。move 是单个事件，表示"从 A 到 B"，中间没有间隙。

4. 完整数据流：从多读头到位置事件

<pre><code>KLM9700 #1 ──┐ KLM9700 #2 ──┤ KLM9700 #3 ──┼──→ 事件抽象层 ──→ 融合层 ──→ 位置追踪 ──→ 位置事件 KLM9700 #4 ──┘ (enter/leave) (zone) (move) (MQTT)</code></pre>

每一层的输入输出：

<pre><code class="lang-yaml">pipeline_stages: - name: "EventAbstraction" input: "TagRead 流 (epc, present, confidence)" output: "业务事件流 (enter/leave per reader)" component: "TagPresenceTracker" - name: "MultiReaderFusion" input: "多个读头的 enter/leave 事件" output: "融合后的位置状态 (epc, zone, confidence)" component: "MultiReaderFusion" - name: "LocationTracking" input: "融合后的位置状态流" output: "位置事件流 (enter_zone/leave_zone/move)" component: "LocationTracker" - name: "MQTTReporting" input: "位置事件" output: "MQTT 消息 (topic: {site}/{zone}/location_event)" component: "MQTTReporter"</code></pre>

5. 部署标定：不是配参数，是测环境

人话版

融合算法的参数（置信度阈值、融合窗口）不是"填个数就完事"。每个仓库的金属环境、天线位置、标签类型都不同。

部署前必须做<strong>现场标定</strong>：在目标场景跑真实标签，统计每个读头对每个区域的检出率，生成 confidence_map。

标定流程

<pre><code class="lang-yaml">calibration_procedure: step_1: "在目标区域放置已知数量的标签（建议 20-50 个）" step_2: "所有读头同时启动，记录 10 分钟的原始盘点数据" step_3: "对每个读头，统计对每个区域的检出率" step_4: "生成 confidence_map" example: zone: "shelf-a" reader: "reader-001" total_reads: 1200 unique_tags_seen: 20 expected_tags: 20 detection_rate: 1.0 # 100% avg_confidence: 0.92 rssi_mean: -55.3 rssi_std: 4.2 output: reader_coverage: reader-001: primary_zone: "shelf-a" confidence_map: shelf-a: 0.92 aisle-1: 0.28 # 信号溢出 warnings: - "如果 detection_rate < 0.8，检查天线位置或功率" - "如果 rssi_std > 8，环境干扰严重，需要硬件排查" - "confidence_map 必须定期复测（建议每季度一次）"</code></pre>

*本系列下一篇：把位置事件接入 WMS——不是"这个标签在 shelf-a"，而是"这个标签对应的 SKU 在 shelf-a/row-3/col-5"。位置事件 + 物品注册表 = 完整的库存可视化。*