YOLOv3+RTMP+PyQt5:交通路口智能监控系统实战架构 简介这套基于YOLOv3与PyQt5的交通路口智能监控系统源码面向具备一定Python基础的开发者、计算机视觉学习者和智慧交通项目实践者解决远端视频流接入、道路目标检测与本地可视化分析三端联动问题。系统由SRS流媒体服务器、GPU服务器和Local客户端组成支持通过RTMP协议传输实时视频对人、车、交通灯等目标进行检测并具备并发处理能力。资源压缩包共113个文件约54.48MB包含32个Python源码、39个编译生成的pyc文件、YOLOv3及tiny-yolo-voc等4个cfg模型配置、8个h5权重模型文件、3个PyQt5界面ui文件以及jpg/png样例图片、flv测试视频和说明文档结构清晰便于直接对照学习。已有130人学习下载。整套资料可直接用于课程设计、毕业设计或交通监控原型验证省去自行搭建流媒体与模型推理环境的繁琐过程适合希望快速掌握YOLO工程化落地与PyQt5客户端整合的读者。1. 推流-检测-显示三层拆开的交通路口监控YOLOv3、RTMP 与 PyQt5 的配合实际部署时最颠簸的环节往往不在模型精度而在视频流进出的方式。“拉流-检测-显示”三件事如果塞进同一个 Python 进程网络抖动会直接把 UI 一起拖死。这个基于 YOLOv3 PyQt5 开发的交通路口智能监控项目把链路拆成了 SRS 流媒体服务器、GPU 检测服务器、PyQt5 客户端三层远端视频通过 RTMP 统一收拢GPU 服务器用 YOLO 模型识别行人、车辆、交通灯客户端负责回显并用自带的三个 OCR 模型识别车牌。对已经跑通目标检测、想接真实监控流的 Python 开发者来说这套源码把工程化路上最绕的流媒体和并发问题完整演示了一遍值得从架构层面逐层拆开看。2. SRS 流媒体接入层RTMP 收流、转推与消费链路2.1 为什么中间要放一台 SRS而不是让 OpenCV 直接拉流如果只在本机跑一个 traffic.flv 做验证OpenCV 的 VideoCapture 直接读文件确实省事。但换到真实路口摄像头大概率以 RTSP 协议挂在远端网段。RTSP 的会话是点对点的一个消费端断开重连会影响采集端多个检测程序同时去拉同一路 RTSP摄像头侧并发压力也明显。更现实的问题是跨网段丢包时OpenCV 的读帧会阻塞在底层 socket 上检测线程和 UI 线程很容易一起挂起。SRSSimple Realtime Server在这套系统里相当于一个视频交换机摄像头或测试视频先把流推到 SRS再由 SRS 转发给任意多个下游。推流端统一用 RTMP 协议消费端也拉 RTMP格式统一断线重连只发生在消费端和 SRS 之间不会反向影响采集端。项目里自带 traffic.flv 和 traffic_2.flv 就是给这件事准备的输入材料原样推流即可模拟两路真实摄像头。提示SRS 的定位是解决“谁在收流、谁能同时看”的问题不要在 GPU 服务器上再跑第二个独立 RTSP 拉流器。2.2 SRS 的配置项与启动方式SRS 通过 srs.conf 描述监听端口、虚拟主机和缓存行为。最简配置是监听 1935 端口、关闭 GOP 缓存换取低延迟listen 1935; max_connections 1000; vhost __defaultVhost__ { gop_cache off; # 关闭 GOP 缓存 hls { enabled off; # 不切片为 HLS } }gop_cache是最关键的一项。打开时 SRS 会缓存关键帧往前的一段 GOP新客户端接入能秒开代价是缓存期间画面延迟明显增加。在客户端只做实时监控和检测回显的场合通常更看重低延迟所以关掉如果后续要给网页端做回放预览再单独开一个 vhost 打开缓存即可。启动方式常规是在编译好的二进制目录下执行./objs/srs -c conf/srs.conf没有现成二进制时也可以直接用官方镜像起一个容器把配置文件挂载进去端口映射到 1935 即可。启动后可以先用 ffmpeg 把项目自带的测试视频推进去模拟两路摄像头ffmpeg -re -i traffic.flv -c copy -f flv rtmp://127.0.0.1:1935/live/road_1 ffmpeg -re -i traffic_2.flv -c copy -f flv rtmp://127.0.0.1:1935/live/road_2-re表示按视频原始帧率读取不加速推流-c copy表示不重新编码直接拷贝CPU 开销几乎为零。实际摄像头接入时只需要把输入源换成本地 RTSP 地址并加-rtsp_transport tcp强制走 TCP避免 UDP 丢包引起的花屏。2.3 GPU 服务器侧拉流ffmpeg 转流代替 VideoCapture(rtmp://)检测程序从 SRS 拉流时很多人第一反应是cv2.VideoCapture(rtmp://127.0.0.1:1935/live/road_1)。这个写法在少数带 RTMP 扩展的 OpenCV 构建里能跑但 pip 安装的 opencv-python 默认并不保证包含 RTMP 解封装实际项目里经常出现读取返回空帧或直接段错误。更稳的做法是先启动一个 ffmpeg 子进程把 RTMP 流解码成原生帧从管道输出再用 numpy 接住。import subprocess import numpy as np import cv2 WIDTH, HEIGHT 1280, 720 cmd [ ffmpeg, -i, rtmp://127.0.0.1:1935/live/road_1, -f, rawvideo, -pix_fmt, bgr24, -s, f{WIDTH}x{HEIGHT}, - ] proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.DEVNULL) while True: raw proc.stdout.read(WIDTH * HEIGHT * 3) if len(raw) ! WIDTH * HEIGHT * 3: break frame np.frombuffer(raw, np.uint8).reshape((HEIGHT, WIDTH, 3)) # 到这里就拿到了与本地读文件一致的 BGR 帧管道输出的尺寸必须和-s指定的一致proc.stdout.read每次读一帧的字节数。stderr丢弃是因为 ffmpeg 的日志会干扰控制台如果遇到拉流异常先把stderr保留下来能快速定位协议错误。实际项目中分辨率不固定的情况可以先解析输入流信息再决定-s缩放目标否则网络分辨率变化会导致读帧错位。这个“ffmpeg 子进程 pipe”模式是后续目标检测的稳定数据入口。3. GPU 服务器的 YOLOv3 检测cfg 选型、OpenCV DNN 前向与目标过滤3.1 四个 cfg 文件的差异和选型依据目标检测这一步围绕 YOLOv3 的 Darknet 模型展开。项目目录里同时出现了 yolov3.cfg、yolo.cfg、yolo-voc.cfg、tiny-yolo-voc.cfg 四个配置它们描述的是同一个检测网络的不同出口形态不能混用。cfg 文件决定网络层数、anchor 尺寸、类别数和输入分辨率权重参数必须与 cfg 一一对应拿 VOC 权重的去加载 yolov3.cfg 会在第一层卷积就报维度错误。配置文件类别数典型输入尺寸适用场景yolov3.cfg80COCO 体系608x608 或 416x416默认检测识别 person、car、traffic lightyolo.cfg80COCO 体系416x416输入尺寸更小、推理更快的对照配置yolo-voc.cfg20VOC 体系416x416只关心行人、车辆等 VOC 类别时使用tiny-yolo-voc.cfg20VOC 体系416x416CPU 或低算力环境兜底交通路口最关注的三个目标——行人、普通车辆、交通灯在 COCO 类别表里分别是 person、car、traffic light这也是主配置选 yolov3.cfg 的原因。如果换用 VOC 体系交通灯就不在 20 类范围内只能退回识别 car、bus 等车辆目标监控维度会少一块。tiny 系列的主干网络更浅检测精度有下降但可以在无独显的机器上跑出可用帧率适合联调阶段做功能验证。提示用[net]段下的width和height字段确认当前 cfg 期望的输入尺寸代码里 blobFromImage 的缩放参数要与之对应权重文件体积大需要自行放到项目目录并保持与 cfg 同名。3.2 OpenCV DNN 加载 Darknet 模型并执行 forward加载和推理可以直接用 OpenCV DNN 模块它原生支持 Darknet 模型不用在 GPU 服务器上再编译一份 darknet 本体import cv2 import numpy as np cfg_path yolov3.cfg weights_path yolov3.weights # 自行放入权重文件 net cv2.dnn.readNetFromDarknet(cfg_path, weights_path) net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV) net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA) # CPU 环境改为 DNN_TARGET_CPU layer_names net.getLayerNames() out_layers [layer_names[i - 1] for i in net.getUnconnectedOutLayers()] blob cv2.dnn.blobFromImage(frame, 1 / 255.0, (416, 416), swapRBTrue, cropFalse) net.setInput(blob) outs net.forward(out_layers)readNetFromDarknet先读 cfg 构建计算图再读 weights 填充权重两步任一步失败都会返回空 net。getUnconnectedOutLayers返回 YOLO 最后三个输出层索引OpenCV 4.5.4 之后的返回类型由 int 改成数组上面这种layer_names[i - 1]的写法对两种版本都兼容。DNN_TARGET_CUDA依赖驱动和 CUDA 环境不对时会在 forward 阶段报错联调时先切回 CPU 排除模型本身的问题。3.3 三尺度输出重塑、置信度过滤与 NMS 去重输入 416x416 时YOLOv3 的三个输出尺度分别是 13x13、26x26、52x52小尺度负责大目标大尺度负责小目标最后叠加过滤def postprocess(outs, orig_w, orig_h, conf_thres0.5, nms_thres0.4): boxes, scores, class_ids [], [], [] for out in outs: for det in out: # det 是 85 维向量 obj_conf det[4] cls_scores det[5:] class_id int(np.argmax(cls_scores)) confidence obj_conf * cls_scores[class_id] if confidence conf_thres: continue # 前 4 个值是相对输入图的归一化坐标 cx, cy, w, h det[:4] * np.array([orig_w, orig_h] * 2) x, y int(cx - w / 2), int(cy - h / 2) boxes.append([x, y, int(w), int(h)]) scores.append(float(confidence)) class_ids.append(class_id) idxs cv2.dnn.NMSBoxes(boxes, scores, conf_thres, nms_thres) results [] if len(idxs) 0: for i in idxs.flatten(): results.append((boxes[i], class_ids[i], scores[i])) return resultsdet[4]是目标存在概率与每个类别的条件概率相乘才是最终置信度只取det[5:]里的最大值是最常见的错误来源。cv2.dnn.NMSBoxes做非极大值抑制去掉同一目标上的重复框。阈值选择上行人和车辆置信度给 0.4 左右够用交通灯相对小且远处模糊常要把 conf 降到 0.3 才不会漏检但误检也会变多。工程上一般把 conf 和 nms 放进配置项而不是写死方便在不同路口视角下调整。后处理出来之后按照 COCO 类别表 person0、car2、traffic light9 的编号把结果送到客户端就完成了检测侧的全部工作。4. PyQt5 客户端回显与车牌识别QThread 刷新、汉字分类和序列模型4.1 为什么视频帧不能在 UI 线程里直接跑PyQt5 客户端的主要任务是把检测结果和原画面同步显示。常见的错误是在窗口的 while 循环里直接做 read、detect、setPixmap这样代码很短但 UI 事件循环被阻塞拖动窗口、点击按钮都会卡住检测一旦变慢整个界面就假死。正确做法是把拉流和处理放进 QThread用信号把结果送回主线程。from PyQt5.QtCore import QThread, pyqtSignal import numpy as np import cv2 class DetectThread(QThread): frame_ready pyqtSignal(np.ndarray, list) # 原始帧 检测框 def __init__(self, stream_url, detect_func, parentNone): super().__init__(parent) self.stream_url stream_url self.detect_func detect_func self._running True def run(self): cap open_stream(self.stream_url) # 内部用 ffmpeg 子进程 while self._running: frame cap.read() if frame is None: continue boxes self.detect_func(frame) # 检测耗时较长 self.frame_ready.emit(frame, boxes)pyqtSignal声明在 QObject 子类内部emit 时若接收方在主线程Qt 会自动切换为队列连接数据传递是线程安全的不需要额外加锁。真正吃时间的detect_func留在子线程主线程只负责把信号里的帧转成 QImage 显示。这里要注意_running的退出条件窗口关闭时必须thread.stop()再thread.wait()否则进程会带着子线程退不干净。4.2 用 QPixmap 回显画面并叠加检测框子线程发过来的 frame 是 ndarrayPyQt5 显示需要先转成 QImagefrom PyQt5.QtGui import QImage, QPixmap def update_frame(self, frame, boxes): rgb cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) h, w, ch rgb.shape qimg QImage(rgb.data, w, h, w * ch, QImage.Format_RGB888).copy() self.video_label.setPixmap( QPixmap.fromImage(qimg).scaled( self.video_label.size(), Qt.KeepAspectRatio) ) self.draw_boxes(boxes)QImage构造时直接用了rgb.data但 ndarray 的内存生命周期不由 QImage 管理这里加.copy()强制深拷贝避免 frame 被回收后出现花屏。检测框的绘制不要在 numpy 上画拿到 QImage 之后用 QPainter 在 label 的 QPixmap 上叠画矩形线条和字体走 Qt 渲染高清屏下不会发虚。若还要叠加实时统计面板可以再挂一个 QWebEngineView 载入本地 HTML用 JavaScript 接收信号同步数据但这会明显增加内存占用视硬件条件取舍。4.3 车牌 OCR 的两条通路char_chi_sim 汉字分类与 GRU/RNN 序列解码项目里三个 h5 文件是车牌识别的核心ocr_plate_all_w_rnn_2.h5、ocr_plate_all_gru.h5是两条可替换的端到端序列识别模型char_chi_sim.h5是中文汉字字符分类模型。它们的配合关系是典型的“整体 局部”两段式识别先从检测画面里把车牌区域抠出来用序列模型识别整块车牌同时对首位汉字用 char_chi_sim 单独分类从省份简称集合里给出结果两边互补纠正。模型文件网络结构承担任务ocr_plate_all_w_rnn_2.h5双向 RNN 序列输出车牌字符概率序列端到端解码ocr_plate_all_gru.h5GRU 序列与上一模型功能重复显存占用更小char_chi_sim.h5单字符分类识别车牌首位的省份汉字及数字字母序列模型的输入是固定尺寸的车牌灰度图预处理常见做法是 resize 到宽高约 272x72 或 168x48 的区间再做归一化送入模型from tensorflow.keras.models import load_model seq_model load_model(ocr_plate_all_gru.h5) char_model load_model(char_chi_sim.h5) plate crop_plate(frame, boxes) # 从检测结果抠出车牌 plate_gray cv2.cvtColor(plate, cv2.COLOR_BGR2GRAY) plate_resized cv2.resize(plate_gray, (272, 72)) x plate_resized.astype(float32) / 255.0 x x[None, ..., None] # (1, 72, 272, 1) pred seq_model.predict(x)[0] # 序列预测 text ctc_greedy_decode(pred) # 贪心解码 去重ctc_greedy_decode是这套流程里最容易被忽略的一步网络每个时间步输出字符概率分布取 argmax 得到字符索引后还要把连续重复的字符合并再删掉空白符号才是可读的车牌字符串。模型训练时的字符表顺序必须与解码表一致否则会出现“识别结果全是乱码但训练准确率很高”的现象换模型文件时第一个要核对的就是类表顺序和 h5 分类层的顺序。5. 多路并发、低延迟调优与 Python 依赖安装排查5.1 并发收流时的批处理与丢帧策略GPU 服务器同时接多路 RTMP 流时一路视频开一个 DetectThread 的做法会浪费显存每路流都复制了一份模型图。常见做法是多个拉流线程共享帧队列聚合线程按时间戳攒够 batch 后用cv2.dnn.blobFromImages一次前向推理多张图blob cv2.dnn.blobFromImages( [f[0] for f in batch_frames], 1 / 255.0, (416, 416), swapRBTrue, cropFalse) net.setInput(blob) preds net.forward(out_layers)batch 推理能显著提升 GPU 利用率但显存占用也按 batch 数线性上涨4 路以下用 batch4 比较稳超过 8 路建议先量化显存余量。检测速度低于输入帧率时服务端主动丢帧比在 UI 端丢更好拉流线程只保留最近一帧检测线程空闲时才取保证消费侧永远用最新画面端到端延迟被压到检测耗时本身而不会累积视频队列。代码上就是queue.Queue(maxsize1)再配get()前清理旧帧不要用默认无界队列。5.2 依赖安装和越用越慢的排查这套系统依赖 PyQt5、opencv-python、numpy、h5py 和对应版本的 TensorFlow/Keras安装时固定 PyQt5 主版本避免 Qt 6 API 差异影响界面代码pip install -i https://pypi.tuna.tsinghua.edu.cn/simple \ pyqt55.15.10 opencv-python numpy h5py uv pip install --system pyqt55.15.10 opencv-pythonuv是依赖解析和下载并行化做得比较快的包管理器解决的是依赖冲突时的耗时问题。如果客户端越跑越慢先看三个位置ffmpeg 子进程是不是每帧都在重建PyQt5 是不是高频调用setPixmap导致 UI 过载SRS 是否积压了下游消费队列。前两个靠对象复用解决最后一个用 SRS 的 HTTP API 看各连接帧率即可定位。端到端延迟可以用 ffprobe 量首帧时间差做基准ffprobe -v error -show_entries formatstart_time -of csvp0 \ rtmp://127.0.0.1:1935/live/road_1把推流端写入时间与客户端收到首帧时间相减就得到整条链路的基础延迟。如果这个值超过 2 秒优先检查 SRS 的gop_cache是否被打开、播放端 ffmpeg 有没有带-fflags nobuffer这两处是 RTMP 链路里最常见的隐性缓存。本文还有配套的精品资源点击获取