Python開發(fā)中OpenCV視頻流的多線程處理方式詳解
前言
在做視覺類項目中,常常需要在Python環(huán)境下使用OpenCV讀取本地的還是網(wǎng)絡(luò)攝像頭的視頻流,之后再調(diào)入各種模型,如目標分類、目標檢測,人臉識別等等。如果使用單線程處理,很多時候會出現(xiàn)比較嚴重的時延,如果算力吃緊,模型推理所占用的更長的話,這種延遲感會更加明顯,會出現(xiàn)卡幀的現(xiàn)象。在這種情況下,往往要把代碼從單線程改為了多線程,即單獨用一個線程實時捕獲視頻幀,主線程在需要時從子線程拷貝最近的幀使用即可。
單線程處理視頻流時,如果目標檢測模型較大或者任務(wù)復(fù)雜,會影響處理速度。而使用多線程,讓視頻捕獲和目標檢測分別在各自的線程中運行,能夠更充分地利用 CPU 的多核心處理能力,提高整體的處理效率和實時性。
在實時視頻處理中,特別是涉及到深度學(xué)習(xí)模型推理這種計算密集型的任務(wù)時,多線程確實能夠帶來顯著的性能提升。通過將視頻捕獲和處理分開,可以避免由于處理時間過長而導(dǎo)致的幀丟失或延遲。
一、多線程
在 Python 中,可以使用 threading 模塊來實現(xiàn)多線程。下面是一個簡單的例子,演示了如何在 Python 中創(chuàng)建和使用多線程:
1、導(dǎo)入 threading 模塊
首先導(dǎo)入 Python 的 threading 模塊,它提供了多線程編程所需的功能。
import threading
2、創(chuàng)建線程執(zhí)行函數(shù)
定義一個函數(shù),作為線程的執(zhí)行體。這個函數(shù)將會在每個線程中運行。在函數(shù)內(nèi)編寫希望線程執(zhí)行的代碼邏輯。
def my_function(): # Your code here pass
3、創(chuàng)建線程對象
使用 threading.Thread() 創(chuàng)建一個線程對象,將目標函數(shù)指定為剛才定義的函數(shù),并傳入所需參數(shù)。
my_thread = threading.Thread(target=my_function, args=(arg1, arg2)) # 傳入?yún)?shù) args
4、動線程
使用 start() 方法啟動線程。
my_thread.start()
5、等待線程執(zhí)行完成
使用 join() 方法等待線程執(zhí)行完畢。這會讓主線程等待子線程的結(jié)束。
my_thread.join()
6、示例
以下是一個簡單的示例,演示了如何使用多線程:
import threading import time # 線程執(zhí)行體函數(shù) def print_numbers(): for i in range(5): print(f"Child Thread: {i}") time.sleep(1) # 創(chuàng)建線程對象 thread = threading.Thread(target=print_numbers) # 啟動線程 thread.start() # 主線程中的其他操作 for i in range(5): print(f"Main Thread: {i}") time.sleep(0.5) # 等待子線程執(zhí)行結(jié)束 thread.join() print("Main Thread exiting...")
print_numbers() 函數(shù)是子線程的執(zhí)行體,在子線程中打印數(shù)字。主線程在啟動子線程后,會同時進行自己的任務(wù)。最后通過 join() 方法等待子線程結(jié)束。
二、視頻處理
一般視頻處理代碼分為兩部分:從相機讀取下一個可用幀以及對幀進行圖像處理,例如把圖像送到Y(jié)olov5目標檢測模型進行檢測。
在沒有多線程的程序中,按順序讀取下一幀并進行處理。程序等待下一幀可用,然后對其進行必要的處理。讀取幀所需的時間主要取決于請求、等待下一個視頻幀,并將其從相機傳輸?shù)絻?nèi)存的時間。無論是在 CPU 還是 GPU 上,對視頻幀進行計算所需的時間都占據(jù)了視頻處理所花費的大部分時間。
然而,在具有多線程的程序中,讀取下一幀并對其進行處理不需要按順序進行。當一個線程執(zhí)行讀取下一幀的任務(wù)時,主線程可以使用 CPU 或 GPU 來處理最后一個讀取的幀。通過這種方式,這兩個任務(wù)可以重疊執(zhí)行,從而減少讀取和處理幀的總時間。
1.視頻單線程處理
# importing required libraries import cv2 import time# opening video capture stream vcap = cv2.VideoCapture(0) if vcap.isOpened() is False : print("[Exiting]: Error accessing webcam stream.") exit(0) fps_input_stream = int(vcap.get(5)) print("FPS of webcam hardware/input stream: {}".format(fps_input_stream)) grabbed, frame = vcap.read() # reading single frame for initialization/ hardware warm-up# processing frames in input stream num_frames_processed = 0 start = time.time() while True : grabbed, frame = vcap.read() if grabbed is False : print('[Exiting] No more frames to read') break# adding a delay for simulating time taken for processing a frame delay = 0.03 # delay value in seconds. so, delay=1 is equivalent to 1 second time.sleep(delay) num_frames_processed += 1cv2.imshow('frame' , frame) key = cv2.waitKey(1) if key == ord('q'): break end = time.time()# printing time elapsed and fps elapsed = end-start fps = num_frames_processed/elapsed print("FPS: {} , Elapsed Time: {} , Frames Processed: {}".format(fps, elapsed, num_frames_processed))# releasing input stream , closing all windows vcap.release() cv2.destroyAllWindows()
2.視頻多線程處理
# importing required libraries import cv2 import time from threading import Thread # library for implementing multi-threaded processing# defining a helper class for implementing multi-threaded processing class WebcamStream : def __init__(self, stream_id=0): self.stream_id = stream_id # default is 0 for primary camera # opening video capture stream self.vcap = cv2.VideoCapture(self.stream_id) if self.vcap.isOpened() is False : print("[Exiting]: Error accessing webcam stream.") exit(0) fps_input_stream = int(self.vcap.get(5)) print("FPS of webcam hardware/input stream: {}".format(fps_input_stream)) # reading a single frame from vcap stream for initializing self.grabbed , self.frame = self.vcap.read() if self.grabbed is False : print('[Exiting] No more frames to read') exit(0)# self.stopped is set to False when frames are being read from self.vcap stream self.stopped = True# reference to the thread for reading next available frame from input stream self.t = Thread(target=self.update, args=()) self.t.daemon = True # daemon threads keep running in the background while the program is executing # method for starting the thread for grabbing next available frame in input stream def start(self): self.stopped = False self.t.start()# method for reading next frame def update(self): while True : if self.stopped is True : break self.grabbed , self.frame = self.vcap.read() if self.grabbed is False : print('[Exiting] No more frames to read') self.stopped = True break self.vcap.release()# method for returning latest read frame def read(self): return self.frame# method called to stop reading frames def stop(self): self.stopped = True# initializing and starting multi-threaded webcam capture input stream webcam_stream = WebcamStream(stream_id=0) # stream_id = 0 is for primary camera webcam_stream.start()# processing frames in input stream num_frames_processed = 0 start = time.time() while True : if webcam_stream.stopped is True : break else : frame = webcam_stream.read()# adding a delay for simulating time taken for processing a frame delay = 0.03 # delay value in seconds. so, delay=1 is equivalent to 1 second time.sleep(delay) num_frames_processed += 1cv2.imshow('frame' , frame) key = cv2.waitKey(1) if key == ord('q'): break end = time.time() webcam_stream.stop() # stop the webcam stream# printing time elapsed and fps elapsed = end-start fps = num_frames_processed/elapsed print("FPS: {} , Elapsed Time: {} , Frames Processed: {}".format(fps, elapsed, num_frames_processed))# closing all windows cv2.destroyAllWindows()
上面的代碼創(chuàng)建了一個 WebcamStream 類,其中包含了多線程讀取相機幀的邏輯。在主循環(huán)中,它仍然以順序方式處理每個幀,但是讀取幀的線程是在后臺運行的。
但上面的代碼提升速度的同時,還有以下有改進的問題:
- 處理多個幀的邏輯: 代碼在主循環(huán)中每次處理幀時都有一個固定的延遲 time.sleep(delay),這并不真實地模擬出幀處理的時間。應(yīng)該考慮記錄每個幀的時間戳,并在處理完幀后計算幀處理的實際時間。
- 多線程下的幀處理: 雖然視頻流讀取部分在單獨的線程中,但是主循環(huán)仍然是順序執(zhí)行的,它等待每個幀進行處理。在多線程環(huán)境中,也許值得考慮在單獨的線程中對幀進行處理。
- 內(nèi)存和資源管理: 確保在程序退出時釋放所有資源,特別是在多線程環(huán)境中,需要小心確保線程的安全退出。
- 代碼結(jié)構(gòu)和注釋: 為了更好地可讀性和維護性,添加一些注釋來解釋每個函數(shù)和方法的作用,以及代碼塊的意圖。
3.多線程代碼優(yōu)化
去除固定延遲的處理方式: 代碼在處理每一幀時都有固定的延遲??紤]使用實際幀處理時間的方法,而不是使用固定的延遲。這可以通過記錄每個幀的時間戳來實現(xiàn)。
并行處理視頻幀: 在主線程中按順序處理每一幀。在多線程環(huán)境下,可以考慮使用多個線程并行處理視頻幀,以加快處理速度。
資源釋放: 在程序結(jié)束時,確保釋放所有資源。這包括在適當?shù)臅r候關(guān)閉視頻流、終止線程等。
import cv2 import time from threading import Thread class WebcamStream: def __init__(self, stream_id=0): self.stream_id = stream_id self.vcap = cv2.VideoCapture(self.stream_id) if not self.vcap.isOpened(): print("[Exiting]: Error accessing webcam stream.") exit(0) self.fps_input_stream = int(self.vcap.get(cv2.CAP_PROP_FPS)) print("FPS of webcam hardware/input stream: {}".format(self.fps_input_stream)) self.grabbed, self.frame = self.vcap.read() if not self.grabbed: print('[Exiting] No more frames to read') exit(0) self.stopped = False self.t = Thread(target=self.update, args=()) self.t.daemon = True self.t.start() def update(self): while not self.stopped: grabbed, frame = self.vcap.read() if not grabbed: print('[Exiting] No more frames to read') self.stopped = True break self.frame = frame def read(self): return self.frame def stop(self): self.stopped = True self.t.join() self.vcap.release() webcam_stream = WebcamStream(stream_id=0) num_frames_processed = 0 start = time.time() while True: frame = webcam_stream.read() if webcam_stream.stopped: break delay = 0.03 time.sleep(delay) num_frames_processed += 1 cv2.imshow('frame', frame) key = cv2.waitKey(1) if key == ord('q'): break end = time.time() webcam_stream.stop() elapsed = end - start fps = num_frames_processed / elapsed print("FPS: {} , Elapsed Time: {} , Frames Processed: {}".format(fps, elapsed, num_frames_processed)) cv2.destroyAllWindows()
以上就是Python開發(fā)中OpenCV視頻流的多線程處理方式詳解的詳細內(nèi)容,更多關(guān)于Python OpenCV視頻流處理的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
使用Python創(chuàng)建快捷方式管理應(yīng)用
在Windows系統(tǒng)中,快速訪問常用程序通常通過“開始菜單”中的“應(yīng)用熱門”功能實現(xiàn),在這篇博客中,我將向你展示如何使用Python和wxPython創(chuàng)建一個GUI應(yīng)用,幫助用戶輕松將桌面上的快捷方式添加到Windows“開始菜單”的“應(yīng)用熱門”中,需要的朋友可以參考下2024-08-08python安裝cxOracle避坑總結(jié)不要直接pip install
這篇文章主要為大家介紹了python安裝cx_Oracle是遇到的一些問題的解決辦法的總結(jié),來幫大家避避坑有需要的朋友可以借鑒參考下,希望能夠有所幫助祝大家多多進步2021-10-10numpy和tensorflow中的各種乘法(點乘和矩陣乘)
這篇文章主要介紹了numpy和tensorflow中的各種乘法(點乘和矩陣乘),文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03python2.7無法使用pip的解決方法(安裝easy_install)
下面小編就為大家分享一篇python2.7無法使用pip的解決方法(安裝easy_install),具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2018-04-04Django admin.py 在修改/添加表單界面顯示額外字段的方法
今天小編就為大家分享一篇Django admin.py 在修改/添加表單界面顯示額外字段的方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-08-08