从零搭一条 RFID 盘点管线：读取、清洗、盘点、标注

承接前作：户口本讲 EPC 身份体系、大脑讲 RFID 状态机、神经讲数据流。这篇文章把三者接成一条能跑的管线——从读头的原始数据，一路走到「谁在场、谁离场」。

本系列用的硬件是我们自己的 KLM9700 固定式读写器（内核 Impinj E710 模块），代码也围绕它来写。如果你手边没有这台设备，MockReader 部分可以零硬件跑通全流程。

**AI 使用说明**：本文含完整的协议规格（YAML）、原始报文摘录（hex dump）、字节偏移表和状态机定义。把这些 code 块喂给任何 AI 编码工具，它应该能直接输出可运行的对接代码，不需要再翻 SDK 文档。

0. 先看一段真实的烂数据

不先讲硬件，先甩数据。这是一台 KLM9700 跑 10 秒吐出来的东西（截段）：

EPC,RSSI,Phase,Antenna,Timestamp
E2003412012A1B2C3D4,-52.3,1.87,1,1726900000.123
E2003412012A1B2C3D4,-51.9,-2.94,1,1726900000.187
E2003412012A1B2C3D4,-78.4,0.31,2,1726900000.502
,,,1,1726900000.551
E2003412012A1B2C3D4,-53.1,1.95,1,1726900000.610
E2003412012A1B2C3D4,-52.7,3.05,1,1726900000.672

看出问题了吗？

重复： 同一个标签 60 毫秒内被读了两次——读头的采样率远高于我们的关心粒度；

跳变： Phase 从 1.87 直接跳到 -2.94，差了整整 2π——相位回绕（wrap），不是它真翻了身；

串读： -78.4 那条是天线 2 从隔壁货架"漏"进来的，并不是本盘点的目标；

空包： 第三行 EPC 直接是空的，硬件偶发。

我们的目标，是把这一堆噪声变成一句话：{"epc": "...", "present": true, "confidence": 0.92}。

先有数据，再有管线。下面每一步都在解决上面某一种脏。

1. 通信与协议：KLM9700 怎么吐数据

人话版

KLM9700 基于 Impinj E710 模块，支持两种物理连接：TCP 网口（默认端口 4001）和 RS232 串口（E710 默认波特率 115200）。固定式部署一般走网口——一根网线搞定，不用操心串口驱动和线缆长度。

它不走标准 LLRP 协议，走的是厂商自定义的二进制帧。你不需要读完协议手册才能干活——下面把你需要知道的字段全部摊开了。

协议完整规格

# KLM9700 / E710 专有二进制协议规格
# 版本: M70X v4.1 | 适用: KLM9700 固定式读写器

transport:
  tcp:
    default_port: 4001
    description: "推荐。固定部署首选，一根网线。"
  serial:
    default_baud: 115200
    data_bits: 8
    stop_bits: 1
    parity: none
    description: "E710 评估板默认。需串口线。"

frame:
  header:
    value: 0xA0
    description: "每帧固定以此字节开头"
  length:
    offset: 1
    size: 1 byte
    meaning: "本字段之后的字节数（不含 header 和 length 自身）"
    formula: "length = addr(1) + cmd(1) + data(N) + check(1)"
  address:
    offset: 2
    size: 1 byte
    default: 0x00
    description: "读写器地址，单机环境固定 0x00"
  command:
    offset: 3
    size: 1 byte
  data:
    offset: 4
    size: "variable = length - 2"
  checksum:
    offset: "length + 1"
    size: 1 byte
    algorithm: "从 header 到 data 最后一个字节逐字节累加，取结果的低 8 位"
    formula: "check = sum(frame[0:frame_len]) & 0xFF"

commands:
  inventory_start:
    code: 0x89
    subcmd: 0x00
    frame: "A0 05 [addr] 89 00 00 [check]"
    description: "启动连续盘存，读头开始主动推送标签数据帧"
  inventory_stop:
    code: 0x89
    subcmd: 0x01
    frame: "A0 05 [addr] 89 01 00 [check]"
    description: "停止盘存"
  inventory_result:
    code: 0x89
    direction: "response (per round)"
    data_layout:
      ant_id:
        offset: 4
        size: 1 byte
      read_rate:
        offset: 5
        size: 2 bytes
        endian: big
        meaning: "本盘存轮次读取速率"
      total_read:
        offset: 7
        size: 4 bytes
        endian: big
        meaning: "本盘存轮次累计读取次数"
    frame: "A0 0A [addr] 89 [AntID] [ReadRate×2] [TotalRead×4] [check]"

tag_data_frame:
  trigger: "inventory_start 后，读头每识别到一个标签主动推送"
  command: 0x89
  layout:
    - field: header
      offset: 0
      size: 1
      value: 0xA0
    - field: length
      offset: 1
      size: 1
      meaning: "后续字节数"
    - field: address
      offset: 2
      size: 1
    - field: command
      offset: 3
      size: 1
      value: 0x89
    - field: freq_ant
      offset: 4
      size: 1
      bit_field:
        high_6bits: "频点编号 (freq_idx = byte >> 2)"
        low_2bits: "天线号 0-based (antenna = (byte & 0x03) + 1)"
    - field: pc
      offset: 5
      size: 2
      endian: big
      meaning: "EPC C1G2 PC 字，通常 0x0000 或含 EPC 长度信息"
    - field: epc
      offset: 7
      size: "variable"
      length_formula: "epc_bytes = length - 8"
      meaning: "标签 EPC 标识，转 HEX 大写即唯一 ID"
    - field: rssi
      offset: "length"
      size: 1
      meaning: "信号强度 dBm"
      conversion: "unsigned byte → signed: if v > 127 then v - 256"
      example: "0xB0 → 176 → 176-256 = -80 dBm"
    - field: checksum
      offset: "length + 1"
      size: 1

total_frame_size: "length + 2 字节（含 header 和 length 自身）"
min_tag_frame: "21 字节（EPC=12 bytes 时，length=19）"
max_antennas: 16
phase_in_basic_frame: false
phase_note: "基础 TCP 帧不含相位。需要相位请用 SDK DLL 模式（见 §2.3）。"

真实报文摘录

下面是一条真实的标签数据帧，抓自 KLM9700 TCP 连接。你可以用它验证自己的 parser：

A0 13 00 89 42 E2 00 00 34 12 01 2A 1B 2C 3D 45 00 08 B0 18
│  │  │  │  │  │              └── EPC (12 bytes) ──┘ │  │
│  │  │  │  │  └── PC (2 bytes)                      │  └── checksum
│  │  │  │  └── FreqAnt: 0x42                        └── RSSI: 0xB0 = -80 dBm
│  │  │  └── Cmd: 0x89 (标签数据)
│  │  └── Addr: 0x00
│  └── Length: 0x13 = 19（后续 19 字节）
└── Header: 0xA0

解析结果:
  FreqAnt  = 0x42 = 0100_0010b
    → 天线号 = (0x42 & 0x03) + 1 = 2 + 1 = 3 号天线
    → 频点号 = (0x42 >> 2) & 0x3F = 16
  PC       = 0xE200
  EPC      = 003412012A1B2C3D450008
  RSSI     = 0xB0 = 176 → 176 - 256 = -80 dBm
  Checksum = (0xA0+0x13+0x00+0x89+0x42+0xE2+0x00+0x00+0x34+0x12+0x01+0x2A
              +0x1B+0x2C+0x3D+0x45+0x00+0x08+0xB0) & 0xFF
           = 0x218 & 0xFF = 0x18 ✓

TCP 粘包处理

TCP 是流协议，一次 recv 可能收到半帧、一帧、或三帧半粘在一起。处理规则只有两条：

tcp_stream_rules:
  rule_1_find_header: "从 buf[0] 开始扫描，找到 0xA0 才开始解析；之前的字节全部丢弃"
  rule_2_wait_complete: "读 length 字段算出总帧长 (length+2)；buf 不够就等下次 recv 再拼"
  buffer_strategy: "用 bytearray 或 bytes 拼接；每次 parse 成功后切掉已消费的前缀"

统一事件结构

不管数据从哪来，下游只认一个结构：

from dataclasses import dataclass

@dataclass
class TagRead:
    epc: str          # 标签唯一身份（"户口本"里的那一页）
    rssi: float       # dBm，信号强度
    phase: float      # rad，原始值落在 [-π, π]（无则填 0.0）
    antenna: int      # 天线号，多天线场景的关键维度
    reader_id: str
    ts: float         # epoch 秒

这个结构屏蔽了硬件差异。不管后面接的是 KLM9700、其他品牌的读写器、还是离线 CSV 回放，下游的清洗和盘点代码一字不改。

2. 读取层：三种 Reader 实现

怎么选

reader_selection:
  rule: "开发调参 → MockReader | 网口部署只要 EPC+RSSI → KL9700TCPReader | 需要相位 → DLLReader"
  table:
    - reader: MockReader
      when: "离线开发、调算法、写单测"
      hardware: "无"
      phase: false
      platform: "全平台"
    - reader: KL9700TCPReader
      when: "生产部署，KLM9700 走网口"
      hardware: "KLM9700 (TCP)"
      phase: false
      platform: "全平台（纯 Python socket）"
    - reader: DLLReader
      when: "需要相位做运动检测/速度估算"
      hardware: "KLM9700 (串口/USB)"
      phase: true
      platform: "仅 Windows（依赖 RFID_API_ver1.dll）"

三者实现同一个接口：

from typing import Iterator

class ReaderBase:
    """所有读头的统一接口：只暴露 stream()，产出 TagRead"""
    def stream(self) -> Iterator[TagRead]:
        raise NotImplementedError

2.1 MockReader：离线回放

import csv

class MockReader(ReaderBase):
    """回放 CSV，离线可跑。开发和调算法时用这个。"""
    def __init__(self, path):
        self.path = path

    def stream(self):
        with open(self.path, newline="", encoding="utf-8") as f:
            for row in csv.DictReader(f):
                if not row.get("EPC"):        # 空包跳过
                    continue
                yield TagRead(
                    epc=row["EPC"],
                    rssi=float(row["RSSI"]),
                    phase=float(row.get("Phase", 0)),
                    antenna=int(row["Antenna"]),
                    reader_id="mock",
                    ts=float(row["Timestamp"]),
                )

这不是玩具——它是你调清洗参数、验证状态机、写单元测试的地基。真机调试的时延和不确定性不应该出现在算法开发阶段。

2.2 KL9700TCPReader：直连真机

这是核心——把第 1 节的协议规格翻译成可运行的 Python 代码。逐行对应 YAML 里的字段定义。

import socket
import time

class KL9700TCPReader(ReaderBase):
    """
    通过 TCP 直连 KLM9700，解析二进制帧。
    协议规格见 §1 YAML。默认端口 4001，帧头 0xA0。
    """
    FRAME_HEAD = 0xA0
    CMD_INV_DATA = 0x89

    def __init__(self, ip, port=4001, address=0x00, reader_id="klm9700"):
        self.ip = ip
        self.port = port
        self.address = address
        self.reader_id = reader_id
        self._sock = None

    def _connect(self):
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._sock.settimeout(5.0)
        self._sock.connect((self.ip, self.port))

    def _send_start_inv(self):
        """发送盘存启动帧: A0 05 [addr] 89 00 00 [check]"""
        frame = bytes([0xA0, 0x05, self.address, 0x89, 0x00, 0x00])
        frame += bytes([self._checksum(frame)])
        self._sock.sendall(frame)

    def _send_stop_inv(self):
        """发送盘存停止帧: A0 05 [addr] 89 01 00 [check]"""
        frame = bytes([0xA0, 0x05, self.address, 0x89, 0x01, 0x00])
        frame += bytes([self._checksum(frame)])
        self._sock.sendall(frame)

    @staticmethod
    def _checksum(frame: bytes) -> int:
        """校验和：从 header 到最后一个 data 字节累加，取低 8 位"""
        return sum(frame) & 0xFF

    def _parse_tag_frame(self, buf: bytes):
        """
        解析一条标签数据帧。对应 §1 tag_data_frame 规格。
        返回 (TagRead | None, consumed_bytes)
          consumed_bytes = 0 → 数据没收齐，等下次 recv
          consumed_bytes = 1 → 无效字节，跳过
          consumed_bytes = N → 成功解析或跳过非目标帧
        """
        # --- rule_1: 找 header ---
        if len(buf) < 2 or buf[0] != self.FRAME_HEAD:
            return None, 1

        frame_len = buf[1]              # length 字段 = 后续字节数
        total_len = frame_len + 2       # 整帧 = header(1) + length(1) + 后续(frame_len)

        # --- rule_2: 等数据收齐 ---
        if len(buf) < total_len:
            return None, 0

        cmd = buf[3]
        if cmd != self.CMD_INV_DATA:
            return None, total_len      # 非标签数据帧，整帧跳过

        # --- 按 §1 tag_data_frame.layout 逐字段提取 ---
        freq_ant = buf[4]
        antenna = (freq_ant & 0x03) + 1          # 低 2 位 → 天线号 (0-based → 1-based)
        # freq_idx = (freq_ant >> 2) & 0x3F      # 高 6 位 → 频点编号（可选）

        # PC = buf[5:7]                          # 2 字节 PC 字（通常不用于盘点）

        epc_len = frame_len - 8                  # 总长 - 固定开销(header+len+addr+cmd+freqant+pc+rssi+check = 8)
        if epc_len <= 0:
            return None, total_len               # 异常帧，跳过

        epc_bytes = buf[7 : 7 + epc_len]
        epc = epc_bytes.hex().upper()

        rssi_raw = buf[7 + epc_len]
        rssi = float(rssi_raw if rssi_raw < 128 else rssi_raw - 256)

        # 校验（可选但推荐：验证 checksum 防脏数据）
        expected_check = self._checksum(buf[:total_len - 1])
        if buf[total_len - 1] != expected_check:
            return None, total_len               # checksum 不匹配，丢弃

        tag = TagRead(
            epc=epc,
            rssi=rssi,
            phase=0.0,        # TCP 基础帧不含相位，填 0
            antenna=antenna,
            reader_id=self.reader_id,
            ts=time.time(),
        )
        return tag, total_len

    def stream(self):
        self._connect()
        self._send_start_inv()
        buf = b""
        try:
            while True:
                try:
                    chunk = self._sock.recv(4096)
                except socket.timeout:
                    continue
                if not chunk:
                    break
                buf += chunk
                while buf:
                    tag, consumed = self._parse_tag_frame(buf)
                    if consumed == 0:
                        break           # 等更多数据
                    buf = buf[consumed:]
                    if tag is not None:
                        yield tag
        finally:
            self._send_stop_inv()
            self._sock.close()

几个容易踩的坑：

RSSI 是有符号字节。 直接读是 0-255 的无符号数，超过 127 的要减 256 才是真实 dBm。-80 dBm 在帧里是 0xB0（176）。

天线号从 0 开始。 协议里低 2 位是 0-based，人类习惯 1-based，所以 +1。忘了这个，你的天线 1 会变成天线 0，后面 AntennaFilter 对不上。

EPC 长度不是固定的。 虽然常见 12 字节（24 位 HEX），但协议允许变长。用 frame_len - 8 算，不要硬编码 12。

checksum 要验证。 射频环境偶尔会有脏帧，不校验的话假 EPC 会污染你的盘点结果。

2.3 DLLReader：完整数据（含相位）

# pip install pythonnet
# 仅 Windows。需要 RFID_API_ver1.dll 在 sys.path 可达的位置。
import clr
import sys
sys.path.append(r"C:\path\to\RFID_API_ver1")
clr.AddReference("RFID_API_ver1")

from RFID_API_ver1 import Reader, ReaderType, Channels

import threading
import queue
import time

class DLLReader(ReaderBase):
    """
    通过官方 RFID_API_ver1.dll 获取完整标签数据（含相位）。
    SDK 连接方式: Reader.Create(ReaderType, Channels, arg1, arg2)
      TCP:  Reader.Create(ReaderType.TCP,    Channels.One, "192.168.1.100", 4001)
      串口: Reader.Create(ReaderType.SERIAL,  Channels.One, "COM3", 115200)
    回调: reader.TagRead += callback  (每读到一个标签触发)
    """
    def __init__(self, conn_type="serial", port="COM3", baud=115200,
                 ip="", tcp_port=4001, reader_id="klm9700-dll"):
        self.conn_type = conn_type
        self.port = port
        self.baud = baud
        self.ip = ip
        self.tcp_port = tcp_port
        self.reader_id = reader_id
        self._queue = queue.Queue()

    def _on_tag(self, sender, e):
        """SDK 回调：每读到一个标签触发一次"""
        td = e.TagData
        self._queue.put(TagRead(
            epc=td.EPC.hex().upper(),
            rssi=float(td.Rssi),
            phase=float(td.Phase),
            antenna=int(td.Antenna),
            reader_id=self.reader_id,
            ts=time.time(),
        ))

    def stream(self):
        if self.conn_type == "tcp":
            reader = Reader.Create(
                ReaderType.TCP, Channels.One,
                self.ip, self.tcp_port
            )
        else:
            reader = Reader.Create(
                ReaderType.SERIAL, Channels.One,
                self.port, self.baud
            )
        reader.TagRead += self._on_tag
        reader.Connect()
        try:
            while True:
                tag = self._queue.get(timeout=1.0)
                yield tag
        finally:
            reader.Disconnect()

3. 原始数据解剖

一个 TagRead 里有 6 个字段，但每个都在不同环节被用：

tagread_fields:
  - field: epc
    type: str
    meaning: "标签唯一身份"
    dirty: "偶发空值（硬件丢包）"
    used_in: "清洗（丢弃空值）"
  - field: rssi
    type: float
    unit: dBm
    meaning: "信号强度，负值，越小越弱"
    dirty: "多径导致野值（如突然 -78）"
    used_in: "清洗（中位数抗噪）+ 置信度计算"
  - field: phase
    type: float
    unit: rad
    range: "[-π, π]"
    meaning: "载波相位"
    dirty: "回绕跳变（π → -π 不连续）"
    used_in: "清洗（解缠）+ 运动检测"
    note: "TCP 基础帧不含此字段，恒为 0.0"
  - field: antenna
    type: int
    meaning: "天线号，1-based"
    dirty: "无"
    used_in: "去重维度 + 串读过滤 + 方向判定"
  - field: reader_id
    type: str
    meaning: "读头标识"
    dirty: "无"
    used_in: "多读头融合"
  - field: ts
    type: float
    unit: "epoch seconds"
    meaning: "读取时间戳"
    dirty: "无"
    used_in: "滑窗、状态机、超时判定"

关键认知：RSSI 是 dBm，是「越小越弱」的负值；而 Phase 原始值在 [-π, π] 循环，用它之前必须先解缠，否则一切基于相位差的判断都是错的。如果你用的是 TCPReader（基础帧不含 Phase），那 phase 字段恒为 0.0，运动检测功能不可用，但盘点功能不受影响。

4. 清洗：去重、抗噪、解缠

4.1 滑动窗口去重（按 EPC × 天线）

同一个标签 60ms 内读两次是浪费，去重窗口要可配，且按 (EPC, 天线) 维度——因为同一标签从不同天线读到，是两条有意义的信息，不能互相吃掉。

import math
from statistics import median

class Cleaner:
    def __init__(self, dedup_window=0.3):
        self.dedup_window = dedup_window          # 秒，可配
        self.state = {}                           # (epc, ant) -> 解缠状态

    def _unwrap(self, key, raw):
        """把 [-π,π] 的回绕相位展开成连续相位"""
        st = self.state.setdefault(key, {})
        if "phase" not in st:
            st["phase"], st["raw"] = raw, raw
            return raw
        d = raw - st["raw"]
        while d >  math.pi: d -= 2 * math.pi      # 回绕修正
        while d < -math.pi: d += 2 * math.pi
        st["phase"] += d
        st["raw"] = raw
        return st["phase"]

    def clean(self, reads):
        """输入一批原始读，输出去重后的干净事件"""
        out, buf = [], {}
        for r in reads:
            if not r.epc:                          # 空包直接丢
                continue
            key = (r.epc, r.antenna)
            st = self.state.setdefault(key, {})
            last = st.get("ts")
            if last is None or r.ts - last > self.dedup_window:
                if key in buf:                     # 旧窗口先落盘
                    out.append(self._flush(key, buf.pop(key)))
                st["ts"] = r.ts
                buf[key] = [r]
            else:
                buf[key].append(r)
        for key, rs in buf.items():
            out.append(self._flush(key, rs))
        return sorted(out, key=lambda x: x.ts)

    def _flush(self, key, rs):
        r0 = rs[0]
        return TagRead(
            epc=r0.epc, antenna=r0.antenna, reader_id=r0.reader_id,
            rssi=median(r.rssi for r in rs),           # 中位数抗异常
            phase=self._unwrap(key, rs[-1].phase),     # 解缠后的相位
            ts=r0.ts,
        )

两个反直觉点：

RSSI 用中位数而不是滑动平均。dBm 里偶尔会插进一条 -78 的多径野值，平均会被它拖偏；中位数直接无视它。

去重窗口长度要可配。默认 300ms 只是经验值——KLM9700 的采样率可以通过 RF 链路配置调整，硬编码 500ms 是给自己挖坑。

4.2 串读治理：按天线过滤

第 0 节那条 -78.4 的串读，根因是多径反射——射频信号在金属环境中弹来弹去，隔壁天线的信号"漏"进了当前天线。

最直接的对策是 RSSI 阈值过滤，但这不够——有些串读的 RSSI 并不低。更稳的做法是按盘点目标定义天线的有效集合，非目标天线的读数直接丢弃：

class AntennaFilter:
    """只保留目标天线的读数，过滤串读"""
    def __init__(self, active_ants: set, rssi_floor=-80.0):
        self.active_ants = active_ants    # 例如 {1, 2} 表示只用天线 1 和 2
        self.rssi_floor = rssi_floor      # RSSI 低于此值的也丢

    def filter(self, tag: TagRead) -> bool:
        return tag.antenna in self.active_ants and tag.rssi >= self.rssi_floor

在 KLM9700 的实际部署中，一台设备最多支持 16 根天线。货架盘点场景通常只用 2-4 根，通道门场景用 2 根（一前一后）。把不用的天线排除在 active_ants 之外，串读就从源头被拦住了。

5. 盘点：状态机与「假离场」治理（★核心）

到这一步，我们有一串干净的事件流。但盘点要的不是事件，是状态：这个标签现在在不在。

天真做法：超时没读到 → 判离场。这是整条管线最大的坑——因为漏读，标签明明在货架上，读头两秒没扫到，你就误判它"离场"了。这叫假离场。

正确做法：连续漏读计数 + 滞回（hysteresis）。

状态机定义

inventory_state_machine:
  states: [ABSENT, PRESENT]
  initial: ABSENT

  transitions:
    - from: ABSENT
      to: PRESENT
      trigger: "连续 enter_hits 个窗口均检测到该 EPC"
      guard: "hits >= enter_hits"
      description: "进场确认——挡住偶发单读的幽灵标签"

    - from: PRESENT
      to: ABSENT
      trigger: "连续 exit_misses 个窗口均未检测到该 EPC"
      guard: "misses >= exit_misses"
      description: "离场确认——挡住漏读导致的假离场"

  per_window_logic:
    on_detect:
      - "hits += 1"
      - "misses = 0"
      - "rssi 列表追加（保留最近 5 个）"
      - "count += 本窗口读取次数"
      - "检查 ABSENT→PRESENT 转换条件"
    on_miss:
      - "hits = 0"
      - "misses += 1"
      - "检查 PRESENT→ABSENT 转换条件"

  confidence:
    formula: "min(1.0, count/10) * exp(-spread/20)"
    where:
      count: "累计有效读取次数"
      spread: "max(rssi) - min(rssi)，最近 5 次"
    interpretation: "读得越多越可信，RSSI 抖动越小越可信"

误判治理对照

error_correction:
  - symptom: "假离场（标签在但判 ABSENT）"
    root_cause: "漏读（射频盲区、标签遮挡）"
    fix: "exit_misses 滞回——连续 N 个窗口没读到才判离场"
  - symptom: "假在场（标签不在但判 PRESENT）"
    root_cause: "串读、幽灵标签"
    fix: "enter_hits 进场门槛 + AntennaFilter + 置信度阈值"
  - symptom: "RSSI 漂移"
    root_cause: "多径、人体遮挡"
    fix: "天线分集——多天线对同一 EPC 取中位数"

实现

class InventoryManager:
    def __init__(self, window=1.0, enter_hits=2, exit_misses=3):
        self.window = window
        self.enter_hits = enter_hits        # 滞回：进场门槛
        self.exit_misses = exit_misses      # 滞回：离场门槛
        self.tags = {}                      # epc -> 状态
        self._win = None
        self._agg = {}                      # 当前窗口聚合

    def update(self, reads, now):
        wid = int(now // self.window)
        if self._win is None:
            self._win = wid
        while self._win < wid:              # 时间推进 → 结算历史窗口
            self._advance()
            self._win += 1
        for r in reads:
            self._agg.setdefault(r.epc, []).append(r.rssi)

    def _advance(self):
        # 新标签先建档（默认 ABSENT，靠连续命中转正）
        for epc in self._agg:
            self.tags.setdefault(
                epc, {"state": "ABSENT", "hits": 0, "misses": 0,
                      "rssi": [], "count": 0})
        for epc, st in self.tags.items():
            if epc in self._agg:
                st["hits"] += 1
                st["misses"] = 0
                st["rssi"] = (st["rssi"] + self._agg[epc])[-5:]   # 仅留最近5次
                st["count"] += len(self._agg[epc])
                if st["state"] == "ABSENT" and st["hits"] >= self.enter_hits:
                    st["state"] = "PRESENT"                       # → 进场
            else:
                st["hits"] = 0
                st["misses"] += 1
                if st["state"] == "PRESENT" and st["misses"] >= self.exit_misses:
                    st["state"] = "ABSENT"                        # → 离场
        self._agg = {}

    def snapshot(self):
        return [{
            "epc": epc,
            "present": st["state"] == "PRESENT",
            "reads": st["count"],
            "rssi_median": median(st["rssi"]) if st["rssi"] else None,
            "confidence": self._confidence(st),
        } for epc, st in self.tags.items()]

    @staticmethod
    def _confidence(st):
        if not st["rssi"]:
            return 0.0
        base = min(1.0, st["count"] / 10)                       # 读得越多越可信
        spread = max(st["rssi"]) - min(st["rssi"])              # 抖动越小越可信
        return round(base * math.exp(-spread / 20), 2)

进阶（各点到为止）：

通道门方向判定——两个天线一前一后安装，按 RSSI 上升沿/下降沿的先后顺序推出"进"还是"出"，适合出入库口。

相位检测运动——静止标签相位稳定、运动标签相位持续变化。前提是第 4 节已把相位解缠，否则回绕会伪装成"剧烈运动"。这需要 DLLReader 模式（基础 TCP 帧不含 Phase）。

6. 从快照到报告：置信度与验证

盘点结果不能只有「在/不在」，得带置信度（见上 _confidence）。

怎么验证准不准？最实的方法不是找算法 ground truth，而是双次人工盘点取交集：人工盘两遍，只在两遍都出现的标签认为"真在场"，用它当基准，去比对你的管线输出，算漏报（明明在、你说不在）和误报（明明不在、你说在）。

def verify(snapshot, ground_truth):
    pred = {t["epc"] for t in snapshot if t["present"] and t["confidence"] >= 0.6}
    gt = set(ground_truth)
    miss = gt - pred          # 漏报：真在场却判离场
    false = pred - gt         # 误报：判在场其实不在
    acc = len(pred & gt) / len(gt | pred) if (gt | pred) else 1.0
    return {"accuracy": round(acc, 3), "missed": sorted(miss), "false": sorted(false)}

调参靶子也在这：enter_hits 调高 → 误报降、漏报升；exit_misses 调高 → 假离场降、但响应变慢。用上面的 missed/false 做 A/B，别凭感觉拍数字。

7. 延伸：上报、标注与边缘部署

7.1 上报

快照 → MQTT，主题 rfid/inventory/{reader_id}，下行给 WMS/ERP。只上报状态变化（PRESENT→ABSENT 或 ABSENT→PRESENT），不上报每次读取——省带宽也省云成本。

# 概念示意，不依赖特定 MQTT 库
def publish_changes(prev_snapshot, curr_snapshot):
    prev = {t["epc"]: t["present"] for t in prev_snapshot}
    for t in curr_snapshot:
        was = prev.get(t["epc"], False)
        if was != t["present"]:
            event = "enter" if t["present"] else "leave"
            # mqtt_publish(f"rfid/inventory/{reader_id}/{t['epc']}",
            #              {"event": event, "ts": time.time(), "confidence": t["confidence"]})

7.2 自动标注

把盘点快照与摄像头帧按时间戳对齐（ts 就近匹配），给画面里的人/物打上"持有哪些 EPC"的标签。听着性感，但涉及视觉-射频时空对齐、遮挡处理，是个独立课题，这里只点到。

7.3 边缘部署

KLM9700 本身是固定式读写器，跑的是嵌入式 Linux。如果你的部署环境有边缘网关（比如一台小工控机），可以把清洗 + 状态机跑在网关上，读头只负责原始数据上报。这样即使网络断了，本地盘点状态依然准确——网络恢复后只需要同步状态变化。

8. 完整管线：串起来跑

rfid_pipeline/
├── models.py        # TagRead 事件结构
├── readers.py       # MockReader / KL9700TCPReader / DLLReader
├── cleaner.py       # 去重 + 中位数抗噪 + 相位解缠 + 天线过滤
├── inventory.py     # InventoryManager 状态机（大脑）
├── report.py        # confidence + verify
├── data/sample.csv  # 第 0 节那批烂数据
└── main.py          # 串起全流程

main.py 的核心循环：

import time

def main():
    # ---- 选择 Reader（见 §2 决策表）----
    # 离线开发：
    reader = MockReader("data/sample.csv")
    # 真机部署（取消注释）：
    # reader = KL9700TCPReader("192.168.1.100", port=4001)

    cleaner = Cleaner(dedup_window=0.3)
    ant_filter = AntennaFilter(active_ants={1, 2}, rssi_floor=-75.0)
    inv = InventoryManager(window=1.0, enter_hits=2, exit_misses=3)

    batch, batch_start = [], time.time()
    BATCH_INTERVAL = 0.5  # 每 0.5 秒清洗一批

    for tag in reader.stream():
        if not ant_filter.filter(tag):
            continue
        batch.append(tag)

        now = time.time()
        if now - batch_start >= BATCH_INTERVAL:
            clean_reads = cleaner.clean(batch)
            inv.update(clean_reads, now)
            batch.clear()
            batch_start = now

        # 每 5 秒打印一次快照
        if int(time.time()) % 5 == 0:
            snap = inv.snapshot()
            for t in snap:
                status = "●" if t["present"] else "○"
                print(f"  {status} {t['epc'][-8:]}  "
                      f"reads={t['reads']}  "
                      f"rssi={t['rssi_median']}  "
                      f"conf={t['confidence']}")

if __name__ == "__main__":
    main()

python main.py 就能跑。先用 MockReader + sample.csv 验证管线逻辑，确认无误后把 Reader 换成 KL9700TCPReader，填上你设备的 IP，就能对接真实硬件。其余代码一字不改。

9. KLM9700 调参指南

别拍脑袋定参数。以下是 KLM9700 在典型场景下的经验值，以及调参方法：

tuning_guide:
  hardware: KLM9700 (Impinj E710)
  max_antennas: 16
  default_rf_profile: "Profile 7 (PR-ASK, Tari 20μs, Miller-4 250kHz)"
  power_range_dBm: "10 ~ 33"

  parameters:
    - name: dedup_window
      default: "0.3s"
      shelf_inventory: "0.3s"
      gate: "0.1s"
      how_to_tune: "看同一标签两次读取的最小间隔"

    - name: window
      default: "1.0s"
      shelf_inventory: "1.0s"
      gate: "0.5s"
      how_to_tune: "标签通过天线的典型时长"

    - name: enter_hits
      default: 2
      shelf_inventory: 2
      gate: 3
      how_to_tune: "调高 → 误报降、漏报升"

    - name: exit_misses
      default: 3
      shelf_inventory: 3
      gate: 2
      how_to_tune: "调高 → 假离场降、响应变慢"

    - name: rssi_floor
      default: "-75 dBm"
      shelf_inventory: "-75 dBm"
      gate: "-70 dBm"
      how_to_tune: "在部署环境空跑 10 秒，看噪声分布"

  tuning_method:
    step_1: "在目标场景放已知数量的标签（比如 20 个）"
    step_2: "跑管线，用 verify() 函数算漏报和误报"
    step_3: "enter_hits × exit_misses 的每组组合跑一遍"
    step_4: "选准确率和响应速度的平衡点"

*本系列下一篇：把这条管线接入真实业务系统——从 MQTT 上报到 WMS 联动。*