如何在PyQt(PySide6)应用中高效稳定处理多摄像头流存储

我正在开发PyQt(PySide6)应用，需同时显示和存储多摄像头流。显示功能正常，但存储操作会导致应用显著卡顿，甚至GUI无法流畅运行。以下是我的最小可复现代码：

import sys
from time import sleep

import av
import numpy as np
import pyqtgraph as pg
from PySide6.QtCore import QThread, Signal, Slot, Qt
from PySide6.QtWidgets import QApplication, QHBoxLayout, QWidget, QVBoxLayout, QPushButton, QGroupBox


class RGBCameraStub(QThread):

    newFrame = Signal(np.ndarray)

    def __init__(self):
        super().__init__()
        self.killSwitch = True

    def stop(self):
        self.killSwitch = False
        self.quit()
        self.wait()

    def run(self):
        self.killSwitch = True
        while self.killSwitch:
            self.newFrame.emit((np.random.rand(1456, 1080, 3) * 255).astype(np.uint8))
            sleep((20 + int(np.random.rand() * 30))/ 1000)


class VideoWriter(QThread):

    def __init__(self):
        super().__init__()
        self.output_container = av.open('output_video.mkv', mode='w')
        self.stream = self.output_container.add_stream('ffv1', rate=None)
        self.stream.width = 1456
        self.stream.height = 1080
        self.stream.pix_fmt = 'yuv420p'

    @Slot(np.ndarray)
    def addFrame(self, frame: np.ndarray):
        av_frame = av.VideoFrame.from_ndarray(frame, format='rgb24')
        av_frame.pts = None # Leave emtpy for auto-handling - variable framerate?
        for packet in self.stream.encode(av_frame):
            self.output_container.mux(packet)

    def stop(self):
        self.output_container.close()
        self.quit()
        self.wait()

    def run(self):
        self.exec()


class VideoBox(QGroupBox):

    def __init__(self, title):
        super().__init__(title=title)
        self.createLayout()
        self.videoWidget.setImage((np.random.rand(1456, 1080, 3) * 255).astype(np.uint8))

    def createLayout(self):
        layout = QVBoxLayout()
        self.videoWidget = pg.RawImageWidget()
        layout.addWidget(self.videoWidget)
        self.setLayout(layout)
        self.setStyleSheet("""QGroupBox {
            border: 1px solid #494B4F;
            margin-top: 8px;
            min-width: 180px;
            min-height: 180px;
            padding: 2px 0px 0px 0px;
            }
        QGroupBox::title {
            color: #aeb0b8;
            subcontrol-origin: margin;
            subcontrol-position: top left;
            left: 20px;
            padding: 0 8px;
        }""")

    def setImage(self, data: np.ndarray):
        self.videoWidget.setImage(data)

class MainWindow(QWidget):

    closeSignal = Signal()

    def __init__(self):
        super().__init__()
        self.setGeometry(0, 0, 900, 720)
        self.createLayout()

    def createLayout(self):
        self.vimbaImage = VideoBox("RGB")
        self.info = self.infoLayout()

        layout = QVBoxLayout()
        layout.addWidget(self.vimbaImage)
        layout.addWidget(self.info)
        self.setLayout(layout)

        self.setAttribute(Qt.WA_StyledBackground, True)
        self.setStyleSheet("MainWindow { background-color: #1e1f22; }")

    def infoLayout(self):
        widget = QWidget()
        layout = QVBoxLayout()

        rgbButtonWidget = QWidget()
        buttonLayout = QHBoxLayout()
        self.connectButton = QPushButton('Connect', parent=self)
        self.disconnectButton = QPushButton('Disconnect', parent=self)
        buttonLayout.addWidget(self.connectButton)
        buttonLayout.addWidget(self.disconnectButton)
        buttonLayout.addStretch()
        rgbButtonWidget.setLayout(buttonLayout)
        layout.addWidget(rgbButtonWidget)

        widget.setLayout(layout)
        return widget

    def closeEvent(self, event):
        self.closeSignal.emit()
        event.accept()



if __name__ == "__main__":
    app = QApplication(sys.argv)

    rgbCamera = RGBCameraStub()
    videoWriter = VideoWriter()
    videoWriter.start()

    main_window = MainWindow()

    # Button connections
    main_window.connectButton.clicked.connect(rgbCamera.start)
    main_window.disconnectButton.clicked.connect(rgbCamera.stop)
    # main_window.disconnectButton.clicked.connect(videoWriter.stop)

    # Display frames
    rgbCamera.newFrame.connect(main_window.vimbaImage.setImage)

    # Write frame to file
    rgbCamera.newFrame.connect(videoWriter.addFrame)

    # Close application
    main_window.closeSignal.connect(rgbCamera.stop)
    main_window.closeSignal.connect(videoWriter.stop)

    main_window.show()
    sys.exit(app.exec())

我的问题 #

1. 如何提升VideoWriter的性能？当前摄像头线程一输出帧就逐帧写入是否并非最优方案？
2. 摄像头帧率不稳定，我设置av_frame.pts = None是否合理？
3. 当前生成的视频文件体积迅速膨胀，有无无质量损失的解决办法？

注：我目前使用FFmpeg的PyAV封装，也接受其他方案建议。

1. 提升VideoWriter性能，优化写入逻辑 #

当前addFrame槽函数直接在信号触发线程执行编码写入，会阻塞主线程导致卡顿。最优方案是用队列缓冲帧+独立线程处理编码：

• 给VideoWriter加线程安全队列，槽函数仅负责存帧，不做耗时操作；
• 在run方法中批量取出队列中的帧，在独立线程完成编码写入。

修改后的VideoWriter示例：

from queue import Queue, Empty
from PySide6.QtCore import QMutex, QMutexLocker

class VideoWriter(QThread):
    def __init__(self):
        super().__init__()
        self.output_container = av.open('output_video.mkv', mode='w')
        self.stream = self.output_container.add_stream('ffv1', rate=None)
        self.stream.width = 1456
        self.stream.height = 1080
        self.stream.pix_fmt = 'yuv420p'
        self.frame_queue = Queue(maxsize=30)  # 限制队列大小防内存溢出
        self.mutex = QMutex()
        self.running = True

    @Slot(np.ndarray)
    def addFrame(self, frame: np.ndarray):
        # 仅存帧，不处理编码
        with QMutexLocker(self.mutex):
            if self.running and not self.frame_queue.full():
                self.frame_queue.put(frame.copy())  # 复制帧避免原数据被修改

    def stop(self):
        self.running = False
        self.frame_queue.put(None)  # 发送终止信号
        self.wait()
        # 编码剩余帧并关闭容器
        for packet in self.stream.encode():
            self.output_container.mux(packet)
        self.output_container.close()

    def run(self):
        while self.running:
            try:
                frame = self.frame_queue.get(timeout=0.1)
                if frame is None:
                    break
                # 在独立线程执行编码写入
                av_frame = av.VideoFrame.from_ndarray(frame, format='rgb24')
                av_frame.pts = None
                for packet in self.stream.encode(av_frame):
                    self.output_container.mux(packet)
            except Empty:
                continue

2. 不稳定帧率下的PTS设置 #

av_frame.pts = None不合理，PyAV不会自动处理PTS，会导致视频播放时序混乱。正确做法是基于实际时间戳计算PTS：

步骤1：修改摄像头线程，传递帧时间戳 #

from time import time

class RGBCameraStub(QThread):
    newFrame = Signal(np.ndarray, float)  # 添加时间戳参数

    def run(self):
        self.killSwitch = True
        while self.killSwitch:
            frame = (np.random.rand(1456, 1080, 3) * 255).astype(np.uint8)
            timestamp = time()  # 获取当前时间戳
            self.newFrame.emit(frame, timestamp)
            sleep((20 + int(np.random.rand() * 30))/ 1000)

步骤2：修改VideoWriter，计算PTS #

class VideoWriter(QThread):
    def __init__(self):
        super().__init__()
        # ... 其他初始化代码 ...
        self.start_time = None
        self.time_base = self.stream.time_base  # 视频流时间基准

    @Slot(np.ndarray, float)
    def addFrame(self, frame: np.ndarray, timestamp: float):
        with QMutexLocker(self.mutex):
            if self.running and not self.frame_queue.full():
                self.frame_queue.put((frame.copy(), timestamp))

    def run(self):
        while self.running:
            try:
                item = self.frame_queue.get(timeout=0.1)
                if item is None:
                    break
                frame, timestamp = item
                if self.start_time is None:
                    self.start_time = timestamp
                # 计算相对时间并转换为PTS
                relative_time = timestamp - self.start_time
                av_frame = av.VideoFrame.from_ndarray(frame, format='rgb24')
                av_frame.pts = int(relative_time / self.time_base)
                for packet in self.stream.encode(av_frame):
                    self.output_container.mux(packet)
            except Empty:
                continue

3. 无质量损失控制文件体积 #

方案1：优化FFV1参数 #

FFV1支持无损压缩级别调整，提升压缩率同时保持画质：

self.stream = self.output_container.add_stream('ffv1', rate=None)
self.stream.options = {'crf': '1'}  # 0-31，0为默认无损，1压缩率更高仍无损

方案2：切换到libx264rgb无损编码 #

libx264rgb的无损模式压缩率远高于FFV1，且专门适配RGB格式：

self.stream = self.output_container.add_stream('libx264rgb', rate=None)
self.stream.options = {'crf': '0', 'preset': 'fast'}  # crf=0为无损，preset控制编码速度

方案3：启用硬件加速编码 #

如果设备支持，用硬件加速（如NVIDIA nvenc、Intel qsv）提升编码速度，同时保持无损：

# NVIDIA nvenc示例
self.stream = self.output_container.add_stream('h264_nvenc', rate=None)
self.stream.options = {'crf': '0', 'preset': 'fast'}

内容的提问来源于stack exchange，提问作者Jeroen De Geeter