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

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

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

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

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

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

<pre><code class="lang-csv">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</code></pre>

看出问题了吗？

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

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

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

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

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

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

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

人话版

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

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

协议完整规格

<pre><code class="lang-yaml"># 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）。"</code></pre>

真实报文摘录

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

<pre><code>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 ✓</code></pre>

TCP 粘包处理

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

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

统一事件结构

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

<pre><code class="lang-python">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 秒</code></pre>

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

2. 读取层：三种 Reader 实现

怎么选

<pre><code class="lang-yaml">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）"</code></pre>

三者实现同一个接口：

<pre><code class="lang-python">from typing import Iterator class ReaderBase: """所有读头的统一接口：只暴露 stream()，产出 TagRead""" def stream(self) -> Iterator[TagRead]: raise NotImplementedError</code></pre>

2.1 MockReader：离线回放

<pre><code class="lang-python">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"]), )</code></pre>

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

2.2 KL9700TCPReader：直连真机

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

<pre><code class="lang-python">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()</code></pre>

几个容易踩的坑：

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

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

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

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

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

<pre><code class="lang-python"># 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()</code></pre>

3. 原始数据解剖

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

<pre><code class="lang-yaml">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: "滑窗、状态机、超时判定"</code></pre>

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

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

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

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

<pre><code class="lang-python">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, )</code></pre>

两个反直觉点：

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

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

4.2 串读治理：按天线过滤

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

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

<pre><code class="lang-python">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</code></pre>

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

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

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

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

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

状态机定义

<pre><code class="lang-yaml">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 抖动越小越可信"</code></pre>

误判治理对照

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

实现

<pre><code class="lang-python">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)</code></pre>

进阶（各点到为止）：

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

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

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

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

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

<pre><code class="lang-python">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)}</code></pre>

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

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

7.1 上报

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

<pre><code class="lang-python"># 概念示意，不依赖特定 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"]})</code></pre>

7.2 自动标注

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

7.3 边缘部署

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

8. 完整管线：串起来跑

<pre><code>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 # 串起全流程</code></pre>

<code>main.py</code> 的核心循环：

<pre><code class="lang-python">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()</code></pre>

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

9. KLM9700 调参指南

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

<pre><code class="lang-yaml">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: "选准确率和响应速度的平衡点"</code></pre>

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