位置事件 + 物品注册表：从"标签在哪"到"库存可视化"

# 位置事件 + 物品注册表：从"标签在哪"到"库存可视化"

承接上篇：我们把多个读头的事件融合成了位置事件——"这个标签从 aisle-1 移动到了 shelf-a"。

但 WMS 不关心"标签在哪"，它关心"SKU-001 还有多少件，在哪个库位"。

标签是物理层的概念。SKU 是业务层的概念。中间需要一个<strong>物品注册表</strong>：把 EPC 映射到 SKU，把位置映射到库位。

这篇讲怎么把位置事件变成库存可视化——从物品注册表设计到实时库存查询接口。

0. 先看一段真实的断层

这是位置事件流（上篇的输出）：

<pre><code>[15:30:45] LocationEvent: epc=E2003412012A1B2C3D4, type=move, from=aisle-1, to=shelf-a [15:30:46] LocationEvent: epc=E2003412012A1B2C3D5, type=enter_zone, zone=inbound-zone [15:30:47] LocationEvent: epc=E2003412012A1B2C3D6, type=leave_zone, zone=shelf-b</code></pre>

这是 WMS 需要的查询：

<pre><code>GET /api/v1/inventory?sku=ITEM-001 { "sku": "ITEM-001", "total_qty": 42, "locations": [ {"zone": "shelf-a", "qty": 30}, {"zone": "inbound-zone", "qty": 12} ] }</code></pre>

看出断层了吗？

<strong>语义鸿沟：</strong> 位置事件说的是 EPC（E2003412012A1B2C3D4），WMS 问的是 SKU（ITEM-001）。中间的映射关系在哪？

<strong>聚合缺失：</strong> WMS 要的是"ITEM-001 在 shelf-a 有 30 件"，不是"这 30 个 EPC 在 shelf-a"。需要按 SKU 聚合。

<strong>状态不一致：</strong> 位置事件是"标签在移动"，库存是"物品在库位"。标签从 shelf-a 移到 aisle-1，库存要从 shelf-a 减 1、 aisle-1 加 1。这是状态机，不是事件流。

我们需要一个<strong>物品注册表</strong>：维护 EPC → SKU 的映射，维护 SKU → 库位 → 数量的聚合状态。

1. 物品注册表：从 EPC 到 SKU

人话版

每个 RFID 标签有个唯一的 EPC。每件商品有个唯一的 SKU。一个 SKU 可能对应多个 EPC（同一款商品有多件）。

物品注册表就是这张映射表：EPC → SKU → 业务属性（名称、规格、批次、保质期）。

注册表结构

<pre><code class="lang-yaml"># 物品注册表 # EPC → SKU → 业务属性 item_registry: # EPC 级别的记录（物理层） items: - epc: "E2003412012A1B2C3D4" sku: "ITEM-001" status: "active" # active | inactive | decommissioned registered_at: "2026-09-01T10:00:00+08:00" metadata: batch: "B20260901" production_date: "2026-08-15" expiry_date: "2027-08-15" unit: "件" - epc: "E2003412012A1B2C3D5" sku: "ITEM-001" status: "active" registered_at: "2026-09-01T10:00:00+08:00" metadata: batch: "B20260901" production_date: "2026-08-15" expiry_date: "2027-08-15" unit: "件" - epc: "E2003412012A1B2C3D6" sku: "ITEM-002" status: "active" registered_at: "2026-09-02T14:30:00+08:00" metadata: batch: "B20260902" production_date: "2026-08-20" expiry_date: "2027-02-20" unit: "件" # SKU 级别的汇总（业务层） sku_summary: ITEM-001: name: "智能RFID标签 KLM9200" category: "电子产品" total_registered: 100 total_active: 98 total_inactive: 2 ITEM-002: name: "UHF读写器 KLM9700" category: "设备" total_registered: 50 total_active: 50 total_inactive: 0</code></pre>

一个坑：

<strong>EPC 必须全局唯一。</strong> 如果两个标签有相同的 EPC，注册表会混乱。Impinj E710 的 EPC 是 96 位，理论上不会重复，但生产线可能出错。注册时要校验 EPC 格式，发现重复要告警。

2. 库存状态机：从位置事件到库存变化

人话版

位置事件说"标签从 A 移到 B"。库存状态机说"SKU-001 在 A 库位减 1，在 B 库位加 1"。

不是每个位置事件都触发库存变化。标签从 shelf-a 移到 aisle-1，可能是"在货架上调整位置"，也可能是"被拿走准备出库"。需要业务规则判断。

库存状态机

<pre><code class="lang-yaml"># 库存状态机 # 输入：位置事件流 (epc, event_type, from_zone, to_zone) # 输出：库存变化事件 (sku, zone, delta) inventory_state_machine: name: "InventoryTracker" states: - IN_STOCK: "在库（某个存储区域）" - IN_TRANSIT: "在途（通道区域）" - PENDING_OUT: "待出库（出库口）" - OUT_OF_STOCK: "已出库" transitions: # 入库 - from: OUT_OF_STOCK to: IN_STOCK trigger: "enter_zone where zone.type = 'storage'" action: "emit INVENTORY_IN (sku, zone, +1)" # 库间移动 - from: IN_STOCK to: IN_STOCK trigger: "move from zone_a to zone_b where both.type = 'storage'" action: "emit INVENTORY_MOVE (sku, from_zone, to_zone)" # 准备出库 - from: IN_STOCK to: PENDING_OUT trigger: "move to zone where zone.type = 'gate' && zone.direction = 'outbound'" action: "emit INVENTORY_RESERVED (sku, zone, -1 pending)" # 确认出库 - from: PENDING_OUT to: OUT_OF_STOCK trigger: "leave_zone where zone.type = 'gate'" action: "emit INVENTORY_OUT (sku, zone, -1 confirmed)" # 取消出库（又移回存储区） - from: PENDING_OUT to: IN_STOCK trigger: "move to zone where zone.type = 'storage'" action: "emit INVENTORY_UNRESERVE (sku, zone, +1)" business_rules: - "入库口读到 → 自动关联入库单（如果有预通知）" - "出库口读到 → 校验出库单（防止错发）" - "库间移动 → 更新库位，不影响总库存"</code></pre>

实现

<pre><code class="lang-python">from dataclasses import dataclass from typing import Dict, Optional from enum import Enum class ItemStatus(Enum): OUT_OF_STOCK = "out_of_stock" IN_STOCK = "in_stock" IN_TRANSIT = "in_transit" PENDING_OUT = "pending_out" @dataclass class InventoryEvent: epc: str sku: str event_type: str # "inventory_in" | "inventory_out" | "inventory_move" | "inventory_reserved" from_zone: Optional[str] to_zone: Optional[str] ts: float class InventoryTracker: def __init__(self, item_registry, zone_config): self.item_registry = item_registry # EPC -> SKU 映射 self.zone_config = zone_config # zone -> type/direction # 当前库存状态：epc -> {status, zone, sku} self.item_status = {} # 聚合库存：sku -> {zone -> qty} self.inventory_by_sku = {} def feed(self, location_event) -> Optional[InventoryEvent]: """输入位置事件，输出库存事件""" epc = location_event.epc event_type = location_event.event_type # 查询 EPC 对应的 SKU sku = self.item_registry.get(epc) if not sku: return None # 未注册的 EPC current = self.item_status.get(epc, { "status": ItemStatus.OUT_OF_STOCK, "zone": None }) current_status = current["status"] current_zone = current["zone"] # 根据位置事件和当前状态，决定库存变化 if event_type == "enter_zone": zone = location_event.to_zone zone_type = self.zone_config[zone]["type"] if zone_type == "storage" and current_status == ItemStatus.OUT_OF_STOCK: # 入库 self._update_inventory(epc, sku, ItemStatus.IN_STOCK, zone) return InventoryEvent(epc, sku, "inventory_in", None, zone, location_event.ts) elif zone_type == "gate" and self.zone_config[zone].get("direction") == "outbound": # 准备出库 self._update_inventory(epc, sku, ItemStatus.PENDING_OUT, zone) return InventoryEvent(epc, sku, "inventory_reserved", current_zone, zone, location_event.ts) elif event_type == "move": from_zone = location_event.from_zone to_zone = location_event.to_zone from_type = self.zone_config[from_zone]["type"] to_type = self.zone_config[to_zone]["type"] if from_type == "storage" and to_type == "storage": # 库间移动 self._update_inventory(epc, sku, ItemStatus.IN_STOCK, to_zone) return InventoryEvent(epc, sku, "inventory_move", from_zone, to_zone, location_event.ts) elif from_type == "storage" and to_type == "gate": # 移向出库口 self._update_inventory(epc, sku, ItemStatus.PENDING_OUT, to_zone) return InventoryEvent(epc, sku, "inventory_reserved", from_zone, to_zone, location_event.ts) elif event_type == "leave_zone": zone = location_event.from_zone zone_type = self.zone_config[zone]["type"] if zone_type == "gate" and current_status == ItemStatus.PENDING_OUT: # 确认出库 self._update_inventory(epc, sku, ItemStatus.OUT_OF_STOCK, None) return InventoryEvent(epc, sku, "inventory_out", zone, None, location_event.ts) return None def _update_inventory(self, epc: str, sku: str, status: ItemStatus, zone: Optional[str]): """更新单品状态和聚合库存""" old = self.item_status.get(epc, {"status": ItemStatus.OUT_OF_STOCK, "zone": None}) old_zone = old["zone"] # 更新单品状态 self.item_status[epc] = {"status": status, "zone": zone} # 更新聚合库存 if sku not in self.inventory_by_sku: self.inventory_by_sku[sku] = {} # 从旧库位减 if old_zone and old_zone in self.inventory_by_sku[sku]: self.inventory_by_sku[sku][old_zone] -= 1 if self.inventory_by_sku[sku][old_zone] == 0: del self.inventory_by_sku[sku][old_zone] # 向新库位加 if zone: self.inventory_by_sku[sku][zone] = self.inventory_by_sku[sku].get(zone, 0) + 1 def query_inventory(self, sku: str) -> dict: """查询某个 SKU 的库存分布""" locations = self.inventory_by_sku.get(sku, {}) total = sum(locations.values()) return { "sku": sku, "total_qty": total, "locations": [ {"zone": zone, "qty": qty} for zone, qty in locations.items() ] }</code></pre>

一个坑：

<strong>库间移动必须原子化。</strong> 不能先减后加，中间有个"库存为 0"的瞬间。如果这时候有查询进来，会看到"缺货"。用事务或者先加后减。

3. 库存查询接口：给 WMS 用的 API

人话版

WMS 不关心"标签在哪"，它关心"SKU-001 还有多少件，在哪个库位"。

查询接口要快（毫秒级），要准（和物理库存一致），要支持批量查询（盘点时用）。

接口协议

<pre><code class="lang-yaml"># 库存查询接口 # REST API，JSON over HTTPS endpoints: query_sku: method: GET path: /api/v1/inventory/{sku} description: "查询单个 SKU 的库存分布" response: 200: sku: "ITEM-001" total_qty: 42 last_update: "2026-09-22T15:30:45.123+08:00" locations: - zone: "shelf-a" qty: 30 last_change: "2026-09-22T15:30:45.123+08:00" - zone: "inbound-zone" qty: 12 last_change: "2026-09-22T15:25:00.000+08:00" 404: error: "SKU not found" query_zone: method: GET path: /api/v1/inventory/zone/{zone} description: "查询某个库位的所有 SKU" response: 200: zone: "shelf-a" items: - sku: "ITEM-001" qty: 30 - sku: "ITEM-002" qty: 15 total_skus: 2 total_qty: 45 batch_query: method: POST path: /api/v1/inventory/batch description: "批量查询多个 SKU" request_body: skus: ["ITEM-001", "ITEM-002", "ITEM-003"] response: 200: results: - sku: "ITEM-001" total_qty: 42 locations: [...] - sku: "ITEM-002" total_qty: 15 locations: [...] not_found: ["ITEM-003"] query_item: method: GET path: /api/v1/inventory/item/{epc} description: "查询单个 EPC 的状态" response: 200: epc: "E2003412012A1B2C3D4" sku: "ITEM-001" status: "in_stock" zone: "shelf-a" last_update: "2026-09-22T15:30:45.123+08:00" metadata: batch: "B20260901" production_date: "2026-08-15" expiry_date: "2027-08-15"</code></pre>

实现

<pre><code class="lang-python">from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import time app = FastAPI() class BatchQueryRequest(BaseModel): skus: List[str] class InventoryAPI: def __init__(self, inventory_tracker): self.tracker = inventory_tracker def query_sku(self, sku: str) -> dict: """查询单个 SKU""" result = self.tracker.query_inventory(sku) if result["total_qty"] == 0 and sku not in self.tracker.inventory_by_sku: raise HTTPException(status_code=404, detail="SKU not found") # 添加最后更新时间 result["last_update"] = self._get_last_update(sku) # 添加每个库位的最后变化时间 for loc in result["locations"]: loc["last_change"] = self._get_zone_last_update(sku, loc["zone"]) return result def query_zone(self, zone: str) -> dict: """查询某个库位""" items = [] for sku, zones in self.tracker.inventory_by_sku.items(): if zone in zones: items.append({ "sku": sku, "qty": zones[zone] }) return { "zone": zone, "items": items, "total_skus": len(items), "total_qty": sum(item["qty"] for item in items) } def batch_query(self, skus: List[str]) -> dict: """批量查询""" results = [] not_found = [] for sku in skus: try: result = self.query_sku(sku) results.append(result) except HTTPException: not_found.append(sku) return { "results": results, "not_found": not_found } def query_item(self, epc: str) -> dict: """查询单个 EPC""" if epc not in self.tracker.item_status: raise HTTPException(status_code=404, detail="EPC not found") status = self.tracker.item_status[epc] sku = self.tracker.item_registry.get(epc) return { "epc": epc, "sku": sku, "status": status["status"].value, "zone": status["zone"], "last_update": self._get_epc_last_update(epc), "metadata": self.tracker.item_registry.get_metadata(epc) } # 路由 api = InventoryAPI(inventory_tracker) @app.get("/api/v1/inventory/{sku}") async def get_inventory(sku: str): return api.query_sku(sku) @app.get("/api/v1/inventory/zone/{zone}") async def get_zone_inventory(zone: str): return api.query_zone(zone) @app.post("/api/v1/inventory/batch") async def batch_query(request: BatchQueryRequest): return api.batch_query(request.skus) @app.get("/api/v1/inventory/item/{epc}") async def get_item(epc: str): return api.query_item(epc)</code></pre>

4. 库存可视化：给人看的界面

人话版

API 是给机器用的。人需要看图表、看库位图、看趋势。

库存可视化不是"把数字显示出来"，而是"让人一眼看出问题"——哪个库位空了，哪个 SKU 快过期了，哪个通道堵了。

可视化组件

<pre><code class="lang-yaml"># 库存可视化组件 # 不是"显示数字"，是"让人一眼看出问题" dashboard_components: - name: "库位热力图" description: "仓库平面图，每个库位用颜色表示占用率" data_source: "query_zone for all zones" visualization: type: "heatmap" color_scale: - value: 0 color: "#e0e0e0" # 灰色，空 - value: 0.5 color: "#4caf50" # 绿色，半满 - value: 1.0 color: "#f44336" # 红色，满 alerts: - condition: "zone.occupancy > 0.9" message: "{zone} 即将满仓" - condition: "zone.occupancy == 0" message: "{zone} 空闲超过 24 小时" - name: "SKU 库存趋势" description: "某个 SKU 过去 7 天的库存变化" data_source: "inventory event log" visualization: type: "line_chart" x_axis: "time (7 days)" y_axis: "quantity" series: - name: "shelf-a" color: "#2196f3" - name: "shelf-b" color: "#9c27b0" alerts: - condition: "qty < safety_stock" message: "{sku} 低于安全库存" - name: "即将过期" description: "按过期日期排序的 SKU 列表" data_source: "item_registry.metadata.expiry_date" visualization: type: "table" columns: - "SKU" - "批次" - "过期日期" - "剩余天数" - "当前库存" sort: "expiry_date ASC" alerts: - condition: "days_remaining < 30" severity: "warning" - condition: "days_remaining < 7" severity: "critical" - name: "异常移动" description: "非预期的位置变化（如出库口读到但未关联出库单）" data_source: "inventory event log + business rules" visualization: type: "timeline" events: - type: "unauthorized_move" color: "#ff9800" - type: "missing_checkout" color: "#f44336" alerts: - condition: "event.type == 'unauthorized_move'" message: "立即检查"</code></pre>

5. 完整数据流：从读头到看板

<pre><code>KLM9700 读头 ↓ (TCP 解析) TagRead 流 ↓ (事件抽象) 业务事件 (enter/leave per reader) ↓ (多读头融合) 位置事件 (enter_zone/leave_zone/move) ↓ (物品注册表) 库存事件 (inventory_in/out/move) ↓ (聚合) 库存状态 (sku -> zone -> qty) ↓ (API) 查询接口 ↓ (可视化) 看板</code></pre>

每一层的输入输出都明确，每一层都可以独立测试。

这就是工程化的终点：不是"一个脚本从读头直连看板"，而是分层解耦，每层可替换，每层可测试。

本系列总结

从 KLM9700 的二进制帧，到看板的库存数字，我们走了 7 篇：

1. <strong>协议解析</strong>：把私有二进制变成干净的 TagRead<br>2. <strong>盘点管线</strong>：从 TagRead 到盘点快照<br>3. <strong>事件抽象</strong>：从盘点快照到 enter/leave 事件<br>4. <strong>MQTT 上报</strong>：把事件推给业务系统<br>5. <strong>多读头融合</strong>：从"谁在场"到"在哪里"<br>6. <strong>库存可视化</strong>：从"标签在哪"到"SKU 有多少"

每一层都是独立的，可以单独替换。不喜欢 MQTT？换成 Kafka。不喜欢 SQLite？换成 PostgreSQL。不喜欢热力图？换成 3D 可视化。

这就是协议的价值：把复杂系统拆成可组合的模块，每个模块都可以独立演进。