爬取 CCTV 即时图像

这篇教学会使用 OpenCV 读取图像串流 MJPEG 的能力，在不透过 Requests 等爬虫函数库的状况下，爬取交通部提供的 CCTV 即时图像开放数据，除了可以即时监看图像串流，也可以将图像串流保留到自己电脑中。

参考：OpenCV 函数库、读取串流图片 MJPEG ( Motion JPEG )

快速导览：

• 使用 CCTV 即时图像 API
• 手动取得即时图像网址
• 下载即时图像为图片
• 下载即时图像为影片

使用 CCTV 即时图像 API

虽然台湾的开放数据已经相当丰富，但却没有统一的管道可以查看所有的即时图像，然而也越来越多的即时图像，也开始使用 youtube 直播的方式播放，下方列出一些有提供即时图像 API 的来源：

• 交通部开放数据
  • CCTV 静态信息
  • 省道交控路侧设备数据 ( 闭路电视摄影监控 )
• 政府数据开放平台
  • 台中市交通即时道路图像
  • 嘉义市即时交通信息－路况监视摄影机CCTV信息
  • 台南市路况摄影机(CCTV)位置数据

以“台中市交通即时道路图像”为例，打开开放数据的网页后，点击并下载 JSON 格式文件。

打开 JSON 文件，文件里 url 的数值，就是即时串流图像的网址，目前有提供即时串流的 API 都能取得串流网址，只在于 JSON 结构有差异。

首先使用浏览器打开网址，就能看到即时图像的串流图片，使用鼠标在图片上按右键，选择“复制图片网址”。

使用下方的程序码，将刚刚复制的图片网址贴到 url 的位置，执行后就能在本机环境查看即时图像。

参考：OpenCV 函数库、读取串流图片 MJPEG ( Motion JPEG )

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 的位置，执行后就能在本机环境查看即时图像。

参考：OpenCV 函数库、读取串流图片 MJPEG ( Motion JPEG )

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。

意见反馈