如何用OpenCV、多进程与Pygame实现多RTSP监控画面展示？

我需要开发一个多CCTV监控查看器，在单个窗口中显示20个不同的RTSP视频源，且不使用PyQt5。此前尝试用cv2.hconcat或np.hstack实现时程序直接崩溃，因此改用多进程（multiprocessing）结合Pygame的方案。

以下是我编写的代码及修正后的版本：

Camera.py #

import cv2
import multiprocessing as mp
import imutils
import pygame
import numpy as np

class CameraWidget():
    """独立摄像头流
    使用多进程在后台抓取IP摄像头帧

    @param width - 视频帧宽度
    @param height - 视频帧高度
    @param stream_link - IP/RTSP/摄像头链接
    @param aspect_ratio - 是否保持帧的宽高比，还是强制适配指定尺寸
    """

    def __init__(self, width, height, stream_link=0, aspect_ratio=False, deque_size=1):
        # 初始化用于存储流读取帧的队列
        self.deque = mp.Queue(deque_size)

        # 预留偏移量（适配布局间隙）
        self.offset = 16
        self.screen_width = width - self.offset
        self.screen_height = height - self.offset
        self.maintain_aspect_ratio = aspect_ratio

        self.camera_stream_link = stream_link

        # 摄像头状态标记
        self.online = False
        self.capture = None

        self.load_network_stream()

        # 启动后台帧抓取进程
        self.get_frame_thread = mp.Process(target=self.get_frame, args=())
        self.get_frame_thread.daemon = True
        self.get_frame_thread.start()

        print(f'已启动摄像头: {self.camera_stream_link}')

    def load_network_stream(self):
        """验证流链接有效性，若有效则打开新流"""

        def load_network_stream_thread():
            if self.verify_network_stream(self.camera_stream_link):
                self.capture = cv2.VideoCapture(self.camera_stream_link)
                self.online = True

        self.load_stream_thread = mp.Process(target=load_network_stream_thread, args=())
        self.load_stream_thread.daemon = True
        self.load_stream_thread.start()

    def verify_network_stream(self, link):
        """尝试从给定链接获取一帧，验证流是否有效"""

        cap = cv2.VideoCapture(link)
        if not cap.isOpened():
            return False
        cap.release()
        return True

    def get_frame(self):
        """读取帧、调整尺寸，并将图像存入队列"""

        while True:
            try:
                if self.capture is not None and self.capture.isOpened() and self.online:
                    # 读取流的下一帧并加入队列
                    status, frame = self.capture.read()
                    if status:
                        # 队列满时自动丢弃旧帧
                        if self.deque.full():
                            self.deque.get()
                        self.deque.put(frame)
                    else:
                        self.capture.release()
                        self.online = False
                else:
                    # 尝试重连
                    print(f'尝试重连摄像头: {self.camera_stream_link}')
                    self.load_network_stream()
            except AttributeError:
                pass

    def get_processed_frame(self):
        """获取处理后的Pygame可用帧"""

        if not self.online:
            return None

        if not self.deque.empty():
            # 获取最新帧
            frame = self.deque.get()

            # 调整帧尺寸
            if self.maintain_aspect_ratio:
                frame = imutils.resize(frame, width=self.screen_width)
            else:
                frame = cv2.resize(frame, dsize=(384, 216), interpolation=cv2.INTER_AREA)

            # 添加时间戳
            cv2.rectangle(frame, (frame.shape[1] - 190, 0), (frame.shape[1], 50), color=(0, 0, 0), thickness=-1)
            cv2.putText(frame, str(cv2.getTickCount()//cv2.getTickFrequency()), 
                        (frame.shape[1]-180, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (255,255,255), 2)

            # 转换为Pygame兼容格式
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            frame = np.rot90(frame)
            return pygame.surfarray.make_surface(frame)
        
        return None

main.py #

import sys
from Camera import CameraWidget
import pygame
from pygame.locals import *

if __name__ == '__main__':
    # 替换为实际的20个RTSP链接
    urls = [
        'rtsp://username:password@192.168.1.43:554/cam/realmonitor?channel=1&subtype=0',
        'rtsp://username:password@192.168.1.45/axis-media/media.amp',
        # 补充剩余18个RTSP链接
    ]

    # 初始化摄像头实例列表
    cams = []
    # 20个摄像头采用5列4行布局，计算单帧尺寸
    cam_width = 1920 // 5
    cam_height = 1080 // 4
    for url in urls:
        cams.append(CameraWidget(cam_width, cam_height, url, aspect_ratio=True))

    # 初始化Pygame
    pygame.init()
    screen = pygame.display.set_mode((1920, 1080), pygame.HWSURFACE | pygame.DOUBLEBUF)
    clock = pygame.time.Clock()

    try:
        while True:
            # 填充背景为黑色
            screen.fill((0,0,0))

            # 遍历所有摄像头，绘制帧到对应位置
            for idx, cam in enumerate(cams):
                frame = cam.get_processed_frame()
                if frame is not None:
                    x = (idx % 5) * cam_width
                    y = (idx // 5) * cam_height
                    screen.blit(frame, (x, y))

            # 更新显示
            pygame.display.flip()
            clock.tick(30)  # 控制帧率

            # 处理退出事件
            for event in pygame.event.get():
                if event.type == QUIT or (event.type == KEYDOWN and event.key == K_ESCAPE):
                    pygame.quit()
                    sys.exit(0)

    except (KeyboardInterrupt, SystemExit):
        pygame.quit()
        sys.exit(0)

原代码问题修正说明 #

1. Camera.py：
   • 移除了无用的PyQt继承相关代码
   • 修复了帧获取与返回的逻辑错误，新增get_processed_frame方法直接输出Pygame可用帧
   • 优化队列操作，避免队列溢出
   • 补充时间戳的文本绘制逻辑
2. main.py：
   • 修复了摄像头实例创建的循环错误
   • 新增20个摄像头的合理布局（5列4行）
   • 修复了未定义帧变量的问题，实现多帧分位置绘制
   • 增加帧率控制与完善的退出事件处理

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