基于PyQt的YOLO与DeepSORT车速与车流量检测系统

内容分享5天前发布
1 0 0

一、研究背景与意义

随着城市化进程的加速,道路交通管理面临着诸多挑战,准确快速地获取车流量数据和车速信息对交通管理和规划具有重要意义。传统的车流量检测方法主要依靠人工统计或地感线圈等硬件设备,存在成本高、覆盖范围有限、维护困难等问题。随着计算机视觉和深度学习技术的发展,基于视频图像的交通监测系统逐渐成为研究热点。

本课题旨在开发一种基于深度学习的车辆检测与跟踪系统,通过整合目前先进的计算机视觉算法YOLO(You Only Look Once)和DeepSORT(Deep Simple Online and Realtime Tracking)实现对车辆的实时检测、跟踪、车速测量和车流量统计,并结合PyQt5开发友好的用户界面,为交通监测提供一种低成本、高效率的解决方案。

该系统的研究意义主要体现在以下几个方面:

技术创新性:将目标检测与目标跟踪算法有机结合,实现车辆的连续跟踪和精确计数;应用价值:可为交通规划、道路设计和交通管理提供数据支持;经济性:相比传统硬件设备,基于视频的检测系统部署成本低;可扩展性:系统架构设计灵活,可根据实际需求进行功能扩展。

二、国内外研究现状

国外研究现状

国外在交通监测领域的研究起步较早,技术较为成熟。近年来,随着深度学习技术的快速发展,基于视频的交通监测系统取得了显著进展:

Joseph Redmon等人提出的YOLO系列算法在目标检测领域取得了突破性进展,最新的YOLOv5、YOLOv7和YOLOv8在检测速度和精度上都达到了较高水平;Nicolai Wojke等人提出的DeepSORT算法通过结合深度学习特征提取和传统的多目标跟踪算法,显著提高了车辆跟踪的稳定性和准确性;美国、德国等发达国家已将类似技术应用于智能交通系统中,用于交通流量监测、车速检测和交通事件检测等领域。

国内研究现状

国内在智能交通领域的研究也取得了较大进展:

清华大学、中国科学院等研究机构在目标检测与跟踪算法改进方面做了大量工作;华为、百度等企业推出了面向智能交通的解决方案,如华为的HiCarSight平台和百度的AI City解决方案;国内学者在针对复杂交通环境下的目标检测和跟踪方面提出了多种改进算法,以适应国内交通状况的特点。

尽管研究取得了很大进展,但目前仍存在一些问题:一是算法性能与实际应用场景的适配性有待提高;二是在复杂环境下(如恶劣天气、夜间等)的检测精度有待提升;三是针对普通用户的开源系统相对缺乏。因此,开发一套易于使用、功能完备的车辆检测与统计系统具有重要意义。

三、研究内容与目标

研究内容

车辆检测模块:基于YOLOv5目标检测算法实现对视频中车辆的实时检测,能够识别不同类型的车辆(如小汽车、摩托车、卡车、公交车等);车辆跟踪模块:利用DeepSORT算法对检测到的车辆进行跟踪,维持车辆ID的一致性,解决遮挡、交叉等复杂场景下的跟踪问题;车速计算模块:基于车辆的位置变化和时间信息,计算车辆的行驶速度,并进行可视化展示;车流量统计模块:设置虚拟检测线,统计通过该线的车辆数量,计算不同时间段的车流量,并生成统计报表;GUI界面设计:使用PyQt5开发用户友好的图形界面,实现视频播放、参数设置、检测结果显示、数据导出等功能;系统测试与优化:针对不同场景(如城市道路、高速公路等)进行测试,评估系统性能,并进行相应优化。

研究目标

功能目标
实现对视频中车辆的实时检测与分类实现车辆的连续跟踪与ID维持准确计算车辆速度统计车流量并生成分析报告提供友好的用户界面
性能目标
车辆检测准确率达到85%以上车辆跟踪成功率达到80%以上车速测量误差控制在10%以内在普通PC上实现实时或近实时处理(≥15fps)
拓展目标
支持摄像头实时输入支持检测结果的存储与回放提供数据导出功能(CSV、Excel等格式)支持系统参数的灵活配置

四、研究方法与技术路线

研究方法

文献调研法:通过查阅国内外相关文献,了解目标检测与跟踪算法的最新发展,为系统设计提供理论依据;模块化设计法:将系统分为检测、跟踪、车速计算、车流量统计等模块,独立开发与测试,最后集成;实验验证法:利用公开数据集和自采集视频进行测试,评估系统性能,不断优化算法;人机交互设计法:根据用户需求设计直观、易用的界面,提高系统可用性。

技术路线

平台与工具选择
编程语言:Python 3.8+深度学习框架:PyTorchGUI框架:PyQt5图像处理库:OpenCV数据处理库:NumPy, Pandas
核心算法选择
目标检测:YOLOv5目标跟踪:DeepSORT车速计算:基于像素位移和时间间隔的速度估算车流量统计:虚拟检测线计数法
系统架构设计
前端:PyQt5构建的GUI界面后端:检测与跟踪算法模块数据处理:车速计算与流量统计模块数据存储:检测结果与统计数据存储模块
实现流程
视频输入→车辆检测→车辆跟踪→车速计算→车流量统计→结果显示→数据存储

五、研究计划与进度安排

本课题的研究计划安排为期5个月,具体进度安排如下:

第一阶段(1个月):需求分析与技术调研

明确系统功能需求调研相关算法与技术搭建开发环境完成开题报告

第二阶段(1.5个月):算法实现与模块开发

实现YOLOv5车辆检测模块实现DeepSORT车辆跟踪模块实现车速计算模块实现车流量统计模块

第三阶段(1个月):GUI界面设计与系统集成

设计并实现PyQt5图形界面集成各功能模块实现数据存储与导出功能

第四阶段(1个月):系统测试与优化

收集测试视频数据进行系统功能测试评估系统性能指标根据测试结果进行优化

第五阶段(0.5个月):论文撰写与答辩准备

撰写毕业论文准备答辩材料完善系统文档

六、预期成果与创新点

预期成果

软件系统:一套完整的基于PyQt5的车速与车流量检测系统,包含源代码和可执行程序;技术文档:系统设计文档、用户手册和开发文档;研究论文:一篇关于系统设计与实现的毕业论文;测试数据:系统测试的视频数据和检测结果。

创新点

算法集成创新:将YOLOv5和DeepSORT算法有机结合,提高车辆检测与跟踪的准确性;应用场景创新:针对车流量统计和车速检测这一特定场景,优化算法参数和处理流程;交互设计创新:设计直观、易用的图形界面,降低系统使用门槛;系统架构创新:采用模块化设计,提高系统的可维护性和可扩展性。

七、可行性分析

技术可行性

算法成熟度:YOLOv5和DeepSORT均为成熟的开源算法,有大量实践案例;开发环境:Python生态系统完善,PyQt5、OpenCV等库功能强大且易于使用;硬件要求:普通PC或笔记本电脑配合GPU加速即可满足实时处理需求。

实现可行性

数据可获取性:可利用公开交通数据集进行算法训练和测试;时间充裕:按照进度安排,5个月时间足以完成系统开发与论文撰写;知识储备:已掌握计算机视觉、深度学习和Python编程等相关知识。

应用可行性

应用价值:系统可应用于交通监测、道路规划等领域;推广潜力:系统部署成本低,易于推广应用;拓展空间:系统架构具有良好的可扩展性,可根据需求增加新功能。

八、参考文献

Redmon, J., & Farhadi, A. (2018). YOLOv3: An incremental improvement. arXiv preprint arXiv:1804.02767.Wojke, N., Bewley, A., & Paulus, D. (2017). Simple online and realtime tracking with a deep association metric. In 2017 IEEE International Conference on Image Processing (ICIP) (pp. 3645-3649).Jocher, G., et al. (2021). YOLOv5. https://github.com/ultralytics/yolov5.Bewley, A., Ge, Z., Ott, L., Ramos, F., & Upcroft, B. (2016). Simple online and realtime tracking. In 2016 IEEE International Conference on Image Processing (ICIP) (pp. 3464-3468).Ren, S., He, K., Girshick, R., & Sun, J. (2015). Faster R-CNN: Towards real-time object detection with region proposal networks. In Advances in neural information processing systems (pp. 91-99).Dalal, N., & Triggs, B. (2005). Histograms of oriented gradients for human detection. In 2005 IEEE Computer Society Conference on Computer Vision and Pattern Recognition (CVPR) (Vol. 1, pp. 886-893).Riverbank Computing. (2021). PyQt5 Reference Guide. PyQt5 Reference Guide — PyQt Documentation v5.15.7.Bradski, G., & Kaehler, A. (2008). Learning OpenCV: Computer vision with the OpenCV library. O'Reilly Media, Inc.张良均, 李振平, 郭春宝. (2019). 基于YOLOv3的车辆检测与计数方法研究. 计算机工程与应用, 55(22), 209-214.王磊, 赵海鸥, 王尔德. (2020). 一种改进的深度学习目标跟踪算法在车流量统计中的应用. 交通信息与安全, 38(3), 107-115.

核心设计部分(仅供学习和参考)

该系统基于PyQt5构建界面,使用YOLOv5进行车辆检测,DeepSORT算法进行目标跟踪,并计算车速和统计车流量。

项目结构

vehicle_detection_system/
├── main.py                    # 主程序入口
├── config.py                  # 配置文件
├── modules/
│   ├── __init__.py           
│   ├── detector.py            # YOLO检测器模块
│   ├── tracker.py             # DeepSORT跟踪模块
│   ├── speed_calculator.py    # 车速计算模块
│   ├── traffic_counter.py     # 车流量统计模块
│   └── data_recorder.py       # 数据记录模块
├── ui/
│   ├── __init__.py
│   ├── main_window.py         # 主窗口UI
│   ├── video_widget.py        # 视频显示组件
│   └── statistics_widget.py   # 统计数据显示组件
└── utils/
    ├── __init__.py
    ├── draw_utils.py          # 绘图工具
    └── video_utils.py         # 视频处理工具

1. 配置文件 (config.py)

# config.py
class Config:
    # 模型配置
    YOLO_MODEL = 'yolov5s'
    YOLO_CONF_THRESH = 0.4
    YOLO_IOU_THRESH = 0.5
    VEHICLE_CLASSES = [2, 3, 5, 7]  # 车辆类别ID (car, motorcycle, bus, truck)
    
    # DeepSORT配置
    MAX_AGE = 30
    N_INIT = 3
    MAX_IOU_DISTANCE = 0.7
    MAX_COSINE_DISTANCE = 0.2
    NMS_MAX_OVERLAP = 0.5
    
    # 车速计算配置
    SPEED_ESTIMATION_FRAMES = 10
    PIXEL_TO_METER_RATIO = 0.05  # 每像素代表的实际距离(米)
    
    # 车流量统计配置
    COUNTING_LINE_POSITION = 0.5  # 计数线位置(屏幕高度的比例)
    
    # 视频配置
    DEFAULT_VIDEO_PATH = 'videos/highway.mp4'
    FRAME_RATE = 30
    
    # UI配置
    UI_UPDATE_INTERVAL = 30  # 毫秒
    MAIN_WINDOW_WIDTH = 1280
    MAIN_WINDOW_HEIGHT = 720
    
    # 数据记录配置
    SAVE_RESULTS = True
    RESULT_FOLDER = 'results/'

2. 检测模块 (modules/detector.py)

# modules/detector.py
import torch
import numpy as np
from config import Config

class YOLODetector:
    def __init__(self):
        """初始化YOLOv5检测器"""
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        print(f"Using device: {self.device}")
        
        # 加载YOLOv5模型
        self.model = torch.hub.load('ultralytics/yolov5', Config.YOLO_MODEL)
        self.model.to(self.device)
        self.model.conf = Config.YOLO_CONF_THRESH
        self.model.iou = Config.YOLO_IOU_THRESH
        self.model.classes = Config.VEHICLE_CLASSES  # 只检测车辆类别
        
        # 设置推理模式
        self.model.eval()
        
    def detect(self, frame):
        """
        对输入帧进行车辆检测
        
        参数:
            frame: 输入的图像帧
            
        返回:
            detections: 检测结果,格式为 [x1, y1, x2, y2, conf, cls]
        """
        # 使用YOLOv5进行推理
        results = self.model(frame)
        
        # 提取检测结果
        detections = results.xyxy[0].cpu().numpy()  # xyxy格式 [x1, y1, x2, y2, conf, cls]
        
        return detections
    
    def get_vehicle_count(self, detections):
        """获取当前帧中检测到的车辆数量"""
        return len(detections)

3. 跟踪模块 (modules/tracker.py)

# modules/tracker.py
import numpy as np
from deep_sort_pytorch.deep_sort import DeepSort
from config import Config

class VehicleTracker:
    def __init__(self):
        """初始化DeepSORT跟踪器"""
        # 初始化DeepSORT
        self.deepsort = DeepSort(
            model_type="resnet50",
            max_age=Config.MAX_AGE,
            n_init=Config.N_INIT,
            max_iou_distance=Config.MAX_IOU_DISTANCE,
            max_cosine_distance=Config.MAX_COSINE_DISTANCE,
            nn_budget=None,
            use_cuda=True
        )
        
        # 跟踪记录,用于车速计算
        self.tracks_history = {}  # 格式: {track_id: [pos1, pos2, ...]}
        
    def update(self, frame, detections):
        """
        使用DeepSORT更新跟踪结果
        
        参数:
            frame: 输入图像帧
            detections: YOLO检测结果,格式为 [x1, y1, x2, y2, conf, cls]
            
        返回:
            tracks: 跟踪结果,格式为 [track_id, x1, y1, x2, y2, cls]
        """
        if len(detections) == 0:
            return np.empty((0, 5))
        
        # 将YOLO检测结果转换为DeepSORT输入格式
        bbox_xywh = []
        confidences = []
        class_ids = []
        
        for x1, y1, x2, y2, conf, cls in detections:
            w = x2 - x1
            h = y2 - y1
            bbox_xywh.append([x1 + w/2, y1 + h/2, w, h])
            confidences.append(conf)
            class_ids.append(int(cls))
        
        bbox_xywh = np.array(bbox_xywh)
        
        # 更新跟踪器
        outputs = self.deepsort.update(bbox_xywh, confidences, class_ids, frame)
        
        # 更新跟踪历史
        for track in outputs:
            track_id = track[4]
            bbox = track[:4]  # [x1, y1, x2, y2]
            center = [(bbox[0] + bbox[2]) / 2, (bbox[1] + bbox[3]) / 2]  # 中心点 [x, y]
            
            if track_id not in self.tracks_history:
                self.tracks_history[track_id] = []
            
            self.tracks_history[track_id].append(center)
            
            # 限制历史记录长度
            if len(self.tracks_history[track_id]) > Config.SPEED_ESTIMATION_FRAMES:
                self.tracks_history[track_id].pop(0)
        
        return outputs

4. 车速计算模块 (modules/speed_calculator.py)

# modules/speed_calculator.py
import numpy as np
import math
from config import Config

class SpeedCalculator:
    def __init__(self, fps=None):
        """
        初始化车速计算器
        
        参数:
            fps: 视频帧率,如果为None则使用配置中的默认值
        """
        self.fps = fps if fps is not None else Config.FRAME_RATE
        self.pixel_to_meter = Config.PIXEL_TO_METER_RATIO
        
    def calculate_speed(self, track_history):
        """
        根据跟踪历史计算车辆速度
        
        参数:
            track_history: 车辆跟踪历史位置列表 [(x1,y1), (x2,y2), ...]
            
        返回:
            speed: 估计的车辆速度(km/h)
        """
        if len(track_history) < 2:
            return 0
        
        # 计算最近N帧的位移
        positions = np.array(track_history[-Config.SPEED_ESTIMATION_FRAMES:])
        if len(positions) < 2:
            return 0
        
        # 计算欧几里得距离
        distances = []
        for i in range(1, len(positions)):
            dist = math.sqrt((positions[i][0] - positions[i-1][0])**2 + 
                             (positions[i][1] - positions[i-1][1])**2)
            distances.append(dist)
        
        # 平均每帧移动的像素距离
        avg_pixel_distance = np.mean(distances)
        
        # 转换为实际距离(米)
        meter_distance = avg_pixel_distance * self.pixel_to_meter
        
        # 计算速度: 距离/时间
        # 时间 = 1/fps 秒/帧
        speed_ms = meter_distance * self.fps  # 米/秒
        speed_kmh = speed_ms * 3.6  # 千米/小时
        
        return speed_kmh
    
    def calibrate_pixel_to_meter(self, known_distance_pixels, known_distance_meters):
        """
        校准像素到实际距离的转换比例
        
        参数:
            known_distance_pixels: 已知的像素距离
            known_distance_meters: 对应的实际距离(米)
        """
        self.pixel_to_meter = known_distance_meters / known_distance_pixels
        print(f"像素到米的转换比例已更新为: {self.pixel_to_meter}")

5. 车流量统计模块 (modules/traffic_counter.py)

# modules/traffic_counter.py
import numpy as np
import time
from collections import defaultdict
from config import Config

class TrafficCounter:
    def __init__(self, frame_height, frame_width):
        """
        初始化车流量统计器
        
        参数:
            frame_height: 视频帧高度
            frame_width: 视频帧宽度
        """
        self.frame_height = frame_height
        self.frame_width = frame_width
        
        # 计数线的y坐标(屏幕高度的一定比例)
        self.counting_line_y = int(frame_height * Config.COUNTING_LINE_POSITION)
        
        # 已计数的车辆ID
        self.counted_vehicles = set()
        
        # 车流量统计
        self.total_count = 0
        self.up_count = 0
        self.down_count = 0
        
        # 时间统计
        self.start_time = time.time()
        self.hourly_counts = defaultdict(int)  # 按小时统计的车流量
        
        # 最近一段时间的车流量
        self.recent_times = []
        self.recent_counts = []
        
    def update(self, tracks, tracks_history):
        """
        更新车流量统计
        
        参数:
            tracks: 当前帧的跟踪结果
            tracks_history: 跟踪历史记录
            
        返回:
            新计数的车辆ID列表
        """
        newly_counted = []
        
        for track in tracks:
            track_id = track[4]
            
            # 如果该车辆已经被计数,则跳过
            if track_id in self.counted_vehicles:
                continue
                
            # 获取跟踪历史
            if track_id not in tracks_history:
                continue
                
            history = tracks_history[track_id]
            if len(history) < 2:
                continue
                
            # 检查是否穿过计数线
            prev_center = history[-2]
            curr_center = history[-1]
            
            prev_y = prev_center[1]
            curr_y = curr_center[1]
            
            # 如果前后两帧跨越了计数线,则计数+1
            if ((prev_y < self.counting_line_y and curr_y >= self.counting_line_y) or
                (prev_y > self.counting_line_y and curr_y <= self.counting_line_y)):
                
                self.counted_vehicles.add(track_id)
                self.total_count += 1
                newly_counted.append(track_id)
                
                # 区分向上和向下的车流
                if prev_y > curr_y:  # 向上移动
                    self.up_count += 1
                else:  # 向下移动
                    self.down_count += 1
                
                # 更新小时统计
                current_hour = time.strftime("%Y-%m-%d %H:00", time.localtime())
                self.hourly_counts[current_hour] += 1
                
                # 更新最近时间统计
                self.recent_times.append(time.time())
                self.recent_counts.append(self.total_count)
                
        return newly_counted
    
    def get_flow_rate(self, minutes=5):
        """
        计算最近n分钟的车流量
        
        参数:
            minutes: 计算最近多少分钟的车流量
            
        返回:
            flow_rate: 车流量(辆/小时)
        """
        now = time.time()
        threshold = now - minutes * 60
        
        # 过滤出最近n分钟的计数
        recent_count = sum(1 for t in self.recent_times if t >= threshold)
        
        # 计算车流量(辆/小时)
        if minutes > 0:
            flow_rate = (recent_count / minutes) * 60
        else:
            flow_rate = 0
            
        return flow_rate
    
    def get_statistics(self):
        """
        获取车流量统计数据
        
        返回:
            stats: 包含各种统计信息的字典
        """
        # 计算运行时间(小时)
        running_hours = (time.time() - self.start_time) / 3600
        
        # 计算平均车流量
        avg_flow_rate = self.total_count / running_hours if running_hours > 0 else 0
        
        stats = {
            'total_count': self.total_count,
            'up_count': self.up_count,
            'down_count': self.down_count,
            'avg_flow_rate': avg_flow_rate,
            'flow_rate_5min': self.get_flow_rate(5),
            'flow_rate_15min': self.get_flow_rate(15),
            'flow_rate_60min': self.get_flow_rate(60),
            'hourly_counts': dict(self.hourly_counts)
        }
        
        return stats

6. 数据记录模块 (modules/data_recorder.py)

# modules/data_recorder.py
import os
import csv
import json
import time
import cv2
import numpy as np
from config import Config

class DataRecorder:
    def __init__(self):
        """初始化数据记录器"""
        self.result_dir = Config.RESULT_FOLDER
        
        # 创建结果目录
        if not os.path.exists(self.result_dir):
            os.makedirs(self.result_dir)
            
        # 创建CSV文件用于记录车辆数据
        self.csv_path = os.path.join(self.result_dir, f'vehicle_data_{int(time.time())}.csv')
        self.video_path = os.path.join(self.result_dir, f'output_{int(time.time())}.mp4')
        
        # 初始化CSV文件
        with open(self.csv_path, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['时间戳', '车辆ID', '类别', '位置X', '位置Y', '速度(km/h)'])
        
        # 视频写入器
        self.video_writer = None
    
    def init_video_writer(self, frame_width, frame_height, fps):
        """初始化视频写入器"""
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        self.video_writer = cv2.VideoWriter(
            self.video_path, fourcc, fps, (frame_width, frame_height)
        )
    
    def record_frame(self, frame):
        """记录处理后的视频帧"""
        if self.video_writer is not None:
            self.video_writer.write(frame)
    
    def record_vehicle_data(self, timestamp, track_id, class_id, position, speed):
        """
        记录车辆数据到CSV文件
        
        参数:
            timestamp: 时间戳
            track_id: 车辆ID
            class_id: 车辆类别
            position: 位置坐标 (x, y)
            speed: 速度(km/h)
        """
        with open(self.csv_path, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                timestamp,
                track_id,
                class_id,
                position[0],
                position[1],
                speed
            ])
    
    def record_statistics(self, stats):
        """
        记录统计数据到JSON文件
        
        参数:
            stats: 统计数据字典
        """
        stats_path = os.path.join(self.result_dir, f'statistics_{int(time.time())}.json')
        with open(stats_path, 'w') as f:
            json.dump(stats, f, indent=4)
    
    def release(self):
        """释放资源"""
        if self.video_writer is not None:
            self.video_writer.release()

7. 绘图工具 (utils/draw_utils.py)

# utils/draw_utils.py
import cv2
import numpy as np
from config import Config

class DrawingUtils:
    # 不同车辆类别的颜色
    COLORS = {
        2: (0, 255, 0),    # 汽车 - 绿色
        3: (255, 0, 0),    # 摩托车 - 蓝色
        5: (0, 0, 255),    # 公交车 - 红色
        7: (255, 255, 0)   # 卡车 - 青色
    }
    
    @staticmethod
    def draw_bounding_boxes(frame, detections, speeds=None):
        """
        在图像上绘制检测框
        
        参数:
            frame: 输入图像
            detections: 检测结果 [x1, y1, x2, y2, conf, cls]
            speeds: 对应的车速列表(可选)
        
        返回:
            frame: 绘制后的图像
        """
        for i, det in enumerate(detections):
            x1, y1, x2, y2, conf, cls = det
            x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
            cls = int(cls)
            
            # 获取对应类别的颜色
            color = DrawingUtils.COLORS.get(cls, (128, 128, 128))
            
            # 绘制边界框
            cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
            
            # 如果有速度信息,则显示
            if speeds is not None and i < len(speeds):
                speed_text = f"{speeds[i]:.1f} km/h"
                cv2.putText(frame, speed_text, (x1, y1 - 10), 
                            cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
            
            # 显示类别和置信度
            cls_names = {2: 'Car', 3: 'Motorcycle', 5: 'Bus', 7: 'Truck'}
            label = f"{cls_names.get(cls, 'Unknown')} {conf:.2f}"
            cv2.putText(frame, label, (x1, y1 + 20), 
                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
            
        return frame
    
    @staticmethod
    def draw_tracks(frame, tracks, tracks_history, speeds=None):
        """
        在图像上绘制跟踪结果和轨迹
        
        参数:
            frame: 输入图像
            tracks: 跟踪结果 [track_id, x1, y1, x2, y2, cls]
            tracks_history: 跟踪历史记录
            speeds: 对应的车速字典 {track_id: speed}
        
        返回:
            frame: 绘制后的图像
        """
        for track in tracks:
            track_id = track[4]
            bbox = track[:4]  # [x1, y1, x2, y2]
            cls = track[5]
            
            x1, y1, x2, y2 = int(bbox[0]), int(bbox[1]), int(bbox[2]), int(bbox[3])
            
            # 获取对应类别的颜色
            color = DrawingUtils.COLORS.get(cls, (128, 128, 128))
            
            # 绘制边界框
            cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
            
            # 绘制ID
            id_text = f"ID: {track_id}"
            cv2.putText(frame, id_text, (x1, y1 - 30), 
                        cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
            
            # 如果有速度信息,则显示
            if speeds is not None and track_id in speeds:
                speed_text = f"{speeds[track_id]:.1f} km/h"
                cv2.putText(frame, speed_text, (x1, y1 - 10), 
                            cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2)
            
            # 绘制轨迹
            if track_id in tracks_history:
                history = tracks_history[track_id]
                for i in range(1, len(history)):
                    if i > 0:
                        cv2.line(frame, 
                                (int(history[i-1][0]), int(history[i-1][1])), 
                                (int(history[i][0]), int(history[i][1])), 
                                color, 2)
        
        return frame
    
    @staticmethod
    def draw_counting_line(frame, line_y):
        """
        绘制车流量计数线
        
        参数:
            frame: 输入图像
            line_y: 计数线的y坐标
            
        返回:
            frame: 绘制后的图像
        """
        h, w = frame.shape[:2]
        cv2.line(frame, (0, line_y), (w, line_y), (255, 0, 255), 2)
        cv2.putText(frame, "Counting Line", (10, line_y - 10), 
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 0, 255), 2)
        
        return frame
    
    @staticmethod
    def draw_statistics(frame, stats):
        """
        在图像上绘制统计信息
        
        参数:
            frame: 输入图像
            stats: 统计信息字典
            
        返回:
            frame: 绘制后的图像
        """
        # 在右上角绘制统计信息
        h, w = frame.shape[:2]
        
        info = [
            f"Total Vehicles: {stats['total_count']}",
            f"Upward: {stats['up_count']}",
            f"Downward: {stats['down_count']}",
            f"Flow Rate (5min): {stats['flow_rate_5min']:.1f} veh/h",
            f"Flow Rate (15min): {stats['flow_rate_15min']:.1f} veh/h",
            f"Flow Rate (60min): {stats['flow_rate_60min']:.1f} veh/h"
        ]
        
        # 创建半透明背景
        overlay = frame.copy()
        cv2.rectangle(overlay, (w - 300, 0), (w, 30 * len(info)), (0, 0, 0), -1)
        alpha = 0.7
        frame = cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0)
        
        # 绘制文本
        for i, text in enumerate(info):
            cv2.putText(frame, text, (w - 290, 25 + i * 30), 
                        cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
        
        return frame

8. 视频处理工具 (utils/video_utils.py)

# utils/video_utils.py
import cv2
import time
import numpy as np

class VideoUtils:
    @staticmethod
    def get_video_properties(video_path):
        """
        获取视频属性
        
        参数:
            video_path: 视频文件路径
            
        返回:
            属性字典 (width, height, fps, frame_count)
        """
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise ValueError(f"无法打开视频: {video_path}")
        
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
        fps = cap.get(cv2.CAP_PROP_FPS)
        frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        
        cap.release()
        
        return {
            'width': width,
            'height': height,
            'fps': fps,
            'frame_count': frame_count
        }
    
    @staticmethod
    def create_blank_frame(width, height, text="No Video Input"):
        """创建空白帧"""
        blank = np.zeros((height, width, 3), dtype=np.uint8)
        
        # 添加文本
        font = cv2.FONT_HERSHEY_SIMPLEX
        text_size = cv2.getTextSize(text, font, 1, 2)[0]
        text_x = (width - text_size[0]) // 2
        text_y = (height + text_size[1]) // 2
        
        cv2.putText(blank, text, (text_x, text_y), font, 1, (255, 255, 255), 2)
        
        return blank
    
    @staticmethod
    def resize_frame(frame, max_width=None, max_height=None):
        """
        调整图像大小,保持宽高比
        
        参数:
            frame: 输入图像
            max_width: 最大宽度
            max_height: 最大高度
            
        返回:
            调整大小后的图像
        """
        h, w = frame.shape[:2]
        
        if max_width is None and max_height is None:
            return frame
        
        if max_width is not None and max_height is not None:
            # 同时考虑最大宽度和高度
            scale = min(max_width / w, max_height / h)
        elif max_width is not None:
            # 只考虑最大宽度
            scale = max_width / w
        else:
            # 只考虑最大高度
            scale = max_height / h
        
        # 如果不需要缩放
        if scale >= 1:
            return frame
        
        new_w = int(w * scale)
        new_h = int(h * scale)
        
        return cv2.resize(frame, (new_w, new_h))
    
    @staticmethod
    def add_timestamp(frame):
        """在图像上添加时间戳"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        cv2.putText(frame, timestamp, (10, 30), 
                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
        return frame

9. 视频显示组件 (ui/video_widget.py)

# ui/video_widget.py
from PyQt5.QtWidgets import QWidget, QLabel, QVBoxLayout
from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtGui import QImage, QPixmap
import cv2
import numpy as np
from config import Config

class VideoDisplayWidget(QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.init_ui()
        
        # 当前显示的图像
        self.current_frame = None
        
    def init_ui(self):
        """初始化UI"""
        # 创建图像标签
        self.image_label = QLabel(self)
        self.image_label.

© 版权声明

相关文章

暂无评论

none
暂无评论...