做售货柜视觉识别项目时,最核心的一条流水线就是 IPC 拉流、抽帧、YOLO 识别。我刚接手这类项目的时候,以为难点全在模型识别率上,结果真正卡住进度的,全是拉流不稳定、抽帧节奏不对、队列把内存打爆这些小问题。这篇文章把我从零搭到稳定运行的完整链路写出来,包括 IPC 选型、RTSP 拉流细节、抽帧策略、YOLO 模型部署和最后的业务判定闭环,给正在做或者准备做类似项目的人一个可以直接参考的路线。
1. 为什么售货柜不能"来一帧识别一帧":从业务需求倒推技术选型
1.1 业务场景对延迟和准确率的硬性约束
售货柜和常规监控项目不一样,它的核心诉求不是"事后查录像",而是在顾客开关柜门的十几秒里判断出:他到底拿走了什么、拿了几个。这个判断结果直接决定订单结算金额,错了就是客诉。
我这边实际跑过的柜子,开门到关门一般是 8 到 20 秒,顾客可能中间犹豫、拿起又放下、遮挡商品。留给识别系统的时间窗口其实不宽裕,但也没有苛刻到必须每帧都判。比较合理的预期是:关门后 2 到 5 秒内给出结果,准确率 95% 以上。延迟卡在 5 秒内,是因为顾客还站在柜子前等手机弹结算通知,等太久体验极差。
另一个硬约束是成本。售货柜本质是零售终端,毛利没那么高,不可能给每台柜子配一台顶配 GPU 服务器。目前主流方案是柜内放一台工业级边缘盒子或者旧工控机,CPU 或者轻量 NPU 去跑模型。算力有限,就意味着每一帧都过 YOLO 是不现实的。
还有一点容易被忽略:相邻帧之间高度冗余。视频 25 帧每秒,顾客拿一罐可乐可能只有 0.8 秒,但前后画面几乎没变化。如果每帧都推理,结果会剧烈抖动——同一瓶饮料有时候检到、有时候没检到,反而让判断逻辑很难写。所以从业务规则上就需要"抽帧"来稳定输入。
1.2 三段式流水线的职责划分,以及为什么不把识别塞进摄像机
我最终搭的流水线,逻辑上分成三层:
- 拉流层:从网络摄像机(IPC)取 RTSP 视频流,负责取到"最新"的一帧,并保证断流后能自动恢复。
- 抽帧层:决定哪些帧值得送进模型,也就是所谓的"采样节奏",避免无效计算。
- 识别层:跑 YOLO 目标检测,输出商品类别、坐标、置信度,再结合业务规则算出哪些商品被拿走。
很多刚入门的人会问,能不能直接在 IPC 里做识别?市面上确实有带智能算法的 IPC,能识别人员闯入、越界这种固定场景,但售货柜里要识别的是几十上百种具体商品,和柜内光线、摆放角度、包装反光强相关,厂商内置算法根本不可能覆盖。所以老老实实把视频流拉到本地,用自训练模型识别,是目前最可落地的方式。
拆成三段还有个好处:每一层可以独立调试。拉流出问题就只查网络和协议,识别率低就只调模型和数据集,不会互相干扰。后面我会按这个顺序一层层讲。
2. IPC选型与RTSP拉流细节:稳定取流是整条流水线的前提
2.1 柜内 IPC 怎么选,重点看哪几个参数
售货柜里的摄像头,一般不是给你自由发挥的空间,柜体结构已经预留了安装位和线束。选型时我重点看四个参数:分辨率、焦距、编码格式、供电方式。
分辨率不用盲目上 400 万像素,200 万像素(1080p)在柜内 1 到 2 米的距离下完全够用,像素太高反而拖慢解码。焦距要选广角,柜内纵深有限,通常 2.8mm 左右比较合适,能覆盖单层货架的大半视野。编码格式优先 H.265,码率比 H.264 省一半以上,对磁带库和带宽都是好事。
供电这块,能选 PoE 就选 PoE,一根网线把电和图像都解决。不过很多旧柜子内部只有 DC 供电口,那就要注意 IPC 的电压规格是不是 12V,别现场再变电。
还有一个细节:买回来一定要先测试 RTSP 协议是否开放。部分消费级摄像头默认开启云平台转发,却不开放 RTSP 或者需要临时开启,这种就不能用,因为后续我们所有取流都走 RTSP。
2.2 RTSP 地址里藏的坑:主码流、子码流与鉴权
拿到一台 IPC,第一件事是拼 RTSP 地址。以海康为例,常见格式是:
rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101admin:password是摄像头的登录账号密码,必须得有访问权限。192.168.1.64是摄像头 IP。554是默认 RTSP 端口。/Streaming/Channels/101表示取通道 1 的主码流,102表示通道 1 的子码流。
不同品牌路径差异很大。大华一般是/cam/realmonitor?channel=1&subtype=0,宇视是/live/ch0。所以选型时要尽量统一品牌,或者在代码里做一张设备型号到 URL 模板的映射表,否则维护成本很高。
这里有个非常容易踩的坑:主码流分辨率高、码流大,清晰度好,但解码耗时也高;子码流分辨率低、延迟小,适合传输和预览,却不适合精细识别。我在实际项目里会拉主码流做识别,因为商品小字、包装图案都需要细节。如果只是想看画面通不通,才用子码流测试。
鉴权信息不要硬编码在代码里。虽然柜子内网相对封闭,但摄像机密码有时候要定期更换,写成配置文件更便于运维。
2.3 用 OpenCV 还是 FFmpeg 拉流,以及自动重连策略
拉 RTSP 流,最简单的是 OpenCV 的VideoCapture:
import cv2 cap = cv2.VideoCapture("rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101") ret, frame = cap.read()这段代码跑通很容易,但问题在cap.read()默认会缓存最近一帧,如果网络抖动,返回的画面可能是几秒前的旧画面,延迟忽高忽低。OpenCV 底层用的还是 FFmpeg 的解码能力,只是接口简化了,所以遇到复杂场景我不太建议直接裸用。
我的做法是分两种情况:
- 快速验证、原型 demo,用 OpenCV 足够。
- 生产环境,用 FFmpeg 拉流管道接 Python,或者直接用支持缓冲控制的 GStreamer 管道。
用 FFmpeg 子进程推流进管道的例子:
ffmpeg -rtsp_transport tcp -i "rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101" -f rawvideo -pix_fmt bgr24 -an -sn - > pipePython 侧从管道读原始帧再转numpy数组。这种方式延迟可控,但代码量比 OpenCV 大不少。
无论用哪种方式,重连策略必须做。IPC 这东西没人敢保证一年不掉线,碰到升级固件、异常断电、网线松动,流就断了。我用的策略是:连续 3 秒拿不到新帧就触发重连;重连失败后间隔指数退避,从 2 秒开始,最多退到 30 秒,避免空转打满 CPU。
import time import cv2 def connect_with_retry(url, max_retries=10): cap = cv2.VideoCapture(url, cv2.CAP_FFMPEG) if not cap.isOpened(): raise RuntimeError("can not open stream") return cap # 使用时持续 read,超时则重建连接3. 抽帧模块设计:不是简单"sleep 一下"就能完事
3.1 抽帧解决的三个问题:冗余、抖动、算力
抽帧,字面上是"从视频流里挑出部分帧"。但它背后解决的问题有三个。
第一是降低算力消耗。YOLOv5s 在一块普通 CPU 上推理一帧可能 200 到 400 毫秒,而 IPC 一秒出 25 帧,如果满负荷跑,CPU 永远追不上,帧会在队列里越积越多。抽帧后推理频率降到每秒 2 到 3 次,负载就完全可控。
第二是消除检测结果抖动。目标检测对画面变化很敏感,视频压缩产生的噪点、光线微变、货架反光都可能导致同一目标在相邻帧里检出又漏掉。抽帧让输入信号稳定,后端的"消失判定"才做得准。
第三是减少重复计算。顾客拿一瓶水,可能 3 秒内画面几乎没变化,重复推理这些帧没有业务价值,纯属浪费。
不过抽帧不是越少越好,抽太少了会漏掉关键动作。我见过有人把抽帧间隔设成 3 秒,结果顾客快速连拿两个商品,只拍到第二个,第一个直接漏判。抽帧节奏必须和业务动作速度匹配。
3.2 基于时间戳的等间隔抽帧实现,别用固定帧号
一开始我犯过一个错:用帧计数来做抽帧,比如每 10 帧送一次推理。问题是 IPC 实际帧率不稳定,繁忙时资源不够会掉帧,固定帧号间隔的时间跨度忽长忽短。后来我改成按时间戳抽帧。
核心逻辑很简单:每收到一帧,记录当前系统时间,如果距离上一次送入推理的时间间隔大于我们设定的阈值(比如 0.5 秒),就送这一帧;否则直接丢弃或者仅用于显示。
import time import threading class FrameSampler: def __init__(self, interval=0.5): self.interval = interval self.last_infer_time = 0.0 self.lock = threading.Lock() def should_infer(self, frame_stamp=None): if frame_stamp is None: frame_stamp = time.time() with self.lock: if frame_stamp - self.last_infer_time >= self.interval: self.last_infer_time = frame_stamp return True return False这里的interval怎么定?我一般先看模型单帧推理耗时t_infer,然后取max(t_infer * 1.5, 0.3)作为抽帧间隔,保证推理不积压,同时又不会为追求低负载丢掉动作细节。如果用的是 GPU 或者 NPU,t_infer 很短,抽帧间隔主要根据业务需要来定。
3.3 抽帧后的画面连续性:录像回放不受影响,判断逻辑要用基准帧
有次和客户对需求,对方很担心抽帧会"把视频弄卡",其实这是概念混淆。我们抽的是"喂给模型做推理的帧",不是"存进硬盘的录像帧"。如果项目里还需要原始录像,完全可以从 IPC 的另一个码流(比如子码流)持续录像,或者在拉流端做副本存储,抽帧只影响识别模块的输入,不影响录像完整性。
但抽帧确实会让业务判断逻辑变复杂。因为不是每一帧都进模型,后端如果要判断"某个商品消失了",就需要在抽帧之间维护一个基准状态。我的做法是:抽帧送入推理后,把识别结果缓存为last_result,下一次新结果出来后和它做对比,得出商品变化。这个基准帧不需要每帧更新,只有新的有效结果才刷新。
4. YOLO 识别:模型选择、后处理与业务规则闭环
4.1 模型尺寸和硬件平台怎么匹配
YOLO 系列到现在已经出了很多版本,我这边实际用过的有 YOLOv5、YOLOv8,也试过 YOLO11。版本本身不是最重要的,关键看部署平台支持什么。
如果是 NVIDIA Jetson 或者有 GPU 的工控机,优先考虑 TensorRT 加速,YOLOv8 可以直接导出为 engine 格式,推理速度很快。如果是瑞芯微 RK3588 这类平台,通常导出 RKNN 格式,YOLOv5 的适配资料最全,YOLOv8 也行但编译工具版本得对应。如果是纯 CPU 环境,老实选 nano 或者 small 尺寸的模型。
模型尺寸选择我给一个参考:
| 场景 | 模型尺寸 | 推理设备 | 单帧耗时参考 |
|---|---|---|---|
| 快速原型 | YOLOv8n | CPU | 80-150ms |
| 量产边缘盒子 | YOLOv8n/s | RK3588 NPU | 20-50ms |
| 更高精度场景 | YOLOv8s/m | Jetson Orin / 低端 GPU | 15-40ms |
| 纯 CPU 老工控机 | YOLOv5nu | CPU | 150-300ms |
有一点非常重要:预训练的 COCO 权重不能直接用于售货柜商品识别。COCO 的 80 类里根本没有"某种口味的饮料""某品牌薯片"这种细分类。所以必须自采数据、自标注、重新训练。我建议每类商品至少采集 500 到 1000 个样本,覆盖不同光照、角度、遮挡情况,只有柜内真实拍摄的照片才有效。
4.2 预处理、NMS 和类别过滤,这些细节决定上限
很多教程演示 YOLO 都是直接调用ultralytics库,一行代码出结果。但在生产环境,我通常不用它的高层 API,而是导出 ONNX 后用 ONNX Runtime 或者 TensorRT 跑,这样部署灵活,也不会被框架版本绑架。
推理流程拆开看:
- 预处理:把摄像头原始帧缩放成模型输入尺寸(比如 640×640),这一步不能直接拉伸,会破坏宽高比,得用 letterbox 方式,在边缘补灰边。
- 推理:
inputs是一张归一化后的张量,outputs是检测框、类别概率、置信度。 - 后处理:先按置信度阈值过滤,再做 NMS(非极大值抑制)去掉重叠框,最后把坐标缩放回原始图像尺寸。
置信度阈值我通常设在 0.25 到 0.45 之间。调低会召回更多目标但误检也变多;调高则相反。售货柜业务里,漏检比误检更难处理,因为商品没被识别=门店损失,所以我一般偏低一点,把误检交给业务层去修正。
4.3 从检测框到"商品被拿走":判定逻辑才是业务闭环的核心
识别模型只是个"眼睛",真正决定这笔订单怎么算的,是后端的判定逻辑。我第一版犯的错误是直接看两帧的检测框数量差,比如前一帧检出 3 瓶可乐,后一帧检出 2 瓶,就判定拿走 1 瓶。但顾客的手伸进去拿东西时,手会遮挡商品,中间若干帧可能少检了,最后其实什么都没拿。
后来我改成"多帧确认 + 形态快照"的思路:
- 柜门刚刚打开时,采集连续 3 到 5 帧有效结果作为基准快照,记录每个类别的数量。
- 柜门关闭前,再采集连续 3 到 5 帧有效结果,计算最终快照。
- 最终快照和基准快照做数量差,才生成"拿走清单"。
- 如果两者数量差异较大,需要结合柜门的开关信号确认是否真的完成了一次购买。
这里的关键点是不用中间帧做决策,中间帧只负责更新当前状态,避免手部遮挡造成的误判。这个逻辑帮我大幅压低了"拿起又放回却扣了钱"的客诉。
5. 整条流水线的工程化落地:线程队列、掉线重连与异常兜底
5.1 生产者-消费者模型:拉流、抽帧、识别三个线程怎么分工
把流水线做成单线程顺序执行,代码最简单,但有一个致命问题:拉流等待 I/O、模型推理都是阻塞操作,如果它们串行执行,拉流慢的时候推理空转,推理慢的时候拉流缓冲积累。实际部署时我用了三个线程,通过队列解耦。
大致结构:
import threading import queue class Pipeline: def __init__(self, rtsp_url, infer_interval=0.5): self.url = rtsp_url self.frame_queue = queue.Queue(maxsize=10) # 原始帧缓冲 self.infer_queue = queue.Queue(maxsize=5) # 待推理帧缓冲 self.sampler = FrameSampler(infer_interval) def pull_loop(self): # 循环拉流,塞入 frame_queue pass def sample_loop(self): # 从 frame_queue 取帧,按时间戳抽帧,塞入 infer_queue pass def infer_loop(self): # 从 infer_queue 取帧,跑 YOLO 后处理,回调结果 pass def start(self): threading.Thread(target=self.pull_loop, daemon=True).start() threading.Thread(target=self.sample_loop, daemon=True).start() threading.Thread(target=self.infer_loop, daemon=True).start()线程消息是 Python,但要注意 GIL 问题。如果你的推理用 ONNX Runtime、TensorRT 这类 C++ 扩展,核心计算会释放 GIL,三个线程并行度还可以。如果纯用 CPU 跑 PyTorch 推理,GIL 会限制并行效果,这时候我建议要么把推理放进子进程,要么干脆退化为"拉流 → 抽帧 → 推理"单线程顺序流,减少上下文切换。
5.2 队列长度、内存释放和延迟的三方平衡
队列设置要非常小心。拉流线程往队列塞帧很快,如果推理速度跟不上,队列会越积越长,延迟越来越大,最后清理的时候内存一路冲高。
我的经验是:队列长度宁可短,不要长。frame_queue设置 maxsize=10,缓冲区满了就把最旧的一帧丢掉,保证永远拿最新的画面。待推理队列 maxsize=5,满了就丢弃当前最新帧,防止内存膨胀。
有人问,丢帧会不会影响业务判定?不会,因为抽帧后的判断本身就不依赖连续帧,丢一两帧不会改变最终数量对比。
内存释放上,Python 的numpy数组复用是个容易被忽略的坑。如果你在循环里反复cap.read()拿到新帧,旧帧在没有引用后会被回收,但为了减少 GC 压力,可以在用完后直接del frame,或者用数组池复用内存。
5.3 掉线重连、看门狗与异常兜底
稳定运行期最大的敌人是网络。IPC 掉线后,拉流线程会抛异常或者阻塞,我做了几个兜底:
- 拉流线程自己维护一个"最近收到帧时间"
last_frame_time,如果超过 5 秒没有新帧,主动销毁当前VideoCapture,然后按 2 秒、4 秒、8 秒……的退避策略重连。 - 识别线程如果连续多次推理失败(比如模型加载异常、推理结果解析失败),不能死循环卡死,要记录日志并继续下一帧。
- 整条流水线加一个外部看门狗,定时检查三个线程是否还活着,如果某个线程意外退出,尝试重启该线程而不是重启整个进程。
这些逻辑看起来繁琐,但实际部署后很有用。我维护的一批柜子,大部分异常都是断网导致的,没有自动重连机制的话,掉一次线就得跑一次现场。
6. 实测性能与调优经验:从 demo 到稳定交付
6.1 一个典型配置下的实测数据
我拿一套典型的量产配置做过压测:1080p IPC,主码流 4Mbps,H.265 编码,边缘盒子是 RK3588,YOLOv8n 导出 RKNN,抽帧间隔 0.5 秒。连续跑了 72 小时,结果如下:
| 项目 | 实测值 | 说明 |
|---|---|---|
| 拉流到拿到帧的平均延迟 | 120-180ms | 局域网 RTSP over TCP |
| 单帧 YOLO 推理耗时 | 25-35ms | RKNN 加速 |
| 抽帧间隔 | 0.5s | 每秒约 2 次识别 |
| 单次拿取动作的判定耗时 | 关门后 1.5-3s | 含尾帧确认 |
| 内存占用 | 400-500MB | Python + ONNX Runtime |
| CPU 占用 | 15-25% | 四核中低频运行 |
这个数据说明什么?说明整个链路没有瓶颈,留给业务判断的时间非常充裕。如果推理耗时超过 100ms,我会怀疑是不是模型太大或者平台驱动没装好。
6.2 我踩过的高频坑和对应方案
最后把几个高频坑列出来,每一个都是真金白银换来的:
- RTSP 拉流地址超时。IPC 默认可能只允许一个 RTSP 会话,如果你同时拉主码流和子码流,有些廉价摄像头会直接拒绝。解决方法是确认设备最大会话数,或者只用一路。
- 柜内反光导致识别率骤降。饮料瓶、薯片包装都容易反光,特别是到了晚上柜内补光灯一照,反光区域直接盖住商品。我后期的做法是把补光灯改成漫反射灯条,同时训练数据里加入傍晚和夜晚样本。
- YOLO 后处理坐标越界。letterbox 缩放后,坐标回原图时需要减去 padding 值再除以缩放比例,很多新手在这里直接乘比例,导致画框偏移。这属于代码细节,但影响很大。
- 误检商品导致扣错款。有顾客手里拿着手机靠近摄像头,模型把手机识别成某个商品。这种情况没法只靠模型解决,我加了区域限制,只在货架区域做目标匹配,超出 ROI 的目标直接忽略。
- 时间戳不一致。不同摄像头的时间不同步,如果一台柜子有多个摄像头,各自抽帧的时间点不同,在做跨镜头的同一商品匹配时会出问题。方案是统一以服务器时间为准,在收到帧时打本地时间戳。
一点收尾经验
售货柜识别项目做到后期,我的体会是模型精度只占成功的一半,另一半是拉流稳定性、抽帧节奏和业务判定逻辑。如果你正准备启动类似项目,我建议先把拉流和重连模块调稳定,再用最少的数据把模型跑通一个端到端 demo,最后才去抠识别率。这个顺序能让你少走非常多弯路。