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

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

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

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

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

0. 先看一段真实的混乱

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

[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

看出问题了吗？

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

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

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

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

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

人话版

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

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

区域定义

# 空间区域模型
# 逻辑区域（业务可见） 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

一个坑：

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

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

人话版

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

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

融合逻辑

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

两个反直觉点：

不是"投票最多的赢"。 如果 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。

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

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

人话版

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

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

位置事件状态机

# 位置事件状态机
# 输入：融合后的位置状态流 (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  # 秒

实现

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

一个坑：

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

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

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

每一层的输入输出：

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"

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

人话版

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

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

标定流程

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 必须定期复测（建议每季度一次）"

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