如何解决PyQt/OpenCV/Python中图像分析速度过慢的问题

我要开发一款程序，实现摄像头画面显示、文本识别、与预设结构对比并在匹配时暂停的功能。目前已经用PyQt做了GUI，通过pyUIC转换，创建了VideoThread(QThread)类获取摄像头图像，MyWindow(QMainWindow)类负责GUI展示和启停按钮处理。

但我写的文本识别函数analyser运行很慢（约300ms），每帧都执行的话画面会和实时脱节。我想改成每次分析完成后用最新的帧再次执行分析（中间可能跳过2-3帧），但不清楚：

• 是否要把这个函数放进类里？
• 是否需要新开线程？
• 怎么让它获取当前最新的帧？

以下是我的代码：

import sys
from PyQt5.QtCore import QThread, pyqtSignal, Qt
from PyQt5.QtMultimedia import QCameraInfo
import cv2
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5 import QtWidgets, QtGui
from PyQt5.QtGui import QPixmap
import numpy as np
import skimage.exposure
from fct_mod import analyser

from Process import Ui_MainWindow

class VideoThread(QThread):
    change_pixmap_signal = pyqtSignal(np.ndarray)

    def __init__(self):
        super().__init__()

    def run(self):
        # 从摄像头捕获画面
        self._run_flag = True
        self.cap = cv2.VideoCapture(1, cv2.CAP_DSHOW)
        while self._run_flag:
            ret, cv_img = self.cap.read()
            if ret:
                self.change_pixmap_signal.emit(cv_img)
        # 关闭捕获系统
        self.cap.release()

    def stop(self):
        """将运行标志设为False并等待线程结束"""
        self._run_flag = False

class MyWindow(QMainWindow):
    def __init__(self):
        super(MyWindow, self).__init__()
        self.available_cameras = QCameraInfo.availableCameras()  # 获取可用摄像头

        self.m_ui = Ui_MainWindow()
        self.m_ui.setupUi(self)

        self.max_expo = 192
        self.initWindow()
        self.cv_img = None

    ########################################################################################################################
    #                                                   窗口初始化                                                         #
    ########################################################################################################################
    def initWindow(self):
        # 创建视频捕获线程
        self.thread = VideoThread()

        # 视频启动按钮
        self.m_ui.lbl_image.setText("点击开始")
        self.m_ui.bt_continuer.setText("开始")

        self.m_ui.bt_continuer.clicked.connect(self.ClickStartVideo)

        self.m_ui.sliderL.setProperty("value", 192)
        self.m_ui.statusbar.showMessage('准备就绪')


    ########################################################################################################################
    #                                                   按钮操作                                                           #
    ########################################################################################################################
    # 启动/停止视频按钮点击时触发（启动操作）
    def ClickStartVideo(self):
        # 更改标签颜色为浅蓝色
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickStartVideo)
        self.m_ui.statusbar.showMessage('视频运行中...')
        # 将按钮改为暂停
        self.m_ui.bt_continuer.setText('暂停')
        self.thread = VideoThread()
        self.thread.change_pixmap_signal.connect(self.update_image)

        # 启动线程
        self.thread.start()
        self.m_ui.bt_continuer.clicked.connect(self.thread.stop)  # 点击按钮停止视频
        self.m_ui.bt_continuer.clicked.connect(self.ClickStopVideo)

        # 断开滑块改变时的图像更新连接
        if self.cv_img is not None:
            self.m_ui.sliderL.valueChanged.disconnect(self.update_image_when_video_paused)

    # 启动/停止视频按钮点击时触发（停止操作）
    def ClickStopVideo(self):
        self.thread.change_pixmap_signal.disconnect()
        self.m_ui.bt_continuer.setText('开始')
        self.m_ui.statusbar.showMessage('准备就绪')
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickStopVideo)
        self.m_ui.bt_continuer.clicked.disconnect(self.thread.stop)
        self.m_ui.bt_continuer.clicked.connect(self.ClickStartVideo)

        # 连接滑块信号以更新图像
        self.m_ui.sliderL.valueChanged.connect(self.update_image_when_video_paused)

    ########################################################################################################################
    #                                                   功能操作                                                           #
    ########################################################################################################################

    def render_processed_image(self, cv_img, min_expo=50, max_expo=255):
        if min_expo != 50:
            print(min_expo)
        if max_expo != self.max_expo:
            print(f"max_expo已更改为 {max_expo}")
            self.max_expo = max_expo
        img_expo = skimage.exposure.rescale_intensity(cv_img, in_range=(min_expo, max_expo),
                                                      out_range=(0, 255)).astype(np.uint8)
        img_analyse, text_analyse = analyser(img_expo) # 该函数每帧运行速度过慢
        qt_img = self.convert_cv_qt(img_expo)
        self.m_ui.lbl_image.setPixmap(qt_img)
        qt_imgOCR = self.convert_cv_qt(img_analyse)
        self.m_ui.lbl_image.setPixmap(qt_imgOCR)

    def update_image(self, cv_img):
        """使用新的OpenCV图像更新图像标签"""
        self.cv_img = cv_img
        self.render_processed_image(self.cv_img, max_expo=self.m_ui.sliderL.value())

    def update_image_when_video_paused(self):
        self.render_processed_image(self.cv_img, max_expo=self.m_ui.sliderL.value())

    def convert_cv_qt(self, cv_img):
        """将OpenCV图像转换为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)
        p = convert_to_Qt_format.scaled(int(self.m_ui.centralwidget.width() / 2),
                                        int(self.m_ui.centralwidget.width() / 2),
                                        Qt.KeepAspectRatio)
        return QPixmap.fromImage(p)


if __name__ == '__main__':
    app = QApplication(sys.argv)
    win = MyWindow()
    win.show()
    sys.exit(app.exec())

核心思路 #

把文本识别任务放到独立线程里，避免阻塞UI和摄像头采集线程；同时维护一个"最新帧"的缓冲区，让识别线程每次取最新的帧处理，而不是逐帧等待。

具体实现步骤 #

1. 创建识别线程类
   新建一个继承QThread的RecognitionThread，专门处理文本识别任务。通过信号传递识别结果，线程内部循环检查是否有新帧需要处理，且保证每次只处理最新的帧。

2. 维护最新帧缓冲区
   在MyWindow里用一个带锁的变量保存最新帧（因为摄像头线程和识别线程会同时访问），避免多线程冲突。

3. 修改摄像头线程的帧传递逻辑
   摄像头线程只负责采集并更新最新帧，同时触发UI更新画面，不再直接调用识别函数。

4. 控制识别线程的启停
   和摄像头线程同步启停，识别完成后自动获取最新帧继续分析，直到匹配成功或用户暂停。

修改后的完整代码 #

import sys
from PyQt5.QtCore import QThread, pyqtSignal, Qt, QMutex, QMutexLocker
from PyQt5.QtMultimedia import QCameraInfo
import cv2
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5 import QtWidgets, QtGui
from PyQt5.QtGui import QPixmap
import numpy as np
import skimage.exposure
from fct_mod import analyser

from Process import Ui_MainWindow

class VideoThread(QThread):
    change_pixmap_signal = pyqtSignal(np.ndarray)

    def __init__(self, frame_mutex, latest_frame):
        super().__init__()
        self.frame_mutex = frame_mutex
        self.latest_frame = latest_frame

    def run(self):
        self._run_flag = True
        self.cap = cv2.VideoCapture(1, cv2.CAP_DSHOW)
        while self._run_flag:
            ret, cv_img = self.cap.read()
            if ret:
                # 更新最新帧，加锁避免冲突
                with QMutexLocker(self.frame_mutex):
                    self.latest_frame[:] = cv_img.copy()
                # 触发UI更新画面
                self.change_pixmap_signal.emit(cv_img)
        self.cap.release()

    def stop(self):
        self._run_flag = False
        self.wait()

class RecognitionThread(QThread):
    recognition_result_signal = pyqtSignal(np.ndarray, str)
    match_found_signal = pyqtSignal()

    def __init__(self, frame_mutex, latest_frame, slider):
        super().__init__()
        self.frame_mutex = frame_mutex
        self.latest_frame = latest_frame
        self.slider = slider
        self._run_flag = True
        self._pause_flag = False

    def run(self):
        while self._run_flag:
            if self._pause_flag:
                self.msleep(50)
                continue
            
            # 获取最新帧
            current_frame = None
            with QMutexLocker(self.frame_mutex):
                if self.latest_frame is not None and self.latest_frame.size > 0:
                    current_frame = self.latest_frame.copy()
            
            if current_frame is not None:
                # 执行图像预处理和识别
                img_expo = skimage.exposure.rescale_intensity(current_frame, 
                                                              in_range=(50, self.slider.value()),
                                                              out_range=(0, 255)).astype(np.uint8)
                img_analyse, text_analyse = analyser(img_expo)
                # 发送识别结果到UI线程
                self.recognition_result_signal.emit(img_analyse, text_analyse)
                
                # 此处添加预设结构对比逻辑，匹配成功则触发暂停
                # if text_analyse == 你的预设内容:
                #     self.match_found_signal.emit()
                #     self._pause_flag = True
            
            # 避免占用过多CPU，可根据需要调整
            self.msleep(10)

    def stop(self):
        self._run_flag = False
        self.wait()

    def resume(self):
        self._pause_flag = False

class MyWindow(QMainWindow):
    def __init__(self):
        super(MyWindow, self).__init__()
        self.available_cameras = QCameraInfo.availableCameras()

        self.m_ui = Ui_MainWindow()
        self.m_ui.setupUi(self)

        # 创建帧互斥锁和最新帧缓冲区
        self.frame_mutex = QMutex()
        self.latest_frame = np.array([])
        self.initWindow()

    def initWindow(self):
        self.m_ui.lbl_image.setText("点击开始")
        self.m_ui.bt_continuer.setText("开始")
        self.m_ui.bt_continuer.clicked.connect(self.ClickStartVideo)
        self.m_ui.sliderL.setProperty("value", 192)
        self.m_ui.statusbar.showMessage('准备就绪')

    def ClickStartVideo(self):
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickStartVideo)
        self.m_ui.statusbar.showMessage('视频运行中...')
        self.m_ui.bt_continuer.setText('暂停')

        # 初始化摄像头线程
        self.video_thread = VideoThread(self.frame_mutex, self.latest_frame)
        self.video_thread.change_pixmap_signal.connect(self.update_image)
        self.video_thread.start()

        # 初始化识别线程
        self.recognition_thread = RecognitionThread(self.frame_mutex, self.latest_frame, self.m_ui.sliderL)
        self.recognition_thread.recognition_result_signal.connect(self.update_recognition_result)
        self.recognition_thread.match_found_signal.connect(self.on_match_found)
        self.recognition_thread.start()

        self.m_ui.bt_continuer.clicked.connect(self.ClickStopVideo)
        if self.latest_frame.size > 0:
            self.m_ui.sliderL.valueChanged.disconnect(self.update_image_when_video_paused)

    def ClickStopVideo(self):
        self.video_thread.stop()
        self.recognition_thread.stop()
        self.video_thread.change_pixmap_signal.disconnect()
        self.recognition_thread.recognition_result_signal.disconnect()
        self.recognition_thread.match_found_signal.disconnect()

        self.m_ui.bt_continuer.setText('开始')
        self.m_ui.statusbar.showMessage('准备就绪')
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickStopVideo)
        self.m_ui.bt_continuer.clicked.connect(self.ClickStartVideo)
        self.m_ui.sliderL.valueChanged.connect(self.update_image_when_video_paused)

    def update_image(self, cv_img):
        qt_img = self.convert_cv_qt(cv_img)
        self.m_ui.lbl_image.setPixmap(qt_img)

    def update_recognition_result(self, img_analyse, text_analyse):
        # 更新识别后的图像显示
        qt_imgOCR = self.convert_cv_qt(img_analyse)
        self.m_ui.lbl_image.setPixmap(qt_imgOCR)
        # 可在此处添加文本识别结果的UI显示逻辑
        # self.m_ui.lbl_text_result.setText(text_analyse)

    def on_match_found(self):
        # 匹配成功后的暂停逻辑
        self.video_thread.stop()
        self.m_ui.statusbar.showMessage('匹配成功，已暂停')
        self.m_ui.bt_continuer.setText('继续')
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickStopVideo)
        self.m_ui.bt_continuer.clicked.connect(self.ClickResumeVideo)

    def ClickResumeVideo(self):
        self.m_ui.bt_continuer.clicked.disconnect(self.ClickResumeVideo)
        self.video_thread = VideoThread(self.frame_mutex, self.latest_frame)
        self.video_thread.change_pixmap_signal.connect(self.update_image)
        self.video_thread.start()
        self.recognition_thread.resume()
        self.m_ui.statusbar.showMessage('视频运行中...')
        self.m_ui.bt_continuer.setText('暂停')
        self.m_ui.bt_continuer.clicked.connect(self.ClickStopVideo)

    def update_image_when_video_paused(self):
        if self.latest_frame.size > 0:
            img_expo = skimage.exposure.rescale_intensity(self.latest_frame, 
                                                          in_range=(50, self.m_ui.sliderL.value()),
                                                          out_range=(0, 255)).astype(np.uint8)
            img_analyse, text_analyse = analyser(img_expo)
            qt_imgOCR = self.convert_cv_qt(img_analyse)
            self.m_ui.lbl_image.setPixmap(qt_imgOCR)

    def convert_cv_qt(self, cv_img):
        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)
        p = convert_to_Qt_format.scaled(int(self.m_ui.centralwidget.width() / 2),
                                        int(self.m_ui.centralwidget.width() / 2),
                                        Qt.KeepAspectRatio)
        return QPixmap.fromImage(p)

if __name__ == '__main__':
    app = QApplication(sys.argv)
    win = MyWindow()
    win.show()
    sys.exit(app.exec())

关键说明 #

• 线程分离：摄像头采集、UI更新、文本识别三个任务分别在不同线程执行，互不阻塞，保证画面流畅。
• 最新帧缓冲区：用QMutex保护最新帧的读写，避免多线程竞争问题，识别线程每次只取当前最新的帧，跳过中间的旧帧。
• 识别逻辑解耦：analyser函数无需修改结构，直接放到识别线程中调用即可。
• 匹配暂停逻辑：通过match_found_signal触发暂停，只需在注释位置添加你的预设结构对比代码。

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