拓冰网站定制 
简介目标检测是计算机视觉的核心任务之一旨在识别图像或视频中的特定物体并定位其位置。其原理是通过深度学习模型学习图像特征从而区分不同类别的目标。这项技术在安防监控、自动驾驶和智慧城市等领域具有极高的技术价值。在智慧交通场景中实时、准确的目标检测是实现车辆计数、行为分析等高级应用的基础。本文聚焦于如何利用YOLOv8这一高效的单阶段目标检测模型结合数据标注、模型训练与调优、以及基于跟踪的业务逻辑如虚拟检测线计数和违章行为规则判断构建一套完整的、可工程化部署的交通感知系统有效解决传统方案成本高、误检率高等痛点。1. 项目缘起从“数车”到“管车”的实战需求几年前我参与过一个智慧园区项目其中有个需求听起来很简单统计园区主干道每小时的车流量。当时团队的第一反应是找现成的交通摄像头方案结果报价单上的数字让人倒吸一口凉气。硬件采购、平台授权、后期维护加起来是个不小的数目。我们转而尝试用传统图像处理比如背景减除加轮廓检测白天光照稳定时还行一到傍晚或者树影晃动误检率就飙升把晃动的影子当成车把并排行驶的车数成一辆是家常便饭。正是这个有点“土法炼钢”的经历让我开始深入研究基于深度学习的目标检测来做交通感知。YOLOYou Only Look Once系列算法进入视野它的“单次检测”特性意味着速度快能满足实时性要求。从最初的YOLOv3到现在的YOLOv11我算是看着它一路迭代过来。这个“基于YOLO的交通流量统计、违章行为检测”项目可以说是我将这些年踩坑、调试、优化的经验打包成的一个可复现的实战方案。它不只是一个算法演示更是一套从数据准备、模型训练、到业务逻辑集成的完整工程化思路。简单说这个项目能帮你用普通的监控摄像头甚至是一些公开的交通流视频自动完成两件事一是数清楚有多少辆车、多少人经过生成流量报表二是识别出一些典型的违章行为比如车辆违规变道、行人闯红灯、机动车占用非机动车道等。它的核心价值在于把学术界前沿的YOLO模型以一种稳定、可落地的方式应用到实际的交通管理场景中成本远低于专用硬件方案灵活性却高得多。2. YOLO模型选型与部署为什么是v8而非盲目追新面对YOLO v5, v7, v8, v9, v10, v11……这么多版本新手最容易犯的错就是盲目追求版本号最高的。我的建议是对于交通检测这类成熟应用YOLOv8是目前平衡性能、易用性和社区支持的最佳选择。v9/v10/v11在特定任务或架构上有创新但v8的生态最为完善从数据标注工具如Roboflow到部署框架如ONNX, TensorRT, OpenVINO的支持都最好坑最少。2.1 环境搭建避开版本冲突的深坑模型训练和部署的环境是第一个拦路虎。很多人卡在torch和torchvision版本不匹配或者CUDA装不上。这里给出一个经过大量项目验证的稳定环境配置以Ubuntu 20.04/22.04或Windows WSL2为例# 1. 创建并激活独立的Python虚拟环境强烈建议避免污染系统环境 conda create -n traffic_yolo python3.8 -y conda activate traffic_yolo # 2. 根据你的CUDA版本安装PyTorch假设CUDA 11.3 pip install torch1.12.1cu113 torchvision0.13.1cu113 torchaudio0.12.1 --extra-index-url https://download.pytorch.org/whl/cu113 # 3. 安装Ultralytics YOLOv8 pip install ultralytics # 4. 安装其他必要的工具包 pip install opencv-python pillow matplotlib pandas seaborn scikit-learn注意ultralytics包已经包含了运行和训练YOLOv8所需的所有依赖。上面单独安装PyTorch是为了锁定一个我们已知稳定的版本组合。直接pip install ultralytics可能会安装最新版的PyTorch有时会引入未知兼容性问题。验证安装是否成功import torch print(torch.__version__) # 应显示 1.12.1 print(torch.cuda.is_available()) # 应显示 True from ultralytics import YOLO model YOLO(yolov8n.pt) # 加载一个纳米级预训练模型 print(model.info()) # 打印模型信息2.2 模型选择从“纳米”到“超大”的权衡YOLOv8提供了不同大小的模型后缀从n(nano)到x(extra large)。对于交通场景yolov8n.pt / yolov8s.pt适合边缘设备部署如Jetson Nano, Nvidia Jetson Orin NX。在1080p视频上使用TensorRT加速后帧率FPS可轻松超过30。缺点是对于小目标如远处车辆或密集场景拥堵车流的检测精度mAP会下降。yolov8m.pt我的主力推荐。它在精度和速度上取得了很好的平衡。在RTX 3060这类消费级显卡上推理速度也能达到每秒几十帧完全满足实时处理需求同时对于车辆、行人的检测精度足够高。yolov8l.pt / yolov8x.pt精度最高但模型体积大速度慢。除非你对检测精度有极端要求例如需要区分非常相似的车型或者用于生成高精度的训练数据自动标注否则在工程中不推荐。对于交通流量统计车辆和行人是主要目标。YOLO官方预训练模型在COCO数据集上训练已经包含了car、truck、bus、person等类别可以直接拿来用初始效果就不错。这也是YOLOv8的一大优势——开箱即用。3. 数据准备与标注你的模型上限由数据决定直接使用COCO预训练模型是快速启动的方法但要想在具体的路口场景中获得最佳效果针对性的数据微调Fine-tuning是必不可少的。因为你的摄像头角度、光照条件、车辆类型比如某个地区特有的三轮车都和通用数据集有差异。3.1 数据收集与清洗数据来源可以是公开数据集BDD100K、UA-DETRAC、Cityscapes。这些数据集质量高标注规范是很好的起点。你需要将其标注格式通常是COCO JSON或VOC XML转换为YOLO格式.txt文件。自采视频用自己的摄像头录制。这是最匹配实际场景的方式。录制时要注意覆盖不同时段早、中、晚、夜、不同天气晴、雨、雾、不同流量稀疏、拥堵。一个关键的清洗步骤是抽帧。没必要对视频每一帧都标注那会累死人。通常每秒抽1-2帧1-2 FPS就足够了。可以使用OpenCV简单实现import cv2 import os video_path your_traffic_video.mp4 output_dir extracted_frames os.makedirs(output_dir, exist_okTrue) cap cv2.VideoCapture(video_path) frame_count 0 saved_count 0 fps cap.get(cv2.CAP_PROP_FPS) extract_interval int(fps) # 每秒取一帧 while True: ret, frame cap.read() if not ret: break if frame_count % extract_interval 0: cv2.imwrite(os.path.join(output_dir, fframe_{saved_count:06d}.jpg), frame) saved_count 1 frame_count 1 cap.release()3.2 标注工具与技巧推荐使用Roboflow或LabelImg。Roboflow是线上平台功能强大支持团队协作和自动预处理、增强LabelImg是本地工具简单快捷。标注时的心得框要贴得紧 bounding box尽可能紧密地包围目标物体减少背景干扰。类别要统一 提前定义好类别比如car,bus,truck,motorcycle,person,bicycle。对于交通场景可以进一步细分car为car,van,SUV但前提是你的业务需要。处理遮挡 对于被部分遮挡的车辆仍然标注整个可见部分。YOLO有能力学习这种部分可见的模式。小目标处理 对于远处很小的车辆或行人如果小于几个像素比如10x10以下标注的意义不大模型也很难学习可以考虑忽略或通过提高输入图像分辨率来改善。标注完成后YOLO格式的每个.txt文件对应一张图片每行表示一个物体class_id x_center y_center width height。坐标和宽高都是相对于图片宽度和高度的归一化值0-1之间。3.3 数据组织与配置文件将数据按如下结构组织traffic_dataset/ ├── images/ │ ├── train/ │ │ ├── frame_000001.jpg │ │ └── ... │ └── val/ │ ├── frame_000501.jpg │ └── ... └── labels/ ├── train/ │ ├── frame_000001.txt │ └── ... └── val/ ├── frame_000501.txt └── ...然后创建一个数据集配置文件traffic_data.yaml# traffic_data.yaml path: /path/to/your/traffic_dataset # 数据集根目录 train: images/train # 训练集图片路径相对path val: images/val # 验证集图片路径相对path # 类别数量和名称 nc: 6 # 例如: car, bus, truck, motorcycle, person, bicycle names: [car, bus, truck, motorcycle, person, bicycle]4. 模型训练与调优不只是跑个命令那么简单有了数据训练似乎就是一行命令的事yolo train datatraffic_data.yaml modelyolov8m.pt epochs100。但要想获得一个鲁棒的模型里面的门道不少。4.1 关键训练参数解析yolo train datatraffic_data.yaml \ modelyolov8m.pt \ epochs100 \ imgsz640 \ batch16 \ workers4 \ patience20 \ lr00.01 \ lrf0.01 \ weight_decay0.0005 \ projecttraffic_detection \ nameexp_v1imgsz输入图像尺寸。默认640。如果你的目标如车辆在图像中占比较大可以尝试增大到832甚至1024有助于提升小目标检测精度但会显著增加显存消耗和训练时间。batch批大小。在显存允许的情况下尽可能设大有助于训练稳定。可以通过--batch -1开启自动批大小。patience早停耐心值。如果验证集指标在连续patience个epoch内没有提升则提前停止训练防止过拟合。lr0 lrf初始学习率和最终学习率因子。lrf0.01意味着最终学习率是初始的1%。对于微调任务lr0可以设小一点比如0.001避免破坏预训练好的特征。weight_decay权重衰减一种正则化手段防止模型过拟合。4.2 训练监控与问题诊断训练开始后Ultralytics会在project/name目录本例中为traffic_detection/exp_v1下生成大量有用文件weights/best.pt效果最好的模型权重。results.csv每个epoch的训练/验证指标记录。args.yaml本次训练的所有参数快照。最重要的看板是TensorBoard或内置的日志。你需要关注两个核心指标损失Losstrain/box_loss,train/cls_loss,val/box_loss。训练损失应稳步下降验证损失在后期应趋于平稳或缓慢上升过拟合迹象。如果验证损失很早就开始上升说明模型过拟合了需要增加数据增强、减小模型容量或加大正则化。精度Metricsmetrics/mAP50-95(B)。这是COCO标准下的平均精度是衡量模型好坏的核心指标。mAP50指IoU阈值为0.5时的精度mAP50-95是多个IoU阈值下的平均值更严格。如果训练指标出现异常比如mAP始终为0检查数据标注首先确认你的标注文件.txt内容是否正确类别ID是否在[0, nc-1]范围内。检查数据路径确保traffic_data.yaml中的路径正确并且images和labels文件夹下的文件能一一对应文件名相同后缀不同。降低学习率过大的学习率可能导致训练发散将lr0从0.01降到0.001或0.0001试试。关闭数据增强在训练命令中加入augmentFalse先用最原始的数据跑几个epoch看模型能否学到东西排除增强导致的干扰。4.3 模型验证与测试训练完成后用最佳模型在验证集上看看效果yolo val modeltraffic_detection/exp_v1/weights/best.pt datatraffic_data.yaml这个命令会输出详细的评估表格包括每个类别的精确率Precision、召回率Recall、mAP等。召回率低意味着很多目标没被检测出来漏检可能需要对这类目标补充训练数据。精确率低意味着很多误检把背景当成车可能需要清理训练数据中的错误标注或增加包含复杂背景的负样本没有目标的图片。5. 流量统计核心逻辑如何把“检测框”变成“统计数字”模型能框出每一帧里的车辆和行人了但这只是第一步。流量统计的核心是跨帧的目标关联也就是判断上一帧的“蓝色轿车”和这一帧的“蓝色轿车”是不是同一辆这样才能跟踪它的轨迹判断它是否穿过了一条我们设定的“计数线”。5.1 基于IOU的简单跟踪器SORT原理简化版对于不太拥挤、速度稳定的场景一个基于交并比IoU的简单跟踪器就够用了。思路是为每个检测到的目标分配一个唯一ID在下一帧计算新检测框与所有已有ID的上一帧框的IoU。如果IoU超过一个阈值如0.3就认为是同一个目标继承ID否则就认为是一个新目标分配新ID。import numpy as np from collections import OrderedDict class SimpleTracker: def __init__(self, iou_threshold0.3, max_disappeared5): self.next_object_id 0 self.objects OrderedDict() # 当前跟踪的对象 {id: [x1, y1, x2, y2]} self.disappeared OrderedDict() # 对象消失的帧数 {id: count} self.iou_thresh iou_threshold self.max_disappear max_disappeared def _calculate_iou(self, boxA, boxB): # 计算两个矩形框的IoU xA max(boxA[0], boxB[0]) yA max(boxA[1], boxB[1]) xB min(boxA[2], boxB[2]) yB min(boxA[3], boxB[3]) interArea max(0, xB - xA) * max(0, yB - yA) boxAArea (boxA[2] - boxA[0]) * (boxA[3] - boxA[1]) boxBArea (boxB[2] - boxB[0]) * (boxB[3] - boxB[1]) iou interArea / float(boxAArea boxBArea - interArea) return iou def update(self, detections): # detections: 当前帧检测到的所有框列表 [[x1,y1,x2,y2], ...] if len(detections) 0: # 没有检测到任何目标所有现有目标消失帧数1 for obj_id in list(self.disappeared.keys()): self.disappeared[obj_id] 1 if self.disappeared[obj_id] self.max_disappear: self._deregister(obj_id) return self.objects # 初始化当前帧的对象字典 current_objects OrderedDict() # 如果当前没有任何跟踪对象直接注册所有检测结果 if len(self.objects) 0: for box in detections: self._register(box, current_objects) else: # 计算现有对象与当前检测框的IoU矩阵 object_ids list(self.objects.keys()) object_boxes list(self.objects.values()) iou_matrix np.zeros((len(object_boxes), len(detections))) for i, obj_box in enumerate(object_boxes): for j, det_box in enumerate(detections): iou_matrix[i, j] self._calculate_iou(obj_box, det_box) # 关联逻辑为每个现有对象找IoU最大的检测框 rows iou_matrix.shape[0] cols iou_matrix.shape[1] used_cols set() for row in range(rows): col np.argmax(iou_matrix[row]) if iou_matrix[row, col] self.iou_thresh and col not in used_cols: obj_id object_ids[row] current_objects[obj_id] detections[col] used_cols.add(col) self.disappeared[obj_id] 0 # 重置消失计数器 else: # 未匹配到的现有对象消失帧数1 obj_id object_ids[row] self.disappeared[obj_id] 1 # 处理未匹配的检测框新出现的目标 for col in range(cols): if col not in used_cols: self._register(detections[col], current_objects) # 检查是否有对象消失太久需要注销 for obj_id in list(self.disappeared.keys()): if self.disappeared[obj_id] self.max_disappear: self._deregister(obj_id) self.objects current_objects return self.objects def _register(self, box, current_objects): object_id self.next_object_id self.objects[object_id] box self.disappeared[object_id] 0 current_objects[object_id] box self.next_object_id 1 def _deregister(self, object_id): del self.objects[object_id] del self.disappeared[object_id]5.2 虚拟检测线与流量计数有了跟踪器我们可以为每个目标保存其历史轨迹最近N帧的中心点坐标。流量计数就转化为一个几何问题判断目标的轨迹线段是否与我们预设的“虚拟线”相交。假设我们在画面中画了一条从点(x1, y1)到点(x2, y2)的计数线方向是从左到右只统计从左向右穿越的车辆。对于每个被跟踪的目标我们取它当前帧的中心点(cx_curr, cy_curr)和上一帧的中心点(cx_prev, cy_prev)形成一条轨迹线段。判断两条线段是否相交可以用向量叉乘的方法。一个更简单实用的方法是方向判断法我们只关心目标中心点从线的“一侧”运动到“另一侧”。可以计算点到直线的有向距离。设定线的方向向量计算点相对于线的位置叉乘的符号。如果当前帧和上一帧的位置符号相反且运动方向符合我们设定的计数方向比如从负到正则认为发生了一次穿越计数器加一。def count_crossing(tracker_objects, line_start, line_end, directionleft_to_right): 统计穿越虚拟线的目标数量。 tracker_objects: SimpleTracker返回的当前跟踪对象字典 {id: box} 我们需要在外部维护每个id的历史轨迹点列表例如 trajectories {id: [(cx, cy), ...]} line_start, line_end: 线的起点和终点 (x, y) direction: 计数方向例如 left_to_right, top_to_bottom count 0 # 计算线的方向向量和法向量 line_vec np.array(line_end) - np.array(line_start) # 对于‘从左到右’的线我们可以简单比较x坐标 if direction left_to_right: # 简化如果目标中心点的x坐标从小于线起点的x变为大于线终点的x则认为穿越 # 这里需要更精确的几何判断简化示例 for obj_id, box in tracker_objects.items(): cx (box[0] box[2]) / 2 cy (box[1] box[3]) / 2 # 获取上一帧中心点 (需要从轨迹历史中获取) prev_point trajectories.get(obj_id, [])[-1] if trajectories.get(obj_id) else None if prev_point: prev_cx, prev_cy prev_point # 简单判断如果上一帧在线左当前帧在线右则计数 if prev_cx line_start[0] and cx line_end[0]: count 1 # 更新轨迹历史 trajectories.setdefault(obj_id, []).append((cx, cy)) # 保持轨迹长度例如只保留最近10帧 if len(trajectories[obj_id]) 10: trajectories[obj_id].pop(0) return count, trajectories在实际工程中你需要维护一个全局的trajectories字典来记录每个ID的轨迹。同时为了避免重复计数一个目标在线附近抖动可以加入一个“冷却”机制比如同一个ID在穿越后的若干帧内不再计数。6. 违章行为检测的逻辑实现流量统计是基础违章检测则是更高阶的应用其核心是轨迹分析与规则判断。这里以“违规变道”和“机动车占用非机动车道”为例。6.1 违规变道检测在真实道路上变道通常伴随着转向灯但基于纯视觉的检测很难判断是否打了灯。我们通常定义一个更简单的规则车辆在短时间内连续穿越两条以上的虚拟车道线。实现步骤车道线标定在视频画面中手动或通过算法标出每条车道的大致分界线同样是两条点定义的线段。假设有 Lane1, Lane2, Lane3。轨迹记录如上一节所述为每辆车记录其中心点轨迹。状态机为每个跟踪ID维护一个状态。初始状态为in_lane_x在某个车道内。当检测到其轨迹穿越了某条车道线时状态变为crossing_from_x_to_y并记录时间戳。规则判断如果在很短的时间窗口内比如1秒一辆车的状态从in_lane_1变为crossing_from_1_to_2然后又迅速变为crossing_from_2_to_3中间没有稳定在in_lane_2的状态则判定为“连续变道”违规。同时可以计算变道时的角度如果角度过于尖锐类似“画龙”也可以作为危险驾驶的嫌疑信号。class LaneChangeDetector: def __init__(self, lane_lines, time_threshold1.0): lane_lines: 列表每个元素是一条车道线的定义例如 [((x1,y1), (x2,y2)), ...] self.lane_lines lane_lines self.time_thresh time_threshold self.vehicle_states {} # {id: {current_lane: idx, last_crossing_time: timestamp, crossing_history: []}} def update(self, tracker_objects, current_time, trajectories): violations [] for obj_id, box in tracker_objects.items(): cx, cy (box[0]box[2])/2, (box[1]box[3])/2 prev_point trajectories.get(obj_id, [])[-2] if len(trajectories.get(obj_id, [])) 2 else None if prev_point: prev_cx, prev_cy prev_point # 判断当前中心点在哪条车道简化根据x坐标区间判断 current_lane self._get_lane_index(cx) # 判断轨迹线段是否穿越了车道线 crossed_line_idx self._check_cross_line((prev_cx, prev_cy), (cx, cy)) if crossed_line_idx is not None: # 发生了穿越 state self.vehicle_states.get(obj_id, {current_lane: None, last_crossing_time: None, crossing_history: []}) # 记录这次穿越 state[crossing_history].append({line: crossed_line_idx, time: current_time}) # 保留最近几次穿越记录 state[crossing_history] state[crossing_history][-3:] # 检查是否在短时间内连续穿越了不同的线例如连续穿越线0和线1 if len(state[crossing_history]) 2: last_cross state[crossing_history][-2] curr_cross state[crossing_history][-1] if (curr_cross[time] - last_cross[time] self.time_thresh and curr_cross[line] ! last_cross[line]): # 疑似违规变道 violations.append(obj_id) self.vehicle_states[obj_id] state return violations6.2 机动车占用非机动车道检测这个规则更直接一些区域划分在画面中通过多边形Polygon标定出非机动车道的区域。类别与位置判断对于每个检测到的目标如果它的类别是car,bus,truck等机动车并且其检测框的底部中心点通常代表车辆的接地点落在了非机动车道多边形区域内。持续时间判断单帧落入可能是误检或正常借道。需要加入时间维度如果一辆机动车在非机动车道区域内连续存在超过N帧例如2秒对应60帧则判定为违章占用。这里的关键是点是否在多边形内的判断可以使用经典的射线法算法。def point_in_polygon(point, polygon): # polygon: 列表多边形的顶点坐标 [(x1,y1), (x2,y2), ...] x, y point n len(polygon) inside False p1x, p1y polygon[0] for i in range(1, n 1): p2x, p2y polygon[i % n] if y min(p1y, p2y): if y max(p1y, p2y): if x max(p1x, p2x): if p1y ! p2y: xinters (y - p1y) * (p2x - p1x) / (p2y - p1y) p1x if p1x p2x or x xinters: inside not inside p1x, p1y p2x, p2y return inside class BikeLaneOccupancyDetector: def __init__(self, bike_lane_polygon, min_frames30): self.bike_lane_poly bike_lane_polygon self.min_frames min_frames self.occupancy_counter {} # {obj_id: count} def update(self, tracker_objects, class_names): violations [] for obj_id, box in tracker_objects.items(): # 获取类别假设tracker_objects也存储了类别信息 cls_id tracker_objects[obj_id].get(class_id, -1) cls_name class_names[cls_id] if cls_id ! -1 else unknown # 只检查机动车 if cls_name in [car, bus, truck]: # 计算检测框底部中心点 bottom_center_x (box[0] box[2]) / 2 bottom_center_y box[3] # 框的底部y坐标 point (bottom_center_x, bottom_center_y) if point_in_polygon(point, self.bike_lane_poly): self.occupancy_counter[obj_id] self.occupancy_counter.get(obj_id, 0) 1 if self.occupancy_counter[obj_id] self.min_frames: violations.append(obj_id) else: # 不在区域内重置计数器 self.occupancy_counter[obj_id] 0 return violations7. 工程化部署与性能优化让模型在单台机器上跑通demo只是第一步要真正实用还需要考虑工程化部署。7.1 模型导出与加速YOLOv8训练出的.pt文件是PyTorch格式部署时最好转换为更高效的格式。导出为ONNXONNX是一个开放的模型格式可以被多种推理引擎支持。yolo export modelbest.pt formatonnx imgsz640 simplifyTrue参数simplifyTrue会应用ONNX Simplifier优化计算图有时能提升推理速度。使用TensorRT加速NVIDIA GPU这是性能提升最显著的一步。可以将ONNX模型进一步转换为TensorRT引擎.engine文件。# 使用trtexec工具TensorRT自带 trtexec --onnxbest.onnx --saveEnginebest.engine --fp16--fp16表示使用半精度浮点数能大幅提升速度且精度损失很小。在Python中可以使用TensorRT的Python API来加载和推理.engine文件。实测下来相比原生PyTorchTensorRT通常能有2-5倍的推理速度提升。使用OpenVINO加速Intel CPU/GPU如果你的部署环境是Intel的CPU或集成显卡OpenVINO是更好的选择。yolo export modelbest.pt formatopenvino这条命令会直接导出为OpenVINO的IR格式.xml和.bin文件然后可以使用OpenVINO的Runtime进行推理。7.2 构建实时处理流水线一个健壮的流水线需要处理视频流的读取、解码、推理、后处理、跟踪、业务逻辑计数/违章判断和结果输出保存/推流等多个环节。我推荐使用生产者-消费者模型和多线程/多进程来避免阻塞。import threading import queue import cv2 from ultralytics import YOLO import time class VideoProcessor: def __init__(self, video_source, model_path, output_pathNone): self.cap cv2.VideoCapture(video_source) self.model YOLO(model_path) # 可以是.pt, .onnx, .engine self.frame_queue queue.Queue(maxsize30) # 缓冲队列 self.result_queue queue.Queue() self.running True self.output_path output_path if output_path: fps self.cap.get(cv2.CAP_PROP_FPS) width int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) self.writer cv2.VideoWriter(output_path, cv2.VideoWriter_fourcc(*mp4v), fps, (width, height)) def frame_producer(self): 生产者线程读取视频帧 while self.running: ret, frame self.cap.read() if not ret: break # 如果队列满了等待 if self.frame_queue.full(): time.sleep(0.001) continue self.frame_queue.put(frame) self.cap.release() def frame_consumer(self): 消费者线程推理与处理 tracker SimpleTracker() flow_counter FlowCounter() # 假设的流量计数器 while self.running or not self.frame_queue.empty(): try: frame self.frame_queue.get(timeout1) except queue.Empty: continue # 1. 推理 results self.model(frame, imgsz640, verboseFalse)[0] detections [] for box in results.boxes: xyxy box.xyxy.cpu().numpy()[0] # 获取框坐标 conf box.conf.cpu().numpy()[0] # 置信度 cls int(box.cls.cpu().numpy()[0]) # 类别ID if conf 0.5: # 置信度阈值 detections.append(xyxy) # 可以在框上画类别和置信度 label f{results.names[cls]} {conf:.2f} cv2.rectangle(frame, (int(xyxy[0]), int(xyxy[1])), (int(xyxy[2]), int(xyxy[3])), (0,255,0), 2) cv2.putText(frame, label, (int(xyxy[0]), int(xyxy[1])-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 2) # 2. 跟踪 tracked_objects tracker.update(detections) # 3. 业务逻辑流量统计、违章检测 count, _ flow_counter.update(tracked_objects) # 将计数信息画在帧上 cv2.putText(frame, fVehicle Count: {count}, (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,0,255), 2) # 4. 将处理后的帧放入结果队列 self.result_queue.put(frame) def result_writer(self): 结果写入线程如果需保存视频 while self.running or not self.result_queue.empty(): try: frame self.result_queue.get(timeout1) except queue.Empty: continue if self.output_path: self.writer.write(frame) # 也可以在这里显示画面 cv2.imshow(Traffic Analysis, frame) if cv2.waitKey(1) 0xFF ord(q): self.running False if self.output_path: self.writer.release() cv2.destroyAllWindows() def run(self): # 启动线程 producer_thread threading.Thread(targetself.frame_producer) consumer_thread threading.Thread(targetself.frame_consumer) writer_thread threading.Thread(targetself.result_writer) producer_thread.start() consumer_thread.start() writer_thread.start() producer_thread.join() consumer_thread.join() writer_thread.join()这个流水线将耗时的I/O读帧和推理过程分离避免了因推理速度慢导致的视频卡顿。在实际部署中你甚至可以将读帧、推理、后处理与跟踪分别放在不同的线程或进程里并用多个队列连接充分利用多核CPU。7.3 性能瓶颈分析与优化当处理高清1080p或更高视频流时性能瓶颈可能出现在模型推理这是最耗时的部分。解决方案使用TensorRT/OpenVINO降低推理尺寸imgsz如从640降到320但会牺牲精度使用更小的模型如YOLOv8n。图像解码对于高分辨率视频cv2.VideoCapture().read()可能成为瓶颈。可以考虑使用硬件解码如NVIDIA的Video Codec SDK或更高效的解码库如ffmpeg-python。Python GILPython的多线程受全局解释器锁限制。对于CPU密集型的后处理、跟踪逻辑可以考虑使用多进程multiprocessing模块或者用Cython/Numba加速关键循环甚至将这部分逻辑用C实现再通过Python调用。一个实用的优化顺序是先确保模型推理用上了最快的引擎TensorRT然后检查视频解码是否占用了过多CPU最后再考虑将跟踪器等后处理逻辑用更高效的语言重写。在大多数情况下前两步优化已经能让单路1080p视频流在主流GPU上达到实时25 FPS处理。本文还有配套的精品资源点击获取 基于YOLOv8的交通流量统计与违章检测实战指南
简介目标检测是计算机视觉的核心任务之一旨在识别图像或视频中的特定物体并定位其位置。其原理是通过深度学习模型学习图像特征从而区分不同类别的目标。这项技术在安防监控、自动驾驶和智慧城市等领域具有极高的技术价值。在智慧交通场景中实时、准确的目标检测是实现车辆计数、行为分析等高级应用的基础。本文聚焦于如何利用YOLOv8这一高效的单阶段目标检测模型结合数据标注、模型训练与调优、以及基于跟踪的业务逻辑如虚拟检测线计数和违章行为规则判断构建一套完整的、可工程化部署的交通感知系统有效解决传统方案成本高、误检率高等痛点。1. 项目缘起从“数车”到“管车”的实战需求几年前我参与过一个智慧园区项目其中有个需求听起来很简单统计园区主干道每小时的车流量。当时团队的第一反应是找现成的交通摄像头方案结果报价单上的数字让人倒吸一口凉气。硬件采购、平台授权、后期维护加起来是个不小的数目。我们转而尝试用传统图像处理比如背景减除加轮廓检测白天光照稳定时还行一到傍晚或者树影晃动误检率就飙升把晃动的影子当成车把并排行驶的车数成一辆是家常便饭。正是这个有点“土法炼钢”的经历让我开始深入研究基于深度学习的目标检测来做交通感知。YOLOYou Only Look Once系列算法进入视野它的“单次检测”特性意味着速度快能满足实时性要求。从最初的YOLOv3到现在的YOLOv11我算是看着它一路迭代过来。这个“基于YOLO的交通流量统计、违章行为检测”项目可以说是我将这些年踩坑、调试、优化的经验打包成的一个可复现的实战方案。它不只是一个算法演示更是一套从数据准备、模型训练、到业务逻辑集成的完整工程化思路。简单说这个项目能帮你用普通的监控摄像头甚至是一些公开的交通流视频自动完成两件事一是数清楚有多少辆车、多少人经过生成流量报表二是识别出一些典型的违章行为比如车辆违规变道、行人闯红灯、机动车占用非机动车道等。它的核心价值在于把学术界前沿的YOLO模型以一种稳定、可落地的方式应用到实际的交通管理场景中成本远低于专用硬件方案灵活性却高得多。2. YOLO模型选型与部署为什么是v8而非盲目追新面对YOLO v5, v7, v8, v9, v10, v11……这么多版本新手最容易犯的错就是盲目追求版本号最高的。我的建议是对于交通检测这类成熟应用YOLOv8是目前平衡性能、易用性和社区支持的最佳选择。v9/v10/v11在特定任务或架构上有创新但v8的生态最为完善从数据标注工具如Roboflow到部署框架如ONNX, TensorRT, OpenVINO的支持都最好坑最少。2.1 环境搭建避开版本冲突的深坑模型训练和部署的环境是第一个拦路虎。很多人卡在torch和torchvision版本不匹配或者CUDA装不上。这里给出一个经过大量项目验证的稳定环境配置以Ubuntu 20.04/22.04或Windows WSL2为例# 1. 创建并激活独立的Python虚拟环境强烈建议避免污染系统环境 conda create -n traffic_yolo python3.8 -y conda activate traffic_yolo # 2. 根据你的CUDA版本安装PyTorch假设CUDA 11.3 pip install torch1.12.1cu113 torchvision0.13.1cu113 torchaudio0.12.1 --extra-index-url https://download.pytorch.org/whl/cu113 # 3. 安装Ultralytics YOLOv8 pip install ultralytics # 4. 安装其他必要的工具包 pip install opencv-python pillow matplotlib pandas seaborn scikit-learn注意ultralytics包已经包含了运行和训练YOLOv8所需的所有依赖。上面单独安装PyTorch是为了锁定一个我们已知稳定的版本组合。直接pip install ultralytics可能会安装最新版的PyTorch有时会引入未知兼容性问题。验证安装是否成功import torch print(torch.__version__) # 应显示 1.12.1 print(torch.cuda.is_available()) # 应显示 True from ultralytics import YOLO model YOLO(yolov8n.pt) # 加载一个纳米级预训练模型 print(model.info()) # 打印模型信息2.2 模型选择从“纳米”到“超大”的权衡YOLOv8提供了不同大小的模型后缀从n(nano)到x(extra large)。对于交通场景yolov8n.pt / yolov8s.pt适合边缘设备部署如Jetson Nano, Nvidia Jetson Orin NX。在1080p视频上使用TensorRT加速后帧率FPS可轻松超过30。缺点是对于小目标如远处车辆或密集场景拥堵车流的检测精度mAP会下降。yolov8m.pt我的主力推荐。它在精度和速度上取得了很好的平衡。在RTX 3060这类消费级显卡上推理速度也能达到每秒几十帧完全满足实时处理需求同时对于车辆、行人的检测精度足够高。yolov8l.pt / yolov8x.pt精度最高但模型体积大速度慢。除非你对检测精度有极端要求例如需要区分非常相似的车型或者用于生成高精度的训练数据自动标注否则在工程中不推荐。对于交通流量统计车辆和行人是主要目标。YOLO官方预训练模型在COCO数据集上训练已经包含了car、truck、bus、person等类别可以直接拿来用初始效果就不错。这也是YOLOv8的一大优势——开箱即用。3. 数据准备与标注你的模型上限由数据决定直接使用COCO预训练模型是快速启动的方法但要想在具体的路口场景中获得最佳效果针对性的数据微调Fine-tuning是必不可少的。因为你的摄像头角度、光照条件、车辆类型比如某个地区特有的三轮车都和通用数据集有差异。3.1 数据收集与清洗数据来源可以是公开数据集BDD100K、UA-DETRAC、Cityscapes。这些数据集质量高标注规范是很好的起点。你需要将其标注格式通常是COCO JSON或VOC XML转换为YOLO格式.txt文件。自采视频用自己的摄像头录制。这是最匹配实际场景的方式。录制时要注意覆盖不同时段早、中、晚、夜、不同天气晴、雨、雾、不同流量稀疏、拥堵。一个关键的清洗步骤是抽帧。没必要对视频每一帧都标注那会累死人。通常每秒抽1-2帧1-2 FPS就足够了。可以使用OpenCV简单实现import cv2 import os video_path your_traffic_video.mp4 output_dir extracted_frames os.makedirs(output_dir, exist_okTrue) cap cv2.VideoCapture(video_path) frame_count 0 saved_count 0 fps cap.get(cv2.CAP_PROP_FPS) extract_interval int(fps) # 每秒取一帧 while True: ret, frame cap.read() if not ret: break if frame_count % extract_interval 0: cv2.imwrite(os.path.join(output_dir, fframe_{saved_count:06d}.jpg), frame) saved_count 1 frame_count 1 cap.release()3.2 标注工具与技巧推荐使用Roboflow或LabelImg。Roboflow是线上平台功能强大支持团队协作和自动预处理、增强LabelImg是本地工具简单快捷。标注时的心得框要贴得紧 bounding box尽可能紧密地包围目标物体减少背景干扰。类别要统一 提前定义好类别比如car,bus,truck,motorcycle,person,bicycle。对于交通场景可以进一步细分car为car,van,SUV但前提是你的业务需要。处理遮挡 对于被部分遮挡的车辆仍然标注整个可见部分。YOLO有能力学习这种部分可见的模式。小目标处理 对于远处很小的车辆或行人如果小于几个像素比如10x10以下标注的意义不大模型也很难学习可以考虑忽略或通过提高输入图像分辨率来改善。标注完成后YOLO格式的每个.txt文件对应一张图片每行表示一个物体class_id x_center y_center width height。坐标和宽高都是相对于图片宽度和高度的归一化值0-1之间。3.3 数据组织与配置文件将数据按如下结构组织traffic_dataset/ ├── images/ │ ├── train/ │ │ ├── frame_000001.jpg │ │ └── ... │ └── val/ │ ├── frame_000501.jpg │ └── ... └── labels/ ├── train/ │ ├── frame_000001.txt │ └── ... └── val/ ├── frame_000501.txt └── ...然后创建一个数据集配置文件traffic_data.yaml# traffic_data.yaml path: /path/to/your/traffic_dataset # 数据集根目录 train: images/train # 训练集图片路径相对path val: images/val # 验证集图片路径相对path # 类别数量和名称 nc: 6 # 例如: car, bus, truck, motorcycle, person, bicycle names: [car, bus, truck, motorcycle, person, bicycle]4. 模型训练与调优不只是跑个命令那么简单有了数据训练似乎就是一行命令的事yolo train datatraffic_data.yaml modelyolov8m.pt epochs100。但要想获得一个鲁棒的模型里面的门道不少。4.1 关键训练参数解析yolo train datatraffic_data.yaml \ modelyolov8m.pt \ epochs100 \ imgsz640 \ batch16 \ workers4 \ patience20 \ lr00.01 \ lrf0.01 \ weight_decay0.0005 \ projecttraffic_detection \ nameexp_v1imgsz输入图像尺寸。默认640。如果你的目标如车辆在图像中占比较大可以尝试增大到832甚至1024有助于提升小目标检测精度但会显著增加显存消耗和训练时间。batch批大小。在显存允许的情况下尽可能设大有助于训练稳定。可以通过--batch -1开启自动批大小。patience早停耐心值。如果验证集指标在连续patience个epoch内没有提升则提前停止训练防止过拟合。lr0 lrf初始学习率和最终学习率因子。lrf0.01意味着最终学习率是初始的1%。对于微调任务lr0可以设小一点比如0.001避免破坏预训练好的特征。weight_decay权重衰减一种正则化手段防止模型过拟合。4.2 训练监控与问题诊断训练开始后Ultralytics会在project/name目录本例中为traffic_detection/exp_v1下生成大量有用文件weights/best.pt效果最好的模型权重。results.csv每个epoch的训练/验证指标记录。args.yaml本次训练的所有参数快照。最重要的看板是TensorBoard或内置的日志。你需要关注两个核心指标损失Losstrain/box_loss,train/cls_loss,val/box_loss。训练损失应稳步下降验证损失在后期应趋于平稳或缓慢上升过拟合迹象。如果验证损失很早就开始上升说明模型过拟合了需要增加数据增强、减小模型容量或加大正则化。精度Metricsmetrics/mAP50-95(B)。这是COCO标准下的平均精度是衡量模型好坏的核心指标。mAP50指IoU阈值为0.5时的精度mAP50-95是多个IoU阈值下的平均值更严格。如果训练指标出现异常比如mAP始终为0检查数据标注首先确认你的标注文件.txt内容是否正确类别ID是否在[0, nc-1]范围内。检查数据路径确保traffic_data.yaml中的路径正确并且images和labels文件夹下的文件能一一对应文件名相同后缀不同。降低学习率过大的学习率可能导致训练发散将lr0从0.01降到0.001或0.0001试试。关闭数据增强在训练命令中加入augmentFalse先用最原始的数据跑几个epoch看模型能否学到东西排除增强导致的干扰。4.3 模型验证与测试训练完成后用最佳模型在验证集上看看效果yolo val modeltraffic_detection/exp_v1/weights/best.pt datatraffic_data.yaml这个命令会输出详细的评估表格包括每个类别的精确率Precision、召回率Recall、mAP等。召回率低意味着很多目标没被检测出来漏检可能需要对这类目标补充训练数据。精确率低意味着很多误检把背景当成车可能需要清理训练数据中的错误标注或增加包含复杂背景的负样本没有目标的图片。5. 流量统计核心逻辑如何把“检测框”变成“统计数字”模型能框出每一帧里的车辆和行人了但这只是第一步。流量统计的核心是跨帧的目标关联也就是判断上一帧的“蓝色轿车”和这一帧的“蓝色轿车”是不是同一辆这样才能跟踪它的轨迹判断它是否穿过了一条我们设定的“计数线”。5.1 基于IOU的简单跟踪器SORT原理简化版对于不太拥挤、速度稳定的场景一个基于交并比IoU的简单跟踪器就够用了。思路是为每个检测到的目标分配一个唯一ID在下一帧计算新检测框与所有已有ID的上一帧框的IoU。如果IoU超过一个阈值如0.3就认为是同一个目标继承ID否则就认为是一个新目标分配新ID。import numpy as np from collections import OrderedDict class SimpleTracker: def __init__(self, iou_threshold0.3, max_disappeared5): self.next_object_id 0 self.objects OrderedDict() # 当前跟踪的对象 {id: [x1, y1, x2, y2]} self.disappeared OrderedDict() # 对象消失的帧数 {id: count} self.iou_thresh iou_threshold self.max_disappear max_disappeared def _calculate_iou(self, boxA, boxB): # 计算两个矩形框的IoU xA max(boxA[0], boxB[0]) yA max(boxA[1], boxB[1]) xB min(boxA[2], boxB[2]) yB min(boxA[3], boxB[3]) interArea max(0, xB - xA) * max(0, yB - yA) boxAArea (boxA[2] - boxA[0]) * (boxA[3] - boxA[1]) boxBArea (boxB[2] - boxB[0]) * (boxB[3] - boxB[1]) iou interArea / float(boxAArea boxBArea - interArea) return iou def update(self, detections): # detections: 当前帧检测到的所有框列表 [[x1,y1,x2,y2], ...] if len(detections) 0: # 没有检测到任何目标所有现有目标消失帧数1 for obj_id in list(self.disappeared.keys()): self.disappeared[obj_id] 1 if self.disappeared[obj_id] self.max_disappear: self._deregister(obj_id) return self.objects # 初始化当前帧的对象字典 current_objects OrderedDict() # 如果当前没有任何跟踪对象直接注册所有检测结果 if len(self.objects) 0: for box in detections: self._register(box, current_objects) else: # 计算现有对象与当前检测框的IoU矩阵 object_ids list(self.objects.keys()) object_boxes list(self.objects.values()) iou_matrix np.zeros((len(object_boxes), len(detections))) for i, obj_box in enumerate(object_boxes): for j, det_box in enumerate(detections): iou_matrix[i, j] self._calculate_iou(obj_box, det_box) # 关联逻辑为每个现有对象找IoU最大的检测框 rows iou_matrix.shape[0] cols iou_matrix.shape[1] used_cols set() for row in range(rows): col np.argmax(iou_matrix[row]) if iou_matrix[row, col] self.iou_thresh and col not in used_cols: obj_id object_ids[row] current_objects[obj_id] detections[col] used_cols.add(col) self.disappeared[obj_id] 0 # 重置消失计数器 else: # 未匹配到的现有对象消失帧数1 obj_id object_ids[row] self.disappeared[obj_id] 1 # 处理未匹配的检测框新出现的目标 for col in range(cols): if col not in used_cols: self._register(detections[col], current_objects) # 检查是否有对象消失太久需要注销 for obj_id in list(self.disappeared.keys()): if self.disappeared[obj_id] self.max_disappear: self._deregister(obj_id) self.objects current_objects return self.objects def _register(self, box, current_objects): object_id self.next_object_id self.objects[object_id] box self.disappeared[object_id] 0 current_objects[object_id] box self.next_object_id 1 def _deregister(self, object_id): del self.objects[object_id] del self.disappeared[object_id]5.2 虚拟检测线与流量计数有了跟踪器我们可以为每个目标保存其历史轨迹最近N帧的中心点坐标。流量计数就转化为一个几何问题判断目标的轨迹线段是否与我们预设的“虚拟线”相交。假设我们在画面中画了一条从点(x1, y1)到点(x2, y2)的计数线方向是从左到右只统计从左向右穿越的车辆。对于每个被跟踪的目标我们取它当前帧的中心点(cx_curr, cy_curr)和上一帧的中心点(cx_prev, cy_prev)形成一条轨迹线段。判断两条线段是否相交可以用向量叉乘的方法。一个更简单实用的方法是方向判断法我们只关心目标中心点从线的“一侧”运动到“另一侧”。可以计算点到直线的有向距离。设定线的方向向量计算点相对于线的位置叉乘的符号。如果当前帧和上一帧的位置符号相反且运动方向符合我们设定的计数方向比如从负到正则认为发生了一次穿越计数器加一。def count_crossing(tracker_objects, line_start, line_end, directionleft_to_right): 统计穿越虚拟线的目标数量。 tracker_objects: SimpleTracker返回的当前跟踪对象字典 {id: box} 我们需要在外部维护每个id的历史轨迹点列表例如 trajectories {id: [(cx, cy), ...]} line_start, line_end: 线的起点和终点 (x, y) direction: 计数方向例如 left_to_right, top_to_bottom count 0 # 计算线的方向向量和法向量 line_vec np.array(line_end) - np.array(line_start) # 对于‘从左到右’的线我们可以简单比较x坐标 if direction left_to_right: # 简化如果目标中心点的x坐标从小于线起点的x变为大于线终点的x则认为穿越 # 这里需要更精确的几何判断简化示例 for obj_id, box in tracker_objects.items(): cx (box[0] box[2]) / 2 cy (box[1] box[3]) / 2 # 获取上一帧中心点 (需要从轨迹历史中获取) prev_point trajectories.get(obj_id, [])[-1] if trajectories.get(obj_id) else None if prev_point: prev_cx, prev_cy prev_point # 简单判断如果上一帧在线左当前帧在线右则计数 if prev_cx line_start[0] and cx line_end[0]: count 1 # 更新轨迹历史 trajectories.setdefault(obj_id, []).append((cx, cy)) # 保持轨迹长度例如只保留最近10帧 if len(trajectories[obj_id]) 10: trajectories[obj_id].pop(0) return count, trajectories在实际工程中你需要维护一个全局的trajectories字典来记录每个ID的轨迹。同时为了避免重复计数一个目标在线附近抖动可以加入一个“冷却”机制比如同一个ID在穿越后的若干帧内不再计数。6. 违章行为检测的逻辑实现流量统计是基础违章检测则是更高阶的应用其核心是轨迹分析与规则判断。这里以“违规变道”和“机动车占用非机动车道”为例。6.1 违规变道检测在真实道路上变道通常伴随着转向灯但基于纯视觉的检测很难判断是否打了灯。我们通常定义一个更简单的规则车辆在短时间内连续穿越两条以上的虚拟车道线。实现步骤车道线标定在视频画面中手动或通过算法标出每条车道的大致分界线同样是两条点定义的线段。假设有 Lane1, Lane2, Lane3。轨迹记录如上一节所述为每辆车记录其中心点轨迹。状态机为每个跟踪ID维护一个状态。初始状态为in_lane_x在某个车道内。当检测到其轨迹穿越了某条车道线时状态变为crossing_from_x_to_y并记录时间戳。规则判断如果在很短的时间窗口内比如1秒一辆车的状态从in_lane_1变为crossing_from_1_to_2然后又迅速变为crossing_from_2_to_3中间没有稳定在in_lane_2的状态则判定为“连续变道”违规。同时可以计算变道时的角度如果角度过于尖锐类似“画龙”也可以作为危险驾驶的嫌疑信号。class LaneChangeDetector: def __init__(self, lane_lines, time_threshold1.0): lane_lines: 列表每个元素是一条车道线的定义例如 [((x1,y1), (x2,y2)), ...] self.lane_lines lane_lines self.time_thresh time_threshold self.vehicle_states {} # {id: {current_lane: idx, last_crossing_time: timestamp, crossing_history: []}} def update(self, tracker_objects, current_time, trajectories): violations [] for obj_id, box in tracker_objects.items(): cx, cy (box[0]box[2])/2, (box[1]box[3])/2 prev_point trajectories.get(obj_id, [])[-2] if len(trajectories.get(obj_id, [])) 2 else None if prev_point: prev_cx, prev_cy prev_point # 判断当前中心点在哪条车道简化根据x坐标区间判断 current_lane self._get_lane_index(cx) # 判断轨迹线段是否穿越了车道线 crossed_line_idx self._check_cross_line((prev_cx, prev_cy), (cx, cy)) if crossed_line_idx is not None: # 发生了穿越 state self.vehicle_states.get(obj_id, {current_lane: None, last_crossing_time: None, crossing_history: []}) # 记录这次穿越 state[crossing_history].append({line: crossed_line_idx, time: current_time}) # 保留最近几次穿越记录 state[crossing_history] state[crossing_history][-3:] # 检查是否在短时间内连续穿越了不同的线例如连续穿越线0和线1 if len(state[crossing_history]) 2: last_cross state[crossing_history][-2] curr_cross state[crossing_history][-1] if (curr_cross[time] - last_cross[time] self.time_thresh and curr_cross[line] ! last_cross[line]): # 疑似违规变道 violations.append(obj_id) self.vehicle_states[obj_id] state return violations6.2 机动车占用非机动车道检测这个规则更直接一些区域划分在画面中通过多边形Polygon标定出非机动车道的区域。类别与位置判断对于每个检测到的目标如果它的类别是car,bus,truck等机动车并且其检测框的底部中心点通常代表车辆的接地点落在了非机动车道多边形区域内。持续时间判断单帧落入可能是误检或正常借道。需要加入时间维度如果一辆机动车在非机动车道区域内连续存在超过N帧例如2秒对应60帧则判定为违章占用。这里的关键是点是否在多边形内的判断可以使用经典的射线法算法。def point_in_polygon(point, polygon): # polygon: 列表多边形的顶点坐标 [(x1,y1), (x2,y2), ...] x, y point n len(polygon) inside False p1x, p1y polygon[0] for i in range(1, n 1): p2x, p2y polygon[i % n] if y min(p1y, p2y): if y max(p1y, p2y): if x max(p1x, p2x): if p1y ! p2y: xinters (y - p1y) * (p2x - p1x) / (p2y - p1y) p1x if p1x p2x or x xinters: inside not inside p1x, p1y p2x, p2y return inside class BikeLaneOccupancyDetector: def __init__(self, bike_lane_polygon, min_frames30): self.bike_lane_poly bike_lane_polygon self.min_frames min_frames self.occupancy_counter {} # {obj_id: count} def update(self, tracker_objects, class_names): violations [] for obj_id, box in tracker_objects.items(): # 获取类别假设tracker_objects也存储了类别信息 cls_id tracker_objects[obj_id].get(class_id, -1) cls_name class_names[cls_id] if cls_id ! -1 else unknown # 只检查机动车 if cls_name in [car, bus, truck]: # 计算检测框底部中心点 bottom_center_x (box[0] box[2]) / 2 bottom_center_y box[3] # 框的底部y坐标 point (bottom_center_x, bottom_center_y) if point_in_polygon(point, self.bike_lane_poly): self.occupancy_counter[obj_id] self.occupancy_counter.get(obj_id, 0) 1 if self.occupancy_counter[obj_id] self.min_frames: violations.append(obj_id) else: # 不在区域内重置计数器 self.occupancy_counter[obj_id] 0 return violations7. 工程化部署与性能优化让模型在单台机器上跑通demo只是第一步要真正实用还需要考虑工程化部署。7.1 模型导出与加速YOLOv8训练出的.pt文件是PyTorch格式部署时最好转换为更高效的格式。导出为ONNXONNX是一个开放的模型格式可以被多种推理引擎支持。yolo export modelbest.pt formatonnx imgsz640 simplifyTrue参数simplifyTrue会应用ONNX Simplifier优化计算图有时能提升推理速度。使用TensorRT加速NVIDIA GPU这是性能提升最显著的一步。可以将ONNX模型进一步转换为TensorRT引擎.engine文件。# 使用trtexec工具TensorRT自带 trtexec --onnxbest.onnx --saveEnginebest.engine --fp16--fp16表示使用半精度浮点数能大幅提升速度且精度损失很小。在Python中可以使用TensorRT的Python API来加载和推理.engine文件。实测下来相比原生PyTorchTensorRT通常能有2-5倍的推理速度提升。使用OpenVINO加速Intel CPU/GPU如果你的部署环境是Intel的CPU或集成显卡OpenVINO是更好的选择。yolo export modelbest.pt formatopenvino这条命令会直接导出为OpenVINO的IR格式.xml和.bin文件然后可以使用OpenVINO的Runtime进行推理。7.2 构建实时处理流水线一个健壮的流水线需要处理视频流的读取、解码、推理、后处理、跟踪、业务逻辑计数/违章判断和结果输出保存/推流等多个环节。我推荐使用生产者-消费者模型和多线程/多进程来避免阻塞。import threading import queue import cv2 from ultralytics import YOLO import time class VideoProcessor: def __init__(self, video_source, model_path, output_pathNone): self.cap cv2.VideoCapture(video_source) self.model YOLO(model_path) # 可以是.pt, .onnx, .engine self.frame_queue queue.Queue(maxsize30) # 缓冲队列 self.result_queue queue.Queue() self.running True self.output_path output_path if output_path: fps self.cap.get(cv2.CAP_PROP_FPS) width int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) self.writer cv2.VideoWriter(output_path, cv2.VideoWriter_fourcc(*mp4v), fps, (width, height)) def frame_producer(self): 生产者线程读取视频帧 while self.running: ret, frame self.cap.read() if not ret: break # 如果队列满了等待 if self.frame_queue.full(): time.sleep(0.001) continue self.frame_queue.put(frame) self.cap.release() def frame_consumer(self): 消费者线程推理与处理 tracker SimpleTracker() flow_counter FlowCounter() # 假设的流量计数器 while self.running or not self.frame_queue.empty(): try: frame self.frame_queue.get(timeout1) except queue.Empty: continue # 1. 推理 results self.model(frame, imgsz640, verboseFalse)[0] detections [] for box in results.boxes: xyxy box.xyxy.cpu().numpy()[0] # 获取框坐标 conf box.conf.cpu().numpy()[0] # 置信度 cls int(box.cls.cpu().numpy()[0]) # 类别ID if conf 0.5: # 置信度阈值 detections.append(xyxy) # 可以在框上画类别和置信度 label f{results.names[cls]} {conf:.2f} cv2.rectangle(frame, (int(xyxy[0]), int(xyxy[1])), (int(xyxy[2]), int(xyxy[3])), (0,255,0), 2) cv2.putText(frame, label, (int(xyxy[0]), int(xyxy[1])-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 2) # 2. 跟踪 tracked_objects tracker.update(detections) # 3. 业务逻辑流量统计、违章检测 count, _ flow_counter.update(tracked_objects) # 将计数信息画在帧上 cv2.putText(frame, fVehicle Count: {count}, (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,0,255), 2) # 4. 将处理后的帧放入结果队列 self.result_queue.put(frame) def result_writer(self): 结果写入线程如果需保存视频 while self.running or not self.result_queue.empty(): try: frame self.result_queue.get(timeout1) except queue.Empty: continue if self.output_path: self.writer.write(frame) # 也可以在这里显示画面 cv2.imshow(Traffic Analysis, frame) if cv2.waitKey(1) 0xFF ord(q): self.running False if self.output_path: self.writer.release() cv2.destroyAllWindows() def run(self): # 启动线程 producer_thread threading.Thread(targetself.frame_producer) consumer_thread threading.Thread(targetself.frame_consumer) writer_thread threading.Thread(targetself.result_writer) producer_thread.start() consumer_thread.start() writer_thread.start() producer_thread.join() consumer_thread.join() writer_thread.join()这个流水线将耗时的I/O读帧和推理过程分离避免了因推理速度慢导致的视频卡顿。在实际部署中你甚至可以将读帧、推理、后处理与跟踪分别放在不同的线程或进程里并用多个队列连接充分利用多核CPU。7.3 性能瓶颈分析与优化当处理高清1080p或更高视频流时性能瓶颈可能出现在模型推理这是最耗时的部分。解决方案使用TensorRT/OpenVINO降低推理尺寸imgsz如从640降到320但会牺牲精度使用更小的模型如YOLOv8n。图像解码对于高分辨率视频cv2.VideoCapture().read()可能成为瓶颈。可以考虑使用硬件解码如NVIDIA的Video Codec SDK或更高效的解码库如ffmpeg-python。Python GILPython的多线程受全局解释器锁限制。对于CPU密集型的后处理、跟踪逻辑可以考虑使用多进程multiprocessing模块或者用Cython/Numba加速关键循环甚至将这部分逻辑用C实现再通过Python调用。一个实用的优化顺序是先确保模型推理用上了最快的引擎TensorRT然后检查视频解码是否占用了过多CPU最后再考虑将跟踪器等后处理逻辑用更高效的语言重写。在大多数情况下前两步优化已经能让单路1080p视频流在主流GPU上达到实时25 FPS处理。本文还有配套的精品资源点击获取