如何在python中使用tkinter使用线程显示多视频? [英] How display multi videos with threading using tkinter in python?
本文介绍了如何在python中使用tkinter使用线程显示多视频?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我最近创建了一个程序，该程序使用 opencv 显示来自 2 个 ip 摄像机的多个视频源.但我决定为我的应用程序创建 UI，现在，我不太清楚如何使用多线程方法实现它.
这是我用来在 TKinter GUI 中只显示一台相机的代码:
导入tkinter导入 cv2导入 PIL.Image、PIL.ImageTk导入时间类应用程序:def __init__(self, window, window_title, video_source=0):self.window = 窗口self.window.title(window_title)self.video_source = video_source# 打开视频源(默认情况下会尝试打开电脑网络摄像头)self.vid = MyVideoCapture(self.video_source)# 创建一个可以适合上述视频源大小的画布self.canvas = tkinter.Canvas(window, width = self.vid.width, height = self.vid.height)self.canvas.pack()# 让用户拍摄快照的按钮self.btn_snapshot=tkinter.Button(窗口，文本=快照"，宽度=50，命令=self.snapshot)self.btn_snapshot.pack(anchor=tkinter.CENTER, expand=True)# 调用一次后，更新方法会每隔延迟毫秒自动调用一次self.delay = 15自我更新()self.window.mainloop()定义快照(自我):# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:cv2.imwrite("frame-" + time.strftime("%d-%m-%Y-%H-%M-%S") + ".jpg", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))定义更新(自我):# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:self.photo = PIL.ImageTk.PhotoImage(image = PIL.Image.fromarray(frame))self.canvas.create_image(0, 0, image = self.photo, anchor = tkinter.NW)self.window.after(self.delay, self.update)类 MyVideoCapture:def __init__(self, video_source=0):# 打开视频源self.vid = cv2.VideoCapture(video_source)如果不是 self.vid.isOpened():raise ValueError(无法打开视频源"，video_source)# 获取视频源宽高self.width = self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)self.height = self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)def get_frame(self):如果 self.vid.isOpened():ret, frame = self.vid.read()如果返回:# 返回一个布尔成功标志并将当前帧转换为 BGR返回 (ret, cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))别的:返回(返回，无)别的:返回(返回，无)# 对象销毁时释放视频源def __del__(self):如果 self.vid.isOpened():self.vid.release()# 创建一个窗口并将其传递给Application对象应用程序(tkinter.Tk()，Tkinter 和 OpenCV")这是我以前的应用程序，它在不同的线程中显示多个视频源:
from threading import Thread导入 cv2导入时间类 VideoWriterWidget(对象):def __init__(self, video_file_name, src=0):# 创建一个 VideoCapture 对象self.frame_name = str(src)self.video_file = video_file_nameself.video_file_name = video_file_name + '.avi'self.capture = cv2.VideoCapture(src)# 获取帧的默认分辨率(取决于系统)self.frame_width = int(self.capture.get(3))self.frame_height = int(self.capture.get(4))# 设置编解码器和输出视频设置self.codec = cv2.VideoWriter_fourcc('M','J','P','G')self.output_video = cv2.VideoWriter(self.video_file_name, self.codec, 30, (self.frame_width, self.frame_height))# 启动线程从视频流中读取帧self.thread = Thread(target=self.update, args=())self.thread.daemon = 真self.thread.start()# 启动另一个线程来显示/保存帧self.start_recording()打印('初始化{}'.格式(self.video_file))定义更新(自我):# 从不同线程的流中读取下一帧为真:如果 self.capture.isOpened():(self.status, self.frame) = self.capture.read()def show_frame(self):# 在主程序中显示帧如果 self.status:cv2.imshow(self.frame_name, self.frame)# 按键盘上的 Q 停止录音键 = cv2.waitKey(1)如果键 == ord('q'):self.capture.release()self.output_video.release()cv2.destroyAllWindows()退出(1)def save_frame(self):# 将获得的帧保存到视频输出文件中self.output_video.write(self.frame)def start_recording(self):# 创建另一个线程来显示/保存帧def start_recording_thread():为真:尝试:self.show_frame()self.save_frame()除了属性错误:经过self.recording_thread = Thread(target=start_recording_thread, args=())self.recording_thread.daemon = Trueself.recording_thread.start()如果 __name__ == '__main__':src1 = '你的链接 1'video_writer_widget1 = VideoWriterWidget('摄像机 1', src1)src2 = '你的链接 2'video_writer_widget2 = VideoWriterWidget('Camera 2', src2)src3 = '你的链接 3'video_writer_widget3 = VideoWriterWidget('Camera 3', src3)# 由于每个视频播放器都在自己的线程中，我们需要保持主线程处于活动状态.# 使用 time.sleep() 继续旋转，以便后台线程继续运行# 线程被设置为 daemon=True 所以它们会自动消亡# 当主线程死亡时为真:时间.sleep(5)有人可以帮助我如何使用带线程的 tkinter 在我的新应用程序中使用我以前的代码(显示多摄像头)?
解决方案
tkinter(像许多其他 GUI 一样)不喜欢在线程中使用小部件，所以首先我会尝试在 main 中运行所有没有线程的进程.
在示例中，我将大部分代码移至基于 tkinter.Frame 的类以创建小部件我可以多次使用不同的流.因为我只有一个摄像头(并且系统不能多次使用同一个摄像头)所以我找到了一些外部流/文件来测试它.因为流发送非常大的图像所以我将大小更改为 400, 300
代码在不需要调整图像大小时运行速度很快.
当它必须调整图像大小时，有时会出现问题，但仍然可以.
导入tkinter导入 cv2导入 PIL.Image、PIL.ImageTk导入时间# 带有画布和相机的小部件类 tkCamera(tkinter.Frame):def __init__(self, window, video_source=0):super().__init__(window)self.window = 窗口#self.window.title(window_title)self.video_source = video_sourceself.vid = MyVideoCapture(self.video_source)self.canvas = tkinter.Canvas(窗口，宽度=self.vid.width，高度=self.vid.height)self.canvas.pack()# 让用户拍摄快照的按钮self.btn_snapshot = tkinter.Button(窗口，文本=快照"，宽度=50，命令=self.snapshot)self.btn_snapshot.pack(anchor=tkinter.CENTER, expand=True)# 调用一次后，更新方法会每隔延迟毫秒自动调用一次self.delay = 15self.update_widget()定义快照(自我):# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:cv2.imwrite("frame-" + time.strftime("%d-%m-%Y-%H-%M-%S") + ".jpg", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))def update_widget(self):# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:self.image = PIL.Image.fromarray(frame)self.photo = PIL.ImageTk.PhotoImage(image=self.image)self.canvas.create_image(0, 0, image = self.photo, anchor = tkinter.NW)self.window.after(self.delay, self.update_widget)类应用程序:def __init__(self, window, window_title, video_source1=0, video_source2=0):self.window = 窗口self.window.title(window_title)# 打开视频源(默认情况下会尝试打开电脑网络摄像头)self.vid1 = tkCamera(窗口，video_source1)self.vid1.pack()self.vid2 = tkCamera(窗口，video_source2)self.vid2.pack()# 创建一个可以适合上述视频源大小的画布self.window.mainloop()类 MyVideoCapture:def __init__(self, video_source=0):# 打开视频源self.vid = cv2.VideoCapture(video_source)如果不是 self.vid.isOpened():raise ValueError(无法打开视频源"，video_source)# 获取视频源宽高self.width = self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)self.height = self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)self.width = 400self.height = 300def get_frame(self):如果 self.vid.isOpened():ret, frame = self.vid.read()如果返回:frame = cv2.resize(frame, (400, 300))# 返回一个布尔成功标志并将当前帧转换为 BGR返回 (ret, cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))别的:返回(返回，无)别的:返回(返回，无)# 对象销毁时释放视频源def __del__(self):如果 self.vid.isOpened():self.vid.release()# 创建一个窗口并将其传递给Application对象App(tkinter.Tk(), Tkinter and OpenCV", 0, 'https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4')如果您打算处理框架 - 即.检测运动或面部 - 然后来自 get_frame 的代码可以在单独的 thead 中运行.线程会一直处理帧并分配给 self.frame 并且 get_frame() 应该只返回当前的 self.frame.
在博客 pyImageSearch 上看到类似的想法
导入tkinter导入 cv2导入 PIL.Image、PIL.ImageTk导入时间进口螺纹类 MyVideoCapture:def __init__(self, video_source=0, width=None, height=None, fps=None):self.video_source = video_sourceself.width = 宽度self.height = 高度self.fps = fps# 打开视频源self.vid = cv2.VideoCapture(video_source)如果不是 self.vid.isOpened():raise ValueError([MyVideoCapture] 无法打开视频源", video_source)# 获取视频源宽高如果不是 self.width:self.width = int(self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)) # 将 float 转换为 int如果不是 self.height:self.height = int(self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)) # 将 float 转换为 int如果不是 self.fps:self.fps = int(self.vid.get(cv2.CAP_PROP_FPS)) # 将浮点数转换为整数# 开始时的默认值self.ret = 错误self.frame = 无# 启动线程self.running = 真self.thread = threading.Thread(target=self.process)self.thread.start()定义过程(自我):同时自我运行:ret, frame = self.vid.read()如果返回:# 处理图像frame = cv2.resize(frame, (self.width, self.height))框架 = cv2.cvtColor(框架，cv2.COLOR_BGR2RGB)别的:print('[MyVideoCapture] 流结束:', self.video_source)# TODO: 重新打开流self.running = False休息# 分配新框架self.ret = retself.frame = 框架#睡眠下一帧time.sleep(1/self.fps)def get_frame(self):返回 self.ret, self.frame# 对象销毁时释放视频源def __del__(self):# 停止线程如果自运行:self.running = Falseself.thread.join()# 发布流如果 self.vid.isOpened():self.vid.release()类 tkCamera(tkinter.Frame):def __init__(self, window, text="", video_source=0, width=None, height=None):super().__init__(window)self.window = 窗口#self.window.title(window_title)self.video_source = video_sourceself.vid = MyVideoCapture(self.video_source, width, height)self.label = tkinter.Label(self, text=text)self.label.pack()self.canvas = tkinter.Canvas(self, width=self.vid.width, height=self.vid.height)self.canvas.pack()# 让用户拍摄快照的按钮self.btn_snapshot = tkinter.Button(self, text=Start", command=self.start)self.btn_snapshot.pack(anchor='center', side='left')self.btn_snapshot = tkinter.Button(self, text="Stop", command=self.stop)self.btn_snapshot.pack(anchor='center', side='left')# 让用户拍摄快照的按钮self.btn_snapshot = tkinter.Button(self, text=Snapshot", command=self.snapshot)self.btn_snapshot.pack(anchor='center', side='left')# 调用一次后，更新方法会每隔延迟毫秒自动调用一次# 使用 `FPS` 计算延迟self.delay = int(1000/self.vid.fps)打印('[tkCamera] 来源:'，self.video_source)打印('[tkCamera] fps:'，self.vid.fps，'延迟:'，self.delay)self.image = 无self.running = 真self.update_frame()定义开始(自我):如果不是 self.running:self.running = 真self.update_frame()定义停止(自我):如果自运行:self.running = False定义快照(自我):# 从视频源获取一帧#ret, frame = self.vid.get_frame()#如果返回:# cv2.imwrite(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"), cv2.cvtColor(self.frame, cv2.COLOR_RGB2BGR))# 将当前帧保存在小部件中 - 不是从相机获取新帧 - 这样它可以在停止时保存正确的图像如果 self.image:self.image.save(time.strftime(frame-%d-%m-%Y-%H-%M-%S.jpg"))def update_frame(self):# tkinter 中的小部件已经有方法 `update()`，所以我必须使用不同的名称 -# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:self.image = PIL.Image.fromarray(frame)self.photo = PIL.ImageTk.PhotoImage(image=self.image)self.canvas.create_image(0, 0, image=self.photo, anchor='nw')如果自运行:self.window.after(self.delay, self.update_frame)类应用程序:def __init__(self, window, window_title, video_sources):self.window = 窗口self.window.title(window_title)self.vids = []列 = 2对于数字，枚举中的来源(video_sources):文本，流 = 源vid = tkCamera(self.window, text, stream, 400, 300)x = 数量 % 列y = number//列vid.grid(行=y，列=x)self.vids.append(vid)self.window.protocol(WM_DELETE_WINDOW", self.on_closure)self.window.mainloop()def on_closure(self, event=None):print('[App] 停止线程')对于 self.vids 中的源:source.vid.running = False打印('[应用程序]退出')self.window.destroy()如果 __name__ == '__main__':来源 = [('我', 0),(波兰扎科帕内"，https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4")，('克拉科夫，波兰'，'https://imageserver.webcamera.pl/rec/krakow4/latest.mp4')，(波兰华沙"，https://imageserver.webcamera.pl/rec/warszawa/latest.mp4")，#('波罗的海，波兰'，'https://imageserver.webcamera.pl/rec/chlopy/latest.mp4'),#('山脉，波兰'，'https://imageserver.webcamera.pl/rec/skolnity/latest.mp4'),]# 创建一个窗口并将其传递给Application对象App(tkinter.Tk(), Tkinter and OpenCV", 来源)编辑
可以录制视频的版本.
cv2 需要带有 BGR 颜色的帧才能正确保存它，所以我必须在将帧转换为 RGB 之前保存它.
我将大部分代码移至 MyVideoCapture，因此即使没有 tkinter 也可以使用它.我还在 MyVideoCapture 中添加选项以获取图像为 cv2 array 或 pillow.image - 所以现在它转换为 pillow> 在 thread 内，所以主线程不必这样做.
导入 tkinter导入 cv2导入 PIL.Image、PIL.ImageTk导入时间进口螺纹类 MyVideoCapture:def __init__(self, video_source=0, width=None, height=None, fps=None):self.video_source = video_sourceself.width = 宽度self.height = 高度self.fps = fps# 打开视频源self.vid = cv2.VideoCapture(video_source)如果不是 self.vid.isOpened():raise ValueError([MyVideoCapture] 无法打开视频源", video_source)# 获取视频源宽高如果不是 self.width:self.width = int(self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)) # 将 float 转换为 int如果不是 self.height:self.height = int(self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)) # 将 float 转换为 int如果不是 self.fps:self.fps = int(self.vid.get(cv2.CAP_PROP_FPS)) # 将浮点数转换为整数# 开始时的默认值self.ret = 错误self.frame = 无self.convert_color = cv2.COLOR_BGR2RGB#self.convert_color = cv2.COLOR_BGR2GRAYself.convert_pillow = True# 记录的默认值self.recording = 假self.recording_filename = 'output.mp4'self.recording_writer = 无# 启动线程self.running = 真self.thread = threading.Thread(target=self.process)self.thread.start()def start_recording(self, filename=None):如果自我记录:print('[MyVideoCapture] 已经录制:', self.recording_filename)别的:# VideoWriter 构造函数#.mp4 = 编解码器 ID 2如果文件名:self.recording_filename = 文件名别的:self.recording_filename = time.strftime("%Y.%m.%d %H.%M.%S", time.localtime()) + ".avi";#fourcc = cv2.VideoWriter_fourcc(*'I420') # .avi#fourcc = cv2.VideoWriter_fourcc(*'MP4V') # .aviFourcc = cv2.VideoWriter_fourcc(*'MP42') # .avi#fourcc = cv2.VideoWriter_fourcc(*'AVC1') # 错误libx264#fourcc = cv2.VideoWriter_fourcc(*'H264') # 错误libx264#fourcc = cv2.VideoWriter_fourcc(*'WRAW') # 错误---没有信息---#fourcc = cv2.VideoWriter_fourcc(*'MPEG') # .avi 30fps#fourcc = cv2.VideoWriter_fourcc(*'MJPG') # .avi#fourcc = cv2.VideoWriter_fourcc(*'XVID') # .avi#fourcc = cv2.VideoWriter_fourcc(*'H265') # 错误self.recording_writer = cv2.VideoWriter(self.recording_filename,fourcc,self.fps,(self.width,self.height))self.recording = Trueprint('[MyVideoCapture] 开始录制:', self.recording_filename)def stop_recording(self):如果不是 self.recording:print('[MyVideoCapture] 未录制')别的:self.recording = 假self.recording_writer.release()print('[MyVideoCapture] 停止录制:', self.recording_filename)定义记录(自我，框架):# 将帧写入文件如果 self.recording_writer 和 self.recording_writer.isOpened():self.recording_writer.write(frame)定义过程(自我):同时自我运行:ret, frame = self.vid.read()如果返回:# 处理图像frame = cv2.resize(frame, (self.width, self.height))# 它必须在转换颜色之前进行记录如果自我记录:self.record(frame)如果 self.convert_pillow:框架 = cv2.cvtColor(框架，cv2.COLOR_BGR2RGB)框架 = PIL.Image.fromarray(frame)别的:print('[MyVideoCapture] 流结束:', self.video_source)# TODO: 重新打开流self.running = False如果自我记录:self.stop_recording()休息# 分配新框架self.ret = retself.frame = 框架#睡眠下一帧time.sleep(1/self.fps)def get_frame(self):返回 self.ret, self.frame# 对象销毁时释放视频源def __del__(self):# 停止线程如果自运行:self.running = Falseself.thread.join()# 发布流如果 self.vid.isOpened():self.vid.release()类 tkCamera(tkinter.Frame):def __init__(self, window, text="", video_source=0, width=None, height=None):super().__init__(window)self.window = 窗口#self.window.title(window_title)self.video_source = video_sourceself.vid = MyVideoCapture(self.video_source, width, height)self.label = tkinter.Label(self, text=text)self.label.pack()self.canvas = tkinter.Canvas(self, width=self.vid.width, height=self.vid.height)self.canvas.pack()# 让用户拍摄快照的按钮self.btn_snapshot = tkinter.Button(self, text=Start", command=self.start)self.btn_snapshot.pack(anchor='center', side='left')self.btn_snapshot = tkinter.Button(self, text="Stop", command=self.stop)self.btn_snapshot.pack(anchor='center', side='left')# 让用户拍摄快照的按钮self.btn_snapshot = tkinter.Button(self, text=Snapshot", command=self.snapshot)self.btn_snapshot.pack(anchor='center', side='left')# 调用一次后，更新方法会每隔延迟毫秒自动调用一次# 使用 `FPS` 计算延迟self.delay = int(1000/self.vid.fps)打印('[tkCamera] 来源:'，self.video_source)打印('[tkCamera] fps:'，self.vid.fps，'延迟:'，self.delay)self.image = 无self.running = 真self.update_frame()定义开始(自我):#如果不是 self.running:# self.running = True# self.update_frame()self.vid.start_recording()定义停止(自我):#if self.running:# self.running = Falseself.vid.stop_recording()定义快照(自我):# 从视频源获取一帧#ret, frame = self.vid.get_frame()#如果返回:# cv2.imwrite(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"), cv2.cvtColor(self.frame, cv2.COLOR_RGB2BGR))# 将当前帧保存在小部件中 - 不是从相机获取新帧 - 这样它可以在停止时保存正确的图像如果 self.image:self.image.save(time.strftime(frame-%d-%m-%Y-%H-%M-%S.jpg"))def update_frame(self):# tkinter 中的小部件已经有方法 `update()`，所以我必须使用不同的名称 -# 从视频源获取一帧ret, frame = self.vid.get_frame()如果返回:#self.image = PIL.Image.fromarray(frame)self.image = 框架self.photo = PIL.ImageTk.PhotoImage(image=self.image)self.canvas.create_image(0, 0, image=self.photo, anchor='nw')如果自运行:self.window.after(self.delay, self.update_frame)类应用程序:def __init__(self, window, window_title, video_sources):self.window = 窗口self.window.title(window_title)self.vids = []列 = 2对于数字，枚举中的来源(video_sources):文本，流 = 源vid = tkCamera(self.window, text, stream, 400, 300)x = 数量 % 列y = number//列vid.grid(行=y，列=x)self.vids.append(vid)self.window.protocol(WM_DELETE_WINDOW", self.on_closure)self.window.mainloop()def on_closure(self, event=None):print('[App] 停止线程')对于 self.vids 中的源:source.vid.running = False打印('[应用程序]退出')self.window.destroy()如果 __name__ == '__main__':来源 = [('我', 0),(波兰扎科帕内"，https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4")，('克拉科夫，波兰'，'https://imageserver.webcamera.pl/rec/krakow4/latest.mp4')，(波兰华沙"，https://imageserver.webcamera.pl/rec/warszawa/latest.mp4")，#('波罗的海，波兰'，'https://imageserver.webcamera.pl/rec/chlopy/latest.mp4'),#('山脉，波兰'，'https://imageserver.webcamera.pl/rec/skolnity/latest.mp4'),]# 创建一个窗口并将其传递给Application对象App(tkinter.Tk(), Tkinter and OpenCV", 来源)我创建了可以选择来源的版本 - 因此它可以显示录制的视频.
这段代码很乱.对话框窗口可以在单独的类中.
我不能在这里输入代码，因为答案限制为 30000 个字符.
我把它放在 GitHub 上:
I recently created a program which displays multi video sources from 2 ip cameras with opencv. but I decided to create UI for my application, and now, It's not so clear for me that how I can implement it using multi threading method.
here is the code I used to display only one camera in TKinter GUI:
import tkinter
import cv2
import PIL.Image, PIL.ImageTk
import time
class App:
def __init__(self, window, window_title, video_source=0):
self.window = window
self.window.title(window_title)
self.video_source = video_source
# open video source (by default this will try to open the computer webcam)
self.vid = MyVideoCapture(self.video_source)
# Create a canvas that can fit the above video source size
self.canvas = tkinter.Canvas(window, width = self.vid.width, height = self.vid.height)
self.canvas.pack()
# Button that lets the user take a snapshot
self.btn_snapshot=tkinter.Button(window, text="Snapshot", width=50, command=self.snapshot)
self.btn_snapshot.pack(anchor=tkinter.CENTER, expand=True)
# After it is called once, the update method will be automatically called every delay milliseconds
self.delay = 15
self.update()
self.window.mainloop()
def snapshot(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
cv2.imwrite("frame-" + time.strftime("%d-%m-%Y-%H-%M-%S") + ".jpg", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
def update(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
self.photo = PIL.ImageTk.PhotoImage(image = PIL.Image.fromarray(frame))
self.canvas.create_image(0, 0, image = self.photo, anchor = tkinter.NW)
self.window.after(self.delay, self.update)
class MyVideoCapture:
def __init__(self, video_source=0):
# Open the video source
self.vid = cv2.VideoCapture(video_source)
if not self.vid.isOpened():
raise ValueError("Unable to open video source", video_source)
# Get video source width and height
self.width = self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)
self.height = self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)
def get_frame(self):
if self.vid.isOpened():
ret, frame = self.vid.read()
if ret:
# Return a boolean success flag and the current frame converted to BGR
return (ret, cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
else:
return (ret, None)
else:
return (ret, None)
# Release the video source when the object is destroyed
def __del__(self):
if self.vid.isOpened():
self.vid.release()
# Create a window and pass it to the Application object
App(tkinter.Tk(), "Tkinter and OpenCV")
and here was my previous application which displays multi video sources in different threads:
from threading import Thread
import cv2
import time
class VideoWriterWidget(object):
def __init__(self, video_file_name, src=0):
# Create a VideoCapture object
self.frame_name = str(src)
self.video_file = video_file_name
self.video_file_name = video_file_name + '.avi'
self.capture = cv2.VideoCapture(src)
# Default resolutions of the frame are obtained (system dependent)
self.frame_width = int(self.capture.get(3))
self.frame_height = int(self.capture.get(4))
# Set up codec and output video settings
self.codec = cv2.VideoWriter_fourcc('M','J','P','G')
self.output_video = cv2.VideoWriter(self.video_file_name, self.codec, 30, (self.frame_width, self.frame_height))
# Start the thread to read frames from the video stream
self.thread = Thread(target=self.update, args=())
self.thread.daemon = True
self.thread.start()
# Start another thread to show/save frames
self.start_recording()
print('initialized {}'.format(self.video_file))
def update(self):
# Read the next frame from the stream in a different thread
while True:
if self.capture.isOpened():
(self.status, self.frame) = self.capture.read()
def show_frame(self):
# Display frames in main program
if self.status:
cv2.imshow(self.frame_name, self.frame)
# Press Q on keyboard to stop recording
key = cv2.waitKey(1)
if key == ord('q'):
self.capture.release()
self.output_video.release()
cv2.destroyAllWindows()
exit(1)
def save_frame(self):
# Save obtained frame into video output file
self.output_video.write(self.frame)
def start_recording(self):
# Create another thread to show/save frames
def start_recording_thread():
while True:
try:
self.show_frame()
self.save_frame()
except AttributeError:
pass
self.recording_thread = Thread(target=start_recording_thread, args=())
self.recording_thread.daemon = True
self.recording_thread.start()
if __name__ == '__main__':
src1 = 'Your link1'
video_writer_widget1 = VideoWriterWidget('Camera 1', src1)
src2 = 'Your link2'
video_writer_widget2 = VideoWriterWidget('Camera 2', src2)
src3 = 'Your link3'
video_writer_widget3 = VideoWriterWidget('Camera 3', src3)
# Since each video player is in its own thread, we need to keep the main thread alive.
# Keep spinning using time.sleep() so the background threads keep running
# Threads are set to daemon=True so they will automatically die
# when the main thread dies
while True:
time.sleep(5)
can someone help me how I can use my previous code (display multi cameras) in my new application using tkinter with threading?
解决方案
tkinter (like many other GUIs) doesn't like to use widgets in threads so first I would try to run all in main process without threads.
In example I moved most of code to class based on tkinter.Frame to create widget
which I can use many times with different streams. Because I have only one camera (and system can't use the same camera many times) so I found some external stream/file to test it. Because stream sends very big image so I change size to 400, 300
Code works fast when it doesn't have to resize image.
When it has to resize image then sometimes it has problem but still it is OK.
import tkinter
import cv2
import PIL.Image, PIL.ImageTk
import time
# widgets with canvas and camera
class tkCamera(tkinter.Frame):
def __init__(self, window, video_source=0):
super().__init__(window)
self.window = window
#self.window.title(window_title)
self.video_source = video_source
self.vid = MyVideoCapture(self.video_source)
self.canvas = tkinter.Canvas(window, width=self.vid.width, height=self.vid.height)
self.canvas.pack()
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(window, text="Snapshot", width=50, command=self.snapshot)
self.btn_snapshot.pack(anchor=tkinter.CENTER, expand=True)
# After it is called once, the update method will be automatically called every delay milliseconds
self.delay = 15
self.update_widget()
def snapshot(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
cv2.imwrite("frame-" + time.strftime("%d-%m-%Y-%H-%M-%S") + ".jpg", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
def update_widget(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
self.image = PIL.Image.fromarray(frame)
self.photo = PIL.ImageTk.PhotoImage(image=self.image)
self.canvas.create_image(0, 0, image = self.photo, anchor = tkinter.NW)
self.window.after(self.delay, self.update_widget)
class App:
def __init__(self, window, window_title, video_source1=0, video_source2=0):
self.window = window
self.window.title(window_title)
# open video source (by default this will try to open the computer webcam)
self.vid1 = tkCamera(window, video_source1)
self.vid1.pack()
self.vid2 = tkCamera(window, video_source2)
self.vid2.pack()
# Create a canvas that can fit the above video source size
self.window.mainloop()
class MyVideoCapture:
def __init__(self, video_source=0):
# Open the video source
self.vid = cv2.VideoCapture(video_source)
if not self.vid.isOpened():
raise ValueError("Unable to open video source", video_source)
# Get video source width and height
self.width = self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)
self.height = self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)
self.width = 400
self.height = 300
def get_frame(self):
if self.vid.isOpened():
ret, frame = self.vid.read()
if ret:
frame = cv2.resize(frame, (400, 300))
# Return a boolean success flag and the current frame converted to BGR
return (ret, cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
else:
return (ret, None)
else:
return (ret, None)
# Release the video source when the object is destroyed
def __del__(self):
if self.vid.isOpened():
self.vid.release()
# Create a window and pass it to the Application object
App(tkinter.Tk(), "Tkinter and OpenCV", 0, 'https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4')
If you plan to process frame - ie. detect motion or faces - then code from get_frame could run in separated thead. Thread would process frames all time and assign to self.frame and get_frame() should only return current self.frame.
See similar idea on blog pyImageSearch in
Increasing webcam FPS with Python and OpenCV.
Probably you could even use
from imutils.video import WebcamVideoStream
EDIT:
Version still without threading but with list of sources so it can display many cameras. But for more then 2 sources it has problem to display - so this would need to use threads.
BTW: widgets and windows in tkinter already have method update() so I renamed it to update_frame()
In snapshot I used pilow.image.save() so I don't have to read new frame and convert to BGR - and I can take snapshot when stream is stoped. Button stops only replacing image on canvas but not stop reading frames from stream in thread - so other function could still process or record stream.
import tkinter
import cv2
import PIL.Image, PIL.ImageTk
import time
class MyVideoCapture:
def __init__(self, video_source=0, width=None, height=None):
# Open the video source
self.vid = cv2.VideoCapture(video_source)
if not self.vid.isOpened():
raise ValueError("Unable to open video source", video_source)
self.width = width
self.height = height
# Get video source width and height
if not self.width:
self.width = int(self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)) # convert float to int
if not self.height:
self.height = int(self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)) # convert float to int
self.ret = False
self.frame = None
def process(self):
ret = False
frame = None
if self.vid.isOpened():
ret, frame = self.vid.read()
if ret:
frame = cv2.resize(frame, (self.width, self.height))
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
self.ret = ret
self.frame = frame
def get_frame(self):
self.process() # later run in thread
return self.ret, self.frame
# Release the video source when the object is destroyed
def __del__(self):
if self.vid.isOpened():
self.vid.release()
class tkCamera(tkinter.Frame):
def __init__(self, window, video_source=0, width=None, height=None):
super().__init__(window)
self.window = window
#self.window.title(window_title)
self.video_source = video_source
self.vid = MyVideoCapture(self.video_source, width, height)
self.canvas = tkinter.Canvas(window, width=self.vid.width, height=self.vid.height)
self.canvas.pack()
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(window, text="Snapshot", width=50, command=self.snapshot)
self.btn_snapshot.pack(anchor='center', expand=True)
# After it is called once, the update method will be automatically called every delay milliseconds
self.delay = 15
self.update_widget()
def snapshot(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
cv2.imwrite("frame-" + time.strftime("%d-%m-%Y-%H-%M-%S") + ".jpg", cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
def update_widget(self):
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
self.image = PIL.Image.fromarray(frame)
self.photo = PIL.ImageTk.PhotoImage(image=self.image)
self.canvas.create_image(0, 0, image = self.photo, anchor = tkinter.NW)
self.window.after(self.delay, self.update_widget)
class App:
def __init__(self, window, window_title, video_sources):
self.window = window
self.window.title(window_title)
self.vids = []
for source in video_sources:
vid = tkCamera(window, source, 400, 300)
vid.pack()
self.vids.append(vid)
# Create a canvas that can fit the above video source size
self.window.mainloop()
if __name__ == '__main__':
sources = [
0,
#'https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4',
#'https://imageserver.webcamera.pl/rec/skolnity/latest.mp4',
'https://imageserver.webcamera.pl/rec/krakow4/latest.mp4',
]
# Create a window and pass it to the Application object
App(tkinter.Tk(), "Tkinter and OpenCV", sources)
EDIT
Version which uses threads to read and process frames. I add time(1/fps) to process it only when it is needed so it works smoother. For delay 15 it freezed sometimes.
I uses sources which have only 24 seconds so after few seconds they stop.
import tkinter
import cv2
import PIL.Image, PIL.ImageTk
import time
import threading
class MyVideoCapture:
def __init__(self, video_source=0, width=None, height=None, fps=None):
self.video_source = video_source
self.width = width
self.height = height
self.fps = fps
# Open the video source
self.vid = cv2.VideoCapture(video_source)
if not self.vid.isOpened():
raise ValueError("[MyVideoCapture] Unable to open video source", video_source)
# Get video source width and height
if not self.width:
self.width = int(self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)) # convert float to int
if not self.height:
self.height = int(self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)) # convert float to int
if not self.fps:
self.fps = int(self.vid.get(cv2.CAP_PROP_FPS)) # convert float to int
# default value at start
self.ret = False
self.frame = None
# start thread
self.running = True
self.thread = threading.Thread(target=self.process)
self.thread.start()
def process(self):
while self.running:
ret, frame = self.vid.read()
if ret:
# process image
frame = cv2.resize(frame, (self.width, self.height))
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
else:
print('[MyVideoCapture] stream end:', self.video_source)
# TODO: reopen stream
self.running = False
break
# assign new frame
self.ret = ret
self.frame = frame
# sleep for next frame
time.sleep(1/self.fps)
def get_frame(self):
return self.ret, self.frame
# Release the video source when the object is destroyed
def __del__(self):
# stop thread
if self.running:
self.running = False
self.thread.join()
# relase stream
if self.vid.isOpened():
self.vid.release()
class tkCamera(tkinter.Frame):
def __init__(self, window, text="", video_source=0, width=None, height=None):
super().__init__(window)
self.window = window
#self.window.title(window_title)
self.video_source = video_source
self.vid = MyVideoCapture(self.video_source, width, height)
self.label = tkinter.Label(self, text=text)
self.label.pack()
self.canvas = tkinter.Canvas(self, width=self.vid.width, height=self.vid.height)
self.canvas.pack()
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(self, text="Start", command=self.start)
self.btn_snapshot.pack(anchor='center', side='left')
self.btn_snapshot = tkinter.Button(self, text="Stop", command=self.stop)
self.btn_snapshot.pack(anchor='center', side='left')
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(self, text="Snapshot", command=self.snapshot)
self.btn_snapshot.pack(anchor='center', side='left')
# After it is called once, the update method will be automatically called every delay milliseconds
# calculate delay using `FPS`
self.delay = int(1000/self.vid.fps)
print('[tkCamera] source:', self.video_source)
print('[tkCamera] fps:', self.vid.fps, 'delay:', self.delay)
self.image = None
self.running = True
self.update_frame()
def start(self):
if not self.running:
self.running = True
self.update_frame()
def stop(self):
if self.running:
self.running = False
def snapshot(self):
# Get a frame from the video source
#ret, frame = self.vid.get_frame()
#if ret:
# cv2.imwrite(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"), cv2.cvtColor(self.frame, cv2.COLOR_RGB2BGR))
# Save current frame in widget - not get new one from camera - so it can save correct image when it stoped
if self.image:
self.image.save(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"))
def update_frame(self):
# widgets in tkinter already have method `update()` so I have to use different name -
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
self.image = PIL.Image.fromarray(frame)
self.photo = PIL.ImageTk.PhotoImage(image=self.image)
self.canvas.create_image(0, 0, image=self.photo, anchor='nw')
if self.running:
self.window.after(self.delay, self.update_frame)
class App:
def __init__(self, window, window_title, video_sources):
self.window = window
self.window.title(window_title)
self.vids = []
columns = 2
for number, source in enumerate(video_sources):
text, stream = source
vid = tkCamera(self.window, text, stream, 400, 300)
x = number % columns
y = number // columns
vid.grid(row=y, column=x)
self.vids.append(vid)
self.window.protocol("WM_DELETE_WINDOW", self.on_closing)
self.window.mainloop()
def on_closing(self, event=None):
print('[App] stoping threads')
for source in self.vids:
source.vid.running = False
print('[App] exit')
self.window.destroy()
if __name__ == '__main__':
sources = [
('me', 0),
('Zakopane, Poland', 'https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4'),
('Kraków, Poland', 'https://imageserver.webcamera.pl/rec/krakow4/latest.mp4'),
('Warszawa, Poland', 'https://imageserver.webcamera.pl/rec/warszawa/latest.mp4'),
#('Baltic See, Poland', 'https://imageserver.webcamera.pl/rec/chlopy/latest.mp4'),
#('Mountains, Poland', 'https://imageserver.webcamera.pl/rec/skolnity/latest.mp4'),
]
# Create a window and pass it to the Application object
App(tkinter.Tk(), "Tkinter and OpenCV", sources)
EDIT
Version which can record video.
cv2 needs frame with BGR color to save it correctly so I had to save it before frame is converted to RGB.
I moved most code to MyVideoCapture so it can be used even without tkinter. I also add option in MyVideoCapture to get image as cv2 array or pillow.image - so now it converts to pillow inside thread so main thread doesn't have to do it.
import tkinter
import cv2
import PIL.Image, PIL.ImageTk
import time
import threading
class MyVideoCapture:
def __init__(self, video_source=0, width=None, height=None, fps=None):
self.video_source = video_source
self.width = width
self.height = height
self.fps = fps
# Open the video source
self.vid = cv2.VideoCapture(video_source)
if not self.vid.isOpened():
raise ValueError("[MyVideoCapture] Unable to open video source", video_source)
# Get video source width and height
if not self.width:
self.width = int(self.vid.get(cv2.CAP_PROP_FRAME_WIDTH)) # convert float to int
if not self.height:
self.height = int(self.vid.get(cv2.CAP_PROP_FRAME_HEIGHT)) # convert float to int
if not self.fps:
self.fps = int(self.vid.get(cv2.CAP_PROP_FPS)) # convert float to int
# default value at start
self.ret = False
self.frame = None
self.convert_color = cv2.COLOR_BGR2RGB
#self.convert_color = cv2.COLOR_BGR2GRAY
self.convert_pillow = True
# default values for recording
self.recording = False
self.recording_filename = 'output.mp4'
self.recording_writer = None
# start thread
self.running = True
self.thread = threading.Thread(target=self.process)
self.thread.start()
def start_recording(self, filename=None):
if self.recording:
print('[MyVideoCapture] already recording:', self.recording_filename)
else:
# VideoWriter constructors
#.mp4 = codec id 2
if filename:
self.recording_filename = filename
else:
self.recording_filename = time.strftime("%Y.%m.%d %H.%M.%S", time.localtime()) + ".avi"
#fourcc = cv2.VideoWriter_fourcc(*'I420') # .avi
#fourcc = cv2.VideoWriter_fourcc(*'MP4V') # .avi
fourcc = cv2.VideoWriter_fourcc(*'MP42') # .avi
#fourcc = cv2.VideoWriter_fourcc(*'AVC1') # error libx264
#fourcc = cv2.VideoWriter_fourcc(*'H264') # error libx264
#fourcc = cv2.VideoWriter_fourcc(*'WRAW') # error --- no information ---
#fourcc = cv2.VideoWriter_fourcc(*'MPEG') # .avi 30fps
#fourcc = cv2.VideoWriter_fourcc(*'MJPG') # .avi
#fourcc = cv2.VideoWriter_fourcc(*'XVID') # .avi
#fourcc = cv2.VideoWriter_fourcc(*'H265') # error
self.recording_writer = cv2.VideoWriter(self.recording_filename, fourcc, self.fps, (self.width, self.height))
self.recording = True
print('[MyVideoCapture] started recording:', self.recording_filename)
def stop_recording(self):
if not self.recording:
print('[MyVideoCapture] not recording')
else:
self.recording = False
self.recording_writer.release()
print('[MyVideoCapture] stop recording:', self.recording_filename)
def record(self, frame):
# write frame to file
if self.recording_writer and self.recording_writer.isOpened():
self.recording_writer.write(frame)
def process(self):
while self.running:
ret, frame = self.vid.read()
if ret:
# process image
frame = cv2.resize(frame, (self.width, self.height))
# it has to record before converting colors
if self.recording:
self.record(frame)
if self.convert_pillow:
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame = PIL.Image.fromarray(frame)
else:
print('[MyVideoCapture] stream end:', self.video_source)
# TODO: reopen stream
self.running = False
if self.recording:
self.stop_recording()
break
# assign new frame
self.ret = ret
self.frame = frame
# sleep for next frame
time.sleep(1/self.fps)
def get_frame(self):
return self.ret, self.frame
# Release the video source when the object is destroyed
def __del__(self):
# stop thread
if self.running:
self.running = False
self.thread.join()
# relase stream
if self.vid.isOpened():
self.vid.release()
class tkCamera(tkinter.Frame):
def __init__(self, window, text="", video_source=0, width=None, height=None):
super().__init__(window)
self.window = window
#self.window.title(window_title)
self.video_source = video_source
self.vid = MyVideoCapture(self.video_source, width, height)
self.label = tkinter.Label(self, text=text)
self.label.pack()
self.canvas = tkinter.Canvas(self, width=self.vid.width, height=self.vid.height)
self.canvas.pack()
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(self, text="Start", command=self.start)
self.btn_snapshot.pack(anchor='center', side='left')
self.btn_snapshot = tkinter.Button(self, text="Stop", command=self.stop)
self.btn_snapshot.pack(anchor='center', side='left')
# Button that lets the user take a snapshot
self.btn_snapshot = tkinter.Button(self, text="Snapshot", command=self.snapshot)
self.btn_snapshot.pack(anchor='center', side='left')
# After it is called once, the update method will be automatically called every delay milliseconds
# calculate delay using `FPS`
self.delay = int(1000/self.vid.fps)
print('[tkCamera] source:', self.video_source)
print('[tkCamera] fps:', self.vid.fps, 'delay:', self.delay)
self.image = None
self.running = True
self.update_frame()
def start(self):
#if not self.running:
# self.running = True
# self.update_frame()
self.vid.start_recording()
def stop(self):
#if self.running:
# self.running = False
self.vid.stop_recording()
def snapshot(self):
# Get a frame from the video source
#ret, frame = self.vid.get_frame()
#if ret:
# cv2.imwrite(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"), cv2.cvtColor(self.frame, cv2.COLOR_RGB2BGR))
# Save current frame in widget - not get new one from camera - so it can save correct image when it stoped
if self.image:
self.image.save(time.strftime("frame-%d-%m-%Y-%H-%M-%S.jpg"))
def update_frame(self):
# widgets in tkinter already have method `update()` so I have to use different name -
# Get a frame from the video source
ret, frame = self.vid.get_frame()
if ret:
#self.image = PIL.Image.fromarray(frame)
self.image = frame
self.photo = PIL.ImageTk.PhotoImage(image=self.image)
self.canvas.create_image(0, 0, image=self.photo, anchor='nw')
if self.running:
self.window.after(self.delay, self.update_frame)
class App:
def __init__(self, window, window_title, video_sources):
self.window = window
self.window.title(window_title)
self.vids = []
columns = 2
for number, source in enumerate(video_sources):
text, stream = source
vid = tkCamera(self.window, text, stream, 400, 300)
x = number % columns
y = number // columns
vid.grid(row=y, column=x)
self.vids.append(vid)
self.window.protocol("WM_DELETE_WINDOW", self.on_closing)
self.window.mainloop()
def on_closing(self, event=None):
print('[App] stoping threads')
for source in self.vids:
source.vid.running = False
print('[App] exit')
self.window.destroy()
if __name__ == '__main__':
sources = [
('me', 0),
('Zakopane, Poland', 'https://imageserver.webcamera.pl/rec/krupowki-srodek/latest.mp4'),
('Kraków, Poland', 'https://imageserver.webcamera.pl/rec/krakow4/latest.mp4'),
('Warszawa, Poland', 'https://imageserver.webcamera.pl/rec/warszawa/latest.mp4'),
#('Baltic See, Poland', 'https://imageserver.webcamera.pl/rec/chlopy/latest.mp4'),
#('Mountains, Poland', 'https://imageserver.webcamera.pl/rec/skolnity/latest.mp4'),
]
# Create a window and pass it to the Application object
App(tkinter.Tk(), "Tkinter and OpenCV", sources)
EDIT:
I created version which can select source - so it can display recorded videos.
This code is chaotic. Dialog window could be in separated class.
I can't put code here because answer has limitation to 30000 characters.
I put it on GitHub: python-cv2-streams-viewer
这篇关于如何在python中使用tkinter使用线程显示多视频?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
