我想要多个闭路电视观众。 在单个窗口中显示 20 个不同 RTSP 源的视频。 不使用PyQt5。 使用cv2.hconcat或np.hstack，程序被强制关闭。 所以我尝试使用多处理和 pygame。
下面是我尝试过的代码。
[相机.py]
import cv2
import multiprocessing as mp
import imutils
import pygame
import numpy as np
class CameraWidget():
"""Independent camera feed
Uses threading to grab IP camera frames in the background
@param width - Width of the video frame
@param height - Height of the video frame
@param stream_link - IP/RTSP/Webcam link
@param aspect_ratio - Whether to maintain frame aspect ratio or force into fraame
"""
def __init__(self, width, height, stream_link=0, aspect_ratio=False, parent=None, deque_size=1):
super(CameraWidget, self).__init__(parent)
# Initialize deque used to store frames read from the stream
self.deque = mp.Queue(deque_size)
# Slight offset is needed since PyQt layouts have a built in padding
# So add offset to counter the padding
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
# Flag to check if camera is valid/working
self.online = False
self.capture = None
self.load_network_stream()
# Start background frame grabbing
self.get_frame_thread = mp.Process(target=self.get_frame, args=())
self.get_frame_thread.daemon = True
self.get_frame_thread.start()
print('Started camera: {}'.format(self.camera_stream_link))
def load_network_stream(self):
"""Verifies stream link and open new stream if valid"""
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):
"""Attempts to receive a frame from given link"""
cap = cv2.VideoCapture(link)
if not cap.isOpened():
return False
cap.release()
return True
def get_frame(self):
"""Reads frame, resizes, and converts image to pixmap"""
while True:
try:
if self.capture.isOpened() and self.online:
# Read next frame from stream and insert into deque
status, frame = self.capture.read()
if status:
self.deque.append(frame)
else:
self.capture.release()
self.online = False
else:
# Attempt to reconnect
print('attempting to reconnect', self.camera_stream_link)
self.load_network_stream()
# self.spin(2)
# self.spin(.001)
except AttributeError:
pass
def set_frame(self):
"""Sets pixmap image to video frame"""
if not self.online:
# self.spin(1)
return
if self.deque and self.online:
# Grab latest frame
frame = self.deque[-1]
# Keep frame aspect ratio
if self.maintain_aspect_ratio:
self.frame = imutils.resize(frame, width=self.screen_width)
# Force resize
else:
self.frame = cv2.resize(frame, dsize=(384, 216), interpolation=cv2.INTER_AREA)
# Add timestamp to cameras
cv2.rectangle(self.frame, (self.screen_width - 190, 0), (self.screen_width, 50), color=(0, 0, 0),
thickness=-1)
frame = cv2.resize(frame, dsize=(384, 216), interpolation=cv2.INTER_AREA)
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = np.rot90(frame)
frame = pygame.surfarray.make_surface(frame)
def get_video_frame(self):
return self.video_frame
[主要.py]
import sys
from Camera import CameraWidget
import pygame
from pygame.locals import *
def exit_application():
"""Exit program event handler"""
sys.exit(1)
if __name__ == '__main__':
urls = ['rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]/axis-media/media.amp',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0']
cams=[]
for i in len(urls):
CameraWidget(1920 // 3, 1080 // 3, urls[i])
pygame.init()
screen = pygame.display.set_mode([1920, 1080])
try:
while True:
screen.blit(frame, (0, 0))
pygame.display.update()
for event in pygame.event.get():
if event.type == KEYDOWN:
sys.exit(0)
except (KeyboardInterrupt,SystemExit):
pygame.quit()
.
这是我参考的代码。
我想用 此代码使用pygame。
我认为你正在尝试创建一个多摄像头查看器、OpenCV 和 pygame。我至少在你的代码中发现了一些错误。 首先设置 pygame 窗口并为指定的 RTSP URL 初始化相机捕获对象。 然后创建单独的线程来捕获每个摄像头的帧并使用 pygame 显示它们。 这些框架在 pygame 窗口中并排排列。 修改后的代码允许您使用 pygame 在单个窗口中查看多个 RTSP 摄像头流。
import sys
import pygame
import cv2
import numpy as np
import imutils
import threading
# Define the RTSP URLs of the cameras you want to view
camera_urls = [
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
'rtsp://username:[email protected]/axis-media/media.amp',
'rtsp://username:[email protected]:554/cam/realmonitor?channel=1&subtype=0',
]
# Set up pygame
pygame.init()
screen_width = 1280
screen_height = 720
screen = pygame.display.set_mode((screen_width, screen_height))
clock = pygame.time.Clock()
# Create a list to store the camera capture objects
capture_objects = []
# Initialize the camera capture objects
for url in camera_urls:
cap = cv2.VideoCapture(url)
if not cap.isOpened():
print(f"Failed to open camera: {url}")
continue
capture_objects.append(cap)
# Create a list to store the camera frames
frames = [None] * len(capture_objects)
# Function to capture frames from each camera
def capture_frames(camera_index):
while True:
ret, frame = capture_objects[camera_index].read()
if not ret:
print(f"Camera {camera_index} disconnected.")
break
frames[camera_index] = frame
# Start threads to capture frames from each camera
threads = []
for i in range(len(capture_objects)):
thread = threading.Thread(target=capture_frames, args=(i,))
thread.daemon = True
thread.start()
threads.append(thread)
try:
running = True
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
screen.fill((0, 0, 0))
# Display frames from each camera
frame_width = screen_width // len(capture_objects)
frame_height = screen_height
for i, frame in enumerate(frames):
if frame is not None:
frame = imutils.resize(frame, width=frame_width)
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = np.rot90(frame)
frame_surface = pygame.surfarray.make_surface(frame)
screen.blit(frame_surface, (i * frame_width, 0))
pygame.display.update()
clock.tick(30) # Adjust the frame rate as needed
except KeyboardInterrupt:
pass
# Release camera capture objects and quit
for cap in capture_objects:
cap.release()
pygame.quit()
sys.exit()
替换 RTSP URL，此代码应该允许您使用 pygame 查看多个摄像头流。