Python 使用 OpenCV 開啟、顯示、處理與儲存 RTSP 串流影片教學與範例

woff

Python OpenCV RTSP 串流影片

介紹如何在 Python 中使用 OpenCV 搭配多行程（multiprocessing）的方式開啟、顯示、處理與儲存 RTSP 串流影片。

測試用 RTSP 串流

網路上有一些測試用的 RTSP 串流伺服器，可用於程式的開發與測試：

• https://github.com/RSATom/rtsp-test-server
• RTSP Stream for Testing
  
  

安裝 OpenCV

安裝 Python 的 OpenCV 模組：

1. # 安裝 Python 的 OpenCV 模組
   
2. pip3 install opencv-python

複製代碼

開啟與顯示 RTSP 串流影像

OpenCV 中的 VideoCapture() 函數可以用來開啟各種不同的影片來源，包含影片檔案、網路攝影機、RTSP 串流等，以下是使用 OpenCV 開啟 RTSP 串流影片的範例。

1. import cv2
   
2. 
   
3. if __name__ == '__main__':
   
4. 
   
5.     # 開啟 RTSP 串流
   
6.     vidCap = cv2.VideoCapture('rtsp://ipcam.stream:8554/bars')
   
7. 
   
8.     # 建立視窗
   
9.     cv2.namedWindow('image_display', cv2.WINDOW_AUTOSIZE)
   
10. 
    
11.     while True:
    
12.         # 從 RTSP 串流讀取一張影像
    
13.         ret, image = vidCap.read()
    
14. 
    
15.         if ret:
    
16.             # 顯示影像
    
17.             cv2.imshow('image_display', image)
    
18.             cv2.waitKey(10)
    
19.         else:
    
20.             # 若沒有影像跳出迴圈
    
21.             break
    
22. 
    
23.     # 釋放資源
    
24.     vidCap.release()
    
25. 
    
26.     # 關閉所有 OpenCV 視窗
    
27.     cv2.destroyAllWindows()

複製代碼

上面這段 Python 指令稿執行之後，就會開啟一個 OpenCV 的視窗，顯示 RTSP 串流的影像：

Python OpenCV RTSP 串流影片

OpenCV 視窗顯示 RTSP 串流影像多行程版本

若在 CPU 處理速度比較慢的機器上，或是遇到影格率（frame rate）較高的影片，可能會出現單一執行緒無法及時處理的問題，這時候就可以考慮使用多行程（multiprocessing）的方式，以一個行程專門收取 RTSP 串流影像，然後將收進來的影像透過佇列（queue）交給另外一個行程來進行顯示，這樣就可以改善整體程式的處理效能，比較不會發生程式處理速度跟不上影片速度的問題。

1. from multiprocessing import Process, Queue
   
2. import cv2
   
3. 
   
4. def image_display(taskqueue):
   
5.     # 建立視窗
   
6.     cv2.namedWindow('image_display', cv2.WINDOW_AUTOSIZE)
   
7. 
   
8.     while True:
   
9.         # 從工作佇列取得影像
   
10.         image = taskqueue.get()
    
11. 
    
12.         # 若沒有影像則終止迴圈
    
13.         if image is None: break
    
14. 
    
15.         # 顯示影像
    
16.         cv2.imshow('image_display', image)
    
17.         cv2.waitKey(10)
    
18. 
    
19. if __name__ == '__main__':
    
20. 
    
21.     # 開啟 RTSP 串流
    
22.     vidCap = cv2.VideoCapture('rtsp://ipcam.stream:8554/bars')
    
23. 
    
24.     # 建立工作佇列
    
25.     taskqueue = Queue()
    
26. 
    
27.     # 建立並執行工作行程
    
28.     proc = Process(target=image_display, args=(taskqueue,))
    
29.     proc.start()
    
30. 
    
31.     while True:
    
32.         # 從 RTSP 串流讀取一張影像
    
33.         ret, image = vidCap.read()
    
34. 
    
35.         if ret:
    
36.             # 將影像放入工作佇列
    
37.             taskqueue.put(image)
    
38.         else:
    
39.             # 若沒有影像跳出迴圈
    
40.             break
    
41. 
    
42.     # 傳入 None 終止工作行程
    
43.     taskqueue.put(None)
    
44. 
    
45.     # 等待工作行程結束
    
46.     proc.join()
    
47. 
    
48.     # 釋放資源
    
49.     vidCap.release()
    
50. 
    
51.     # 關閉所有 OpenCV 視窗
    
52.     cv2.destroyAllWindows()

複製代碼

儲存 RTSP 串流影像

以下是使用 OpenCV 讀取 RTSP 串流影像之後，將其儲存成影片檔案的範例，在儲存影片時維持原影片的解析度與影格率，而編碼則採用 mp4v，靠著計算影格數的方式錄製 10 秒鐘的 RTSP 串流影像，儲存為 output.mp4：

1. from multiprocessing import Process, Queue
   
2. import cv2
   
3. 
   
4. def image_save(taskqueue, width, height, fps):
   
5. 
   
6.     # 指定影片編碼
   
7.     #fourcc = cv2.VideoWriter_fourcc(*'XVID')
   
8.     #fourcc = cv2.VideoWriter_fourcc(*'H264')
   
9.     fourcc = cv2.VideoWriter_fourcc(*'mp4v')
   
10. 
    
11.     # 建立 VideoWriter 物件
    
12.     writer = cv2.VideoWriter('output.mp4', fourcc, fps, (width, height))
    
13. 
    
14.     while True:
    
15.         # 從工作佇列取得影像
    
16.         image = taskqueue.get()
    
17. 
    
18.         # 若沒有影像則終止迴圈
    
19.         if image is None: break
    
20. 
    
21.         # 儲存影像
    
22.         writer.write(image)
    
23. 
    
24.     # 釋放資源
    
25.     writer.release()
    
26. 
    
27. if __name__ == '__main__':
    
28. 
    
29.     # 開啟 RTSP 串流
    
30.     vidCap = cv2.VideoCapture('rtsp://ipcam.stream:8554/bars')
    
31. 
    
32.     # 取得影像的尺寸大小
    
33.     width = int(vidCap.get(cv2.CAP_PROP_FRAME_WIDTH))
    
34.     height = int(vidCap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    
35. 
    
36.     # 取得影格率
    
37.     fps = vidCap.get(cv2.CAP_PROP_FPS)
    
38. 
    
39.     # 建立工作佇列
    
40.     taskqueue = Queue()
    
41. 
    
42.     # 建立並執行工作行程
    
43.     proc = Process(target=image_save, args=(taskqueue, width, height, fps))
    
44.     proc.start()
    
45. 
    
46.     # 計數器
    
47.     frame_counter = 0
    
48. 
    
49.     # 總錄製幀數（10 秒鐘）
    
50.     total_frames = fps * 10
    
51. 
    
52.     while frame_counter < total_frames:
    
53.         # 從 RTSP 串流讀取一張影像
    
54.         ret, image = vidCap.read()
    
55. 
    
56.         if ret:
    
57.             # 將影像放入工作佇列
    
58.             taskqueue.put(image)
    
59.             frame_counter += 1
    
60.         else:
    
61.             # 若沒有影像跳出迴圈
    
62.             break
    
63. 
    
64.     # 傳入 None 終止工作行程
    
65.     taskqueue.put(None)
    
66. 
    
67.     # 等待工作行程結束
    
68.     proc.join()
    
69. 
    
70.     # 釋放資源
    
71.     vidCap.release()

複製代碼

儲存長時間串流影片

若需要儲存長時間的 RTSP 串流影片，可以利用以下範例，將串流切成固定長度的影片，依照編號或時間戳記來命名儲存的影片檔案，方便後續處理。

1. from multiprocessing import Process, Queue
   
2. import cv2
   
3. from datetime import datetime
   
4. 
   
5. def image_save(taskqueue, width, height, fps, frames_per_file):
   
6. 
   
7.     # 指定影片編碼
   
8.     #fourcc = cv2.VideoWriter_fourcc(*'XVID')
   
9.     #fourcc = cv2.VideoWriter_fourcc(*'H264')
   
10.     fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    
11. 
    
12.     writer = None
    
13. 
    
14.     while True:
    
15.         # 從工作佇列取得影像
    
16.         image, frame_counter = taskqueue.get()
    
17. 
    
18.         # 若沒有影像則終止迴圈
    
19.         if image is None: break
    
20. 
    
21.         if frame_counter % frames_per_file == 0:
    
22. 
    
23.             if writer: writer.release()
    
24. 
    
25.             # 建立 VideoWriter 物件（以數字編號）
    
26.             # index = int(frame_counter // frames_per_file)
    
27.             # writer = cv2.VideoWriter(f'output-{index}.mp4', fourcc, fps, (width, height))
    
28. 
    
29.             # 建立 VideoWriter 物件（以時間命名）
    
30.             now = datetime.now()
    
31.             timestamp = now.strftime("%Y-%m-%d-%H-%M-%S")
    
32.             writer = cv2.VideoWriter(f'output-{timestamp}.mp4', fourcc, fps, (width, height))
    
33. 
    
34.         # 儲存影像
    
35.         writer.write(image)
    
36. 
    
37.     # 釋放資源
    
38.     writer.release()
    
39. 
    
40. if __name__ == '__main__':
    
41. 
    
42.     # 開啟 RTSP 串流
    
43.     vidCap = cv2.VideoCapture('rtsp://ipcam.stream:8554/bars')
    
44. 
    
45.     # 取得影像的尺寸大小
    
46.     width = int(vidCap.get(cv2.CAP_PROP_FRAME_WIDTH))
    
47.     height = int(vidCap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    
48. 
    
49.     # 取得影格率
    
50.     fps = vidCap.get(cv2.CAP_PROP_FPS)
    
51. 
    
52.     # 建立工作佇列
    
53.     taskqueue = Queue()
    
54. 
    
55.     # 計數器
    
56.     frame_counter = 0
    
57. 
    
58.     # 總錄製幀數（30 秒鐘）
    
59.     total_frames = fps * 30
    
60. 
    
61.     # 每個檔案的幀數（10 秒鐘）
    
62.     frames_per_file = fps * 10
    
63. 
    
64.     # 建立並執行工作行程
    
65.     proc = Process(target=image_save, args=(taskqueue, width, height, fps, frames_per_file))
    
66.     proc.start()
    
67. 
    
68.     while frame_counter < total_frames:
    
69.         # 從 RTSP 串流讀取一張影像
    
70.         ret, image = vidCap.read()
    
71. 
    
72.         if ret:
    
73.             # 將影像放入工作佇列
    
74.             taskqueue.put((image, frame_counter))
    
75.             frame_counter += 1
    
76.         else:
    
77.             # 若沒有影像跳出迴圈
    
78.             break
    
79. 
    
80.     # 傳入 None 終止工作行程
    
81.     taskqueue.put((None, None))
    
82. 
    
83.     # 等待工作行程結束
    
84.     proc.join()
    
85. 
    
86.     # 釋放資源
    
87.     vidCap.release()

複製代碼

這裡為了方便起見，檔案名稱用的時間戳記是放在工作行程中產生的，所以時間可能會有一些誤差，如果需要非常精確的時間，就要將時間戳記改為接收串流時同時取得時間。

錄製機器作動影片

這裡我的應用場景是將一台網路攝影機架設在工廠內，鏡頭對準特定的機台，然後我希望透過 RTSP 串流影片監看機台的狀況，當機台有動作時自動將作動過程的影片錄製下來，以利後續的 AI 分析，以下是我在開發過程中所用的基本架構範例。

1. from multiprocessing import Process, Queue
   
2. import cv2
   
3. from datetime import datetime
   
4. import numpy as np
   
5. 
   
6. # 開發模式
   
7. DEV_MODE = False
   
8. 
   
9. def printLogMsg(msg):
   
10.     now = datetime.now()
    
11.     timestamp = now.strftime("%Y/%m/%d %H:%M:%S")
    
12.     print("[{}] {}".format(timestamp, msg))
    
13. 
    
14. def image_save(taskqueue, width, height, fps):
    
15. 
    
16.     # 指定影片編碼
    
17.     fourcc = cv2.VideoWriter_fourcc(*'XVID')
    
18.     #fourcc = cv2.VideoWriter_fourcc(*'H264')
    
19.     #fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    
20. 
    
21.     writer = None
    
22. 
    
23.     while True:
    
24.         # 從工作佇列取得影像
    
25.         image, frameCounter = taskqueue.get()
    
26. 
    
27.         # 若沒有影像則終止迴圈
    
28.         if image is None: break
    
29. 
    
30.         if frameCounter == 0:
    
31. 
    
32.             if writer: writer.release()
    
33. 
    
34.             # 建立 VideoWriter 物件（以時間命名）
    
35.             now = datetime.now()
    
36.             timestamp = now.strftime("%Y%m%d-%H%M%S")
    
37.             writer = cv2.VideoWriter(f'test-output-{timestamp}.avi', fourcc, fps, (width, height))
    
38. 
    
39.         # 儲存影像
    
40.         writer.write(image)
    
41. 
    
42.     # 釋放資源
    
43.     if writer: writer.release()
    
44. 
    
45. def captureRTSP(src):
    
46.     # 開啟 RTSP 串流
    
47.     vidCap = cv2.VideoCapture(src)
    
48. 
    
49.     # 取得影像的尺寸大小
    
50.     imgWidth = int(vidCap.get(cv2.CAP_PROP_FRAME_WIDTH))
    
51.     imgHeight = int(vidCap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    
52.     printLogMsg(f"Size: {imgWidth} x {imgHeight}")
    
53. 
    
54.     # 取得影格率
    
55.     fps = vidCap.get(cv2.CAP_PROP_FPS)
    
56.     printLogMsg(f"FPS: {fps}")
    
57. 
    
58.     # 建立工作佇列
    
59.     taskqueue = Queue()
    
60. 
    
61.     # 目標影像 FPS
    
62.     desired_fps = 15
    
63. 
    
64.     # 建立並執行工作行程
    
65.     proc = Process(target=image_save, args=(taskqueue, imgWidth, imgHeight, desired_fps))
    
66.     proc.start()
    
67. 
    
68.     avgImage = None
    
69.     avgImageFloat = None
    
70. 
    
71.     # 計數器
    
72.     frameCounter = 0
    
73. 
    
74.     while vidCap.isOpened():
    
75.         # 從 RTSP 串流讀取一張影像
    
76.         ret, image = vidCap.read()
    
77. 
    
78.         if ret:
    
79.             # 忽略黑白影像（紅外線）
    
80.             if np.array_equal(image[:,:,0], image[:,:,1]) and np.array_equal(image[:,:,0], image[:,:,2]):
    
81.                 avgImage = None
    
82.                 avgImageFloat = None
    
83.                 if frameCounter:
    
84.                     printLogMsg(f"Stop recording (frame counter = {frameCounter})")
    
85.                 frameCounter = 0
    
86.                 continue
    
87. 
    
88.             # 取出中央部分影像，作為變動依據
    
89.             cropImage = image[(imgHeight//4):(imgHeight//4*3), (imgWidth//8*3):(imgWidth//8*5)]
    
90.             centerImage = cv2.resize(cropImage, (imgHeight//4, imgWidth//8))
    
91.             centerArea = imgHeight//4 * imgWidth//8
    
92. 
    
93.             # 初始化平均影像
    
94.             if avgImage is None:
    
95.                 avgImage = cv2.blur(centerImage, (4, 4))
    
96.                 avgImageFloat = np.float32(avgImage)
    
97.                 printLogMsg("avgImage initialization")
    
98.                 continue
    
99. 
    
100.             # 模糊處理
     
101.             blurImage = cv2.blur(centerImage, (4, 4))
     
102. 
     
103.             # 計算目前影格與平均影像的差異值
     
104.             diffImage = cv2.absdiff(avgImage, blurImage)
     
105. 
     
106.             # 將圖片轉為灰階
     
107.             grayImage = cv2.cvtColor(diffImage, cv2.COLOR_BGR2GRAY)
     
108. 
     
109.             # 篩選出變動程度大於門檻值的區域
     
110.             ret, binImage = cv2.threshold(grayImage, 5, 255, cv2.THRESH_BINARY)
     
111. 
     
112.             # 使用型態轉換函數去除雜訊
     
113.             kernel = np.ones((5, 5), np.uint8)
     
114.             binImage = cv2.morphologyEx(binImage, cv2.MORPH_OPEN, kernel, iterations=2)
     
115.             binImage = cv2.morphologyEx(binImage, cv2.MORPH_CLOSE, kernel, iterations=2)
     
116. 
     
117.             # 產生等高線
     
118.             countours, _ = cv2.findContours(binImage, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
     
119. 
     
120.             detected = False
     
121.             for c in countours:
     
122.                 if cv2.contourArea(c) > centerArea / 5:
     
123.                     detected = True
     
124.                     if DEV_MODE:
     
125.                         # 計算等高線的外框範圍
     
126.                         (x, y, w, h) = cv2.boundingRect(c)
     
127. 
     
128.                         # 畫出外框
     
129.                         cv2.rectangle(centerImage, (x, y), (x + w, y + h), (0, 255, 0), 2)
     
130.                     else:
     
131.                         break
     
132. 
     
133.             if DEV_MODE:
     
134.                 # 顯示偵測結果影像
     
135.                 cv2.imshow('image', centerImage)
     
136.                 cv2.imshow('avgImage', avgImage)
     
137.                 cv2.imshow('blurImage', blurImage)
     
138.                 if cv2.waitKey(1) & 0xFF == ord('q'):
     
139.                     # 傳入 None 終止工作行程
     
140.                     taskqueue.put((None, None))
     
141.                     break
     
142. 
     
143.             # 更新平均影像
     
144.             cv2.accumulateWeighted(blurImage, avgImageFloat, 0.1)
     
145.             avgImage = cv2.convertScaleAbs(avgImageFloat)
     
146. 
     
147.             # 將影像放入工作佇列
     
148.             if detected:
     
149.                 if DEV_MODE == False:
     
150.                     taskqueue.put((image, frameCounter))
     
151.                 if frameCounter == 0:
     
152.                     printLogMsg("Start recording")
     
153.                 frameCounter += 1
     
154.             else:
     
155.                 if frameCounter:
     
156.                     printLogMsg(f"Stop recording (frame counter = {frameCounter})")
     
157.                 frameCounter = 0
     
158.                 continue
     
159.         else:
     
160.             # 若沒有影像跳出迴圈
     
161.             printLogMsg("no frame")
     
162.             break
     
163. 
     
164.     # 傳入 None 終止工作行程
     
165.     taskqueue.put((None, None))
     
166. 
     
167.     # 等待工作行程結束
     
168.     proc.join()
     
169. 
     
170.     # 釋放資源
     
171.     vidCap.release()
     
172. 
     
173.     # 關閉所有 OpenCV 視窗
     
174.     cv2.destroyAllWindows()
     
175. 
     
176. if __name__ == '__main__':
     
177.     captureRTSP('rtsp://ipcam.stream:8554/bars')

複製代碼

在現場的網路攝影機是 24 小時全天運作的，但是機器作動的頻率很低，大部分的時間都是處於閒置狀態，而在機器閒置的時間，現場的燈也都是關閉的。正常開燈的時候網路攝影機使用一般的攝像頭，對應的影格率（FPS）為 15，但當網路攝影機遇到關燈的狀態時，會自動遷換為紅外線攝像頭，影像雖然還是 RGB 格式，但是色彩會轉為灰階，而此時對應的影格率（FPS）就會降為 10。

由於機器只會在開燈的時候作動，所以我在程式中加上判斷影像色彩的條件，若整張影像的 RGB 三個 channels 都完全有相同的值，就判定為紅外線影像，將其直接忽略，減少不必要的計算。

當取得串流影像之後，我們將畫面中央的部分取出來，並降低解析度（節省計算量），參考 Python 與 OpenCV 實作移動偵測程式教學，偵測影像中央是否有變動，當出現影像大幅度變動時，就將串流影片儲存下來。

另外為了方便開發與測試，我們靠著 DEV_MODE 變數設定開發模式，當程式處於開發模式的時候，會開啟監看視窗，顯示幾種關鍵的影像內容，同時開發模式也不會儲存任何影像。

在儲存影片檔案時，會自動將獨立的影片片段分開儲存，檔案名稱則自動以開始錄製的時間來命名，影片的解析度、影格率都維持跟來源影片相同。

https://officeguide.cc/python-opencv-open-display-save-rtsp-stream-tutorial-examples/

參考資料

• StackOverflow：Storing RTSP stream as video file with OpenCV VideoWriter
• StackOverflow：Opencv multiprocessing in python – queue sync