PyQt5视频显示与图像捕获双线程卡顿问题求助

问题现象 #

启动摄像头实时显示后，点击“Capture and Detect”按钮执行图像捕获与检测时，摄像头显示线程出现冻结。

问题根源 #

两个线程各自初始化cv2.VideoCapture(0)，同时抢占同一个摄像头设备资源。摄像头作为独占型设备，无法被多个进程/线程同时访问，导致显示线程无法正常读取帧，出现冻结。

修复方案 #

1. 仅在摄像头显示线程中打开摄像头，实时保存当前帧的副本
2. 捕获检测线程直接使用显示线程保存的帧，无需重新打开摄像头
3. 优化线程管理逻辑，避免重复创建线程和资源泄漏

修改后的完整代码（main.py） #

import sys
import cv2
import numpy as np
import time

from PyQt5 import QtGui, QtCore, QtWidgets
from PyQt5.QtCore import pyqtSlot, Qt, QThread, pyqtSignal, QMutex
from PyQt5.QtGui import QPixmap
from PyQt5.QtWidgets import QApplication, QMainWindow
from capture import Ui_MainWindow
from ultralytics import YOLO

model = YOLO("last.pt")

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.uic = Ui_MainWindow()
        self.uic.setupUi(self)

        self.uic.camera.clicked.connect(self.toggle_camera)
        self.uic.capture.clicked.connect(self.capture_and_detect)

        self.thread = {}
        self.current_frame = None  # 保存当前摄像头帧
        self.frame_mutex = QMutex()  # 线程锁，保证帧读写安全

    def toggle_camera(self):
        # 切换摄像头状态：未启动则启动，已启动则停止
        if 1 in self.thread and self.thread[1].isRunning():
            self.thread[1].stop()
            self.thread[1].wait()
            del self.thread[1]
            self.uic.camera.setText("Camera")
        else:
            self.thread[1] = CameraThread()
            self.thread[1].frame_signal.connect(self.update_camera_display)
            self.thread[1].start()
            self.uic.camera.setText("Stop")

    def capture_and_detect(self):
        # 仅当摄像头线程运行时才执行捕获
        if 1 not in self.thread or not self.thread[1].isRunning():
            return
        
        # 启动检测线程，传入当前帧
        if 2 in self.thread and self.thread[2].isRunning():
            return
        self.thread[2] = DetectionThread(self.current_frame)
        self.thread[2].result_signal.connect(self.update_detection_display)
        self.thread[2].start()

    def update_camera_display(self, frame):
        # 线程安全地更新当前帧
        self.frame_mutex.lock()
        self.current_frame = frame.copy()
        self.frame_mutex.unlock()
        
        qt_img = convert_cv_qt(frame)
        self.uic.screen_cam.setPixmap(qt_img)

    def update_detection_display(self, detected_frame):
        qt_img = convert_cv_qt(detected_frame)
        self.uic.screen_cap.setPixmap(qt_img)
        # 检测完成后清理线程
        if 2 in self.thread:
            self.thread[2].wait()
            del self.thread[2]

def convert_cv_qt(cv_img):
    """Convert from an opencv image to QPixmap"""
    rgb_image = cv2.cvtColor(cv_img, cv2.COLOR_BGR2RGB)
    h, w, ch = rgb_image.shape
    bytes_per_line = ch * w
    convert_to_Qt_format = QtGui.QImage(rgb_image.data, w, h, bytes_per_line, QtGui.QImage.Format_RGB888)
    return QPixmap.fromImage(convert_to_Qt_format)

class CameraThread(QThread):
    frame_signal = pyqtSignal(np.ndarray)
    def __init__(self):
        super().__init__()
        self.is_running = True

    def run(self):
        cap = cv2.VideoCapture(0)
        if not cap.isOpened():
            print("无法打开摄像头")
            return
        
        frame_count = 0
        start_time = time.time()
        while self.is_running:
            success, frame = cap.read()
            if not success:
                break
            
            # 计算并绘制FPS
            frame_count += 1
            elapsed_time = time.time() - start_time
            fps = frame_count / elapsed_time
            frame_with_fps = cv2.putText(frame, f'FPS: {round(fps, 2)}', 
                                       (50, 80), cv2.FONT_HERSHEY_SIMPLEX, 
                                       1, (255, 0, 0), 2)  # 缩小字体避免遮挡
            
            self.frame_signal.emit(frame_with_fps)
            # 控制帧率，避免占用过高资源
            self.msleep(16)  # 约60fps
        
        cap.release()
        print("摄像头线程已停止")

    def stop(self):
        self.is_running = False

class DetectionThread(QThread):
    result_signal = pyqtSignal(np.ndarray)
    def __init__(self, frame):
        super().__init__()
        self.frame = frame.copy()  # 复制帧，避免原帧被修改

    def run(self):
        if self.frame is None:
            return
        
        # 执行YOLO检测
        results = model(self.frame)
        detected_frame = results[0].plot()
        self.result_signal.emit(detected_frame)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    main_win = MainWindow()
    main_win.show()
    sys.exit(app.exec())

关键改动说明 #

• 共享摄像头资源：仅CameraThread负责打开和读取摄像头，实时保存当前帧到MainWindow的current_frame变量
• 线程安全：使用QMutex保证多线程对current_frame的读写操作安全，避免数据竞争
• 检测线程优化：DetectionThread初始化时直接传入当前帧，无需重新打开摄像头，避免资源冲突
• 状态切换：摄像头按钮改为切换状态（启动/停止），避免重复创建线程
• 帧率控制：在摄像头线程中加入msleep(16)，控制帧率约60fps，减少资源占用

内容的提问来源于stack exchange，提问作者QueueTi