監視カメラビューアをPySimpleGuiでつくる

ネットワークWi-Fiカメラ - Tapo C200を購入しました。
ビューアアプリを作ったので、紹介します。

💡作成するアプリ

以下が概要です。

1. 接続情報を入力する
2. "Connect"を押すと、接続情報に従い接続
   • カメラ画面が表示される
3. 矢印ボタンでパンチルトを操作
4. "Disconnect"を押すと、切断する

⚙事前準備

アカウントの設定

購入したカメラは、↓となります。

https://www.amazon.co.jp/gp/product/B07YG7RNF2/ref=ppx_yo_dt_b_asin_title_o04_s00?ie=UTF8&psc=1

本カメラは、スマホの専用アプリより映像を見たり、パンチルトの操作ができます。

カメラの映像配信には、ユーザ名とパスワードが必要になります。
ユーザ名とパスワードのアカウント設定は、下記の公式サイトのFAQの

https://www.tapo.com/jp/faq/34/

ステップ1: Tapoアプリでアカウントを作成

を参照願います。

💻環境

開発環境

• windows10 64bit Home
  • Python 3.8.10
  • Python 3.10.2

Pythonモジュール

Pythonに依存するモジュールは以下となります。

> python -m venv env
(env) > pip install rtsp
(env) > pip install pysimplegui
(env) > pip install --upgrade onvif_zeep

📝手順

• カメラ映像を受信する
• パンチルト対応を行う
• パンチルト対応を行う(AbsoluteMove)

カメラ映像を受信する

RTSPライブストリームURL

TapoカメラのRTSPライブストリームURLは以下の様になっています。

• 高画質な1080P (1920*1080) ストリームの場合
  • rtsp://username:password@IP Address:554/stream1
• 低画質な360P (640*360) ストリームの場合
  • rtsp://username:password@IP Address:554/stream2

username, passwordは、事前準備のアカウント設定のものとなります。

コード

コードは以下となります。

view.py

import PySimpleGUI as sg
import rtsp
from PIL import Image, ImageTk

sg.theme('Dark Brown')

USER = "username"    # <=自分の環境に合わせる
PASS = "password"    # <=自分の環境に合わせる
IPADDR = "192.168.11.xx"    # <=自分の環境に合わせる
PORT = "554"
STREAM = "stream2"
ONVIF_PORT = "2020"

layout = [
        # カメラ接続情報
        [   sg.Text('IPADDR: ', size=(12, 1)), sg.InputText(default_text=IPADDR,  size=(20, 1),key='-ipaddr-'),
            sg.Text('PORT: ', size=(12, 1)), sg.InputText(default_text=PORT,  size=(20, 1),key='-port-'),
            sg.Text('STREAM: ', size=(12, 1)), sg.InputText(default_text=STREAM,  size=(20, 1),key='-stream-')],
        [   sg.Text('ONVIF_PORT: ', size=(12, 1)), sg.InputText(default_text=ONVIF_PORT,  size=(20, 1),key='-onvif_port-')],
        [   sg.Text('USER: ', size=(12, 1)), sg.InputText(default_text=USER,  size=(20, 1),key='-user-'),
            sg.Text('PASS: ', size=(12, 1)), sg.InputText(default_text=PASS,  size=(20, 1),key='-pass-')],

        # 画面表示
        [   sg.Image(filename='', key='image')],

        # 接続, 切断
        [   sg.Button('Connect', size=(10, 1), key ='-start-'),
            sg.Button('Disconnect', size=(10, 1), key = '-stop-')],
]

is_streaming = False
window = sg.Window('cam viewer', layout, location=(32, 32))

while True:
    event, values = window.read(timeout=20)
    if event in (None, '-exit-'):
        break

    elif event == '-start-':
        rtsp_url = f"rtsp://{values['-user-']}:{values['-pass-']}@{values['-ipaddr-']}:{values['-port-']}/{values['-stream-']}"
        client = rtsp.Client(rtsp_server_uri=rtsp_url, verbose=True)
        is_streaming = True

    elif event == '-stop-':
        is_streaming = False
        client.close()
        img = Image.new("RGB", (640, 360), color=0)
        window['image'].update(data=ImageTk.PhotoImage(img))

    if is_streaming:
        frame = client.read()
        if frame is not None:
            window['image'].update(data=ImageTk.PhotoImage(frame))
        else:
            print("none")

window.close()

実行手順

(env) > python view.py

"Connect"ボタンで映像配信ができます。

パンチルト対応を行う

パンチルト操作

パンチルトは、ONVIF - PTZサービスを利用します。

ONVIFとは、ネットワークカメラのインターフェースに互換性をもたせるために設立された標準化フォーラム

Pythonモジュールのonvif_zeepを使用します。

コード

UIの部分とカメラ部分をわけて記載します。

• UI部分

ptz_viewer.py

import PySimpleGUI as sg
from camera import CamPtz
from PIL import Image, ImageTk

sg.theme('Dark Brown')

USER = "username"    # <=自分の環境に合わせる
PASS = "password"    # <=自分の環境に合わせる
IPADDR = "192.168.11.xx"    # <=自分の環境に合わせる
PORT = "554"
STREAM = "stream2"
ONVIF_PORT = "2020"

layout = [
        # カメラ接続情報
        [   sg.Text('IPADDR: ', size=(12, 1)), sg.InputText(default_text=IPADDR,  size=(20, 1),key='-ipaddr-'),
            sg.Text('PORT: ', size=(12, 1)), sg.InputText(default_text=PORT,  size=(20, 1),key='-port-'),
            sg.Text('STREAM: ', size=(12, 1)), sg.InputText(default_text=STREAM,  size=(20, 1),key='-stream-')],
        [   sg.Text('ONVIF_PORT: ', size=(12, 1)), sg.InputText(default_text=ONVIF_PORT,  size=(20, 1),key='-onvif_port-')],
        [   sg.Text('USER: ', size=(12, 1)), sg.InputText(default_text=USER,  size=(20, 1),key='-user-'),
            sg.Text('PASS: ', size=(12, 1)), sg.InputText(default_text=PASS,  size=(20, 1),key='-pass-')],

        # 画面表示
        [   sg.Image(filename='', key='image')],

        # 接続, 切断
        [   sg.Button('Connect', size=(10, 1), key ='-start-'),
            sg.Button('Disconnect', size=(10, 1), key = '-stop-')],

        # パンチルト制御
        [   sg.Button('↖', size=(4, 1), font='Helvetica 14',key ='-pt_topleft-'),
            sg.Button('↑', size=(4, 1), font='Helvetica 14',key = '-pt_topcenter-'),
            sg.Button('↗', size=(4, 1), font='Helvetica 14', key='-pt_topright-'), ],
        [   sg.Button('←', size=(4, 1), font='Helvetica 14',key ='-pt_left-'),
            sg.Button('〇', size=(4, 1), font='Helvetica 14',key = '-pt_center-'),
            sg.Button('→', size=(4, 1), font='Helvetica 14', key='-pt_right-'), ],
        [   sg.Button('↙', size=(4, 1), font='Helvetica 14',key ='-pt_btmleft-'),
            sg.Button('↓', size=(4, 1), font='Helvetica 14',key = '-pt_btmcenter-'),
            sg.Button('↘', size=(4, 1), font='Helvetica 14', key='-pt_btmright-'), ],
]

is_streaming = False
window = sg.Window('cam viewer', layout, location=(32, 32))

while True:
    event, values = window.read(timeout=20)
    if event in (None, '-exit-'):
        break

    elif event == '-start-':
        cam_ptz = CamPtz(
            user = values['-user-'],
            pwd = values['-pass-'],
            ipaddr = values['-ipaddr-'],
            port = values['-port-'],
            stream = values['-stream-'],
            onvif_port = values['-onvif_port-']
        )
        # streaming
        cam_ptz.open()
        # onvif - move
        cam_ptz.setup_ptz()
        is_streaming = True

    elif event == '-stop-':
        is_streaming = False
        cam_ptz.close()
        img = Image.new("RGB", (640, 360), color=0)
        window['image'].update(data=ImageTk.PhotoImage(img))

    elif "-pt_" in event:
        x, y = 0, 0
        x = 1 if "right" in event else x
        x = -1 if "left" in event else x
        y = 1 if "top" in event else y
        y = -1 if "btm" in event else y
        print(event, "x:", x, "y:", y)
        if is_streaming:
            cam_ptz.move(x, y)

    if is_streaming:
        frame = cam_ptz.read()
        if frame is not None:
            window['image'].update(data=ImageTk.PhotoImage(frame))

window.close()

• カメラ部分

camera.py

import rtsp
from onvif import ONVIFCamera
import time

class CamStream(object):
    def __init__(self, **kwargs):
        self.user = kwargs["user"]
        self.pwd = kwargs["pwd"]
        self.ipaddr = kwargs["ipaddr"]
        self.port = kwargs["port"]
        self.stream = kwargs["stream"]
        self.onvif_port = kwargs["onvif_port"]
        self.is_connect = False


    def open(self):
        rtsp_url = f"rtsp://{self.user}:{self.pwd}@{self.ipaddr}:{self.port}/{self.stream}"
        self.client = rtsp.Client(rtsp_server_uri=rtsp_url, verbose=True)
        self.is_connect = True


    def read(self):
        if self.is_connect is False:
            return None
        return self.client.read()


    def close(self):
        if self.is_connect:
            self.client.close()
            self.is_connect = False
        return



class CamPtz(CamStream):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        pass
        self.XMAX = 1
        self.XMIN = -1
        self.YMAX = 1
        self.YMIN = -1
        self.moverequest = None
        self.ptz = None
        self.is_ptz_active = False


    def setup_ptz(self):
        mycam = ONVIFCamera(self.ipaddr, self.onvif_port, self.user, self.pwd)
        # Create media service object
        media = mycam.create_media_service()
        # Create ptz service object
        self.ptz = mycam.create_ptz_service()
        # Get target profile
        media_profile = media.GetProfiles()[0]
        # Get PTZ configuration options for getting continuous move range
        request = self.ptz.create_type('GetConfigurationOptions')
        request.ConfigurationToken = media_profile.PTZConfiguration.token
        ptz_configuration_options = self.ptz.GetConfigurationOptions(request)

        self.moverequest = self.ptz.create_type('ContinuousMove')
        self.moverequest.ProfileToken = media_profile.token
        if self.moverequest.Velocity is None:
            self.moverequest.Velocity = self.ptz.GetStatus({'ProfileToken': media_profile.token}).Position

        # Get range of pan and tilt
        # NOTE: X and Y are velocity vector
        self.XMAX = ptz_configuration_options.Spaces.ContinuousPanTiltVelocitySpace[0].XRange.Max
        self.XMIN = ptz_configuration_options.Spaces.ContinuousPanTiltVelocitySpace[0].XRange.Min
        self.YMAX = ptz_configuration_options.Spaces.ContinuousPanTiltVelocitySpace[0].YRange.Max
        self.YMIN = ptz_configuration_options.Spaces.ContinuousPanTiltVelocitySpace[0].YRange.Min
        return


    def move(self, x, y):
        self.moverequest.Velocity.PanTilt.x = x
        self.moverequest.Velocity.PanTilt.y = y
        if self.is_ptz_active:
            self.ptz.Stop({'ProfileToken': self.moverequest.ProfileToken})
        active = True
        self.ptz.ContinuousMove(self.moverequest)

実行手順

(env) > python ptz_viewer.py

"Connect"ボタンで映像配信ができます。
矢印ボタンでパンチルト操作ができます。

パンチルト対応を行う(AbsoluteMove)

コメントで教えていただいた方法について記載します。

パンチルト操作

パンチルトは、ONVIF - AbsoluteMoveを使ってパンチルトを行います。

コード

UIの部分とカメラ部分をわけて記載します。

• UI部分

ptz_viewer.py

import PySimpleGUI as sg
import rtsp
from PIL import Image, ImageTk

from onvifreq import OnvifRequest
import urllib.request


sg.theme('Dark Brown')

# 接続情報
USER = "username"    # <=自分の環境に合わせる
PASS = "password"    # <=自分の環境に合わせる
IPADDR = "192.168.11.xx"    # <=自分の環境に合わせる
PORT = "554"
STREAM = "stream2"
ONVIF_PORT = "2020"

# パンチルトパラメータ
X_MIN, X_MAX, X_STP = -170, 170, 10
Y_MIN, Y_MAX, Y_STP = -30, 30, 5
cur_x = 0
cur_y = 0
inited_ptz = False


layout = [
        # カメラ接続情報
        [   sg.Text('IPADDR: ', size=(12, 1)), sg.InputText(default_text=IPADDR,  size=(20, 1),key='-ipaddr-'),
            sg.Text('PORT: ', size=(12, 1)), sg.InputText(default_text=PORT,  size=(20, 1),key='-port-'),
            sg.Text('STREAM: ', size=(12, 1)), sg.InputText(default_text=STREAM,  size=(20, 1),key='-stream-')],
        [   sg.Text('ONVIF_PORT: ', size=(12, 1)), sg.InputText(default_text=ONVIF_PORT,  size=(20, 1),key='-onvif_port-')],
        [   sg.Text('USER: ', size=(12, 1)), sg.InputText(default_text=USER,  size=(20, 1),key='-user-'),
            sg.Text('PASS: ', size=(12, 1)), sg.InputText(default_text=PASS,  size=(20, 1),key='-pass-')],

        # 画面表示
        [   sg.Image(filename='', key='image')],

        # 接続, 切断
        [   sg.Button('Connect', size=(10, 1), key ='-start-'),
            sg.Button('Disconnect', size=(10, 1), key = '-stop-')],

        # パンチルト制御
        [   sg.Button('↖', size=(4, 1), font='Helvetica 14',key ='-pt_topleft-'),
            sg.Button('↑', size=(4, 1), font='Helvetica 14',key = '-pt_topcenter-'),
            sg.Button('↗', size=(4, 1), font='Helvetica 14', key='-pt_topright-'), ],
        [   sg.Button('←', size=(4, 1), font='Helvetica 14',key ='-pt_left-'),
            sg.Button('〇', size=(4, 1), font='Helvetica 14',key = '-pt_center-'),
            sg.Button('→', size=(4, 1), font='Helvetica 14', key='-pt_right-'), ],
        [   sg.Button('↙', size=(4, 1), font='Helvetica 14',key ='-pt_btmleft-'),
            sg.Button('↓', size=(4, 1), font='Helvetica 14',key = '-pt_btmcenter-'),
            sg.Button('↘', size=(4, 1), font='Helvetica 14', key='-pt_btmright-'), ],
]

is_streaming = False
window = sg.Window('cam viewer', layout, location=(32, 32))

while True:
    event, values = window.read(timeout=20)
    if event in (None, '-exit-'):
        break

    elif event == '-start-':
        rtsp_url = f"rtsp://{values['-user-']}:{values['-pass-']}@{values['-ipaddr-']}:{values['-port-']}/{values['-stream-']}"
        client = rtsp.Client(rtsp_server_uri=rtsp_url, verbose=True)
        onvif_request = OnvifRequest(username=values['-user-'], password=values['-pass-'])
        url = f"http://{values['-ipaddr-']}:{values['-onvif_port-']}/"
        headers = {'Content-Type': 'text/xml; charset=utf-8'}
        is_streaming = True
        inited_ptz = True

    elif event == '-stop-':
        is_streaming = False
        img = Image.new("RGB", (640, 360), color=0)
        window['image'].update(data=ImageTk.PhotoImage(img))

    elif "-pt_" in event or inited_ptz:
        x, y = cur_x, cur_y
        x = (x + X_STP) if "right" in event else x
        x = (x - X_STP) if "left" in event else x
        y = (y + Y_STP) if "top" in event else y
        y = (y - Y_STP) if "btm" in event else y
        x = X_MAX if x > X_MAX else x
        cur_x = X_MIN if x < X_MIN else x
        y = Y_MAX if y > Y_MAX else y
        cur_y = Y_MIN if y < Y_MIN else y

        print(event, "x:", cur_x, "y:", cur_y)
        if is_streaming:
            oreq = onvif_request.absolute_move(cur_x, cur_y)
            req = urllib.request.Request(url, data=oreq.encode(), method='POST', headers=headers)
            try:
                with urllib.request.urlopen(req) as response:
                    pass
                    # obody = response.read()
                    # oheaders = response.getheaders()
                    # ostatus = response.getcode()
                    # print(oheaders)
                    # print(ostatus)
                    # print(obody)
            except urllib.error.URLError as e:
                print(e.reason)
            inited_ptz = False


    if is_streaming:
        frame = client.read()
        if frame is not None:
            window['image'].update(data=ImageTk.PhotoImage(frame))
        else:
            print("none")

window.close()

• パンチルト部分

onvifreq.py

import hashlib
import os
import base64
from datetime import datetime


class OnvifRequest:
    def __init__(self, username, password):
        self.username = username
        self.password = password

    def absolute_move(self, x, y):
        command = """
        <AbsoluteMove xmlns="http://www.onvif.org/ver20/ptz/wsdl">
            <ProfileToken>profile_1</ProfileToken>
            <Position>
                <PanTilt x="{x}" y="{y}" xmlns="http://www.onvif.org/ver10/schema" space="http://www.onvif.org/ver10/tptz/PanTiltSpaces/PositionGenericSpace"></PanTilt>
                <Zoom x="0.0" xmlns="http://www.onvif.org/ver10/schema" space="http://www.onvif.org/ver10/tptz/ZoomSpaces/PositionGenericSpace"></Zoom>
            </Position>
            <Speed>
                <PanTilt x="1" y="1" xmlns="http://www.onvif.org/ver10/schema" space="http://www.onvif.org/ver10/tptz/PanTiltSpaces/GenericSpeedSpace"/>
                <Zoom x="1.0" xmlns="http://www.onvif.org/ver10/schema" space="http://www.onvif.org/ver10/tptz/ZoomSpaces/ZoomGenericSpeedSpace"/>
            </Speed>
        </AbsoluteMove>
        """
        return self.request(self.username, self.password, command.format(x=x, y=y))

    def request(self, username, password, command):
        created = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
        raw_nonce = os.urandom(20)
        nonce = base64.b64encode(raw_nonce)
        sha1 = hashlib.sha1()
        sha1.update(raw_nonce + created.encode("utf8") + password.encode("utf8"))
        raw_digest = sha1.digest()
        digest = base64.b64encode(raw_digest)

        request = """
        <s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope">
        <s:Header>
            <Security xmlns="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd" s:mustUnderstand="1">
            <UsernameToken>
                <Username>{username}</Username>
                <Password Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest">{digest}</Password>
                <Nonce EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary">{nonce}</Nonce>
                <Created xmlns="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">{created}</Created>
            </UsernameToken>
            </Security>
        </s:Header>
        <s:Body xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
            {command}
        </s:Body>
        </s:Envelope>
        """
        return request.format(
            username=username,
            nonce=nonce.decode("utf8"),
            created=created,
            digest=digest.decode("utf8"),
            command=command,
        )

実行手順

(env) > python ptz_viewer.py

"Connect"ボタンで映像配信ができます。
矢印ボタンでパンチルト操作ができます。

さいごに

カメラの映像の取得やパンチルトができるといろいろできそうです。