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

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

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

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

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

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

0. 先看一段真实的断层

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

[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

这是 WMS 需要的查询：

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

看出断层了吗？

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

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

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

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

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

人话版

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

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

注册表结构

# 物品注册表
# 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

一个坑：

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

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

人话版

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

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

库存状态机

# 库存状态机
# 输入：位置事件流 (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:
    - "入库口读到 → 自动关联入库单（如果有预通知）"
    - "出库口读到 → 校验出库单（防止错发）"
    - "库间移动 → 更新库位，不影响总库存"

实现

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()
            ]
        }

一个坑：

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

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

人话版

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

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

接口协议

# 库存查询接口
# 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"

实现

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)

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

人话版

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

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

可视化组件

# 库存可视化组件
# 不是"显示数字"，是"让人一眼看出问题"

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: "立即检查"

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

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

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

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

本系列总结

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

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

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

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