映像デバイス¶
Sora Python SDK では映像デバイスを扱う機能を提供していません。
代わりに、OpenCV の Python バインディングを使用することで、 映像デバイスのキャプチャや表示などを行うことができます。
OpenCV のインストール¶
pip¶
$ pip install opencv-python
rye¶
$ rye add opencv-python
$ rye 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 の記事が参考になりますので、ご参考ください。
映像デバイスをキャプチャして表示する¶
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 typing import Any, Dict, List, Optional
import cv2
from sora_sdk import Sora, SoraConnection, SoraSignalingErrorCode, SoraVideoSource
class Sendonly:
def __init__(self, signaling_urls: List[str], channel_id: str, device_id: int):
"""
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: Optional[str] = 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 connect(self) -> "Sendonly":
"""
Sora に接続し、ビデオ入力ループを開始する
"""
# Sora へ接続
self._connection.connect()
self._video_input_thread = threading.Thread(
target=self._video_input_loop, daemon=True
)
self._video_input_thread.start()
# 接続が成功するまで待つ
assert self._connected.wait(10), "接続に失敗しました"
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(frame)
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.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 へ接続
sample.connect()
# 接続の維持する場合は sample.disconnect() の代わりに sample.run() を呼ぶ
# sample.run()
time.sleep(3)
sample.disconnect()
if __name__ == "__main__":
main()