映像デバイス¶

OpenCV のインストール¶

$ uv add opencv-python
$ uv sync

利用可能な映像デバイス一覧を取得する¶

利用可能な映像デバイス一覧を取得するには、 cv2.VideoCapture クラスを使用します。

import cv2


def get_available_video_devices(max_video_devices: int = 10) -> list[int]:
    """
    利用可能なビデオデバイスのリストを取得する。

    :param max_video_devices: チェックする最大のデバイス番号
    :return: 利用可能なビデオデバイスの番号のリスト
    """
    available_video_devices: list[int] = []

    for i in range(max_video_devices):
        cap: cv2.VideoCapture = cv2.VideoCapture(i)
        if cap is None or not cap.isOpened():
            print(f"カメラが利用できません: {i}")
        else:
            print(f"カメラが利用できます: {i}")
            available_video_devices.append(i)
        cap.release()

    return available_video_devices


if __name__ == "__main__":
    # スクリプトが直接実行された場合、利用可能なビデオデバイスを検出して表示
    available_devices: list[int] = get_available_video_devices()
    print(f"利用可能なビデオデバイス: {available_devices}")

OpenCV ではカメラデバイス名を取得することができません。 カメラデバイス名まで取得したい場合は FFmpeg の記事が参考になりますので、ご参考ください。

Capture/Webcam - FFmpeg

映像デバイスをキャプチャして表示する¶

cv2.imshow を利用する事で、映像デバイスからキャプチャした映像を表示することができます。

import cv2


def capture_and_play(camera_id: int, width: int, height: int) -> None:
    """
    指定されたカメラデバイスからビデオをキャプチャし、再生する。

    :param camera_id: 使用するカメラデバイスのID
    :param width: キャプチャする映像の幅
    :param height: キャプチャする映像の高さ
    """
    # カメラデバイスを開く
    cap: cv2.VideoCapture = cv2.VideoCapture(camera_id)
    if not cap.isOpened():
        print(f"Cannot open camera {camera_id}")
        return

    # 解像度を設定
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)

    try:
        while True:
            # フレームをキャプチャ
            ret: bool
            frame: cv2.typing.MatLike
            ret, frame = cap.read()

            # 正しくフレームが読み込まれた場合のみ表示
            if ret:
                cv2.imshow("Camera Feed", frame)

            # 'q' キーが押されたらループを抜ける
            if cv2.waitKey(1) == ord("q"):
                break
    except KeyboardInterrupt:
        print("Interrupted by user")
    finally:
        # キャプチャの後始末とウィンドウをすべて閉じる
        cap.release()
        cv2.destroyAllWindows()


if __name__ == "__main__":
    # 例: カメラデバイス ID 0 を使用し、解像度を 640x480 に設定
    capture_and_play(camera_id=0, width=640, height=480)

映像デバイスをキャプチャして送信する¶

create_video_source を利用する事で、映像デバイスからキャプチャした映像を Sora に送信することができます。

import json
import os
import threading
import time
from threading import Event
from types import TracebackType
from typing import Any

import cv2
import numpy as np

from sora_sdk import Sora, SoraConnection, SoraSignalingErrorCode, SoraVideoSource


class Sendonly:
    def __init__(self, signaling_urls: list[str], channel_id: str, device_id: int) -> None:
        """
        Sendonly クラスの初期化

        :param signaling_urls: Sora シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        :param device_id: 使用するカメラデバイスの ID
        """
        self._signaling_urls: list[str] = signaling_urls
        self._channel_id: str = channel_id

        self._connection_id: str | None = None

        self._connected: Event = Event()
        self._closed: bool = False

        self._video_height: int = 480
        self._video_width: int = 640

        self._sora: Sora = Sora()
        self._video_source: SoraVideoSource = self._sora.create_video_source()
        self._video_capture: cv2.VideoCapture = cv2.VideoCapture(device_id)
        if not self._video_capture.isOpened():
            raise RuntimeError(f"カメラが開けません: camera_id={device_id}")

        # Sora への接続設定
        self._connection: SoraConnection = self._sora.create_connection(
            signaling_urls=signaling_urls,
            role="sendonly",
            channel_id=channel_id,
            audio=False,
            video=True,
            video_source=self._video_source,
        )

        self._connection.on_set_offer = self._on_set_offer
        self._connection.on_notify = self._on_notify
        self._connection.on_disconnect = self._on_disconnect

    def __enter__(self) -> "Sendonly":
        return self.connect()

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.disconnect()

    def connect(self) -> "Sendonly":
        """
        ビデオ入力ループを開始し、Sora に接続する
        """
        self._video_input_thread = threading.Thread(target=self._video_input_loop, daemon=True)
        self._video_input_thread.start()

        try:
            # Sora へ接続
            self._connection.connect()

            # 接続が成功するまで待つ
            assert self._connected.wait(10), "接続に失敗しました"
        except Exception as e:
            # connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
            self._connection.disconnect()
            raise e

        return self

    def _video_input_loop(self) -> None:
        """
        ビデオフレームを継続的に取得し、Sora に送信するループ
        """
        while not self._closed:
            ret, frame = self._video_capture.read()
            if ret:
                self._video_source.on_captured(np.asarray(frame, dtype=np.uint8))

            if cv2.waitKey(1) == ord("q"):
                break

    def disconnect(self) -> None:
        """
        Sora との接続を切断し、リソースを解放する
        """
        self._connection.disconnect()
        self._video_input_thread.join(10)

        # キャプチャの後始末とウィンドウを全て閉じる
        self._video_capture.release()
        cv2.destroyAllWindows()

    def _on_notify(self, raw_message: str) -> None:
        """
        シグナリング通知のコールバック

        :param raw_message: JSON 形式の生のメッセージ
        """
        message: dict[str, Any] = json.loads(raw_message)
        # event_type が connection.created で、
        # connection_id が自分の connection_id と一致する場合、接続が成功
        if (
            message["type"] == "notify"
            and message["event_type"] == "connection.created"
            and message["connection_id"] == self._connection_id
        ):
            print(f"Sora に接続しました: connection_id={self._connection_id}")
            # 接続が成功したら connected をセット
            self._connected.set()

    def _on_set_offer(self, raw_message: str) -> None:
        """
        シグナリング type: offer のコールバック

        :param raw_message: JSON 形式の生のメッセージ
        """
        message: dict[str, Any] = json.loads(raw_message)
        # "type": offer に自分の connection_id が入ってくるので取得しておく
        if message["type"] == "offer":
            self._connection_id = message["connection_id"]

    def _on_disconnect(self, error_code: SoraSignalingErrorCode, message: str) -> None:
        """
        切断時のコールバック

        :param error_code: 切断の理由を示すエラーコード
        :param message: エラーメッセージ
        """
        print(f"Sora から切断されました: error_code={error_code}, message={message}")
        self._closed = True
        # 切断完了で connected をクリア
        self._connected.clear()

    def run(self) -> None:
        """
        メインループ。接続を維持し、キーボード割り込みを処理する
        """
        try:
            # 接続を維持
            while not self._closed:
                pass
        except KeyboardInterrupt:
            # キーボード割り込みの場合
            pass
        finally:
            # 接続の切断
            if self._connection:
                self._connection.disconnect()


def main() -> None:
    """
    メイン関数。Sendonly インスタンスを作成し、Sora に接続する
    """
    # 環境変数からシグナリング URL とチャネル ID を取得
    signaling_url: str | None = os.getenv("SORA_SIGNALING_URL")
    if not signaling_url:
        raise ValueError("環境変数 SORA_SIGNALING_URL が設定されていません")
    channel_id: str | None = os.getenv("SORA_CHANNEL_ID")
    if not channel_id:
        raise ValueError("環境変数 SORA_CHANNEL_ID が設定されていません")

    # signaling_url はリストである必要があるので、リストに変換
    signaling_urls: list[str] = [signaling_url]

    # device_id は 0 で固定
    sample: Sendonly = Sendonly(signaling_urls, channel_id, 0)

    # Sora へ接続
    with sample.connect():
        time.sleep(5)

    # 接続の維持する場合は sample.disconnect() の代わりに sample.run() を呼ぶ
    # sample.run()


if __name__ == "__main__":
    main()