爬取交通部 CCTV 即時影像
這篇教學會使用 OpenCV 讀取影像串流 MJPEG 的能力,在不透過 Requests 等爬蟲函式庫的狀況下,爬取交通部提供的 CCTV 即時影像開放資料,除了可以即時監看影像串流,也可以將影像串流保留到自己電腦中。
快速導覽:
使用 CCTV 即時影像 API
雖然台灣的開放資料已經相當豐富,但卻沒有統一的管道可以查看所有的即時影像,然而也越來越多的即時影像,也開始使用 youtube 直播的方式播放,下方列出一些有提供即時影像 API 的來源:
- 交通部開放資料
- 政府資料開放平臺
以「臺中市交通即時道路影像」為例,開啟開放資料的網頁後,點擊並下載 JSON 格式檔案。
開啟 JSON 檔案,檔案裡 url 的數值,就是即時串流影像的網址,目前有提供即時串流的 API 都能取得串流網址,只在於 JSON 結構有差異。
首先使用瀏覽器開啟網址,就能看到即時影像的串流圖片,使用滑鼠在圖片上按右鍵,選擇「複製圖片網址」。
使用下方的程式碼,將剛剛複製的圖片網址貼到 url 的位置,執行後就能在本機環境查看即時影像。
import cv2
# 來源串流網址
url = 'https://tcnvr4.taichung.gov.tw:7001/media/00-0F-7C-13-DA-83.mpjpeg?resolution=240p&auth=cHVibGljdmlld2VyOjYxNmU2NmIxM2RkMjg6YWJiMzZlNzMxNmUzM2M2MjdhMjg5MzY2M2Y4MjhmOGY'
cap = cv2.VideoCapture(url) # 讀取來源
if not cap.isOpened():
print("Cannot open camera")
exit()
while True:
ret, frame = cap.read() # 讀取影片的每一幀
if not ret:
print("Cannot receive frame") # 如果讀取錯誤,印出訊息
cap = cv2.VideoCapture(url) # 有時候串流間隔時間較久會中斷,中斷時重新讀取
continue
cv2.imshow('oxxostudio', frame) # 如果讀取成功,顯示該幀的畫面
if cv2.waitKey(1) == ord('q'): # 每一毫秒更新一次,直到按下 q 結束
break
cap.release() # 所有作業都完成後,釋放資源
cv2.destroyAllWindows() # 結束所有視窗
手動取得即時影像網址
如果找不到開放資料的 API,則需要手動讀取網址,下方列出一些可以手動抓取即時影像網址的來源:
以「高雄市即時交通資訊服務網」為例,開啟網站後,可以點擊任一個地圖裡的「攝影機」圖示查看即時影像,開啟即時影像後,使用滑鼠在圖片上按右鍵,選擇「複製圖片網址」。
使用下方的程式碼,將剛剛複製的圖片網址貼到 url 的位置,執行後就能在本機環境查看即時影像。
import cv2
# 來源串流網址
url = 'https://cctv1.kctmc.nat.gov.tw/27e5c086/snapshot?t=1714029708895'
cap = cv2.VideoCapture(url) # 讀取來源
if not cap.isOpened():
print("Cannot open camera")
exit()
while True:
ret, frame = cap.read() # 讀取影片的每一幀
if not ret:
print("Cannot receive frame") # 如果讀取錯誤,印出訊息
cap = cv2.VideoCapture(url) # 有時候串流間隔時間較久會中斷,中斷時重新讀取
continue
cv2.imshow('oxxostudio', frame) # 如果讀取成功,顯示該幀的畫面
if cv2.waitKey(1) == ord('q'): # 每一毫秒更新一次,直到按下 q 結束
break
cap.release() # 所有作業都完成後,釋放資源
cv2.destroyAllWindows() # 結束所有視窗
下載即時影像為圖片
既然能夠讀取影像,就一定能夠下載影像,參考「寫入並儲存圖片」教學,使用下方程式碼,就能在讀取到即時影像時,將影像儲存為 jpg 檔案。
import cv2
url = 'https://cctv1.kctmc.nat.gov.tw/27e5c086/snapshot?t=1714029708895'
cap = cv2.VideoCapture(url)
ret, frame = cap.read() # 讀取影片的每一幀
cv2.imwrite('test-jpg.jpg', frame) # 存成 jpg
如果要將程式改得更加彈性好用,可以使用下方的程式碼,當按下鍵盤 a 的時候就會拍照儲存,並使用 time 模組取得當下的時間作為檔名 ( 也可以改成自動拍照,例如間隔多久就儲存一張 )。
import cv2
import time
url = 'https://cctv1.kctmc.nat.gov.tw/27e5c086/snapshot?t=1714029708895'
cap = cv2.VideoCapture(url)
if not cap.isOpened():
print("Cannot open camera")
exit()
while True:
ret, frame = cap.read() # 讀取影片的每一幀
if ret:
print('ok')
else:
print("Cannot receive frame") # 如果讀取錯誤,印出訊息
cap = cv2.VideoCapture(url)
continue
key = cv2.waitKey(100) # 每 0.1 秒更新一次
if key == ord('q'): # 按下 q 結束
break
elif key == ord('a'): # 按下 a 儲存當下影格
cv2.imwrite(f'test{time.time_ns()}.jpg', frame) # 存成 jpg,取得當下時間作為檔名
cv2.imshow('oxxostudio', frame) # 顯示畫面
cap.release() # 所有作業都完成後,釋放資源
cv2.destroyAllWindows() # 結束所有視窗
下載即時影像為影片
除了將即時影像下載為單張圖片,也可以參考「寫入並儲存影片」的做法,將即時影像下載為「影片」,下載時可以設定影片每秒的格數 ( 幀數 ),下方程式碼執行後,會將高雄夢時代路口的即時影像,下載為每秒 10 格的影片。
import cv2
import time
url = 'https://cctv1.kctmc.nat.gov.tw/27e5c086/snapshot?t=1714029708895'
cap = cv2.VideoCapture(url)
init = False # 建立空影片只需一次,所以使用一個變數做判斷
if not cap.isOpened():
print("Cannot open camera")
exit()
while True:
ret, frame = cap.read() # 讀取影片的每一幀
if ret:
if not init:
init = True # 第一次先建立空影片
w = frame.shape[1] # 取得影片寬度
h = frame.shape[0] # 取得影片高度
fourcc = cv2.VideoWriter_fourcc(*'MJPG') # 設定影片的格式為 MJPG
out = cv2.VideoWriter('output.mp4', fourcc, 10.0, (w, h)) # 產生空的影片,一秒 10 格
out.write(frame) # 寫入影片
print('ok')
else:
print("Cannot receive frame") # 如果讀取錯誤,印出訊息
cap = cv2.VideoCapture(url)
continue
key = cv2.waitKey(100)
if key == ord('q'): # 每一毫秒更新一次,直到按下 q 結束
break
elif key == ord('a'): # 按下 a 儲存當下影格
cv2.imwrite(f'test{time.time_ns()}.jpg', frame) # 存成 jpg,取得當下時間作為檔名
cv2.imshow('oxxostudio', frame) # 如果讀取成功,顯示該幀的畫面
cap.release() # 所有作業都完成後,釋放資源
cv2.destroyAllWindows() # 結束所有視窗
完成後,不僅會自動儲存影片,按下鍵盤的 a,也會自動將該幀的影像儲存為 jpg。
意見回饋
如果有任何建議或問題,可傳送「意見表單」給我,謝謝~