python 使用 process 多线程读取摄像头

多线程读取摄像头"/>

python 使用 process 多线程读取摄像头

要点：

参考： process读取多进程

参考：实时视频传入深度学习目标检测模型进行检测 - 知乎

一 实时读取视频流

解决实时读取延迟卡顿问题，实时读取多个网络摄像头。

读取视频：

import cv2
import time
import multiprocessing as mp# 导入必要的模块
def image_put(q, name, pwd, ip, channel=1):"""图像获取函数，将摄像头的图像放入队列中:param q: 队列，用于存储摄像头图像:param name: 摄像头用户名:param pwd: 摄像头密码:param ip: 摄像头IP地址:param channel: 摄像头通道号，默认为1"""# 尝试使用 HIKVISION 协议连接摄像头cap = cv2.VideoCapture("rtsp://%s:%s@%s//Streaming/Channels/%d" % (name, pwd, ip, channel))if cap.isOpened():print('HIKVISION')else:# 如果连接失败，则尝试使用 DaHua 协议连接摄像头cap = cv2.VideoCapture("rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (name, pwd, ip, channel))print('DaHua')while True:q.put(cap.read()[1])  # 从摄像头读取图像并放入队列q.get() if q.qsize() > 1 else time.sleep(0.01)  # 控制队列大小和图像获取的间隔def image_get(q, window_name):"""图像显示函数，从队列中获取图像并显示:param q: 队列，用于存储摄像头图像:param window_name: 窗口名称，用于显示图像"""cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)  # 创建一个命名窗口while True:frame = q.get()  # 从队列中获取图像cv2.imshow(window_name, frame)  # 显示图像cv2.waitKey(1)  # 等待按键响应，每隔1毫秒刷新窗口def run_multi_camera():"""多摄像头并行处理函数，用于设置摄像头的参数并启动多个图像获取和显示进程"""user_name, user_pwd = "admin", "admin123456"  # 设置摄像头的用户名和密码camera_ip_l = ["172.20.114.26",  # 摄像头IP地址列表，可以包含多个IP地址"[fe80::3aaf:29ff:fed3:d260]",  # 如果是IPv6地址，需要加上中括号# 添加你的摄像头IP地址到这里]mp.set_start_method(method='spawn')  # 设置多进程的启动方式为spawnqueues = [mp.Queue(maxsize=4) for _ in camera_ip_l]  # 创建多个队列，用于存储不同摄像头的图像processes = []for queue, camera_ip in zip(queues, camera_ip_l):# 创建图像获取和显示进程，并传递相应的参数processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))  # 创建图像获取进程processes.append(mp.Process(target=image_get, args=(queue, camera_ip)))  # 创建图像显示进程for process in processes:process.daemon = True  # 设置进程为守护进程，即主进程退出时自动结束子进程process.start()  # 启动进程for process in processes:process.join()  # 等待所有进程结束if __name__ == '__main__':run_multi_camera()  # 执行主函数

在 run_multi_camera 函数中，首先设置摄像头的用户名和密码，并创建一个存储不同摄像头图像的队列列表。然后，使用 mp.set_start_method 函数设置多进程的启动方式为 spawn 。接下来，通过循环遍历摄像头IP地址列表，为每个摄像头创建一个图像获取进程和一个图像显示进程，并将队列和相应的参数传递给它们。然后，将所有进程设置为守护进程，并依次启动它们。最后，使用 process.join() 等待所有进程结束。

1.1 解决实时读取延迟卡顿

关键代码如下，我使用Python自带的多线程队列：

import multiprocessing as mp
...
img_queues = [mp.Queue(maxsize=2) for _ in camera_ip_l]  # 创建多个队列，每个摄像头对应一个队列
...
q.put(frame) if is_opened else None  # 如果摄像头打开成功，则将帧放入队列中，线程A负责将图片放入队列
q.get() if q.qsize() > 1 else time.sleep(0.01)  # 如果队列中的帧数量大于1，则从队列中移除旧的帧，线程A负责移除队列中的旧图
...

这段代码使用了 multiprocessing.Queue 创建了多个队列，每个摄像头对应一个队列。这些队列用于在不同的进程之间传递帧数据。

在第一个注释中，img_queues是一个列表，通过列表推导式创建了多个队列，其中maxsize=2表示每个队列的最大容量为2，即最多可以存放2帧数据。

在第二个注释中，根据条件判断是否将帧放入队列中。is_opened表示摄像头是否打开成功，如果打开成功，则将当前帧frame放入对应的队列q中。

在第三个注释中，根据队列的大小判断是否需要从队列中移除旧的帧。q.qsize()返回队列中的帧数量，如果大于1，则调用q.get()从队列中移除旧的帧；否则，通过time.sleep(0.01)暂停0.01秒，等待新的帧放入队列。这样可以保持队列中始终只有最新的帧数据，避免队列过大导致内存占用过高。

这些操作是在多个线程或进程之间并行执行的，通过队列进行线程间通信，实现了帧数据的生产和消费，保证了实时的视频显示。

二 详细解读

2.1 入门OpenCV读取方式

def run_opencv_camera():video_stream_path = 0  # local camera (e.g. the front camera of laptop)cap = cv2.VideoCapture(video_stream_path)while cap.isOpened():is_opened, frame = cap.read()cv2.imshow('frame', frame)cv2.waitKey(1)cap.release()

当我们需要读取网络摄像头的时候，我们可以对 cap = cv2.VideoCapture(括号里面的东西进行修改)，填写上我们想要读取的视频流，它可以是：

• 数字0，代表计算机的默认摄像头（例如上面提及的笔记本前置摄像头）
• video.avi 视频文件的路径，支持其他格式的视频文件
• rtsp路径（不同品牌的路径一般是不同的，如下面举出的海康与大华，详细情况查看 附录的「关于rtsp协议」）

user, pwd, ip, channel = "admin", "admin123456", "172.20.114.26", 1video_stream_path = 0  # local camera (e.g. the front camera of laptop)
video_stream_path = 'video.avi'  # the path of video file
video_stream_path = "rtsp://%s:%s@%s/h265/ch%s/main/av_stream" % (user, pwd, ip, channel)  # HIKIVISION old version 2015
video_stream_path = "rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel)  # HIKIVISION new version 2017
video_stream_path = "rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel)  # dahuacap = cv2.VideoCapture(video_stream_path)

直接使用参考官网写出来的简单版Demo有延迟卡顿问题，如果读取速度低于视频流的输出速度，窗口显示的图片是好几秒钟前的内容。一段时间过后，缓存区将会爆满，程序报错，我可以使用rtsp读取摄像头：

2.2 使用rtsp读取摄像头

def run_opencv_camera():video_stream_path = 0  # local camera (e.g. the front camera of laptop)cap = cv2.VideoCapture(video_stream_path)while cap.isOpened():is_opened, frame = cap.read()cv2.imshow('frame', frame)cv2.waitKey(1000)  # wait for 1000ms(1s) HERE!!!!!!!!!!!!!!cap.release()

2.3 使用多线程队列，解决延迟卡顿问题

import cv2
import time
import multiprocessing as mpdef image_put(q, user, pwd, ip, channel=1):# 使用rtsp协议连接摄像头cap = cv2.VideoCapture("rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel))if cap.isOpened():print('HIKVISION')else:# 如果连接失败，尝试使用不同的URL格式连接摄像头cap = cv2.VideoCapture("rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel))print('DaHua')while True:# 读取视频帧并将其放入队列中q.put(cap.read()[1])# 如果队列中的帧数超过阈值，移除最早的帧q.get() if q.qsize() > 1 else time.sleep(0.01)def image_get(q, window_name):# 创建窗口来显示视频帧cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)while True:# 从队列中获取视频帧并显示在窗口中frame = q.get()cv2.imshow(window_name, frame)cv2.waitKey(1)def run_single_camera():user_name, user_pwd, camera_ip = "admin", "admin123456", "172.20.114.26"mp.set_start_method(method='spawn')  # 初始化多进程方法queue = mp.Queue(maxsize=2)  # 创建一个队列，用于存储视频帧processes = [mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)),mp.Process(target=image_get, args=(queue, camera_ip))]# 启动并等待子进程完成[process.start() for process in processes][process.join() for process in processes]def run_multi_camera():user_name, user_pwd = "admin", "admin123456"camera_ip_l = ["172.20.114.26",  # ipv4"[fe80::3aaf:29ff:fed3:d260]",  # ipv6]mp.set_start_method(method='spawn')  # 初始化多进程方法queues = [mp.Queue(maxsize=4) for _ in camera_ip_l]  # 创建多个队列，用于存储视频帧processes = []for queue, camera_ip in zip(queues, camera_ip_l):processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))processes.append(mp.Process(target=image_get, args=(queue, camera_ip)))# 启动并等待子进程完成for process in processes:process.daemon = Trueprocess.start()for process in processes:process.join()if __name__ == '__main__':# 运行单个摄像头示例# run_single_camera()# 运行多个摄像头示例run_multi_camera()pass

使用Python3自带的多线程模块，创建一个队列，线程A从通过rtsp协议从视频流中读取出每一帧，并放入队列中，线程B从队列中将图片取出，处理后进行显示。线程A如果发现队列里有两张图片（证明线程B的读取速度跟不上线程A），那么线程A主动将队列里面的旧图片删掉，换上新图片。通过多线程的方法：

• 线程A的读取速度始终不收线程B的影响，防止网络摄像头的缓存区爆满
• 线程A更新了队列中的图片，使线程B始终读取到最新的画面，降低了延迟

import multiprocessing as mp
...
img_queues = [mp.Queue(maxsize=2) for _ in camera_ip_l]  # queue
...
q.put(frame) if is_opened else None  # 线程A不仅将图片放入队列
q.get() if q.qsize() > 1 else time.sleep(0.01) # 线程A还负责移除队列中的旧图
...

三 常见问题

3.1 关于rtsp协议

通过rtsp协议读取视频流，下面依次是我在网络上查到的海康与大华 rtsp 读取路径。经过测试，我手头的海康摄像头支持前面两种读取方式（新旧两种）。大华摄像头用第三种读取方式。

video_stream_path = "rtsp://%s:%s@%s/h264/ch%s/main/av_stream" % (user, pwd, ip, channel)  # HIKIVISION old version 2015
video_stream_path = "rtsp://%s:%s@%s//Streaming/Channels/%d" % (user, pwd, ip, channel)  # HIKIVISION new version 2017
video_stream_path = "rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (user, pwd, ip, channel)  # dahua

1 海康、大华IpCamera RTSP地址和格式

2 海康摄像机、NVR、流媒体服务器、回放取流RTSP地址规则说明

3.2 如果你需要收集多个摄像头拍摄的画面

（也就是将5个子进程拍摄到的图片收集 (collect) 到一个进程中去），那么我们需要这样子处理：

run_multi_camera_in_a_window() 修改自 run_multi_camera()
image_collect() 修改自 image_get()def image_collect(queue_list, camera_ip_l):import numpy as np  # 实际使用的时候记得放在外面"""show in single opencv-imshow window"""window_name = "%s_and_so_no" % camera_ip_l[0]cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)while True:imgs = [q.get() for q in queue_list]imgs = np.concatenate(imgs, axis=1)cv2.imshow(window_name, imgs)cv2.waitKey(1)# """show in multiple opencv-imshow windows""" # [cv2.namedWindow(window_name, flags=cv2.WINDOW_FREERATIO)#  for window_name in camera_ip_l]# while True:#     for window_name, q in zip(camera_ip_l, queue_list):#         cv2.imshow(window_name, q.get())#         cv2.waitKey(1)def run_multi_camera_in_a_window():user_name, user_pwd = "admin", "admin123456"camera_ip_l = ["172.20.114.196",  # ipv4"[fe80::3aaf:29ff:fed3:d260]",  # ipv6# 我在这里分别用ipv4 与ipv6 打开了两个摄像头，只有一个摄像头的话就填写一个IP]mp.set_start_method(method='spawn')  # initqueues = [mp.Queue(maxsize=4) for _ in camera_ip_l]processes = [mp.Process(target=image_collect, args=(queues, camera_ip_l))]for queue, camera_ip in zip(queues, camera_ip_l):processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip)))for process in processes:process.daemon = True  # setattr(process, 'deamon', True)process.start()for process in processes:process.join()

3.3 视频读取已经结束， 而程序没有自动退出

cap = cv2.VideoCapture(video_stream_path)
is_opened = cap.isOpened()while is_opened:  # 如果视频读取完毕，那么 is_opened 为False，循环自动跳出is_opened, frame = cap.read()cv2.imshow('frame', frame)cv2.waitKey(1)
cap.release()