先交代一下背景:我在做智能售货柜识别这块折腾了快两年,最常见的需求就是“顾客从柜子里拿走商品,系统在关门后准确判断拿走了什么、放回了什么,然后自动扣款”。这个业务听起来简单,落到技术底层其实就是一条很朴素的流水线:IPC 摄像头拉流、按策略抽帧、YOLO 识别目标,最后把结果回传给业务端。这篇文章会把这条链路完整拆开讲一遍,从 RTSP 拉流到抽帧再到 YOLO 推理,包括多进程通信设计和踩坑记录。适合正在做边缘视觉项目、想搞懂端到端识别流程的朋友参考,尤其是智能零售、自助设备、门禁监控这一类场景。
1. 项目整体设计与链路拆解
1.1 先搞清楚业务要什么,再谈技术
售货柜识别的核心业务目标不是“识别出画面里有什么”,而是判定一段事件:顾客开门、拿走商品、放回商品、关门之后,系统要给出准确的交易结果。这和单纯的图片分类完全不同,所以技术链路不能只堆一个 YOLO 模型,而是要把取流、选帧、检测、事件判定串起来。
早期不少方案用重力感应托盘,通过层板重量变化判断商品被取走还是放回,但遇到多件商品同时被拿走、商品被拿起又放回原位、或者顾客故意调换位置的情况,重力方案基本全废。视觉方案的补位核心就是这套“IPC 拉流 → 抽帧 → YOLO 识别 → 事件判定”流水线。其中拉流解决数据从哪来,抽帧解决什么时候识别,YOLO 解决识别具体目标,事件判定负责把多帧结果变成业务结果。
我在最开始犯过一个错误:一上来就调模型,把 YOLO 精度刷得不错,结果接到真实柜机上,拉流卡顿、抽帧时机不对、进程崩掉,整条链路根本跑不起来。所以说,模型只是流水线的一环,链路稳定才是上线的关键。这也是我写这篇文章的初衷,把每个环节的选型和坑都讲透。
1.2 硬件与模型选型怎么搭
摄像头选择上有几个参数要注意。分辨率一般 200 万到 400 万像素足够,帧率 25fps 就够用,再高纯属浪费算力。焦距要选短焦广角,比如 2.8mm 或更小,否则柜内空间窄,镜头看不到每层货架的前方区域。安装位置建议在柜体顶部斜向下,视角需要覆盖每一层的取货通道,而不是只盯着货架表面。
边缘主机常见的三个选择:RK3588、Jetson 系列、x86 小主机。我的对比经验如下:
| 方案 | 算力 | 功耗 | 部署难度 | 适合场景 |
|---|---|---|---|---|
| RK3588 | 6 TOPS NPU | 5-15W | 中等,需要交叉编译或 RKNN 工具链 | 单柜机、成本敏感项目 |
| Jetson Orin NX | 100 TOPS | 10-25W | 相对简单,CUDA 生态成熟 | 中高端柜机、多路摄像头 |
| x86 小主机 | 依 GPU 而定 | 50W+ | 最低,直接用 CUDA/OpenVINO | 开发调试、离线验证 |
模型方面,我在 RK3588 上常用 YOLOv5s 或 YOLOv8n,输入分辨率 640x640,单帧推理大概 10-20ms。如果商品类别多、外形相似,可以尝试把输入分辨率提到 960 或 1280,识别小物体会更稳,但算力消耗会明显上升,需要提前压测。版本上 YOLOv8 和 YOLOv11 的 API 差异不大,部署工具链也更友好,新项目建议直接用较新的版本,没必要从老版本迁移。
2. 拉流环节:RTSP 协议与 IPC 接入
2.1 RTSP 拉流的基础概念
RTSP 是网络摄像机的标准控制协议,但它本身不传输媒体数据,真正推流走的是 RTP,媒体参数由 SDP 描述。很多刚接触的人会混淆这一点,以为 RTSP 地址就是视频流地址,实际上 RTSP 只是帮客户端和服务端建立会话,协商好转码方式后,RTP 包才在 UDP 或 TCP 上源源不断传过来。
IPC 通常暴露两个码流:主码流和子码流。主码流分辨率高、码率高,适合本地录像和事后回放;子码流分辨率低、码率低,适合实时预览和算法识别。我们的识别链路建议直接取子码流,因为解码压力小、网络占用低,子码流的画质对 YOLO 检测来说通常足够。RTSP 地址格式一般是:
rtsp://用户名:密码@IP地址:554/Streaming/Channels/101不同厂商路径有差异,但大同小异。我在项目里习惯建一个摄像头配置表,把每路设备的主码流和子码流地址都列出来,代码里只传设备 ID,避免硬编码地址造成维护地狱。
2.2 拉流实战:OpenCV 和 GStreamer 两种方式
OpenCV 的 VideoCapture 是最快上手的方案,Python 里几行代码就能读到帧。
import cv2 RTSP_URL = "rtsp://admin:password@192.168.1.64:554/Streaming/Channels/102" cap = cv2.VideoCapture(RTSP_URL, cv2.CAP_FFMPEG) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # 关键:减少内部缓冲,避免延迟累积 while True: ret, frame = cap.read() if not ret: print("拉流失败,准备重连") break # 这里拿到的是 BGR 格式的 frame,交给后续抽帧模块注意CAP_PROP_BUFFERSIZE这个参数,默认情况下 VideoCapture 内部会缓冲好几帧,如果你的处理速度跟不上拉流速度,读到的就是几秒前的旧画面,实时性会变差。实际项目里我一般把它设为 1,只保留最新一帧,宁可丢帧也不能要延迟。
GStreamer 是另一个常用方案,适合需要精细控制延迟和硬件解码的场景。比如 RK3588 上可以通过 GStreamer 插件走硬件解码,大幅降低 CPU 占用。一个典型的 pipeline 长这样:
gst-launch-1.0 rtspsrc location=rtsp://admin:password@192.168.1.64:554/Streaming/Channels/102 latency=0 ! rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! appsink其中latency=0是为了关闭缓冲延迟,对实时性要求高的识别场景很关键。GStreamer 调试起来比 OpenCV 直观,出问题可以用GST_DEBUG=3环境变量拉日志,能看清楚是网络问题、解码问题还是渲染问题。
开发阶段如果手头没有真实摄像头,可以用 FFmpeg 把本地视频文件推成 RTSP 流,或者用厂商提供的 IPC 模拟器,先把流水线跑通,等硬件到位再切换真实地址。这个习惯能省很多联调时间。
2.3 拉流时容易踩的坑
拉流环节我踩过三个比较深的坑,这里详细说一下。
第一个是断流重连。IPC 在弱网环境下或者长时间运行后,偶发 RTP 丢包、RTSP 会话超时都很常见。表现是cap.read()开始返回 False,或者画面卡住不动。最烂的处理是程序直接退出,正确的做法是捕获到失败后,sleep1 到 2 秒,然后重新创建 VideoCapture,并且加一个重连次数上限,避免设备彻底下线后进程空转。
第二个是花屏和绿屏。这通常是 RTP 包在 UDP 传输过程中丢失导致的,解码出来就是花屏。一个简单有效的办法是把传输协议改成 TCP,RTSP 地址里加?tcp参数,或者用 OpenCV 设置cv2.CAP_PROP_RTSP_TRANSPORT为cv2.CAP_PROP_RTSP_TRANSPORT_TCP。代价是延迟会略微上升,但画面完整性好得多,识别场景我更看重画面完整。
第三个是多路拉流并行。一个柜机可能装 2 到 3 个摄像头,如果放在同一个线程里顺序读取,一路网络抖动会拖垮其他几路。我通常一个摄像头一个拉流进程,各进程独立维护自己的 VideoCapture 和帧队列,互不干扰。这样即使某一路断流重连,也不影响其他路的识别。
3. 抽帧策略:怎么抽帧才不会影响画面和识别
3.1 为什么必须抽帧
很多第一次接触这个场景的人会问:IPC 是 25fps,我直接把每一帧都送去 YOLO 识别不就行了?理论可行,实际上问题很大。25fps 全量识别意味着模型每秒推理 25 次,哪怕单帧推理只要 20ms,一轮下来也得 500ms,CPU 和 NPU 基本打满,功耗、发热、成本全上来。而且业务上根本不需要这么高的识别频率,商品拿取动作持续半秒到一秒,识别间隔 100-200ms 已经足够。
抽帧本质上不是对码流做处理,而是在解码后的帧序列里“选帧”。原始视频流还在正常录制和预览,画面不会因为抽帧而变卡或变糊。识别链路拿到的只是均匀采样或按需采样后的帧子集,用来降低计算开销,同时保证关键动作不丢。
3.2 三种抽帧策略
我实际用过三种策略,各有适用场景。
第一种是定时抽帧,最简单,每 N 帧取 1 帧,比如每 3 帧取 1 帧,相当于把 25fps 降到 8.3fps。这种方式适合识别节奏固定的场景,比如门口的人形检测、通道的流量统计。缺点是业务高峰期和空闲期用同样的频率,有点浪费算力。
第二种是业务状态触发式抽帧,这是售货柜场景的核心。平时待机时,5 秒抽 1 帧做异常监控就够;用户开门后,马上把抽帧频率提升到每 2 到 3 帧抽 1 帧;关门瞬间,连续密集抽帧 20 到 30 帧,把拿取和放回动作完整记录下来。这种方式能兼顾功耗和识别准确率,强烈推荐做业务系统的人使用。
第三种是运动检测触发式抽帧,通过帧差法或者光流法检测画面变化,变化超过阈值才送识别。适合待机时画面长时间静止的场景,可以进一步降低功耗。缺点是运动检测本身也要消耗一点算力,而且如果检测阈值设得不好,容易误触发或者漏触发。
组合拳是最优解。我的实现里,状态机决定抽帧频率:关门状态低频,开门状态高频,关门瞬间全速,再配合定时兜底,防止状态机漏判。
3.3 抽帧参数怎么定
抽帧参数不是拍脑袋定的,而是根据业务动作时长反推。顾客拿货动作大概 0.5 到 1 秒,假设 IPC 是 25fps,每 3 帧抽 1 帧,采样间隔约 120ms,一个 0.5 秒的动作能采到 4 张以上有效帧,足以覆盖遮挡瞬间。如果算力紧张改成每 5 帧抽 1 帧,采样间隔变成 200ms,手完全挡住商品的瞬间有可能漏掉,关门后判定就会犹豫。
所以参数底线是:采样间隔不能超过关键动作周期的四分之一。拿 0.5 秒动作算,间隔最好小于 125ms,也就是保守能支持 8fps 的抽帧频率。再往上提升频率,收益会边际递减,但算力消耗却线性增加。
代码骨架大致长这样:
class FrameSampler: def __init__(self, mode="idle"): self.mode = mode self.frame_count = 0 # idle: 每25帧取1帧; active: 每2帧取1帧; closing: 每帧都取 self.strategy_map = { "idle": 25, "active": 2, "closing": 1, } def should_sample(self) -> bool: step = self.strategy_map[self.mode] if step == 1: return True self.frame_count += 1 if self.frame_count % step == 0: self.frame_count = 0 return True return False实际使用中,状态切换要加一个延时稳定逻辑,避免开关门瞬间状态抖动导致抽帧频率来回跳。
4. YOLO 识别模型与数据准备
4.1 模型选型:S、M、N 到底怎么选
YOLO 家族从 v5 到 v11,模型规模都有 n/s/m/l/x 几档。售货柜识别属于典型的小目标、相似外观、遮挡频繁的任务,我通常推荐从 n 或 s 开始。n模型推理最快,适合 RK3588 这类中低端边缘设备;s模型精度更高,适合 Jetson 或 x86 带 GPU 的场景。m以上在柜机场景性价比不高,除非你的商品类别超过 50 类且外形高度相似。
版本选择上,如果你从零开始,直接用 YOLOv8 或 YOLOv11 就行,ultralytics 库把训练、验证、导出一条龙做完了,部署导出 ONNX 或 TensorRT 都很方便。如果项目里有老代码基于 YOLOv5,也不是不能用,v5 的生态资料多,部署方案成熟。
模型训练不能用随机初始化的权重硬训,必须用 COCO 预训练权重做迁移学习。冻结前几层骨干网络,用柜内商品数据集微调,收敛快且不容易过拟合。常用超参数我一般这么设:图片大小 640,batch size 根据显存来,epoch 100 到 300,早停 patience 20。训练时重点盯 val loss,如果 val loss 一直不降,先检查数据和标签,不要盲目加 epoch。
4.2 数据集准备与格式转换
数据是售货柜识别里最容易翻车的环节。只拍整齐摆放的商品照远远不够,必须采集真实柜内角度、不同光照、手遮挡、商品被拿起瞬间的照片。负样本同样重要:空手开门、手伸进去但没有拿商品、顾客手里拿了别的东西,这些场景如果不在训练集里,模型很容易误报。
标注格式是 YOLO 的 txt 文件,每行对应一个目标,格式如下:
class_id x_center y_center width height坐标都是归一化到 0 到 1 的小数,x_center和y_center是目标中心点相对图像宽高的比例,width和height是目标宽高比例。比如一张 1920x1080 的图,一个框左上角在 (480, 270),右下角在 (960, 810),归一化后就是class_id 0.375 0.5 0.25 0.5。
实际项目里很少直接手写 txt,通常是从其他标注格式转换过来。常见的有 COCO 的 json、Labelme 的 json、KITTI 的 txt、TACO 这类细分数据集。转换逻辑不难:解析源格式拿到像素坐标,再除以图像宽高得到归一化坐标。以 COCO json 为例,核心就是把bbox的[x, y, width, height]转成[ (x + width/2) / img_w, (y + height/2) / img_h, width / img_w, height / img_h ]。
数据集划分我习惯按 80% 训练、15% 验证、5% 测试,随机打乱。重点提醒:划分前检查每个类别的样本数量,防止某个类别全跑到验证集里。小样本类别可以手动做水平翻转、亮度调整、随机裁剪等增强,让模型更鲁棒。
4.3 YOLO 后处理流程
模型输出的不是可以直接使用的坐标列表,而是一堆原始张量。以 YOLOv8 为例,输出维度是[1, 4 + num_classes, num_anchors],需要解析出边界框、置信度、类别概率,再经过置信度过滤和非极大值抑制(NMS)才得到最终结果。很多人调不动模型,就是卡在后处理这一步。
我用 OpenCV DNN 模块跑 YOLO 时,后处理代码大致是这个样子:
import cv2 import numpy as np def postprocess(outputs, conf_threshold=0.25, iou_threshold=0.45): boxes, scores, class_ids = [], [], [] for output in outputs: for detection in output: scores_det = detection[4:] class_id = np.argmax(scores_det) score = scores_det[class_id] if score < conf_threshold: continue cx, cy, w, h = detection[:4] boxes.append([int(cx - w/2), int(cy - h/2), int(w), int(h)]) scores.append(float(score)) class_ids.append(class_id) idxs = cv2.dnn.NMSBoxes(boxes, scores, conf_threshold, iou_threshold) return [boxes[i] for i in idxs.flatten()] if len(idxs) else []NMS 的作用是抑制重叠框,IOU 阈值一般 0.45 到 0.5。如果两个同类商品贴得很近,容易互相遮挡,阈值不要调太低,否则同一目标会出现重复框,也会把相邻目标误伤。置信度阈值则要分场景调,后面我在问题排查章节会展开讲。
顺带提一句训练时的损失函数。YOLO 系列的损失由三部分组成:边界框回归损失、分类损失、置信度损失。训练日志里如果 box_loss 降不下去,通常是标注框质量差;cls_loss 高则可能是类别特征不够明显,比如不同口味的同品牌饮料外观太接近。搞清楚损失函数里每一项在说什么,排查问题会快很多。
5. 多进程流水线与进程通信设计
5.1 为什么用多进程而不是多线程
完整流水线里有三个独立环节:拉流解码、抽帧预处理、YOLO 推理。拉流是 I/O 密集,YOLO 推理是 CPU/NPU 密集,如果全部塞进一个线程,只要网络抖一下或者推理慢一帧,整个链路就卡住了。Python 多线程又有 GIL 限制,没法真正并行执行 CPU 密集任务,多线程在这里基本是摆设。
实际情况是拉流和识别速度天然不匹配,拉流 25fps,推理可能只能跑 8fps,中间必须有一层缓冲来解耦。用多进程加队列,生产者进程只管把抽帧结果放进队列,消费者进程只管从队列取帧做推理,两边互不拖累,即使消费者崩溃,拉流进程还能继续工作,配合守护进程自动重启,稳定性提升非常明显。
还有个彩蛋是标题里的 IPC 双关:前一个 IPC 是网络摄像机,后一个 IPC 是进程间通信。这两个 IPC 正好是这条流水线的两端,一个管取数据,一个管传数据。
5.2 进程通信方案怎么选
单机边缘部署下,我对比过三种进程通信方式:
| 方案 | 数据拷贝 | 延迟 | 复杂度 | 适用场景 |
|---|---|---|---|---|
| multiprocessing.Queue | 需要序列化 | 中等 | 低 | 帧率不高、快速开发 |
| 共享内存 shared_memory | 零拷贝 | 低 | 中等 | 高帧率、大分辨率图像 |
| Redis / 消息队列 | 网络序列化 | 高 | 高 | 多机分布式、上云 |
如果你的抽帧频率只有 5 到 8fps,图像是 720P,直接用multiprocessing.Queue就够了,代码简单,问题也好排查。但如果你用多路摄像头、帧率又拉满,Queue 的序列化和复制开销就会很夸张,这时候共享内存更合适。
共享内存的思路是:拉流进程把图像数据写入一整块预先分配好的共享内存区域,队列里只传帧序号和时间戳这样的元数据,识别进程拿到序号后从共享内存对应位置读图像,避免整帧数据在网络和队列之间反复复制。代码上可以用 Python 的multiprocessing.shared_memory模块,但要注意加锁和生命周期管理,防止两个进程同时写同一块区域导致数据错乱。
5.3 队列背压与丢帧策略
生产者比消费者快的时候,队列一定会满。处理策略只有两种:丢新帧或者丢旧帧。我强烈建议丢旧帧。对于实时识别来说,最新画面比老画面有价值得多,如果队列满时把新帧丢了,等到推理进程腾出手来,读到的还是几秒前的旧画面,那识别结果就没有实时性了。
实现上不建议用queue.put()阻塞等待,而是用非阻塞的put_nowait(),捕获queue.Full异常后再用get_nowait()丢掉一个旧帧,然后重新put_nowait()。这样能保证队列里的帧永远是最新的。
我整理了一个简化的丢旧帧实现:
import queue frame_queue = queue.Queue(maxsize=2) def producer(frame): try: frame_queue.put_nowait(frame) except queue.Full: # 队列满,弃旧留新 try: frame_queue.get_nowait() frame_queue.put_nowait(frame) except Exception: pass这里队列长度为什么只设 2?因为太长的队列只会增加延迟,不会提高准确率。识别进程拿到的永远是最近两帧,哪怕某一帧推理时被覆盖,下一帧也会立刻补上。这个思路在实时视频流水线里叫“有界队列 + 最新帧优先”,做边缘实时识别的朋友可以记一下。
6. 常见问题与排查技巧实录
6.1 高频问题速查表
| 现象 | 可能原因 | 解决办法 |
|---|---|---|
| 画面花屏 / 绿屏 | RTP 丢包 | RTSP 改 TCP 传输,或降低码率 |
| 延迟越来越大 | VideoCapture 内部缓冲堆积 | 设置 CAP_PROP_BUFFERSIZE 为 1,或定期清空队列 |
| CPU 长时间跑满 | 抽帧频率过高,或推理线程退化 | 降低抽帧频率,确认推理是否走了 NPU/GPU 加速 |
| 识别结果时好时坏 | 光照变化剧烈、商品反光 | 增加训练数据增强,考虑做图像预处理白平衡 |
| 画面颜色怪异 | BGR / RGB 通道顺序搞反 | 推理前执行 cv2.COLOR_BGR2RGB 转换 |
| 多路画面卡顿 | 多路拉流共用同一线程 | 每路摄像头独立进程,独立队列 |
| 排队偶发不扣款 | 关门瞬间没有密集抽帧 | 状态机在关门事件触发全速抽帧,保留关键动作帧 |
这张表是我在实际项目里沉淀下来的排障清单,遇到问题先对照一遍,能省很多排查时间。
6.2 三个让我印象深刻的实战坑
第一个坑是 BGR/RGB 颜色反转。OpenCV 读出来的图像默认是 BGR 通道顺序,但 YOLO 训练时用的是 RGB,如果直接把 BGR 图像喂给模型,颜色通道被对调,模型会把红色商品识别成蓝色,蓝色识别成红色,严重影响分类结果。这个问题隐蔽在“整体识别率还挺高,但总觉得某些细节不对”,排查时很容易忽略。解决办法很简单,推理前加一行cv2.cvtColor(frame, cv2.COLOR_BGR2RGB),或者确保训练和推理走同一套预处理流程。
第二个坑是长时间运行后延迟越来越大。现象是刚开机识别很灵敏,跑了一天,关门后扣款来越来越慢。查了半天发现是 VideoCapture 内部缓冲在累积,我的处理速度跟不上拉流速度,导致读到的画面越来越旧。解决办法就是前面提到的把CAP_PROP_BUFFERSIZE设为 1,并且在进程通信队列里也使用丢旧帧策略。这个坑非常典型,几乎每个做实时视频处理的项目都会遇到。
第三个坑是 NMS 和置信度阈值一刀切导致的误检漏检。售货柜场景里,瓶装水表面反光、易拉罐标签图案复杂,单一置信度阈值很难完美兼顾。阈值设低了,反光区域会被识别成商品;设高了,被手挡住一半的商品很容易漏检。我的做法是分类别设置阈值:外形规整、特征清晰的商品用较高阈值;容易受反光影响、外形相似的类别用较低阈值。另外,对“放回”事件不要依赖单帧判定,连续两帧都识别到同一目标再上报,准确率会稳很多。
最后再分享一个贯穿整个项目的小技巧:关门事件触发时,在开始识别前先清空上一轮遗留的识别结果缓存。我发现很多误扣款都来自拿着上一波识别结果和当前帧结果做融合计算,结果把开门时摆放在货架上的商品也算进本次拿取列表。清空缓存后再做事件判定,这个问题的出现率几乎降到了零。整个项目稳定跑下来,我最深的体会是:识别精度只是木桶中的一块板,拉流稳定、抽帧合理、进程通信不崩,才是真正决定上线体验的关键环节。