简介基于Flask框架的实时视频流目标检测项目整合PicoDet/YOLO系列ONNX模型与COCO类别识别面向具备Python基础、想快速掌握视频流检测应用搭建的开发者与学习者。包内共14个文件5个Python脚本覆盖相机采集、视频流处理、Flask应用入口等关键模块4个ONNX推理模型提供320/416两种输入分辨率另有HTML页面、依赖清单、README说明文档以及coco类别映射文件压缩包整体22.91MB结构清晰、易于部署复现。已有153人学习下载。借助该项目可完整了解从视频流接入、目标检测推理到结果页面实时展示的流程demo图片与README能辅助验证效果并支持基于PicoDet的二次开发适合作为入门目标检测Web应用的综合练习素材。1. Flask做实时视频流CPU上也能跑的COCO目标检测闭环之前做一个厂区智能巡检的临时方案时手头只有一台不带GPU的普通服务器要求把监控摄像头画面推到内网页面同时画出人和车辆的检测框。最初试了YOLOv8sCPU单帧推理接近300毫秒画面肉眼可见地卡。换成都不到一半体积的PicoDet ONNX Runtime320输入尺寸下单帧推理大约80到100毫秒配合Flask的MJPEG流浏览器打开就是一个帧率虽低但持续实时的检测画面。这套资源正是这样一个完整落地套件camera.py / camera_opencv.py负责取帧PicoDet.py加载ONNX模型做COCO 80类目标检测app.py通过Flask以multipart/x-mixed-replace方式推流templates/index.html提供浏览器端显示页面。适合两类人一是后端工程师想快速搭一套视觉检测演示系统不想碰TensorRT和DeepStream二是研究目标检测的学生需要一份能改、能跑、能看指标的轻量参考实现。它解决的核心问题不是把检测精度刷到多高而是把“摄像头取帧→模型推理→Web展示”这一条链路在低配环境里可靠地跑起来。2. 视频采集层base_camera.py与camera.py的抽象取舍2.1 BaseCamera后台线程与帧缓冲的通用骨架这套项目里base_camera.py是整个采集层的核心抽象。它不是一个真正去读摄像头的实现而是一个“已经开好了通用线程子类只需要提供取帧方法”的模板。很多第一次接触Flask视频流的同学会直接把cv2.VideoCapture放在Flask路由的while循环里结果每个客户端连接都会重新打开一次摄像头两台电脑一访问设备就被占满。BaseCamera通过类级别共享线程解决了这个并发问题。# base_camera.py import threading import time class BaseCamera: thread None # 类共享线程所有实例共用 frame None # 最新一帧的缓存 _lock threading.Lock() def __init__(self): if BaseCamera.thread is None: BaseCamera.thread threading.Thread(targetself._thread) BaseCamera.thread.daemon True BaseCamera.thread.start() self._wait_for_ready() def get_frame(self): # 加锁读取保证不会读到写了一半的帧 with BaseCamera._lock: return BaseCamera.frame classmethod def _thread(cls): camera cls._open_camera() while True: time.sleep(0.01) # 限频防止空转 frame cls._read_frame(camera) with BaseCamera._lock: BaseCamera.frame frame classmethod def _open_camera(cls): raise NotImplementedError classmethod def _read_frame(cls, camera): raise NotImplementedErrorget_frame加锁是因为OpenCV的imencode和线程替换帧是两步操作如果不加锁Flask生成器在压缩图片时可能拿到一个正在被替换的数组轻则花屏重则出现尺寸不匹配的异常。线程设成daemon的意义在于主进程退出时不会因为这个后台线程而阻塞——视频流服务经常用CtrlC停掉不是daemon的话会出现“卡住不退出”的现象这在调试时非常常见。2.2 camera.py与camera_opencv.py两种采集实现项目里同时提供了camera.py和camera_opencv.py两个文件。前者是基于OpenCV VideoCapture最简单直白的写法适合从视频文件读取或验证流程后者加入了单例模式操作同一摄像头设备适合长时间运行的服务场景。# camera_opencv.py 核心逻辑 import cv2 from base_camera import BaseCamera class Camera(BaseCamera): _camera None # 类变量保证只打开一次设备 classmethod def _open_camera(cls): if cls._camera is None: cls._camera cv2.VideoCapture(0) cls._camera.set(cv2.CAP_PROP_FRAME_WIDTH, 640) cls._camera.set(cv2.CAP_PROP_FRAME_HEIGHT, 480) cls._camera.set(cv2.CAP_PROP_FPS, 15) return cls._camera classmethod def _read_frame(cls, camera): ok, frame camera.read() if not ok: # 摄像头拔出、休眠都会导致空帧 return BaseCamera.frame return frame采集参数上我一般会把分辨率设在640×480而不是默认的1920×1080。PicoDet 320输入经过letterbox后高分辨率大图的缩放开销反而成为瓶颈而且640×480的JPEG压缩体积是1080p的1/4MJPEG流传输负担更小。FPS降到15而不是默认30也是因为检测端本来每秒也就10帧左右采集30帧只会让帧缓冲里的画面时刻比推理结果“新”一点没有实际意义。参数推荐值说明CAP_PROP_FRAME_WIDTH640越高则取帧和压缩耗时越长CAP_PROP_FRAME_HEIGHT480保持4:3避免letterbox失真CAP_PROP_FPS15略高于检测的期望帧率即可CAP_PROP_FOURCCMJPG部分USB摄像头默认YUYV会拉低采集速度2.3 采集层踩坑空帧、权限与热插拔摄像头类应用最常见的异常就是camera.read()返回False。Windows上摄像头被其他程序占用时OpenCV不会报错而是持续返回空帧Linux下/dev/video0权限不足时V4L2会直接拒绝。我的判断顺序是先确认camera.isOpened()为真再在read失败时连续尝试重连连续失败超过3秒就返回一张黑色占位帧至少保证页面上的浏览器连接不断开。项目中camera_opencv.py的_read_frame里加一层None保护就是为这个实际部署时可以在外面包一层重连逻辑而不是把异常传进Flask生成器导致整个视频流500。3. PicoDet.py推理封装ONNX模型解析与后处理3.1 为什么选PicoDet而不是YOLO系列目标检测算法主要分两条路线以Faster R-CNN为代表的Two-stage先找候选区域再分类精度高但速度慢以YOLO、SSD为代表的One-stage直接回归边界框和类别速度优势明显。PicoDet属于后者但它有两点不一样一是用了解耦检测头分类分支和回归分支分开训练时收敛更稳二是结构上一路轻量化设计到极致即使是最大型号的mONNX文件也只有二十多MB。对比同一量级的YOLOv5nPicoDet在CPU上实测的单帧推理更快而且后处理并不复杂——这也是这套资源把它作为CPU演示首选的根本原因。3.2 模型文件与输入输出形状resources目录下给了四个已经导出的ONNX文件picodet_m_320_coco.onnx、picodet_m_416_coco.onnx、picodet_s_320_coco.onnx、picodet_s_416_coco.onnx。文件名里的数字是训练导出时的输入边长。s和m是模型规模m特征通道更宽、精度略好320与416是输入尺寸对应速度与精度的两档取舍。模型文件输入尺寸适用场景picodet_s_320_coco.onnx320×320CPU弱机、追求最高帧率picodet_m_320_coco.onnx320×320均衡型大多数部署选这个picodet_s_416_coco.onnx416×416小目标偏多但算力尚可picodet_m_416_coco.onnx416×416精度优先的离线分析PicoDet颈部网络是FPN结构下采样倍数为8、16、32输入320时三张特征图的尺寸是40×40、20×20、10×10归一化到单条序列后合计2100个预测点。ONNX导出时如果合入了decode输出张量形状通常是[N, 2100, 6]即每个点存cx、cy、w、h、类别分数、类别下标如果只导出裸网络则要自己做解码。拿到模型先用代码确认输出层形状而不是直接写死后处理逻辑这是我跑任何检测模型的第一件事。import onnxruntime as ort session ort.InferenceSession(picodet_m_320_coco.onnx) for inp in session.get_inputs(): print(input:, inp.name, inp.shape) for out in session.get_outputs(): print(output:, out.name, out.shape)3.3 预处理与后处理实现PicoDet.py的完整推理流程是读图→letterbox缩放→归一化→ONNX推理→解码→NMS→映射类别。letterbox是这里最容易出错的环节直接cv2.resize会把图像拉变形原始比例变了训练时学到的锚框分布全部错位小目标漏检会非常明显。# PicoDet.py 核心推理类 import cv2 import numpy as np import onnxruntime as ort class PicoDet: def __init__(self, onnx_path, score_thr0.4, iou_thr0.5): self.session ort.InferenceSession(onnx_path) self.input_name self.session.get_inputs()[0].name self.input_size self.session.get_inputs()[0].shape[2:] # [H, W] self.score_thr score_thr self.iou_thr iou_thr self.names self._load_names(coco.names) staticmethod def _letterbox(img, size(320, 320)): h, w img.shape[:2] scale min(size[0] / h, size[1] / w) nh, nw round(h * scale), round(w * scale) resized cv2.resize(img, (nw, nh)) canvas np.full((size[0], size[1], 3), 114, dtypenp.uint8) x_off (size[1] - nw) // 2 y_off (size[0] - nh) // 2 canvas[y_off:y_off nh, x_off:x_off nw] resized return canvas, scale, x_off, y_off def predict(self, img): canvas, scale, x_off, y_off self._letterbox(img, self.input_size) blob canvas.astype(np.float32) / 255.0 # 归一化到0~1 blob np.transpose(blob, (2, 0, 1))[None, ...] # HWC - NCHW outputs self.session.run(None, {self.input_name: blob}) boxes outputs[0][0] # 假设输出为 [1, 2100, 6] dets boxes[boxes[:, 4] self.score_thr] result [] for cx, cy, w, h, score, cls_id in dets: x1 (cx - w / 2 - x_off) / scale # 映射回原图坐标 y1 (cy - h / 2 - y_off) / scale x2 (cx w / 2 - x_off) / scale y2 (cy h / 2 - y_off) / scale result.append([int(x1), int(y1), int(x2), int(y2), float(score), int(cls_id)]) keep cv2.dnn.NMSBoxes( [r[:4] for r in result], [r[4] for r in result], self.score_thr, self.iou_thr ) return [result[i] for i in keep.flatten()]阈值参数直接放在构造函数里score_thr控制置信度调低会放过更多目标但也引入误检与画面闪烁调高则只保留最有把握的目标。iou_thr用于NMS去重默认0.5对人群密集场景偏紧行人重叠严重的场景建议调高到0.6。注意每个输出框的坐标是从letterbox画布上还原的缩放和平移因子在_letterbox里同步返回映射回原图后再送进cv2.rectangle画框。4. Flask与MJPEG流app.py路由与index.html实时显示4.1 multipart/x-mixed-replace长连接原理浏览器实现Live画面不需要WebSocket——Flask用一个永远不结束的生成器不断往HTTP响应里拼JPEG帧响应头指定multipart/x-mixed-replace; boundaryframe浏览器就会把后续的每个JPEG块当成新的一帧替换显示。这是MJPEG的最简单实现方式代码只有几行代价是每个客户端独占一个长连接并发量高了服务器压力会明显起来。4.2 app.py路由与生成器实现# app.py import cv2 from flask import Flask, Response, render_template, request from camera_opencv import Camera from PicoDet import PicoDet app Flask(__name__) detector PicoDet(picodet_m_320_coco.onnx) camera Camera() def generate_frames(): while True: frame camera.get_frame() if frame is None: continue dets detector.predict(frame) for x1, y1, x2, y2, score, cls_id in dets: cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2) label f{detector.names[cls_id]} {score:.2f} cv2.putText(frame, label, (x1, y1 - 6), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) ok, jpeg cv2.imencode(.jpg, frame) if ok: yield (b--frame\r\n bContent-Type: image/jpeg\r\n\r\n jpeg.tobytes() b\r\n) app.route(/video_feed) def video_feed(): return Response(generate_frames(), mimetypemultipart/x-mixed-replace; boundaryframe) app.route(/) def index(): return render_template(index.html) if __name__ __main__: app.run(host0.0.0.0, port5000, threadedTrue)yield拼接的boundary字符串必须与Response的mimetype一致这是最容易踩的低级错误。threadedTrue开启多线程否则Flask开发服务器收到浏览器对/video_feed的长连接后其他路由包括首页都会阻塞。flask框架的生成器在客户端断开时会抛异常predict里如果因为异常退出整个视频流会停掉——所以在generate_frames外层常见做法是加try/except捕获BrokenPipeError否则Flask会刷一屏幕Traceback。4.3 前端index.html的设计页面用一个img标签直接指向/video_feed即可这是MJPEG的天然优势不需要canvas和JavaScript自动刷新。项目中templates/index.html还会再叠加一层用户控件比如阈值调节通过URL查询参数传回Flask再在生成器里引用。!-- templates/index.html -- !DOCTYPE html html head meta charsetutf-8 title实时目标检测/title /head body h3PicoDet COCO目标检测/h3 img src/video_feed width640 height480 p置信度阈值 0.35 | 模型 picodet_m_320_coco.onnx/p /body /html路由方法作用/GET返回视频流页面/video_feedGET持续输出MJPEG帧浏览器对同一页面的连接数有限制每开一个页面就多一个视频流连接调试时频繁刷新会让旧连接没有立刻释放页面卡在加载中。解决方式是后端的生成器里记录最后一次访问时间超过5秒自动退出循环把连接归还给服务器。5. 输入尺寸、阈值与线程安全实时检测的关键调优5.1 320与416的取舍CPU推理时输入尺寸直接决定了计算量320边长下的算子计算量约为416的0.6倍实测了一批常见CPU之后320大约比416快60%左右而COCO验证集上的精度差距通常在1到2个mAP点。截取一段实时画面单独保存成图片分别用320和416跑一遍看漏检情况比看公开指标更直接。天花板低的工业机不建议硬上416画面跳动比精度损失更让人难受。5.2 一个可复用的耗时验证脚本每次改动模型文件或输入尺寸都要量化一下单帧耗时。避免主观感受脚本逻辑是预热10帧丢弃再连续推理100帧取百分位数。import time import numpy as np import cv2 from PicoDet import PicoDet det PicoDet(picodet_m_320_coco.onnx) sample cv2.imread(imgs/demo.png) for _ in range(10): # 预热触发内存申请 det.predict(sample) costs [] for _ in range(100): # 正式测试 t0 time.perf_counter() det.predict(sample) costs.append((time.perf_counter() - t0) * 1000) print(fp50: {np.percentile(costs, 50):.1f} ms, fp90: {np.percentile(costs, 90):.1f} ms, fmax: {np.max(costs):.1f} ms)首次推理会包含ONNX Runtime的线程池初始化时间高出正常值一个量级必须丢弃。p50是常规帧率参考p90代表画面卡顿的严重程度——如果p90超过200毫秒实时视频流的体感就会明显开裂。5.3 线程模型与GIL的边界前端显示的刷新速度取决于三个环节中最慢的一个摄像头采集、模型推理、图像编码。采集和编码是C扩展会释放GIL真正占用Python的是PicoDet.py里的后处理和cv2画框。实际部署时可以在BaseCamera线程内提前完成目标检测画框Flask生成器只负责取已经画好框的帧然后imencode这样推理不会阻塞生成器的推流。如果要用多进程分配多个摄像头每个进程都要单独加载一次ONNX模型内存占用乘以进程数这个在规划机器配置时要先算清楚。视频流检测服务持续跑几小时后如果发现内存缓慢上涨优先检查每帧是否都是新申请的numpy数组且没有引用泄漏。常见做法是限制待处理帧队列长度不超过3超出的帧直接丢弃——宁可画面偶尔跳帧也不能让延迟持续累积到几十秒。把p50耗时、队列深度一起打进结构化日志用超过阈值时自动记录当帧图像排错效率比看控制台输出高得多。本文还有配套的精品资源点击获取