RTSP で画像を取得する

1. RTSP 表記仕様

[概要]

RTSP で接続するための表記を以下に記載します。

「ネットワークカメラCGIコマンドインターフェース仕様書 統合版」[1] で下記に記載されている情報を元に加筆などしています。

• 「10.5. 汎用ビューアで H.264 を見る」
• 「10.6. 汎用ビューアで H.265 を見る」

i-PRO カメラ

(例) rtsp://admin:password@192.168.0.10/MediaInput/stream_1

補足： カメラ側の映像圧縮方式設定が H.264/H.265 のいずれの場合も上記で接続できました。

マルチセンサーカメラ（X8570 / S8530）

i-PRO (Panasonic) 従来 model

(例) rtsp://admin:password@192.168.0.10/MediaInput/h264/stream_1

注意事項

• RTSPを使用するためには「インターネットモード(over HTTP) の設定を On にする」と[1]に記載されています。

2. i-PRO カメラと RTSP 接続して映像を表示してみる

[概要]

Python で i-PRO カメラ（i-PRO mini (WV-S7130W)）と RTSP 接続して映像を表示してみます。
とりあえず映像を取得してPC画面に表示するまでをやってみます。

• Stream1 はデフォルトで 1920x1024、H.265 となっています。
  必要に応じて H.264 へ変更した方が処理は軽くなりそうです。（推奨）
• それでもまだ処理が重たい場合は、接続先を stream1 から stream2 または stream3 などへ変更してみてください。
• 遊んでみた印象ですが、リアルタイム性（映像遅延性）という観点では極力フレームレートを高めにしておく方が良さそうです。特にブラウザ表示、フレームレートを下げると遅延が大きくなりました。Iフレームを1秒周期で送信しているようなので、恐らく固定枚数分の映像蓄積を行って映像表示を滑らかにしているのだろうと私は予想しています。10枚前後っぽいかな。

[評価環境1]

[評価環境2]

[プログラム]

[プログラムソース "connect_with_rtsp_1.py"]

'''
[abstract]
    RTSP で i-PRO カメラと接続してみる 
[library install]
    pip install opencv-python 
'''

import cv2
user_id = "user-id"     # ご使用のカメラ設定に合わせて変更
user_pw = "password"    # ご使用のカメラ設定に合わせて変更
host = "192.168.0.10"   # ご使用のカメラ設定に合わせて変更
cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")

while(True):
    try:
        ret, frame = cap.read()
        if ret == True:
            # 使っているPC画面に適当に収まる表示大きさへリサイズ
            frame2 = cv2.resize(frame, (1280, 760))
            cv2.imshow('VIDEO', frame2)

        # コンソール上で ctrl-c を押すとプログラムを終了する。
        # GUI 上で ctrl-c してもプログラムを終了できない。
        cv2.waitKey(1)    # 注意： imshow() 関数は、この cv2.waitkey() が無いと画面表示してくれない！

    except KeyboardInterrupt:
        print("KeyboardInterrupt")
        break

cap.release()
cv2.destroyAllWindows()

上記プログラムを動かしてみます。実行はこんな感じで行います。

python connect_with_rtsp_1.py

Windows環境で複数の Python バージョンをインストールしている場合、下図のような感じで実行バージョンを指定することもできます。
こちらはバージョン 3.10 の Python で実行を指示する例です。

py -3.10 connect_with_rtsp_1.py

上記プログラムを動かした様子を動画で示します。

こんなに簡単なプログラムでとても快適な映像表示を実現することができました。

[動画] RTSP でカメラと接続して映像表示してみた様子

i-PRO カメラの IP アドレスを調べるには、下記「EasyIP Setup Tool Plus」を使ってみてください。

EasyIP Setup Tool Plus - リモートカメラシステム - 放送・業務用映像システム - Panasonic

本プログラムでは下記ライブラリを使用しています。概要だけ記載します。

3.プログラムを改善する

[概要]

前章で作成したプログラムはとても簡単に作成できましたが、いろいろと課題がありました。
とりあえず下記３つの課題を解決してみます。

課題１
プログラムを起動するたびにウィンドウ位置が変わる。場合によっては画面外へ表示する場合もあって不便。
適当に画面内に収まる場所に表示してほしい。
⇨ 指定する場所にウィンドウを表示するようにします。

課題２
プログラムを終了するのが大変。
ウィンドウ右上の×を押すとウィンドウがいったん消えるが、すぐに再表示されて終われない。
⇨ ウィンドウ右上の✕ボタンでプログラムを終了できるようにします。

課題３
同様に、任意のキー入力でプログラムを終了できるとうれしい。
⇨ "z" キー押下でプログラムを終了できるようにします。

[評価環境1]

[評価環境2]

[プログラム]

[プログラムソース "connect_with_rtsp_2.py"]

'''
[abstract]
    i-PRO mini (WV-S7130W) と接続して映像を表示してみる

[library install]
    pip install opencv-python
'''

import cv2

user_id = "user-id"     # ご使用のカメラ設定に合わせて変更
user_pw = "password"    # ご使用のカメラ設定に合わせて変更
host = "192.168.0.10"   # ご使用のカメラ設定に合わせて変更
winname = "VIDEO"

cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")

#
windowInitialized = False

# Exception 定義
BackendError = type('BackendError', (Exception,), {})

'''
[Abstract]
    対象ウィンドウが存在するかを確認する。
[Param]
    winname :       ウィンドウタイトル
[Return]
    True :          対象ウィンドウは存在する
    False :         対象ウィンドウは存在しない
[Exception]
    BackendError :  バックエンドで使用している Qt でエラー発生
'''
def IsWindowVisible(winname):
    try:
        ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
        if ret == -1:
            raise BackendError('Use Qt as backend to check whether window is visible or not.')

        return bool(ret)

    except cv2.error:
        return False


while(True):
    try:
        ret, frame = cap.read()
        if ret == True:
            frame2 = cv2.resize(frame, (1280, 760))
            cv2.imshow(winname, frame2)

            if windowInitialized==False:
                # ウィンドウ表示位置が安定しないので、最初の起動時のみ表示場所を指定
                cv2.moveWindow(winname, 100, 100)
                windowInitialized = True

        # "z" キーを押されていたら終了
        k = cv2.waitKey(1)    # 注意： imshow() 関数は、この cv2.waitkey() が無いと画面表示してくれない！
        if k == ord("z"):
            break
        
        # 指定ウィンドウが無かったら終了
        if not IsWindowVisible(winname):
            break

    except KeyboardInterrupt:
        print("KeyboardInterrupt")
        break

cap.release()
cv2.destroyAllWindows()

4. OpenCV で顔検知を加えてみる

JPEG による実装と同様に、RTSP の実装でも OpenCV による顔検知を実装してみます。

RTSP 接続では映像情報は受け身です。このため高解像度、高フレームレートの映像を処理したとき、OpenCV の処理が追いつくかが心配な部分です。

[評価環境1]

[評価環境2]

4-1. まずは単純にやってみる

とにかくまずはやってみます。

JPEG による実装と同様に、映像を受信するたびの OpenCV で毎回認識処理を行ってみます。

[プログラムソース "connect_with_rtsp_3_1.py"]

'''
[abstract]
    i-PRO mini (WV-S7130W) と接続して映像を表示してみる。
    ここでは opencv を使って顔検知を追加してみます。

[library install]
    pip install opencv-python
'''

import cv2

user_id = "user-id"     # ご使用のカメラ設定に合わせて変更
user_pw = "password"    # ご使用のカメラ設定に合わせて変更
host = "192.168.0.10"   # ご使用のカメラ設定に合わせて変更
winname = "VIDEO"

# Exception 定義
BackendError = type('BackendError', (Exception,), {})

'''
[Abstract]
    対象ウィンドウが存在するかを確認する。
[Param]
    winname :       ウィンドウタイトル
[Return]
    True :          対象ウィンドウは存在する
    False :         対象ウィンドウは存在しない
[Exception]
    BackendError :  バックエンドで使用している Qt でエラー発生
'''
def IsWindowVisible(winname):
    try:
        ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
        if ret == -1:
            raise BackendError('Use Qt as backend to check whether window is visible or not.')

        return bool(ret)

    except cv2.error:
        return False

'''
[Abstract]
    顔検知して認識結果を返す
[Param]
    cascade :       OpenCV の CascadeClassifierオブジェクト
    image :         OpenCV 形式の画像
[Return]
    認識結果
'''
def DetectFaces(cascade, image):
    # 顔検出のためにグレイスケール画像に変換
    img_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

    # 顔を画像から検出 
    face_list = cascade.detectMultiScale(img_gray, minSize=(100, 100))

    # 検出結果を返す
    return face_list


'''
[Abstract]
    検出された顔枠情報リストを使って、image 上に赤枠を描画する。
[Param]
    image :         OpenCV 形式の画像
    face_list :     検出された顔枠情報リスト
[Return]
    無し
'''
def DrawFaceRectangles(image, face_list):
    # 検出した顔の数だけ赤枠を描画
    if len(face_list) != 0:
        for (pos_x, pos_y, w, h) in face_list:
            print(f"pos_x = {pos_x}, pos_y = {pos_y}, w = {w}, h = {h}")
            cv2.rectangle(image, (pos_x, pos_y), (pos_x + w, pos_y + h), (0,0,255), thickness=5)


'''
[Abstract]
    main 関数
'''
if __name__ == '__main__':

    cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")

    #
    windowInitialized = False

    # 顔を識別するためのファイル
    cascade_file = "haarcascade_frontalface_alt2.xml"       # 顔
    #cascade_file = "haarcascade_eye.xml"                   # 目？
    #cascade_file = "haarcascade_eye_tree_eyeglasses.xml"   # 目？
    cascade = cv2.CascadeClassifier(cascade_file)

    while(True):
        try:
            ret, frame = cap.read()
            if ret == True:
                # 顔検知
                face_list = DetectFaces(cascade, frame)

                # 検出した顔枠を描画
                DrawFaceRectangles(frame, face_list)

                # PC画面サイズに合わせて適当にリサイズ後、表示
                frame2 = cv2.resize(frame, (1280, 760))
                cv2.imshow(winname, frame2)

                if windowInitialized==False:
                    # ウィンドウ表示位置が安定しないので、最初の起動時のみ表示場所を指定
                    cv2.moveWindow(winname, 100, 100)
                    windowInitialized = True

            # "z" キーを押されていたら終了
            k = cv2.waitKey(1)      # 注意： imshow() 関数は、この cv2.waitkey() が無いと画面表示してくれない！
            if k == ord("z"):
                break
            
            # 指定ウィンドウが無かったら終了
            if not IsWindowVisible(winname):
                break


        except KeyboardInterrupt:
            print("KeyboardInterrupt")
            break

    cap.release()
    cv2.destroyAllWindows()

結果：

私のゲーミングPCではこれでもそこそこ動作しました。思ったより動く、という感想です。

が、それでもだんだん映像が遅れていきます。
顔検知処理と描画の部分をコメントアウトすると、映像表示の遅れはなくなります。
やはり顔検知処理は PC にとって結構重たい処理のようです。

ちょっと残念。何か改善策を考えてみたいところです。

4-2. 顔検知部分を別プロセスの処理にしてみる

そこで、顔検知部分を別タスクに分離することで、映像受信と映像デコード処理を止めずにできるだけ顔検知をやってみる、という感じにプログラムを修正してみます。

NOTE

• 別タスクというと一般的なプログラムでは "スレッド" というテクニックを使いますが、どうやら CPython と呼ばれるプラットフォームの場合はスレッドは複数の処理を同時に実行してくれないらしいです。そこで、ここではプロセスを使用します。
• Queue というIOでプロセス間で情報をやり取りします。
• 顔検知を行った映像と赤枠表示する映像は異なるものになりますが、ここではこれを許容することとします。

[プログラムソース "connect_with_rtsp_3_2.py"]

'''
[abstract]
    i-PRO mini (WV-S7130W) と接続して映像を表示してみる。
    ここでは opencv を使って顔検知を追加してみます。

[library install]
    pip install opencv-python
'''

import cv2
import multiprocessing as mp
from queue import Empty


user_id = "user-id"     # ご使用のカメラ設定に合わせて変更
user_pw = "password"    # ご使用のカメラ設定に合わせて変更
host = "192.168.0.10"   # ご使用のカメラ設定に合わせて変更
winname = "VIDEO"

# 顔を識別するためのファイル
cascade_file = "haarcascade_frontalface_alt2.xml"       # 顔
#cascade_file = "haarcascade_eye.xml"                   # 目？
#cascade_file = "haarcascade_eye_tree_eyeglasses.xml"   # 目？
cascade = cv2.CascadeClassifier(cascade_file)


# Exception 定義
BackendError = type('BackendError', (Exception,), {})

'''
[Abstract]
    対象ウィンドウが存在するかを確認する。
[Param]
    winname :       ウィンドウタイトル
[Return]
    True :          対象ウィンドウは存在する
    False :         対象ウィンドウは存在しない
[Exception]
    BackendError :  バックエンドで使用している Qt でエラー発生
'''
def IsWindowVisible(winname):
    try:
        ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
        if ret == -1:
            raise BackendError('Use Qt as backend to check whether window is visible or not.')

        return bool(ret)

    except cv2.error:
        return False


'''
[Abstract]
    顔検知して認識結果を返す
[Param]
    cascade :       OpenCV の CascadeClassifierオブジェクト
    image :         OpenCV 形式の画像
[Return]
    検出結果
'''
def DetectFaces(cascade, image):
    # 顔検出のためにグレイスケール画像に変換
    img_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

    # 検出した顔の位置情報を取得
    face_list = cascade.detectMultiScale(img_gray, minSize=(100, 100))

    return face_list


'''
[Abstract]
    顔検知タスク
[Param]
    q1 :        [i] 顔検知する画像を保存する Queue
    q2 :        [o] 顔検知した結果を保存する Queue
[Return]
    無し
'''
# def DetectFacesProcess(cascade, q1, q2):
def DetectFacesProcess(q1, q2):
    while True:
        try:
            image = q1.get(True, 10)

            # 終了処理： q1.get から取得したものが int で -1 なら終了
            if type(image) == int:
                if image == -1:
                    break

            # 顔検知
            face_list = DetectFaces(cascade, image)

            q2.put(face_list)
        except Empty: # timeout of q1.get()
            print("Timeout happen.")

    print("Finish DetectFacesProcess()")    


'''
[Abstract]
    検出された顔枠情報リストを使って、image 上に赤枠を描画する。
[Param]
    image :         OpenCV 形式の画像
    face_list :     検出された顔枠情報リスト
[Return]
    無し
'''
def DrawFaceRectangles(image, face_list):
    # 検出した顔の数だけ赤枠を描画
    if len(face_list) != 0:
        for (pos_x, pos_y, w, h) in face_list:
            print(f"pos_x = {pos_x}, pos_y = {pos_y}, w = {w}, h = {h}")
            cv2.rectangle(frame, (pos_x, pos_y), (pos_x + w, pos_y + h), (0,0,255), thickness=5)


'''
[Abstract]
    main 関数 
'''
if __name__ == '__main__':

    cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")

    #
    windowInitialized = False

    q1 = mp.Queue()
    q2 = mp.Queue()

    # "cannot pickle object" というエラーが出て解決できなかったので、args に cascade を加えるのを断念
    # 合わせて cascade をグローバル変数に。悔しい。
    p = mp.Process(target=DetectFacesProcess, args=(q1, q2))
    # p = mp.Process(target=DetectFacesProcess, args=(cascade, q1, q2))
    p.daemon = True
    p.start()

    init = False

    while(True):
        try:
            ret, frame = cap.read()
            if ret == True:
                # 顔検知
                if (q1.qsize() <= 1) and (q2.qsize() <= 1):
                    q1.put(frame)

                if q2.qsize() != 0:
                    face_list = q2.get()
                    init = True

                if init == True:
                    # 検出した顔枠を描画
                    DrawFaceRectangles(frame, face_list)

                # PC画面サイズに合わせて適当にリサイズ後、表示
                frame2 = cv2.resize(frame, (1280, 760))
                cv2.imshow(winname, frame2)

                if windowInitialized==False:
                    # ウィンドウ表示位置が安定しないので、最初の起動時のみ表示場所を指定
                    cv2.moveWindow(winname, 100, 100)
                    windowInitialized = True

            # "z" キーを押されていたら終了
            k = cv2.waitKey(1)      # 注意： imshow() 関数は、この cv2.waitkey() が無いと画面表示してくれない！
            if k == ord("z"):
                break
            
            # 指定ウィンドウが無かったら終了
            if not IsWindowVisible(winname):
                break


        except KeyboardInterrupt:
            print("KeyboardInterrupt")
            break

    # Terminate process p
    q1.put(-1)
    # Waiting for process p to finish
    p.join()

    print("Finish main()")
    cap.release()
    cv2.destroyAllWindows()

結果：

期待する動作をしてくれるようになりました。

プロセス起動の引数として cascade を一緒に渡したかったのですが、"cannot pickle object" というエラーを発生して実現できませんでした。残念ながら cascade をグローバル変数へ変更することで問題を回避しています。
対応策がわかったら記事をアップデートしたいと思います。

[動画] OpenCV で顔検知してみた様子

5. 連番の JPEG ファイルで保存する

RTSP で受信した映像を連番の JPEG ファイルで保存してみます。

• "image" という名称でフォルダを作成し、ここに "image_000001.jpg" というようなファイル名で画像を保存します。
• OpenCV の関数 imwrite を使って保存します。imwrite はファイル名の拡張子から保存形式を自動的に判断してくれるのでとても便利です。'.jpg' の部分を '.png' へ変えるなど実験してみてください。
• 映像受信と表示を優先して処理するために、JPEGファイル保存の処理は別プロセスを作成してここで行うようにします。
  別プロセスの起動には数秒を要するようなので、プログラム起動から 10秒 経過してから JPEGファイル保存を開始するようにしています。
• ファイル保存プロセスとのIFには再び multiprocessing.queue を使用し、画像データとファイル名を渡します。
  queue へ多くの情報が蓄積されるとメモリ不足になる恐れがあるので、ここでは 30 という登録上限を設けています。queue が上限に達している場合は受信画像のファイル保存を行いません。
  → 私の環境では 15fps までは全く問題なし。30fpsにするとファイル保存処理が追い付かず queue の残が少しずつ増えていきました。この場合でも保存枚数 100枚 であれば全て保存できました。
• 無制限にファイル保存すると面倒なので上限枚数を設けます。下記プログラムでは 100枚 を保存するとプログラム終了するようにします。

[評価環境1]

[評価環境2]

[プログラムソース "connect_with_rtsp_4.py"]

'''
[Abstract]
    Try connecting to an i-PRO camera with RTSP.
    RTSP で i-PRO カメラと接続してみる

[Details]
    Save the received image as a JPEG file.
    Add a 6-digit number to the end of the file name and save it as a serial number.
    Exit the program after saving 100 files.

    プログラム起動から10秒経過後から、受信した映像を JPEG ファイル保存します。
    ファイル名の末尾に６ケタの番号を付けて連番で保存します。
    100枚 保存したらプログラムを終了します。

[Note]
    If you set a high number of frame rates or high resolution for the camera,
    the program may not save the file because the program's file saving process may not be in time.

    カメラ側の設定でフレームレート数を高くする、解像度を高くする、などした場合は
    ファイル保存の処理が追い付かなくなってバッファが満杯になり、
    ファイル保存が間引かれる場合があります。

[Library install]
    pip install opencv-python
'''
import cv2
import multiprocessing as mp
from queue import Empty
import os
import datetime

user_id     = "user-id"         # Change to match your camera setting
user_pw     = "password"        # Change to match your camera setting
host        = "192.168.0.10"    # Change to match your camera setting
winname     = "VIDEO"           # Window title
pathOut     = 'image'           # Image file save folder name
queue_max   = 30                # Maximum number of queues
save_max    = 100               # Number of files to save


# Exception 定義
BackendError = type('BackendError', (Exception,), {})

'''
[Abstract]
    対象ウィンドウが存在するかを確認する。
[Param]
    winname :       ウィンドウタイトル
[Return]
    True :          対象ウィンドウは存在する
    False :         対象ウィンドウは存在しない
[Exception]
    BackendError :
'''
def IsWindowVisible(winname):
    try:
        ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
        if ret == -1:
            raise BackendError('Use Qt as backend to check whether window is visible or not.')

        return bool(ret)

    except cv2.error:
        return False


'''
[Abstract]
    画像ファイル保存タスク
[Param]
    imageQueue :    [i] 保存する画像を蓄える Queue
[Return]
    無し
'''
def SaveImageProcess(imageQueue):
    while True:
        try:
            image, filename = imageQueue.get(True, 10)  # timeout 10 sec.

            # 終了処理： imageQueue.get から取得したものが int で -1 なら終了
            if type(image) == int:
                if image == -1:
                    break

            # ファイル保存
            cv2.imwrite(filename, image)

        except Empty: # timeout of q1.get()
            print("Timeout happen.")

    print("Finish SaveImageProcess()")


'''
[Abstract]
    main 関数
'''
if __name__ == '__main__':

    cap = cv2.VideoCapture(f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1")

    windowInitialized = False
    count = 0
    imageQueue = mp.Queue()
    starttime = datetime.datetime.now()

    p = mp.Process(target=SaveImageProcess, args=(imageQueue,))
    p.start()

    while True:
        try:
            ret, frame = cap.read()
            if ret == True:

                # プログラム起動から 10秒 以上経過したらファイル保存
                if (datetime.datetime.now() - starttime).seconds > 10:
                    # queue サイズが上限以内であればファイル保存
                    # print(imageQueue.qsize())
                    if imageQueue.qsize() < queue_max:
                        # Save jpeg file.
                        count += 1
                        filename = os.path.join(pathOut, 'image_{:06d}.jpg'.format(count))
                        print(filename)
                        imageQueue.put([frame, filename])
                        if count >= save_max:
                            break

                # 使っているPC画面に適当に収まる表示大きさへリサイズ
                frame2 = cv2.resize(frame, (1280, 760))
                
                # 映像表示
                cv2.imshow(winname, frame2)

                if windowInitialized==False:
                    # 最初の起動時のみ表示位置を指定
                    cv2.moveWindow(winname, 100, 100)
                    windowInitialized = True

            # "z" キーを押されていたら終了
            k = cv2.waitKey(1)      # 注意： imshow() 関数は、この cv2.waitkey() が無いと画面表示してくれない！
            if k == ord("z"):
                break
            
            # 指定ウィンドウが無かったら終了
            if not IsWindowVisible(winname):
                break

        except KeyboardInterrupt:
            # コンソール上で ctrl-c を押すとプログラムを終了する。
            print("KeyboardInterrupt")
            break

    # Terminate process p
    imageQueue.put([-1,-1])
    print("Wait for process p to finish")
    p.join()
    print("Finish main()")

    cap.release()
    cv2.destroyAllWindows()

6. 映像切断時の再接続処理を追加

ここまでのプログラムは、カメラとの接続が30秒以上切断すると接続が復活しませんでした。
OpenCV の read() 関数のタイムアウトは30秒となっているようです。30秒以内に接続が復活していれば自動的に再接続してくれるのですが、30秒を超えると自動的には復活しません。

"connect_with_rtsp_2.py" を元に再接続処理を追加してこの問題を解決してみたいと思います。

ポイント

• read() の戻り値が False であったら再接続を行います。
• release() で切断後、cv2.VideoCapture() で再接続します。

NOTE

• 切断の試験はカメラの電源をOff/Onする、通信をOff/Onする、などで試験してください。
• カメラとの接続が正常状態に戻ってから再接続までは最長30秒かかります。気長にお待ちください。
• read() のタイムアウト時間を変更する API を確認できませんでした。おそらく固定値で変更できないと思われます。

[評価環境]

[プログラムソース "connect_with_rtsp_5.py"]

'''
[Abstract]
    Try connecting to an i-PRO camera with RTSP.
    RTSP で i-PRO カメラと接続してみる。

[Details]
    Add reconnection when video is disconnected.
    映像切断時の再接続処理を追加する。

[Library install]
    pip install opencv-python
'''

from nturl2path import url2pathname
import cv2

user_id     = "user-id"         # Change to match your camera setting
user_pw     = "password"        # Change to match your camera setting
host        = "192.168.0.10"    # Change to match your camera setting
winname     = "VIDEO"           # Window title
url         = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"

#
windowInitialized = False

# Exception definition.
BackendError = type('BackendError', (Exception,), {})

'''
[Abstract]
    Check if the target window exists.
    対象ウィンドウが存在するかを確認する。
[Param]
    winname :       Window title
[Return]
    True :          exist
                    存在する
    False :         not exist
                    存在しない
[Exception]
    BackendError :
'''
def IsWindowVisible(winname):
    try:
        ret = cv2.getWindowProperty(winname, cv2.WND_PROP_VISIBLE)
        if ret == -1:
            raise BackendError('Use Qt as backend to check whether window is visible or not.')

        return bool(ret)

    except cv2.error:
        return False


'''
[Abstract]
    main function.
'''
if __name__ == '__main__':

    cap = cv2.VideoCapture(url)

    while True:
        try:
            ret, frame = cap.read()
            if ret == True:
                # Please modify the value to fit your PC screen size.
                frame2 = cv2.resize(frame, (1280, 760))
                # Display video.
                cv2.imshow(winname, frame2)

                if windowInitialized==False:
                    # Specify window position only once at startup.
                    cv2.moveWindow(winname, 100, 100)
                    windowInitialized = True
            else:
                print("cap.read() return False.")
                # The timeout period seems to be 30 seconds.
                # And there seems to be no API to change the timeout value.
                
                # Reconnect
                cap.release()
                cap = cv2.VideoCapture(url)

            # Press the "z" key to finish.
            k = cv2.waitKey(1)
            if k == ord("z"):
                break
            
            # Exit the program if there is no specified window.
            if not IsWindowVisible(winname):
                break
        
        except KeyboardInterrupt:
            # Press'[ctrl] + [c]' on the console to exit the program.
            print("KeyboardInterrupt")
            break

    cap.release()
    cv2.destroyAllWindows()

7. GUIで映像表示してみる（tkinter）

ここまでのプログラムは全てOpenCVが作成するウィンドウ表示でした。
ここでは独自の GUI を作成してここに映像表示する例を示します。

GUI 表示の実現方法もいろいろありますが、ここでは Python 標準の tkinter を使用してみます。

tkinter のインストール方法は環境により異なるようです。各人の環境にあった方法をインターネットで調べて実施してください。

ポイント

• tkinter で動画を表示するときは、after() 関数で繰り返し処理を行います。
• tkinter で表示するために ImageTk.PhotoImage という型に変換する必要があります。
  numpy.ndarray → PIL.Image → ImageTk.PhotoImage という順に変換して tkinter で表示します。
• OpenCVの画像は BGR の並び順になっています。tkinker で表示するために BGR データを RGB へ並び替える必要があります。

[評価環境]

7-1. まずは単純にやってみる

とにかくまずはやってみます。

• プログラム起動により自動的にカメラと接続して映像表示を開始します。
• 画面（Canvvas）をマウス左ボタンクリックする度に、映像表示を停止、映像表示を開始、を反復します。
• 画像フォーマットの変換はそれぞれ下記場所で行っています。ソースコード中のコメントを探してみてください。
  • (1) BGR データを RGB へ並び替える
  • (2) numpy.ndarray → PIL.Image
  • (3) PIL.Image → ImageTk.PhotoImage

[プログラムソース "connect_with_rtsp_6_1.py"]

'''
[Abstract]
    Try connecting to an i-PRO camera with RTSP.
    RTSP で i-PRO カメラと接続してみる。

[Details]
    Display the video with GUI using tkinter.
    tkinter を使ったGUIで映像を表示します。

[Library install]
    pip install opencv-python
'''

import cv2
import time
import tkinter as tk
from PIL import Image, ImageTk, ImageOps


user_id     = "user-id"         # Change to match your camera setting
user_pw     = "password"        # Change to match your camera setting
host        = "192.168.0.10"    # Change to match your camera setting
winname     = "VIDEO"           # Window title
url         = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"


class Application(tk.Frame):
    def __init__(self, master = None):
        super().__init__(master)
        self.pack()

        # Window settings.
        self.master.title("Display i-PRO camera with tkinter")      # Window title
        self.master.geometry("800x600+100+100")                     # Window size, position
        
        # Create canvas.
        self.canvas = tk.Canvas(self.master)

        # Add mouse click event to canvas.
        self.canvas.bind('<Button-1>', self.canvas_click)

        # Place canvas.
        self.canvas.pack(expand = True, fill = tk.BOTH)

        # Raise a video display event (disp_image) after 500m
        self.cap = None
        self.disp_id = self.after(500, self.disp_image)

    def canvas_click(self, event):
        ''' Event handling with mouse clicks on canvas '''
        if self.disp_id is None:
            # Connect camera.
            self.cap = cv2.VideoCapture(url)
            # Display image.
            self.disp_image()
        else:
            # Release camera.
            self.after_cancel(self.disp_id)
            self.disp_id = None
            self.cap.release()

    def disp_image(self):
        ''' Display image on Canvas '''

        if self.cap == None:
            # Connect camera.
            self.cap = cv2.VideoCapture(url)

        # Get frame.
        ret, frame = self.cap.read()

        if ret == True:
            # (1) Convert image from BGR to RGB.
            cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

            # (2) Convert image from ndarray to PIL.Image.
            pil_image = Image.fromarray(cv_image)

            # Get canvas size.
            canvas_width = self.canvas.winfo_width()
            canvas_height = self.canvas.winfo_height()

            # Resize the image to the size of the canvas without changing the aspect ratio.
            # アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
            pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))

            # (3) Convert image from Pillow Image to PhotoImage
            # PIL.Image から PhotoImage へ変換する
            self.photo_image = ImageTk.PhotoImage(image=pil_image)

            # Display image on the canvas.
            self.canvas.create_image(
                canvas_width / 2,       # Image display position (center of the canvas)
                canvas_height / 2,                   
                image=self.photo_image  # image data
                )
            
        else:
            print("cap.read() return False.")
            # The timeout period seems to be 30 seconds.
            # And there seems to be no API to change the timeout value.
            time.sleep(1)

            # Reconnect
            self.cap.release()
            self.cap = cv2.VideoCapture(url)

        # Raise a video display event (disp_image) after 1ms.
        self.disp_id = self.after(1, self.disp_image)


if __name__ == "__main__":
    root = tk.Tk()
    app = Application(master = root)
    app.mainloop()

概ね期待に近い動作をしてくれました。
しかし残念ながら、私のPC環境では 30fps の映像を表示すると少しずつ映像が遅延してしまいました。
OpenCV のみで表示したときは問題なかったので、映像受信後の画像フォーマット変換および表示の処理が追い付いていないと予想されます。

何かしら改善を行う必要がありそうです。

7-2. 映像受信を別プロセスにしてパフォーマンス改善

プログラム "connect_with_rtsp_6_1.py" は概ね良好に動作しましたが、私のPC環境では 30fps の映像を表示すると少しずつ映像が遅延してしまいました。
OpenCV のみで表示したときは問題なかったので、映像受信後の画像フォーマット変換および表示の処理が追い付いていないと予想されます。

ここでは、映像受信を別プロセスにしてパフォーマンス改善を試みてみます。
受信した映像を Queue に蓄積すれば、Queue の蓄積数を確認して以降の処理を間引くこともできそうです。これも合わせてやってみましょう。

ポイント

• 映像受信処理を別プロセスにする
• Queue に蓄積された映像枚数が一定数より多い場合、取得した映像の画像変換と表示を行わない。

[プログラムソース "connect_with_rtsp_6_2.py"]

'''
[Abstract]
    Try connecting to an i-PRO camera with RTSP.
    RTSP で i-PRO カメラと接続してみる。

[Details]
    Display the video with GUI using tkinter.
    Try to improve performance by creating a video receiving process.
    
    tkinter を使ったGUIで映像を表示します。
    映像受信プロセスを作成することでパフォーマンス改善を試みます。

    BGR → RGB
    numpy.ndarray → PIL.Image → ImageTk.PhotoImage
    (1) BGR → RGB
    (2) numpy.ndarray → PIL.Image
    (3) PIL.Image → ImageTk.PhotoImage

[Library install]
    pip install opencv-python
'''

import cv2
import time
import tkinter as tk
from tkinter import messagebox
from PIL import Image, ImageTk, ImageOps
import multiprocessing as mp


user_id     = "user-id"         # Change to match your camera setting
user_pw     = "password"        # Change to match your camera setting
host        = "192.168.0.10"    # Change to match your camera setting
winname     = "VIDEO"           # Window title
url         = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"


class Application(tk.Frame):
    def __init__(self, master = None):
        super().__init__(master)
        self.pack()

        # Window settings.
        self.master.title("Display i-PRO camera with tkinter")      # Window title
        self.master.geometry("800x600+100+100")                     # Window size, position

        # Event registration for window termination.
        self.master.protocol("WM_DELETE_WINDOW", self.on_closing_window)
        
        # Create canvas.
        self.canvas = tk.Canvas(self.master)

        # Add mouse click event to canvas.
        self.canvas.bind('<Button-1>', self.canvas_click)

        # Place canvas.
        self.canvas.pack(expand = True, fill = tk.BOTH)

        # Create image receiving process and queue
        self.imageQueue = mp.Queue()
        self.request = mp.Value('i', 0)     # -1 : Exit ReceiveImageProcess.
                                            #  0 : Normal.
                                            #  1 : Connect camera.
                                            #  2 : Release camera.
        self.p = mp.Process(target=ReceiveImageProcess, args=(self.imageQueue, self.request))
        self.p.start()

        # Raise a video display event (disp_image) after 500m
        self.disp_id = self.after(500, self.disp_image)

    def on_closing_window(self):
        ''' Window closing event. '''

        if messagebox.askokcancel("QUIT", "Do you want to quit?"):
            # Request terminate process self.p.
            self.request.value = -1

            # Waiting for process p to finish
            time.sleep(1)

            # Flash buffer.
            # The program cannot complete p.join() unless the imageQueue is emptied.
            for i in range(self.imageQueue.qsize()):
                pil_image = self.imageQueue.get()

            # Wait for process p to be terminated.
            self.p.join()
            self.master.destroy()
            print("Finish Application.")

    def canvas_click(self, event):
        ''' Event handling with mouse clicks on canvas '''

        if self.disp_id is None:
            # Connect camera.
            self.request.value = 1
            # Display image.
            self.disp_image()

        else:
            # Release camera.
            self.request.value = 2
            # Cancel scheduling
            self.after_cancel(self.disp_id)
            self.disp_id = None

    def disp_image(self):
        ''' Display image on Canvas '''

        # If there is data in the imageQueue, the program receives the data and displays the video.
        num = self.imageQueue.qsize()
        if num > 0:
            if (num > 5):
                num -= 1
            for i in range(num):
                cv_image = self.imageQueue.get()

            # (2) Convert image from ndarray to PIL.Image.
            pil_image = Image.fromarray(cv_image)

            # Get canvas size.
            canvas_width = self.canvas.winfo_width()
            canvas_height = self.canvas.winfo_height()

            # Resize the image to the size of the canvas without changing the aspect ratio.
            # アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
            pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))

            # (3) Convert image from PIL.Image to PhotoImage
            # PIL.Image から PhotoImage へ変換する
            self.photo_image = ImageTk.PhotoImage(image=pil_image)

            # Display image on the canvas.
            self.canvas.create_image(
                canvas_width / 2,       # Image display position (center of the canvas)
                canvas_height / 2,                   
                image=self.photo_image  # image data
                )
            
        else:
            pass

        # Raise a video display event (disp_image) after 1ms.
        self.disp_id = self.after(1, self.disp_image)


def ReceiveImageProcess(imageQueue, request):
    """
    Receive Image Process.
    
    Args:
        imageQueue      [o] This process stores the received image data in the imageQueue.
        request         [i] Shared memory for receiving requests from the main process.
                            -1: Terminate process.
                             0: Nothing.
                             1: Connect camera.
                             2: Release camera connection.
    Returns:
        None
    Raises
        None
    """
    # Connect camera.
    cap = cv2.VideoCapture(url)
    
    while True:
        if cap != None:
            # Get frame.
            ret, frame = cap.read()
            
            if ret == True:
                # (1) Convert image from BGR to RGB.
                cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                
                if imageQueue.qsize() < 10:
                    imageQueue.put(cv_image)
            
            else:
                print("cap.read() return False.")
                # The timeout period seems to be 30 seconds.
                # And there seems to be no API to change the timeout value.
                time.sleep(1)
                
                # Reconnect
                cap.release()
                cap = cv2.VideoCapture(url)
        else:
            time.sleep(0.1)
        
        # Check process termination request.
        if request.value == -1:
            # Terminate process.
            cap.release()
            request.value = 0
            break
        
        # Check connect request.
        if request.value == 1:
            cap = cv2.VideoCapture(url)
            request.value = 0
        
        # Check release request.
        if request.value == 2:
            cap.release()
            cap = None
            request.value = 0
    
    print("Terminate SaveImageProcess().")


if __name__ == "__main__":
    root = tk.Tk()
    app = Application(master = root)
    app.mainloop()

7-3. メニュー・ボタンを追加して GUI アプリらしくしてみる

プログラム "connect_with_rtsp_6_2.py" により GUI で映像表示をそれなりにできるようになりました。

ここではメニュー、ボタンをそれぞれ追加することで GUI アプリっぽくしてみます。

ポイント

• [File] > [Quit] メニューを追加する
• [Quit] ボタンを追加する

[プログラムソース "connect_with_rtsp_6_3.py"]

'''
[Abstract]
    Try connecting to an i-PRO camera with RTSP.
    RTSP で i-PRO カメラと接続してみる。

[Details]
    Display the video with GUI using tkinter.
    Add menus and buttons to make it look like a GUI app.
    
    tkinter を使ったGUIで映像を表示します。
    メニューとボタンを追加してGUIアプリらしくします。

    BGR → RGB
    numpy.ndarray → PIL.Image → ImageTk.PhotoImage
    (1) BGR → RGB
    (2) numpy.ndarray → PIL.Image
    (3) PIL.Image → ImageTk.PhotoImage
    
[Library install]
    pip install opencv-python
'''

from re import X
from turtle import window_height
import cv2
import time
import tkinter as tk
from tkinter import messagebox
from PIL import Image, ImageTk, ImageOps
import multiprocessing as mp


user_id     = "user-id"         # Change to match your camera setting
user_pw     = "password"        # Change to match your camera setting
host        = "192.168.0.10"    # Change to match your camera setting
winname     = "VIDEO"           # Window title
url         = f"rtsp://{user_id}:{user_pw}@{host}/MediaInput/stream_1"


class Application(tk.Frame):
    def __init__(self, master = None):
        super().__init__(master)
        self.pack()

        # Window settings.
        self.master.title("Display i-PRO camera with tkinter")      # Window title
        self.master.geometry("800x600+100+100")                     # Window size, position

        # Event registration for window termination.
        self.master.protocol("WM_DELETE_WINDOW", self.on_closing_window)

        # Create menu.
        menubar = tk.Menu(self.master)
        self.master.configure(menu=menubar)
        filemenu = tk.Menu(menubar)
        menubar.add_cascade(label='File', menu=filemenu)
        filemenu.add_command(label='Quit', command = self.on_closing_window)

        # Create button_frame
        self.button_frame = tk.Frame(self.master, padx=10, pady=10, relief=tk.RAISED, bd=2)
        self.button_frame.pack(side = tk.BOTTOM, fill=tk.X)

        # Create quit_button
        self.quit_button = tk.Button(self.button_frame, text='Quit', width=10, command = self.on_closing_window)
        self.quit_button.pack(side=tk.RIGHT)
        
        # Create canvas.
        self.canvas = tk.Canvas(self.master)

        # Add mouse click event to canvas.
        self.canvas.bind('<Button-1>', self.canvas_click)

        # Place canvas.
        self.canvas.pack(expand = True, fill = tk.BOTH)

        # Create image receiving process and queue
        self.imageQueue = mp.Queue()
        self.request = mp.Value('i', 0)     # -1 : Exit ReceiveImageProcess.
                                            #  0 : Normal.
                                            #  1 : Connect camera.
                                            #  2 : Release camera.
        self.p = mp.Process(target=ReceiveImageProcess, args=(self.imageQueue, self.request))
        self.p.start()

        # Raise a video display event (disp_image) after 500m
        self.disp_id = self.after(500, self.disp_image)

    def on_closing_window(self):
        ''' Window closing event. '''

        if messagebox.askokcancel("QUIT", "Do you want to quit?"):
            # Request terminate process self.p.
            self.request.value = -1

            # Waiting for process p to finish
            time.sleep(1)

            # Flash buffer.
            # The program cannot complete p.join() unless the imageQueue is emptied.
            for i in range(self.imageQueue.qsize()):
                pil_image = self.imageQueue.get()

            # Wait for process p to be terminated.
            self.p.join()
            self.master.destroy()
            print("Finish Application.")

    def canvas_click(self, event):
        ''' Event handling with mouse clicks on canvas '''

        if self.disp_id is None:
            # Connect camera.
            self.request.value = 1
            # Display image.
            self.disp_image()

        else:
            # Release camera.
            self.request.value = 2
            # Cancel scheduling
            self.after_cancel(self.disp_id)
            self.disp_id = None

    def disp_image(self):
        ''' Display image on Canvas '''

        # If there is data in the imageQueue, the program receives the data and displays the video.
        num = self.imageQueue.qsize()
        if num > 0:
            if (num > 5):
                num -= 1
            for i in range(num):
                cv_image = self.imageQueue.get()

            # (2) Convert image from ndarray to PIL.Image.
            pil_image = Image.fromarray(cv_image)

            # Get canvas size.
            canvas_width = self.canvas.winfo_width()
            canvas_height = self.canvas.winfo_height()

            # Resize the image to the size of the canvas without changing the aspect ratio.
            # アスペクトを維持したまま画像を Canvas と同じサイズにリサイズ
            pil_image = ImageOps.pad(pil_image, (canvas_width, canvas_height))

            # (3) Convert image from PIL.Image to PhotoImage
            # PIL.Image から PhotoImage へ変換する
            self.photo_image = ImageTk.PhotoImage(image=pil_image)

            # Display image on the canvas.
            self.canvas.create_image(
                canvas_width / 2,       # Image display position (center of the canvas)
                canvas_height / 2,                   
                image=self.photo_image  # image data
                )
            
        else:
            pass

        # Raise a video display event (disp_image) after 1ms.
        self.disp_id = self.after(1, self.disp_image)


def ReceiveImageProcess(imageQueue, request):
    """
    Receive Image Process.

    Args:
        imageQueue      [o] This process stores the received image data in the imageQueue.
        request         [i] Shared memory for receiving requests from the main process.
                            -1: Terminate process.
                             0: Nothing.
                             1: Connect camera.
                             2: Release camera connection.
    Returns:
        None
    Raises
        None
    """

    # Connect camera.
    cap = cv2.VideoCapture(url)

    while True:
        if cap != None:
            # Get frame.
            ret, frame = cap.read()

            if ret == True:
                # (1) Convert image from BGR to RGB.
                cv_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

                if imageQueue.qsize() < 10:
                    imageQueue.put(cv_image)

            else:
                print("cap.read() return False.")
                # The timeout period seems to be 30 seconds.
                # And there seems to be no API to change the timeout value.
                time.sleep(1)

                # Reconnect
                cap.release()
                cap = cv2.VideoCapture(url)
        else:
            time.sleep(0.1)
                
        # Check process termination request.
        if request.value == -1:
            # Terminate process.
            cap.release()
            request.value = 0
            break

        # Check connect request.
        if request.value == 1:
            cap = cv2.VideoCapture(url)
            request.value = 0

        # Check release request.
        if request.value == 2:
            cap.release()
            cap = None
            request.value = 0

    print("Terminate SaveImageProcess().")


if __name__ == "__main__":
    root = tk.Tk()
    app = Application(master = root)
    app.mainloop()

[図] tkinter で作成した GUI アプリ

NOTE

• Frame のサイズ（width, height）を指定することができますが、デフォルトではこれらの値を指定しても無視されます。
  pack_propagate(False) または propagate(False) を実行することで意図通りに設定できるようになります。

ライセンス

本ページの情報は、特記無い限り下記 MIT ライセンスで提供されます。

参考

• [1] ネットワークカメラCGIコマンドインターフェース仕様書　統合版
  https://sol.panasonic.biz/security/cgi-bin/ipro/download/tbookmarka_m.cgi?m=%20&mm=2012100910461872
• [2] [python] ネットワークカメラの画像を RTSP で取得して表示・保存 - へっぽこプログラマーの備忘録 (kuttsun.blogspot.com)
  https://kuttsun.blogspot.com/2021/08/python-rtsp.html
• [3]【Python/tkinter】OpenCVのカメラ動画をCanvasに表示する | イメージングソリューション (imagingsolution.net)
  https://imagingsolution.net/program/python/tkinter/display_opencv_video_canvas/
• [4] コラム - ゼロから歩くPythonの道 | 第19回　tkinterを使って画面を作る…前にmainloopの役割と使えるウィジェットを確認してみた｜CTC教育サービス 研修/トレーニング (ctc-g.co.jp)
  https://www.school.ctc-g.co.jp/columns/hishinuma/hishinuma19.html
• [5]【Python tkinter】Frame（フレーム）の使い方：作成・配置・オプション一覧 | OFFICE54
  https://office54.net/python/tkinter/python-tkinter-frame
• [6] ImageOps Module — Pillow (PIL Fork) 9.1.0 documentation
  https://pillow.readthedocs.io/en/stable/reference/ImageOps.html#imageops-module
• [7] c++ — ネットワークカメラを使用したOpenCV (web-dev-qa-db-ja.com)
  https://www.web-dev-qa-db-ja.com/ja/c++/ネットワークカメラを使用したopencv/957939846/
• [8] OpenCV|を使用してIPカメラからRTSPストリームをキャプチャするリンデフス (lindevs.com)
  https://lindevs.com/capture-rtsp-stream-from-ip-camera-using-opencv

記載

Programming Items トップページ

プライバシーポリシー