Sora Python SDK ドキュメント¶

このドキュメントは Sora Python SDK バージョン 2024.3.0 に対応しています。

Sora 自体の製品お問い合わせは sora at shiguredo dot jp までお願い致します。 (このメールアドレスへの特定電子メールの送信を拒否いたします)

問い合わせについて¶

Sora Python SDK の質問などについては Discord の #sora-sdk-faq をご利用ください。ただし、 Sora のライセンス契約の有無に関わらず、応答時間と問題の解決を保証しませんのでご了承ください。

https://discord.gg/shiguredo

Sora Python SDK に対する有償のサポートについては提供しておりません。

リリースノート¶

CHANGE: 後方互換性のない変更
UPDATE: 後方互換性がある変更
ADD: 後方互換性がある追加
FIX: バグ修正

NVIDIA Jetson JetPack SDK での利用は NVIDIA Jetson JetPack SDK をご確認ください。

2024.3.0¶

日時:: 2024-08-05
対応 Sora C++ SDK:: 2024.7.0
対応 Sora:: 2023.2.x / 2024.1.x
対応 Python:: 3.10 / 3.11 / 3.12

[CHANGE] Python 3.8 と 3.9 のサポートを終了しました
- 今後は優先実装でのサポートとなります
[CHANGE] NVIDIA Jetson 向け sora_sdk バイナリを PyPI から削除しました
- support/jetson-jetpack-6 ブランチでメンテナンスを継続します
- 今後は PyPI 経由ではなく、GitHub Releases から whl ファイルをダウンロードして利用してください
[UPDATE] Sora C++ SDK を 2024.7.0 にアップデートしました
[UPDATE] nanobind を 2.0.0 にアップデートしました
[UPDATE] cmake を 3.29.6 にアップデートしました
[UPDATE] libwebrtc のバージョンを m127.6533.1.1 にアップデートしました
[UPDATE] Windows 向けビルドを Windows Server 2022 x86_64 に変更しました
[ADD] macOS 向けビルドに macOS 14 arm64 を追加しました
[ADD] sora_sdk に型を追加しました
[ADD] SoraConnection に get_stats 関数を追加しました
[ADD] Sora C++ SDK と libwebrtc のローカルビルドを利用できるようにしました
[FIX] SoraAudioSink.read が timeout を無視して失敗を返す問題を修正しました
[FIX] MSVC 内部コンパイルエラーにより Windows で nanobind のビルドができない問題を修正しました

2024.2.0¶

日時:: 2024-04-09
対応 Sora C++ SDK:: 2024.6.0
対応 Sora:: 2023.2.x
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11 / 3.12

[CHANGE] Lyra のサポートを廃止し、以下のオプションを削除しました
- audio_codec_lyra_bitrate
- audio_codec_lyra_usedtx
- check_lyra_version
[UPDATE] Sora C++ SDK のバージョンを 2024.6.0 に上げました
[UPDATE] WEBRTC_BUILD_VERSION を m122.6261.1.0 に上げました
[UPDATE] nanobind のバージョンを 1.9.2 に上げて固定したバージョンとしました
[UPDATE] ruff の最小バージョンを 0.3.0 に上げました
[UPDATE] BOOST_VERSION を 1.84.0 に上げました
[UPDATE] Intel VPL を利用した H.265 に対応しました
[ADD] Sora Python SDK Samples を examples に移動しました
[ADD] シグナリング "type": "switched" メッセージの受信を通知する on_switched コールバックを追加しました
[FIX] Ubuntu 20.04 arm64 NVIDIA Jetson 5.1.2 で AV1 が正常に配信されない問題を修正しました

既知の問題¶

Intel VPL を利用したときに H.264 の映像が正常に受信できない
- Intel VPL の問題と Sora C++ SDK の両方に問題が発生しています
- Sora C++ SDK の問題は修正され、動作が改善する予定です
- 完全な修正は Intel VPL の不具合が修正される必要があります

2024.1.0¶

日時:: 2024-02-20
対応 Sora C++ SDK:: 2024.1.0
対応 Sora:: 2023.2.x
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11 / 3.12

[CHANGE] Python フォーマッターを Ruff に変更しました
[CHANGE] SoraAudioSource.on_data の引数名を変更しました
[CHANGE] SoraVideoSource.on_captured の引数名を変更しました
[CHANGE] SoraVAD.analyze の引数名を変更しました
[CHANGE] SoraConnection.on_track の引数を SoraMediaTrack に変更しました
[UPDATE] Python 3.12 に対応しました
[UPDATE] nanobind の最小を 1.8.0 に上げました
[UPDATE] Sora C++ SDK のバージョンを 2024.1.0 に上げました
- WebRTC m116 で cricket::Codec は protected になったので cricket::CreateVideoCodec に修正しました
- WebRTC m118 でパッケージディレクトリが変更されたためそれに追従しました
- WebRTC m120 の webrtc::EncodedImage API の変更に追従しました
- WEBRTC_BUILD_VERSION を m120.6099.1.2 に上げました
- BOOST_VERSION を 1.83.0 に上げました
- CMAKE_VERSION を 3.27.7 に上げました
[UPDATE] ForwardingFilter に version と metadata を追加しました
- WebRTC SFU Sora の 2023.2.x へ追従しました
[UPDATE] NVIDIA JetPack SDK を 5.1.2 に上げました
[ADD] SoraMediaTrack を追加しました
[ADD] 発話区間の検出が可能な SoraVAD を追加しました
[ADD] リアルタイム性を重視した AudioStreamSink を追加しました
[ADD] AudioStreamSink が返す音声フレームとして pickle が可能な AudioFrame を追加しました

既知の問題¶

C++ SDK の問題で NVIDIA JetPack SDK 5.1.2 での AV1 配信が正常に行えません
- 2024.2.0 で問題は解消されています

2023.3.1¶

日時:: 2023-07-13
対応 Sora C++ SDK:: 2023.7.1
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[FIX] C++ SDK のバージョンを 2023.7.2 にアップデートしました
- 特定のタイミングで切断が発生すると Closing 状態で止まってしまう問題を修正しました

2023.3.0¶

日時:: 2023-07-06
対応 Sora C++ SDK:: 2023.7.1
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[CHANGE] signaling_url を signaling_urls へ変更しました
- 引数の型を str から List[str] に変更しました

2023.2.0¶

日時:: 2023-07-03
対応 Sora C++ SDK:: 2023.7.1
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[ADD] Ubuntu 22.04 x86_64 で OpenH264 を利用した H.264 のデコードとエンコードが利用可能になりました
- 今までは HWA が無いと H.264 が利用できませんでしたが、 OpenH264 のバイナリをダウンロードし、設定することで利用可能になりました

2023.1.2¶

日時:: 2023-06-28
対応 Sora C++ SDK:: 2023.7.1
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[FIX] Windows 版のバイナリが正常に動作しない問題を修正しました

2023.1.1¶

日時:: 2023-06-27
対応 Sora C++ SDK:: 2023.7.1
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[FIX] connect 直後に disconnect すると落ちる問題を修正しました

2023.1.0¶

日時:: 2023-06-19
対応 Sora C++ SDK:: 2023.7.0
対応 Sora:: 2023.1.x 以降
対応 Python:: 3.8 / 3.9 / 3.10 / 3.11

[ADD] PyPI に sora_sdk として公開しました
[ADD] macOS 13 arm64 向けの whl パッケージへ対応しました
- Python 3.8 / 3.9 / 3.10 / 3.11 の Python のバージョンに対応しました
[ADD] Windows 11 x86_64 向けの whl パッケージへ対応しました
- Python 3.8 / 3.9 / 3.10 / 3.11 の Python のバージョンに対応しました
[ADD] Ubuntu 22.04 x86_64 向けの whl パッケージへ対応しました
- Python 3.8 / 3.9 / 3.10 / 3.11 の Python のバージョンに対応しました
[ADD] Ubuntu 20.04 arm64 向けの whl パッケージへ対応しました
- NVIDIA Jetson のみに対応しています
- JetPack SDK 5 系のみに対応しています
- Python 3.8 にのみ対応しています
[ADD] HTTP Proxy に対応しました
[ADD] クライアント認証 (mTLS) に対応しました
[ADD] サイマルキャスト機能に対応しました
[ADD] スポットライト機能に対応しました
[ADD] 転送フィルター機能に対応しました
[ADD] 音声コーデック Lyra に対応しました
[ADD] 映像コーデック VP9 と AV1 と H.264 のパラメーター設定に対応しました
[ADD] 音声ストリーミングの言語コードに対応しました
[ADD] データチャネルメッセージング対応しました
- data_channel_signaling を指定できるようにしました
- ignore_disconnect_websocket を指定できるようにしました
[ADD] データチャネルシグナリング対応しました
- on_message コールバック有効化しました
- data_channels を指定できるようにしました
[ADD] 音声と映像の無効化を利用可能にしました
[ADD] 音声と映像の送受信 / 送信 / 受信サンプルを追加しました
- https://github.com/shiguredo/sora-python-sdk-samples
[ADD] メッセージング機能の送受信 / 送信 / 受信サンプルを追加しました
- https://github.com/shiguredo/sora-python-sdk-samples

Sora Python SDK 概要¶

Sora Python SDK は株式会社時雨堂の WebRTC SFU Sora の Python 向けクライアントフレームワークです。

特徴¶

Sora Python SDK は Sora C++ SDK をラップした Python 向けのライブラリです。そのため、 Sora C++ SDK がサポートする機能のほとんどを利用する事ができます。

ハードウェアアクセラレーター¶

Sora C++ SDK が libwebrtc を利用している事もあり、ブラウザで利用する WebRTC と同じ機能が利用できます。また、時雨堂が独自に様々なハードウェアアクセラレーターに対応することで、CPU 負荷を抑えて高画質な映像配信を行う事ができます。

Apple macOS Video Toolbox
- H.264 / H.265
NVIDIA Video Codec SDK (NVENC / NVDEC)
- VP9 / H.264 / H.265
Intel VPL (Intel Media SDK の後継)
- AV1 / H.264 / H.265

ソフトウェアコーデック¶

OpenH264 を利用する事でハードウェアアクセラレーターが利用できない環境でも H.264 を利用する事ができます。

OpenH264 は Ubuntu x86_64 と macOS arm64 環境で利用できます。

注釈

OpenH264 は Baseline Profile のみに対応しています

NVIDIA Jetson JetPack SDK 対応¶

NVIDIA Jetson JetPack SDK で利用できる wheel ファイルを提供しています。

NVIDIA Jetson

Ubuntu 22.04 arm64
- NVIDIA Jetson JetPack 6 に対応
- Python 3.10 のみ対応

詳細は NVIDIA Jetson JetPack SDK をご確認ください。

ソースコード¶

https://github.com/shiguredo/sora-python-sdk

サンプルソースコード¶

https://github.com/shiguredo/sora-python-sdk の examples ディレクトリ

動作環境¶

対応 Python バージョン¶

Python 3.10 以降

対応プラットフォーム¶

Windows 11 x86_64
Windows Server 2022 x86_64
macOS Sonoma 14 arm64
macOS Ventura 13 arm64
Ubuntu 22.04 LTS x86_64
Ubuntu 24.04 LTS x86_64
Ubuntu 22.04 LTS arm64 (NVIDIA Jetson JetPack 6)

対応 Sora¶

リリースノートをご確認ください

対応プラットフォームサポート方針¶

サポート終了後も優先実装にて対応が可能ですので、 Sora サポートまでお問い合わせください。

Windows サポート方針¶

最新の 1 バージョンのみをサポートします。

Windows 11
- Windows 12 リリース後、12 ヶ月以内に通常サポート終了します
Windows Server 2022
- Windows Server 2025 リリース後、12 ヶ月以内に通常サポート終了します

macOS サポート方針¶

最新の 2 バージョンのみをサポートします。

macOS 14
- macOS 16 リリース後、12 ヶ月以内に通常サポート終了します
macOS 13
- macOS 15 リリース後、12 ヶ月以内に通常サポート終了します

Ubuntu LTS サポート方針¶

最新の 2 バージョンのみをサポートします。

Ubuntu 24.04 LTS
- Ubuntu 28.04 リリース後、12 ヶ月以内に通常サポート終了します
Ubuntu 22.04 LTS
- Ubuntu 26.04 リリース後、12 ヶ月以内に通常サポート終了します

NVIDIA Jetson JetPack SDK サポート方針¶

最新の 1 バージョンのみをサポートします。

Jetson JetPack SDK 6
- Jetson JetPack SDK 7 リリース後、12 ヶ月以内に通常サポート終了します

問い合わせについて¶

Sora Python SDK の質問などについては Discord の #sora-sdk-faq チャンネルにお願いします。ただし、 Sora のライセンス契約の有無に関わらず、応答時間と問題の解決を保証しませんのでご了承ください。

https://discord.gg/shiguredo

またビルドやパッケージングに関する質問に対しては、コミュニティ管理者は回答は行いません。

ガイド¶

実装上の注意¶

Sora Python SDK のコールバックメソッドは、Python ランタイムのスレッドではなく、 C++ で実装された処理を実行するために別に立てたスレッドから呼び出されるため、以下の点に注意する必要があります:
- コールバックの中で例外を使う場合には、必ずコールバック内でキャッチして外に漏らしてはいけません（例外が外に漏れると Python プログラムが異常終了します）
- コールバック処理の中にブロックする処理を記述してはいけません（コールバック時呼び出しスレッド上では WebRTC 通信を実現する諸々の処理も走っているので、ブロックするとそれらの実行を阻害してしまう）
一度切断された Sora インスタンスを使い回して、新しい接続を始めることはできません

FAQ¶

Sora Python SDK のライセンスはなんですか？¶

Apache License, Version 2.0 です。

Sora Python SDK のサポートは提供していますか？¶

サポートは提供しておりません。

Sora Python SDK のサンプルは提供していますか？¶

Python Sora SDK サンプル集にて提供しています。

統計情報は取得できますか？¶

統計情報は SoraConnection.get_stats() で取得できます。

コールバックの引数¶

Sora Python SDK のコールバックの引数を間違えると、引数ミスが原因とわからないエラーが出力される場合があります。

そのためコールバックの引数の指定には注意してください。

ハードウェアアクセラレータが無くても H.264 は利用できますか？¶

はい、OpenH264 を利用する事でハードウェアアクセラレータが無くても H.264 を利用する事ができます。

2024 年 8 月時点での最新版の OpenH264 バイナリは以下からダウンロード可能です。

Release Release Version 2.4.1 · cisco/openh264

参考¶

バージョン¶

正式版のバージョンについて¶

時雨堂では YYYY.MAJOR.FIX という形式のフォーマットを採用しています。

YYYY: リリースした年
MAJOR: メジャーバージョンのリリース回数で 1 から始まります
FIX: バグフィックスのリリースの回数で 0 から始まります

開発版のバージョンについて¶

開発版には .devN というプレフィックスが追加されます。

2024.1.0.dev0 といった形式で N は 0 から始まります。

これは Python の PEP 440 フォーマット準拠です。 Python SDK では a / b / rc は利用せず .devN のみを採用しています。

開発バージョンも PyPI にアップロードされ、 pip や rye で --pre を指定することで利用可能です。

既知の問題¶

Sora Python SDK では現在既知の問題はありません。

インストール¶

ここでは pip と Rye の例を記載しています。

PyPI 経由でインストールする¶

pip を利用している場合¶

$ pip install sora_sdk

Rye を利用している場合¶

$ rye add sora_sdk
$ rye sync

Whl パッケージ経由でインストールする¶

警告

非推奨です。PyPI 経由でインストールしてください。

$ rye add sora_sdk --path <パッケージファイル PATH>
$ rye sync

URL 経由でインストールする¶

警告

非推奨です。PyPI 経由でインストールしてください。

$ rye add sora_sdk --url <パッケージファイル URL>
$ rye sync

Ubuntu x86_64 で利用する際にインストールが必要なライブラリ¶

Ubuntu x86_64 で利用する際は、以下のライブラリをインストールする必要があります。

$ sudo apt install libva2 libdrm2

NVIDIA Jetson JetPack SDK¶

NVIDIA Jetson JetPack SDK 向けの対応は PyPI に登録されている sora_sdk では対応していません。

NVIDIA Jetson JetPack SDK をご確認ください。

チュートリアル¶

すぐに利用できる例を用意しています。

shiguredo/sora-python-sdk-examples

$ git clone https://github.com/shiguredo/sora-python-sdk-examples

Python Sora SDK 例のセットアップ¶

Python Sora SDK のサンプルがある examples ディレクトリに移動して、設定ファイルの準備をします。

$ cd sora-python-sdk-examples
$ cp .env.template .env

.env ファイルに設定値を記入します。

SORA_SIGNALING_URLS: Sora のシグナリング URL を指定してください
SORA_CHANNEL_ID: Sora のチャネル ID を指定してください

接続先の用意¶

接続先は時雨堂が開発、販売している WebRTC SFU Sora を利用します。

検証目的であれば Sora Labo を利用することで、 Sora を無料で試すことが可能です。

GitHub アカウントを用意して Sora Labo のドキュメントを読んだ後、 https://sora-labo.shiguredo.jp/ にサインアップしてください。

uv のインストール¶

Sora Python SDK の例では uv を利用しています。

Installation | uv を参考に uv をインストールしてください。

Windows または macOS で Sora Python SDK を使ってみる¶

$ uv sync
$ uv run src/media_recvonly.py

Linux で Sora Python SDK を使ってみる¶

$ sudo apt install libportaudio2
$ uv sync
$ uv run src/media_recvonly.py

サンプル¶

Python バージョン:: 3.10 以降

音声と映像を送受信¶

import json
import os
import threading
import time
from threading import Event
from typing import Any, Dict, List, Optional

import numpy as np
from sora_sdk import (
    Sora,
    SoraAudioSink,
    SoraAudioSource,
    SoraConnection,
    SoraSignalingErrorCode,
    SoraTrackInterface,
    SoraVideoSink,
    SoraVideoSource,
)


class Sendrecv:
    def __init__(self, signaling_urls: List[str], channel_id: str):
        """
        Sendrecv クラスのコンストラクタ

        :param signaling_urls: シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        """
        # シグナリング URL、ロール、チャネル ID を初期化
        self._signaling_urls: List[str] = signaling_urls
        self._channel_id: str = channel_id

        self._role: str = "sendrecv"

        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._audio_sink: Optional[SoraAudioSink] = None
        self._video_sink: Optional[SoraVideoSink] = None

        # Sora インスタンスの生成
        self._sora: Sora = Sora()

        self._audio_source: SoraAudioSource = self._sora.create_audio_source(
            sample_rate=48000, channels=1
        )
        self._video_source: SoraVideoSource = self._sora.create_video_source()

        # Sora への接続設定
        self._connection: SoraConnection = self._sora.create_connection(
            signaling_urls=self._signaling_urls,
            role=self._role,
            channel_id=self._channel_id,
            # create_connection するタイミングで audio_source と video_source を指定する
            audio_source=self._audio_source,
            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
        self._connection.on_track = self._on_track

    def connect(self) -> "Sendrecv":
        """
        Sora へ接続する

        :return: 接続が成功した場合、自身のインスタンス
        :raises AssertionError: 接続に失敗した場合
        """
        # Sora へ接続
        self._connection.connect()

        self._audio_input_thread = threading.Thread(
            target=self._audio_input_loop, daemon=True
        )
        self._audio_input_thread.start()

        self._video_input_thread = threading.Thread(
            target=self._video_input_loop, daemon=True
        )
        self._video_input_thread.start()

        # 接続完了まで待機
        assert self._connected.wait(10), "接続に失敗しました"

        # 統計情報の定期取得用スレッド
        self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)

        return self

    def _audio_input_loop(self) -> None:
        """ダミー音声を生成するループ"""
        # パラメータ
        sample_rate = 16000  # サンプリングレート (Hz)
        freq = 440  # 周波数 (Hz)
        duration = 0.02  # 時間 (秒)
        amplitude = 0.25  # 振幅

        # 時間配列を生成
        t = np.arange(int(sample_rate * duration)) / sample_rate

        while not self._closed:
            # sin 波を生成
            sin_wave = np.sin(2 * np.pi * freq * t)

            # sin 波を 16 ビット整数に変換
            sin_wave_int16 = np.int16(sin_wave * 32767 * amplitude)

            # sin 波を送信
            self._audio_source.on_data(sin_wave_int16.reshape(-1, 1))

            # 次のサイクルのために時間を進める
            t += duration

    def _video_input_loop(self) -> None:
        """ダミー映像を生成するループ"""
        while not self._closed:
            time.sleep(1.0 / 30)
            self._video_source.on_captured(
                np.zeros((self._video_height, self._video_width, 3), dtype=np.uint8)
            )

    def _stats_loop(self) -> None:
        """統計情報を取得するループ"""
        while not self._closed:
            # 取得する統計情報は文字列
            raw_json = self._connection.get_stats()
            # 変換する
            # 統計情報を表示したい場合は stats を表示する
            _stats = json.loads(raw_json)
            # webrtc stats の仕様そのまま
            # https://www.w3.org/TR/webrtc-stats/
            # [{"type": "inbound-rtp", ...}, ...]
            time.sleep(60)

    def disconnect(self) -> None:
        """Sora から切断する"""
        # Sora から切断
        self._connection.disconnect()
        # スレッドの終了を待機
        self._audio_input_thread.join(10)
        self._video_input_thread.join(10)

    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.get("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 _on_track(self, track: SoraTrackInterface) -> None:
        """
        トラック受信時のコールバック

        :param track: 受信したトラック
        """
        if track.kind == "audio":
            self._audio_sink = SoraAudioSink(
                track=track, output_frequency=16000, output_channels=1
            )

        if track.kind == "video":
            self._video_sink = SoraVideoSink(track=track)

    def run(self) -> None:
        """接続を維持し、必要に応じて切断する"""
        try:
            # 接続を維持
            while not self._closed:
                pass
        except KeyboardInterrupt:
            # キーボード割り込みの場合
            pass
        finally:
            # 接続の切断
            if self._connection:
                self._connection.disconnect()


def main() -> None:
    """メイン関数: Sora への接続と切断を行う"""
    # 環境変数からシグナリング URL とチャネル ID を取得
    signaling_url = os.getenv("SORA_SIGNALING_URL")
    if not signaling_url:
        raise ValueError("環境変数 SORA_SIGNALING_URL が設定されていません")
    channel_id = os.getenv("SORA_CHANNEL_ID")
    if not channel_id:
        raise ValueError("環境変数 SORA_CHANNEL_ID が設定されていません")

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

    sample: Sendrecv = Sendrecv(signaling_urls, channel_id)

    # Sora へ接続
    sample.connect()
    # 接続の維持する場合は sample.connect().run() を呼ぶ
    # sample.connect().run()

    time.sleep(3)

    sample.disconnect()


if __name__ == "__main__":
    main()

音声と映像を送信¶

import json
import os
import threading
import time
from threading import Event
from typing import Any, Dict, List, Optional

import numpy as np
from sora_sdk import (
    Sora,
    SoraAudioSource,
    SoraConnection,
    SoraSignalingErrorCode,
    SoraVideoSource,
)


class Sendonly:
    def __init__(self, signaling_urls: List[str], channel_id: str):
        """
        Sendonly クラスのコンストラクタ

        :param signaling_urls: シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        """
        # シグナリング URL とチャネル ID を初期化
        self._signaling_urls: List[str] = signaling_urls
        self._channel_id: str = channel_id

        self._role: str = "sendonly"

        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._audio_source: SoraAudioSource = self._sora.create_audio_source(
            sample_rate=48000, channels=1
        )
        self._video_source: SoraVideoSource = self._sora.create_video_source()

        # Sora への接続設定
        self._connection: SoraConnection = self._sora.create_connection(
            signaling_urls=self._signaling_urls,
            role=self._role,
            channel_id=self._channel_id,
            audio=True,
            audio_source=self._audio_source,
            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 へ接続する

        :return: 接続が成功した場合、自身のインスタンス
        :raises AssertionError: 接続に失敗した場合
        """
        # Sora へ接続
        self._connection.connect()

        self._audio_input_thread = threading.Thread(
            target=self._audio_input_loop, daemon=True
        )
        self._audio_input_thread.start()

        self._video_input_thread = threading.Thread(
            target=self._video_input_loop, daemon=True
        )
        self._video_input_thread.start()

        # 接続が成功するまで待つ
        assert self._connected.wait(10), "接続に失敗しました"

        # 統計情報の定期取得用スレッド
        self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)

        return self

    def _audio_input_loop(self) -> None:
        """ダミー音声を生成するループ"""
        while not self._closed:
            time.sleep(0.02)
            self._audio_source.on_data(np.zeros((320, 1), dtype=np.int16))

    def _video_input_loop(self) -> None:
        """ダミー映像を生成するループ"""
        while not self._closed:
            time.sleep(1.0 / 30)
            self._video_source.on_captured(
                np.zeros((self._video_height, self._video_width, 3), dtype=np.uint8)
            )

    def _stats_loop(self) -> None:
        """統計情報を取得するループ"""
        while not self._closed:
            # 取得する統計情報は文字列
            raw_json = self._connection.get_stats()
            # 変換する
            # 統計情報を表示したい場合は stats を表示する
            _stats = json.loads(raw_json)
            # webrtc stats の仕様そのまま
            # https://www.w3.org/TR/webrtc-stats/
            # [{"type": "inbound-rtp", ...}, ...]
            time.sleep(60)

    def disconnect(self) -> None:
        """Sora から切断する"""
        self._connection.disconnect()
        self._audio_input_thread.join(10)
        self._video_input_thread.join(10)

    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.get("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:
    """メイン関数: Sora への接続と切断を行う"""
    # 環境変数からシグナリング URL とチャネル ID を取得
    signaling_url = os.getenv("SORA_SIGNALING_URL")
    if not signaling_url:
        raise ValueError("環境変数 SORA_SIGNALING_URL が設定されていません")
    channel_id = os.getenv("SORA_CHANNEL_ID")
    if not channel_id:
        raise ValueError("環境変数 SORA_CHANNEL_ID が設定されていません")

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

    sample: Sendonly = Sendonly(signaling_urls, channel_id)

    # Sora へ接続
    sample.connect()
    # 接続の維持する場合は sample.connect().run() を呼ぶ
    # sample.connect().run()

    time.sleep(3)

    sample.disconnect()


if __name__ == "__main__":
    main()

音声と映像を受信¶

import json
import os
import threading
import time
from threading import Event
from typing import Any, Dict, List, Optional

from sora_sdk import (
    Sora,
    SoraAudioSink,
    SoraConnection,
    SoraSignalingErrorCode,
    SoraTrackInterface,
    SoraVideoSink,
)


class Recvonly:
    def __init__(self, signaling_urls: List[str], channel_id: str):
        """
        Recvonly クラスのコンストラクタ

        :param signaling_urls: シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        """
        # シグナリング URL とチャネル ID を初期化
        self._signaling_urls: List[str] = signaling_urls
        self._channel_id: str = channel_id

        self._role: str = "recvonly"

        self._connection_id: Optional[str] = None

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

        self._audio_sink: Optional[SoraAudioSink] = None
        self._video_sink: Optional[SoraVideoSink] = None

        self._sora: Sora = Sora()
        # Sora への接続設定
        self._connection: SoraConnection = self._sora.create_connection(
            signaling_urls=self._signaling_urls,
            role=self._role,
            channel_id=self._channel_id,
        )

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

    def connect(self) -> "Recvonly":
        """
        Sora へ接続する

        :return: 接続が成功した場合、自身のインスタンス
        :raises AssertionError: 接続に失敗した場合
        """
        # Sora へ接続
        self._connection.connect()

        # 接続が成功するまで待つ
        assert self._connected.wait(10), "接続に失敗しました"

        # 統計情報の定期取得用スレッド
        self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)

        return self

    def _stats_loop(self) -> None:
        """統計情報を取得するループ"""
        while not self._closed:
            # 取得する統計情報は文字列
            raw_json = self._connection.get_stats()
            # 変換する
            # 統計情報を表示したい場合は stats を表示する
            _stats = json.loads(raw_json)
            # webrtc stats の仕様そのまま
            # https://www.w3.org/TR/webrtc-stats/
            # [{"type": "inbound-rtp", ...}, ...]
            time.sleep(60)

    def disconnect(self) -> None:
        """
        Sora から切断する
        """
        self._connection.disconnect()

    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.get("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 _on_track(self, track: SoraTrackInterface) -> None:
        """
        トラック受信時のコールバック

        :param track: 受信したトラック
        """
        if track.kind == "audio":
            self._audio_sink = SoraAudioSink(
                track=track, output_frequency=16000, output_channels=1
            )

        if track.kind == "video":
            self._video_sink = SoraVideoSink(track=track)

    def run(self) -> None:
        """
        接続を維持し、必要に応じて切断する
        """
        try:
            # 接続を維持
            while not self._closed:
                pass
        except KeyboardInterrupt:
            # キーボード割り込みの場合
            pass
        finally:
            # 接続の切断
            if self._connection:
                self._connection.disconnect()


def main() -> None:
    """
    メイン関数: Sora への接続と切断を行う
    """
    # 環境変数からシグナリング URL とチャネル ID を取得
    signaling_url: Optional[str] = os.getenv("SORA_SIGNALING_URL")
    if not signaling_url:
        raise ValueError("環境変数 SORA_SIGNALING_URL が設定されていません")
    channel_id: Optional[str] = os.getenv("SORA_CHANNEL_ID")
    if not channel_id:
        raise ValueError("環境変数 SORA_CHANNEL_ID が設定されていません")

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

    sample: Recvonly = Recvonly(signaling_urls, channel_id)

    # Sora へ接続
    sample.connect()
    # 接続の維持する場合は sample.connect().run() を呼ぶ
    # sample.connect().run()

    time.sleep(3)

    sample.disconnect()


if __name__ == "__main__":
    main()

受信した音声を VAD を利用して判定¶

import json
import os
import time
from threading import Event
from typing import Any, Dict, List, Optional

from sora_sdk import (
    Sora,
    SoraAudioFrame,
    SoraAudioStreamSink,
    SoraConnection,
    SoraMediaTrack,
    SoraSignalingErrorCode,
    SoraVAD,
)


class RecvonlyVAD:
    def __init__(self, signaling_urls: List[str], channel_id: str):
        """
        RecvonlyVAD クラスの初期化

        :param signaling_urls: Sora シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        """
        self._vad: SoraVAD = SoraVAD()
        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._audio_stream_sink: SoraAudioStreamSink
        self._audio_output_frequency: int = 24000
        self._audio_output_channels: int = 1

        self._sora: Sora = Sora()

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

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

    def connect(self) -> "RecvonlyVAD":
        """
        Sora に接続する
        """
        self._connection.connect()

        # 接続が成功するまで待つ
        assert self._connected.wait(10), "接続に失敗しました"

        return self

    def disconnect(self) -> None:
        """
        Sora との接続を切断する
        """
        self._connection.disconnect()

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

        :param raw_message: JSON 形式の生のメッセージ
        """
        message: Dict[str, Any] = json.loads(raw_message)
        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}")
            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)
        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
        self._connected.clear()

    def _on_frame(self, frame: SoraAudioFrame) -> None:
        """
        音声フレーム受信時のコールバック

        :param frame: 受信した音声フレーム
        """
        voice_probability: float = self._vad.analyze(frame)
        if voice_probability > 0.95:  # 0.95 は libwebrtc の判定値
            print(f"Voice! voice_probability={voice_probability}")

    def _on_track(self, track: SoraMediaTrack) -> None:
        """
        トラック受信時のコールバック

        :param track: 受信したメディアトラック
        """
        if track.kind == "audio":
            self._audio_stream_sink = SoraAudioStreamSink(
                track, self._audio_output_frequency, self._audio_output_channels
            )
            self._audio_stream_sink.on_frame = self._on_frame

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


def main() -> None:
    """
    メイン関数。RecvonlyVAD インスタンスを作成し、Sora に接続する
    """
    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_urls: List[str] = [signaling_url]

    sample: RecvonlyVAD = RecvonlyVAD(signaling_urls, channel_id)

    sample.connect()
    # 接続を維持する場合は sample.connect().run() を呼ぶ
    # sample.connect().run()

    time.sleep(3)

    sample.disconnect()


if __name__ == "__main__":
    main()

NVIDIA Jetson JetPack SDK¶

PyPI 経由でインストールする Sora Python SDK は NVIDIA Jetson には対応していません。

NVIDIA Jetson 向けの Sora Python SDK は GitHub Releases から whl ファイルをインストールしてください。

https://github.com/shiguredo/sora-python-sdk/releases

ブランチやタグも通常とは異なります。

リリースノート¶

2024.3.0-jetson-jetpack-6.0.0.0¶

日時:: 2024-08-20
対応 Sora C++ SDK:: 2024.7.0
対応 Sora:: 2023.2.x / 2024.1.x
対応 Python:: 3.10

[ADD] NVIDIA Jetson JetPack SDK 6.0.0 に対応する

メンテナンス¶

NVIDIA Jetson JetPack 向けの Sora Python SDK は標準メンテナンス対象外です。メンテナンスは有償で提供しておりますので、ご希望される方は Sora のサポートまでご連絡ください。

GitHub ブランチ¶

support/jetson-jetpack-{platform-major-version}

GitHub タグ¶

{sora-python-sdk-version}-jetson-jetpack-{platform-version}.{release}

wheel ファイル¶

Jetson JetPack 向けに wheel ファイルを作成しています。

sora_sdk_jetson_jetpack_{platform-version}.{release}-{sora-python-sdk-version}-cp{python-version}-cp{python-version}-manylinux_2_17_aarch64.manylinux2014_aarch64.whl

NVIDIA Jetson JetPack 6 系¶

ブランチ:: support/jetson-jetpack-6
タグ例:: 2024.3.0-jetson-jetpack-6.0.0.0
wheel ファイル名例:: sora_sdk_jetson_jetpack_6.0.0.0-2024.3.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl

rye を利用する場合¶

ファイル名は適宜変更してください。

$ rye pin 3.10
$ rye add sora-sdk --path sora_sdk_jetson_jetpack_6.0.0.0-2024.3.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
$ rye sync

NVIDIA Jetson JetPack 5 系¶

優先実装にて検討可能です。 Sora サポートまでご相談ください。

NVIDIA Jetson JetPack 4 系¶

優先実装にて検討可能です。 Sora サポートまでご相談ください。

音声デバイス¶

Sora Python SDK では音声デバイスを扱う機能を提供していません。

代わりに、 PortAudio の Python バインディングである sounddevice を使用することで、音声デバイスのキャプチャや再生などを行うことができます。

sounddevice のインストール¶

pip¶

$ pip install sounddevice

rye¶

$ rye add sounddevice
$ rye sync

macOS¶

$ brew install portaudio

Ubuntu¶

$ sudo apt install libportaudio2

音声デバイス一覧を取得する¶

sounddevice.query_devices を利用する事で、音声デバイス一覧を取得することができます

from typing import Any, Dict, List

import sounddevice as sd


def get_available_audio_devices() -> None:
    """
    利用可能なオーディオデバイスの情報をシンプルなkey-value形式で表示する関数

    >>> get_available_audio_devices()
    利用可能なオーディオデバイス:
    1:
        name: DELL U4021QW
        host: Core Audio
        max_input_channels: 0
        max_output_channels: 2
        default_samplerate: 48000.0
        default_low_input_latency: 0.010000
        default_high_input_latency: 0.100000
    2:
        name: Logitech StreamCam
        host: Core Audio
        max_input_channels: 2
        max_output_channels: 0
        default_samplerate: 48000.0
        default_low_input_latency: 0.004583
        default_high_input_latency: 0.013917
    ...
    """
    devices: List[Dict[str, Any]] = sd.query_devices()

    print("利用可能なオーディオデバイス:")
    for i, device in enumerate(devices, 1):
        print(f"{i}:")
        print(f"    name: {device['name']}")

        host_api: Dict[str, Any] = sd.query_hostapis(device["hostapi"])
        print(f"    host: {host_api['name']}")

        print(f"    max_input_channels: {device['max_input_channels']}")
        print(f"    max_output_channels: {device['max_output_channels']}")
        print(f"    default_samplerate: {device['default_samplerate']}")
        print(
            f"    default_low_input_latency: {device['default_low_input_latency']:.6f}"
        )
        print(
            f"    default_high_input_latency: {device['default_high_input_latency']:.6f}"
        )
        print()  # デバイス間に空行を挿入


def main():
    get_available_audio_devices()


if __name__ == "__main__":
    main()

音声デバイスから音声キャプチャして再生する¶

sounddevice.play を利用する事で、音声デバイスから音声キャプチャして音声を表示することができます。

import time
from typing import Dict, List

import numpy as np
import sounddevice as sd


def list_audio_devices() -> List[Dict]:
    """
    利用可能なすべてのオーディオデバイスをリストアップし、表示します。

    Returns:
        List[Dict]: 利用可能なオーディオデバイスのリスト
    """
    devices: List[Dict] = sd.query_devices()
    print("利用可能なオーディオデバイス:")
    for i, device in enumerate(devices):
        print(
            f"{i}: {device['name']} (入力チャンネル: {device['max_input_channels']}, 出力チャンネル: {device['max_output_channels']})"
        )
    return devices


def get_valid_device_index(devices: List[Dict], is_input: bool = True) -> int:
    """
    ユーザーに有効なデバイスインデックスの入力を求めます。

    Args:
        devices (List[Dict]): 利用可能なオーディオデバイスのリスト
        is_input (bool): 入力デバイスを選択する場合は True 、出力デバイスの場合は False

    Returns:
        int: 選択されたデバイスのインデックス
    """
    while True:
        try:
            device_index: int = int(
                input(
                    f"{'入力' if is_input else '出力'}デバイスのインデックスを入力してください: "
                )
            )
            if 0 <= device_index < len(devices):
                if (is_input and devices[device_index]["max_input_channels"] > 0) or (
                    not is_input and devices[device_index]["max_output_channels"] > 0
                ):
                    return device_index
            print(
                f"無効なインデックスです。正しい{'入力' if is_input else '出力'}デバイスのインデックスを入力してください。"
            )
        except ValueError:
            print("数値を入力してください。")


def display_volume(volume: float) -> str:
    """
    音量レベルを視覚的に表示するための文字列を生成します。

    Args:
        volume (float): 0.0から1.0の範囲の音量レベル

    Returns:
        str: 音量レベルを表す文字列（バーグラフ付き）
    """
    amplification: int = 5  # 音量の増幅係数（表示を強調するため）
    max_bar_length: int = 50  # バーの最大長さ
    bar_length: int = int(min(volume * amplification * max_bar_length, max_bar_length))
    return f"Volume: [{'|' * bar_length}{' ' * (max_bar_length - bar_length)}] {volume:.4f}"


def capture_audio(input_device_index: int, duration: int = 5) -> np.ndarray:
    """
    指定されたデバイスから音声をキャプチャし、リアルタイムで音量を表示します。

    Args:
        input_device_index (int): 入力デバイスのインデックス
        duration (int): 録音時間（秒）

    Returns:
        np.ndarray: キャプチャされた音声データ
    """
    sample_rate: int = 48000  # サンプリングレート（Hz）
    channels: int = 1  # モノラル録音
    block_size: int = 1024  # 一度に処理するサンプル数

    print(
        f"デバイス '{sd.query_devices(device=input_device_index)['name']}' からの音声をキャプチャしています..."
    )
    print(f"サンプリングレート: {sample_rate} Hz")

    audio_data: List[np.ndarray] = []
    start_time: float = time.time()

    def audio_callback(
        indata: np.ndarray, frames: int, time_info: Dict, status: sd.CallbackFlags
    ) -> None:
        """
        音声データが利用可能になるたびに呼び出されるコールバック関数。

        Args:
            indata (np.ndarray): 入力音声データ
            frames (int): フレーム数
            time_info (Dict): タイムスタンプ情報
            status (sd.CallbackFlags): ステータスフラグ
        """
        if status:
            print(status)  # エラーがあれば表示
        audio_data.append(indata.copy())  # 音声データをリストに追加
        volume: float = np.sqrt(np.mean(indata**2))  # RMS音量を計算
        elapsed_time: int = int(time.time() - start_time)
        print(
            f"\r{display_volume(volume)} 録音中: {elapsed_time}/{duration} 秒", end=""
        )

    # InputStream を使用して音声をキャプチャ
    with sd.InputStream(
        device=input_device_index,
        channels=channels,
        samplerate=sample_rate,
        callback=audio_callback,
        blocksize=block_size,
    ):
        sd.sleep(duration * 1000)  # ミリ秒単位で待機

    print("\n録音完了")
    return np.concatenate(audio_data)  # 全ての音声データを1つの配列に結合


def play_audio(output_device_index: int, audio_data: np.ndarray) -> None:
    """
    キャプチャした音声データを再生します。

    Args:
        output_device_index (int): 出力デバイスのインデックス
        audio_data (np.ndarray): 再生する音声データ
    """
    sample_rate: int = 48000  # キャプチャ時と同じサンプリングレート
    print(
        f"デバイス '{sd.query_devices(device=output_device_index)['name']}' で音声を再生しています..."
    )
    print(f"サンプリングレート: {sample_rate} Hz")
    sd.play(audio_data, samplerate=sample_rate, device=output_device_index)
    sd.wait()  # 再生が完了するまで待機
    print("再生完了")


def main():
    # メインの実行フロー
    devices: List[Dict] = list_audio_devices()
    input_device_index: int = get_valid_device_index(devices, is_input=True)
    output_device_index: int = get_valid_device_index(devices, is_input=False)

    audio_data: np.ndarray = capture_audio(input_device_index)
    play_audio(output_device_index, audio_data)


if __name__ == "__main__":
    main()

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

Sora.create_audio_source() を利用する事で、音声デバイスからキャプチャした音声を Sora に送信することができます。

import json
import os
import time
from threading import Event
from typing import Any, Dict, List, Optional

import numpy as np
import sounddevice as sd
from sora_sdk import Sora, SoraAudioSource, SoraConnection, SoraSignalingErrorCode


class SendonlyAudio:
    def __init__(self, signaling_urls: List[str], channel_id: str, device_id: int):
        """
        SendonlyAudio クラスのコンストラクタ

        :param signaling_urls: Sora シグナリングサーバーの URL リスト
        :param channel_id: 接続するチャンネルの ID
        :param device_id: 使用するオーディオデバイスの ID（デフォルト: None）
        """
        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._sample_rate: int = 48000
        self._channels: int = 1
        self._device_id: int = device_id

        # Sora SDK の初期化
        self._sora: Sora = Sora()
        self._audio_source: SoraAudioSource = self._sora.create_audio_source(
            channels=self._channels, sample_rate=self._sample_rate
        )

        # Sora への接続設定
        self._connection: SoraConnection = self._sora.create_connection(
            signaling_urls=signaling_urls,
            role="sendonly",
            channel_id=channel_id,
            audio=True,
            video=False,
            audio_source=self._audio_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) -> "SendonlyAudio":
        """
        Sora サーバーに接続し、オーディオ入力ストリームを開始する

        :return: self (メソッドチェーン用)
        """
        # Sora へ接続
        self._connection.connect()

        # オーディオ入力ストリームを開始
        self._audio_stream = sd.InputStream(
            samplerate=self._sample_rate,
            channels=self._channels,
            dtype="int16",
            device=self._device_id,
            callback=self._audio_callback,
        )
        self._audio_stream.start()

        # 接続が成功するまで待つ
        assert self._connected.wait(10), "接続に失敗しました"
        return self

    def _audio_callback(
        self, indata: np.ndarray, frames: int, time: Any, status: Any
    ) -> None:
        """
        オーディオ入力コールバック関数

        :param indata: 入力オーディオデータ
        :param frames: フレーム数
        :param time: タイムスタンプ
        :param status: ストリームのステータス
        """
        self._audio_source.on_data(indata)

    def disconnect(self) -> None:
        """
        Sora サーバーから切断し、リソースを解放する
        """
        self._connection.disconnect()
        if hasattr(self, "_audio_stream"):
            self._audio_stream.stop()
            self._audio_stream.close()

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

        :param raw_message: サーバーからの生の通知メッセージ
        """
        message: Dict[str, Any] = json.loads(raw_message)
        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}")
            self._connected.set()

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

        :param raw_message: サーバーからのオファーメッセージ
        """
        message: Dict[str, Any] = json.loads(raw_message)
        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
        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:
    """
    メイン関数: SendonlyAudio インスタンスを作成し、実行する
    """
    signaling_url = os.getenv("SORA_SIGNALING_URL")
    if not signaling_url:
        raise ValueError("環境変数 SORA_SIGNALING_URL が設定されていません")
    channel_id = os.getenv("SORA_CHANNEL_ID")
    if not channel_id:
        raise ValueError("環境変数 SORA_CHANNEL_ID が設定されていません")

    signaling_urls: List[str] = [signaling_url]

    sample: SendonlyAudio = SendonlyAudio(signaling_urls, channel_id, 1)

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

    time.sleep(3)

    sample.disconnect()


if __name__ == "__main__":
    main()

映像デバイス¶

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 の記事が参考になりますので、ご参考ください。

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 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()

`sora_sdk` モジュール¶

概要¶

sora_sdk ライブラリは Sora C++ SDK をラップしたライブラリです。 Sora のシグナリングなどを意識する必要はありません。

音声と映像の扱い¶

Sora Python SDK で音声や映像をデバイスから取得したり、画面に描画する機能を提供していません。その部分は OpenCV や sounddevice などのライブラリを別途利用する必要があります。

libwebrtc ログ指定¶

libwebrtc のログ出力を指定できます。

sora_sdk.enable_libwebrtc_log¶

Type:: SoraLoggingSeverity.VERBOSE | SoraLoggingSeverity.INFO | SoraLoggingSeverity.WARNING | SoraLoggingSeverity.ERROR | SoraLoggingSeverity.NONE

SoraLoggingSeverity のログレベルを指定します。デフォルトは未指定です。

以下のように Sora クライアントを生成するまえに libwebrtc のログレベルを指定してください。

import sora_sdk

if __name__ == '__main__':
   sora_sdk.enable_libwebrtc_log(sora_sdk.SoraLoggingSeverity.VERBOSE)

`Sora` オブジェクト¶

class sora_sdk.Sora(use_hardware_encoder=True, openh264=None)¶

param Optional[bool] use_hardware_encoder:

ハードウェアアクセラレーターを使用するかどうか (デフォルトは True)

param Optional[str] openh264:

OpenH264 ライブラリのパスを指定する (デフォルトは None)

rtype:

None

Sora SDK の主要なクラスです。このクラスを使用して Sora への接続や、音声とビデオのソースを作成します。

create_audio_source(channels, sample_rate)¶

パラメータ:

channels (int) -- チャンネル

sample_rate (int) -- サンプルレート

戻り値の型:

sora_sdk.SoraAudioSource

Sora に音声を送るための SoraAudioSource を作成します。

SoraAudioSource は送信する際に、ここで指定したチャネル数とサンプリングレートに入力されたデータを変換した上で送信します。

create_video_source()¶

戻り値の型:

sora_sdk.SoraVideoSource

Sora に映像を送るための SoraVideoSource を作成します。

create_connection(signaling_urls, role, channel_id, client_id=None, bundle_id=None, metadata=None, signaling_notify_metadata=None, audio_source=None, video_source=None audio=True, video=True, audio_codec_type=None, video_codec_type=None, audio_bit_rate=None, video_bit_rate=None, video_vp9_params=None, video_av1_params=None, video_h264_params=None, simulcast=None, spotlight=None, spotlight_number=None, simulcast_rid=None, spotlight_focus_rid=None, spotlight_unfocus_rid=None, forwarding_filter=None, data_channels=None, data_channel_signaling=None, ignore_disconnect_websocket=None, data_channel_signaling_timeout=None, disconnect_wait_timeout=None, websocket_close_timeout=None, websocket_connection_timeout=None, audio_streaming_language_code=None, insecure=None, client_cert=None, client_key=None, proxy_url=None, proxy_username=None, proxy_password=None, proxy_agent=None)¶

パラメータ:

signaling_urls (List[str]) -- シグナリング URLのリスト
role (str) -- ロール
channel_id (str) -- チャネル ID
client_id (Optional[str]) -- クライアント ID
bundle_id (Optional[str]) -- バンドル ID
metadata (Optional[object]) -- 認証メタデータ
signaling_notify_metadata (Optional[object]) -- シグナリング通知メタデータ
audio_source (Optional[sora_sdk.SoraTrackInterface]) -- 音声ソース
video_source (Optional[sora_sdk.SoraTrackInterface]) -- 映像ソース
audio (Optional[bool]) -- 音声
video (Optional[bool]) -- 映像
audio_codec_type (Optional[str]) -- 音声コーデックタイプ
video_codec_type (Optional[str]) -- 映像コーデックタイプ
audio_bit_rate (Optional[int]) -- 音声ビットレート
video_bit_rate (Optional[int]) -- 映像ビットレート
video_vp9_params (Optional[object]) -- ビデオの VP9 設定指定
video_av1_params (Optional[object]) -- ビデオの AV1 設定指定
video_h264_params (Optional[object]) -- ビデオの H.264 設定指定
simulcast (Optional[bool]) -- サイマルキャスト
spotlight (Optional[bool]) -- スポットライト
spotlight_number (Optional[int]) -- スポットライトナンバー
simulcast_rid (Optional[str]) -- サイマルキャスト RID
spotlight_focus_rid (Optional[str]) -- スポットライトフォーカス RID
spotlight_unfocus_rid (Optional[str]) -- スポットライトアンフォーカス RID
forwarding_filter (Optional[object]) -- 転送フィルター
data_channels (Optional[object]) -- データチャネル
data_channel_signaling (Optional[bool]) -- データチャネルシグナリング
ignore_disconnect_websocket (Optional[bool]) -- ウェブソケット切断無視
data_channel_signaling_timeout (Optional[int]) -- データチャネルシグナリングタイムアウト
disconnect_wait_timeout (Optional[int]) -- 切断待ちタイムアウト
websocket_close_timeout (Optional[int]) -- ウェブソケットクローズタイムアウト
websocket_connection_timeout (Optional[int]) -- ウェブソケット接続タイムアウト
audio_streaming_language_code (Optional[str]) -- 音声ストリーミング言語コード
insecure (Optional[bool]) -- インセキュア
client_cert (Optional[str]) -- クライアント証明書
client_key (Optional[str]) -- クライアントシークレットキー
proxy_url (Optional[str]) -- Proxy URL
proxy_username (Optional[str]) -- Proxy ユーザ名
proxy_password (Optional[str]) -- Proxy パスワード
proxy_agent (Optional[str]) -- Proxy エージェント

戻り値の型:

sora_sdk.SoraConnection

Sora と接続するための SoraConnection を作成します。

接続して切断する最小コード¶

ただ繫いで、切断するだけのコードです。

import time

from sora_sdk import Sora

if __name__ == "__main__":
    # Sora インスタンスを生成
    sora = Sora()
    # Sora に接続する
    conn = sora.create_connection(
        signaling_urls=["wss://example.com/signaling"],
        role="sendonly",
        channel_id="sora",
    )
    conn.connect()

    # 接続するまで待つ
    time.sleep(3)

    # Sora から切断する
    conn.disconnect()

metadata にアクセストークンを指定する最小コード¶

Sora Cloud や Sora Labo を利用する際に metadata にアクセストークンを指定してください。

import time

from sora_sdk import Sora

if __name__ == "__main__":
    # Sora インスタンスを生成
    sora = Sora()
    # Sora に接続する
    conn = sora.create_connection(
        signaling_urls=["wss://example.com/signaling"],
        role="sendonly",
        channel_id="sora",
        # dict をそのまま指定してください
        metadata={"access_token": "access_token"},
    )
    conn.connect()

    # 接続するまで待つ
    time.sleep(3)

    # Sora から切断する
    conn.disconnect()

`SoraConnection` オブジェクト¶

class sora_sdk.SoraConnection¶

Sora との接続を制御する機能を提供します。

Sora.create_connection() から生成します。

connect()¶

戻り値:: None

Sora に接続します。

disconnect()¶

戻り値:: None

Sora から切断します。

get_stats()¶

戻り値:: W3C 準拠の WebRTC 統計情報。
戻り値の型:: str

クライアントの WebRTC 統計情報を取得します。形式は Identifiers for WebRTC's Statistics API に準拠しています。

import json
import time

from sora_sdk import Sora

if __name__ == "__main__":
    sora = Sora()
    conn = sora.create_connection(
        signaling_urls=["wss://sora.example.com/signaling"],
        role="sendrecv",
        channel_id="sora",
    )

    conn.connect()
    # 接続確立まで少し待つ
    time.sleep(5)

    # 取得する統計情報は文字列
    raw_stats = conn.get_stats()
    # 文字列から JSON オブジェクトにする
    stats = json.loads(raw_stats)
    print(stats)

on_disconnect(error_code, message)¶

Type:: Callable[[sora_sdk.SoraSignalingErrorCode, str], None]

コネクション切断時のコールバックです。 SoraSignalingErrorCode で表されるエラーコードとメッセージを引数に取ります。

from sora_sdk import Sora, SoraSignalingErrorCode


def on_disconnect(error_code: SoraSignalingErrorCode, message: str):
    print(f"error_code: {error_code}, message: {message}")


if __name__ == "__main__":
    sora = Sora()
    conn = sora.create_connection(
        signaling_urls=["wss://sora.example.com/signaling"],
        role="sendrecv",
        channel_id="sora",
    )
    # 切断コールバックを登録する
    conn.on_disconnect = on_disconnect

    conn.connect()

send_data_channel(label, data)¶

パラメータ:

label (str) -- データチャンネルのラベル。
data (bytes) -- 送信するデータ。

戻り値:

成功した場合は True 、それ以外の場合は False 。

戻り値の型:

bool

Sora にデータチャンネルを介してデータを送信します。

on_data_channel(label)¶

Type:: Callable[[label: str], None]

データチャネルが追加された際に呼び出されるコールバックです。追加されたデータチャネルのラベルを引数に取ります。

on_message(label, data)¶

Type:: Callable[[label: str, data: bytes], None]

データチャネルでメッセージ通知を受信した際に呼び出されるコールバックです。受信したデータチャネルのラベルとデータを引数に取ります。

from sora_sdk import Sora, SoraConnection


# データチャネルが利用可能になったら呼ばれる
def on_data_channel(label: str):
    print(f"label: {label}")


# データチャネルのメッセージを受信すると呼ばれる
def on_message(label: str, data: bytes):
    print(f"label: {label}, data: {data.decode()}")


if __name__ == "__main__":
    sora: Sora = Sora()
    conn: SoraConnection = sora.create_connection(
        signaling_urls=["wss://sora.example.com/signaling"],
        role="sendrecv",
        channel_id="sora",
        data_channel_signaling=True,
        data_channels=[{"label": "#spam", "direction": "recvonly"}],
    )
    # データチャネルコールバックを登録する
    conn.on_data_channel = on_data_channel

    # データチャネルメッセージコールバックを登録する
    conn.on_message = on_message

    conn.connect()

on_notify(message)¶

Type:: Callable[[str], None]

シグナリング通知を受信した際に呼び出されるコールバックです。受信したメッセージを引数に取ります。

import json
from typing import Any, Dict

from sora_sdk import Sora


def on_notify(raw_message: str):
    # JSON デコード
    message: Dict[str, Any] = json.loads(raw_message)
    if message["type"] == "connection.created":
        print(f"connection.created: {message}")


if __name__ == "__main__":
    sora = Sora()
    conn = sora.create_connection(
        signaling_urls=["wss://sora.example.com/signaling"],
        role="sendrecv",
        channel_id="sora",
    )
    # シグナリング通知コールバックを登録する
    conn.on_notify = on_notify

    conn.connect()

on_push(message)¶

Type:: Callable[[message: str], None]

シグナリングプッシュを受信した際に呼び出されるコールバックです。受信したメッセージを引数に取ります。

on_set_offer¶

Type:: Callable[[message: str], None]

Sora から受け取った Offer を設定した際に呼び出されるコールバックです。受信した Offer を引数に取ります。

on_track(track)¶

Type:: Callable[[track: sora_sdk.SoraMediaTrack], None]

Track が追加された際に呼び出されるコールバックです。追加された Track の SoraMediaTrack を引数に取ります。

コールバックで受け取った後に SoraTrackInterface.kind から video か audio かを判別し、適切な Sink に渡すことで受信した映像、音声データを受け取ることができます。

on_switched(message)¶

Type:: Callable[[message: str], None]

シグナリングが DataChannel に切り替わったタイミングで呼ばれるコールバックです。受信したメッセージを引数に取ります。

`SoraAudioSource` オブジェクト¶

class sora_sdk.SoraAudioSource¶

Sora に Python で生成した音声データを送る機能を提供します。このクラスは SoraTrackInterface を継承しています。

Sora.create_audio_source() から生成します。

on_data(ndarray)¶

パラメータ:: ndarray (ndarray[dtype=int16, shape=(*, *), order='C', device='cpu']) -- チャンネルごとのサンプル数 x チャンネル数になっている音声データ。
戻り値の型:: None

Sora に送る音声データを渡します。

タイムスタンプが引数にないオーバーロードです。タイムスタンプは先に受け取ったデータと連続になっていると想定してサンプル数から自動生成します。

on_data(ndarray, timestamp)¶

パラメータ:

ndarray (ndarray[dtype=int16, shape=(*, *), order='C', device='cpu']) -- チャンネルごとのサンプル数 x チャンネル数になっている音声データ。
timestamp (float) -- time.time() で取得できるエポック秒で表されるフレームのタイムスタンプ。

戻り値の型:

None

危険

このメソッドは非推奨です。

Sora に送る音声データを渡します。

on_data(data, samples_per_channel)¶

パラメータ:

data (int) -- 16bit PCM データのポインタ。
samples_per_channel (int) -- 送信するチャネル毎のサンプル数。

戻り値の型:

None

Sora に送る音声データを渡します。

on_data(data, samples_per_channel, timestamp)¶

パラメータ:

data (int) -- 16bit PCM データのポインタ。
samples_per_channel (int) -- 送信するチャンネル毎のサンプル数
timestamp (float) -- タイムスタンプ

戻り値の型:

None

危険

このメソッドは非推奨です。

Sora に送る音声データを渡します。

`SoraAudioSink` オブジェクト¶

class sora_sdk.SoraAudioSink(track, output_frequency=-1, output_channels=0)¶

パラメータ:

track (sora_sdk.SoraTrackInterface) -- 音声データを受け取る SoraTrackInterface オブジェクト。
output_frequency (int) -- 出力サンプリングレート。指定されたサンプリングレートに出力する音声データをリサンプリングします。 -1 の場合はリサンプリングせず受信したまま出力します。デフォルトは -1。
output_channels (int) -- 出力チャンネル数。指定されたチャネル数に出力する音声データをミキシングします。 0 の場合はミキシングせず受信したまま出力します。デフォルトは 0。

戻り値の型:

None

AudioTrack から音声データを取り出す機能を提供します。

受け取った音声はコンストラクタで設定したサンプリングレートとチャネル数に変換し、内部のバッファに溜め込まれるため、任意のタイミングで音声を取り出すことができます。

read(frames=0, timeout=1)¶

パラメータ:

frames (int) -- 受け取るチャンネルごとのサンプル数。0 を指定した場合には、受信済みのすべてのサンプルを返します。デフォルトは 0。
timeout (float) -- 溜まっているサンプル数が frames で指定した数を満たさない場合の待ち時間。秒単位で指定します。デフォルトは 1。

戻り値:

タプルでインデックス 0 には bool で成否を、成功した場合のみインデックス 1 に ndarray[dtype=int16, shape=(*, *)] でチャンネルごとのサンプル数 x チャンネル数になっている音声データを返します。

戻り値の型:

tuple

受信済みのデータをバッファから読み出します。

on_data¶

Type:: Callable[[numpy.ndarray[dtype=int16, shape=(*, *)]], None]

音声データを AudioTrack から受け取った際に呼び出されるコールバックです。音声データが格納された ndarray でチャンネルごとのサンプル数 x チャンネル数になっている音声データを引数にとります。

このコールバック関数内では重い処理は行わないでください。

このコールバックは廃止予定です。

on_format(sample_rate, number_of_channels)¶

Type:: Callable[[int, int], None]

AudioTrack から受け取った音声データのサンプリングレートもしくはチャネル数が変化した場合に呼び出されるコールバックです。新しいサンプリングレートとチャネル数を引数にとります。

このコールバック関数内では重い処理は行わないでください。

このコールバックは廃止予定です。

from sora_sdk import SoraAudioSink

audio_sink = None


def on_track(track):
    # 音声トラックの場合
    if track.kind == "audio":
        # 音声トラックのため込み場所を指定
        _audio_sink = SoraAudioSink(
            track=track,
            # 音声出力デバイスが利用可能な値を指定してください
            output_frequency=16000,
            # 音声出力デバイスが利用可能な値を指定してください
            output_channels=1,
        )

音声を取り出す場合は SoraAudioSink.read() を利用してください。

# frames はバッファサイズです
success, data = audio_sink.read(frames)
if success:
else:

`SoraVideoSource` オブジェクト¶

class sora_sdk.SoraVideoSource¶

Sora に Python で生成した映像のフレームデータを送る機能を提供します。このクラスは SoraTrackInterface を継承しています。

Sora.create_video_source() から生成します。

on_captured(ndarray)¶

パラメータ:: ndarray (ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']) -- H x W x BGR になっているフレームデータ。
戻り値の型:: None

この関数が呼び出された時点のタイムスタンプでフレームを送信します。

映像になるように一定のタイミングで呼び出さない場合、受信側でコマ送りになります。

on_captured(ndarray, timestamp)¶

パラメータ:

ndarray (ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']) -- H x W x BGR になっているフレームデータ。
timestamp (float) -- time.time() で取得できるエポック秒で表されるフレームのタイムスタンプ。

戻り値の型:

None

危険

このメソッドは非推奨です。

timestamp 引数で渡されたタイムスタンプでフレームを送信します。

フレームのタイムスタンプを指定できるようにするため用意したオーバーロードです。

timestamp が映像になるように一定の時間差がない場合、受信側で正しく表示されない場合があります。

表示側で音声データの timestamp と同期を取るため遅延が発生する場合があります。

on_captured(ndarray, timestamp_us)¶

パラメータ:

ndarray (ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']) -- H x W x BGR になっているフレームデータ。
timestamp_us (int) -- マイクロ秒単位の整数で表されるフレームのタイムスタンプ。

戻り値の型:

None

危険

このメソッドは非推奨です。

timestamp_us 引数で渡されたマイクロ秒精度の整数で表されるタイムスタンプでフレームを送信します。

libWebRTC のタイムスタンプはマイクロ秒精度のため用意したオーバーロードです。

timestamp が映像になるように一定の時間差がない場合、受信側で正しく表示されない場合があります。

表示側で音声データの timestamp と同期を取るため遅延が発生する場合があります。

`SoraTrackInterface` オブジェクト¶

class sora_sdk.SoraTrackInterface¶

AudioTrack もしくは VideoTrack を表すために使用されます。

SoraConnection.on_track コールバックで引数として渡されます。

set_enabled(enable)¶

パラメータ:: enable (bool) -- 有効または無効を表すブール値。
戻り値の型:: bool

トラックの有効/無効を設定します。

enabled¶

Type:: bool
Readonly:: True

トラックが有効かどうかを示すブール値。

id¶

Type:: str
Readonly:: True

トラックの ID を示す文字列。

kind¶

Type:: str
Readonly:: True

トラックの種類を示す文字列。 video もしくは audio が入っています。 AudioTrack か VideoTrack かを判別することができます。

state¶

Type:: sora_sdk.SoraTrackState
Readonly:: True

トラックの状態。

`SoraAudioFrame` オブジェクト¶

class sora_sdk.SoraAudioFrame¶

SoraAudioStreamSink から受け取った音声データを扱うために提供されます。

SoraAudioStreamSink.on_frame コールバックで引数として渡されます。

libwebrtc 内の音声処理単位である 10ms の音声データが格納されています。 picklable です。

data()¶

戻り値:: numpy.ndarray でサンプル数 x チャンネル数になっている音声データ。
戻り値の型:: numpy.ndarray[dtype=int16, shape=(*, *)]

SoraAudioFrame 内のフレームデータへの numpy.ndarray での参照を返します。

samples_per_channel¶

Type:: int

チャネルあたりのサンプル数。

num_channels¶

Type:: int

チャンネル数 (モノラルの場合は 1、ステレオの場合は 2)。

sample_rate_hz¶

Type:: int

サンプリングレート。

absolute_capture_timestamp_ms¶

Type:: int

キャプチャした際のタイムスタンプ。ない場合は None。

`SoraAudioStreamSink` オブジェクト¶

class sora_sdk.SoraAudioStreamSink(track, output_frequency=-1, output_channels=0)¶

パラメータ:

track (sora_sdk.SoraTrackInterface) -- 音声データを受け取る SoraTrackInterface オブジェクト。
output_frequency (int) -- 出力サンプリングレート。指定されたサンプリングレートに出力する音声データをリサンプリングします。 -1 の場合はリサンプリングせず受信したまま出力します。デフォルトは -1。
output_channels (int) -- 出力チャンネル数。指定されたチャネル数に出力する音声データをミキシングします。 0 の場合はミキシングせず受信したまま出力します。デフォルトは 0。

戻り値の型:

None

AudioTrack の音声データを受け取って逐次 Python に渡す on_frame コールバックを提供します。

出力には on_frame にコールバック関数を指定する必要があります。

on_frame¶

Type:: Callable[[track: sora_sdk.SoraAudioFrame], None]

音声データを AudioTrack から受け取った際に呼び出されるコールバック。音声データが格納された SoraAudioFrame オブジェクトを引数に取ります。

from sora_sdk import SoraAudioStreamSink

audio_stream_sink = None


def on_frame(frame):
    pass


def on_track(track):
    # 映像トラックの場合
    if track.kind == "audio":
        # トラックの音声データの受け口として SoraAudioStreamSink を生成
        audio_stream_sink = SoraAudioStreamSink(
            track=track,
            # 24kHz にリサンプリングするように指定
            output_frequency=24000,
            # モノラルにミキシングするように指定
            output_channels=1,
        )
        # 音声データが来た際のコールバックを指定
        audio_stream_sink.on_frame = on_frame

`SoraMediaTrack` オブジェクト¶

SoraConnection.on_track コールバックで引数として渡されます。

class sora_sdk.SoraMediaTrack¶

Sora から受け取った AudioTrack もしくは VideoTrack を扱うために提供されます。このクラスは SoraTrackInterface を継承しています。

stream_id¶

Type:: str

この Track の Stream ID を返します。

`SoraVAD` オブジェクト¶

class sora_sdk.SoraVAD¶

SoraAudioFrame の音声データが音声である率を返す Voice Activity Detection (VAD) を提供します。

analyze(frame)¶

パラメータ:: frame (sora_sdk.SoraAudioFrame) -- 音声である確率を求める SoraAudioFrame オブジェクト。
戻り値:: 0 - 1 で表される音声である確率。
戻り値の型:: int

SoraAudioFrame 内の音声データが音声である確率を返します。

frame 内の音声データが 24 kHz 以外の時 24kHz にリサンプリングを 2 チャンネル以上の時 1 チャンネルにミキシングをそれぞれ変換した上で VAD を行います。 VAD しか使わない場合は SoraAudioStreamSink で事前に変換しておくことをお勧めします。

SoraAudioStreamSink と組み合わせて以下のように使います。

from sora_sdk import SoraAudioStreamSink, SoraVAD

audio_stream_sink = None
vad = SoraVAD()


def on_frame(frame):
    # frame が音声である確率を求める
    voice_probability = vad.analyze(frame)
    if voice_probability > 0.95:  # 0.95 は libwebrtc の判定値
        print(f"Voice! voice_probability={voice_probability}")
    else:
        print("Not a voice!")


def on_track(track):
    if track.kind == "audio":
        audio_stream_sink = SoraAudioStreamSink(
            track=track,
            # VAD はサンプリングレートが 24 kHz 以外の時 24kHz にリサンプリングするので、 24kHz にしておく。
            output_frequency=24000,
            # VAD はチャネル数が 2 以上の時 1 チャンネルにミキシングするので、 1 チャンネルにしておく。
            output_channels=1,
        )
        audio_stream_sink.on_frame = on_frame

`SoraVideoFrame` オブジェクト¶

class sora_sdk.SoraVideoFrame¶

SoraVideoSink から受け取ったフレームデータを扱うために提供されます。

SoraVideoSink.on_frame コールバックで引数として渡されます。

data()¶

戻り値:: H x W x BGR になっているフレームデータ。
戻り値の型:: numpy.ndarray[dtype=uint8, shape=(*, *, 3)]

SoraVideoFrame 内のフレームデータへの numpy.ndarray での参照を返します。

OpenCV の cv2.imshow にそのまま渡すと表示されるように成型されています。

`SoraVideoSink` オブジェクト¶

class sora_sdk.SoraVideoSink(track)¶

パラメータ:: track -- フレームを受け取る SoraTrackInterface オブジェクト。
戻り値の型:: None

VideoTrack からフレームデータを受け取って、逐次 Python に渡す on_frame コールバックを提供します。

出力には on_frame にコールバック関数を指定する必要があります。

on_frame¶

Type:: Callable[[track: sora_sdk.SoraVideoFrame], None]

フレームデータを VideoTrack から受け取った際に呼び出されるコールバックです。フレームデータが格納された SoraVideoFrame オブジェクトを引数に取ります。

このコールバック関数内では重い処理は行わないでください。 サンプルを参考に queue を利用するなどの対応を推奨します。

この関数はメインスレッドから呼び出されないため、関数内で OpenCV の cv2.imshow を実行しても macOS の場合は表示されません。

from sora_sdk import SoraVideoSink

video_sink = None


def on_frame(frame):
    pass


def on_track(track):
    # 映像トラックの場合
    if track.kind == "video":
        # 映像トラックのため込み場所を指定
        video_sink = SoraVideoSink(track=track)
        # 映像フレームコールバックを指定
        video_sink.on_frame = on_frame

OpenCV¶

Sora Python SDK サンプルでは OpenCV を利用して映像を出力しています。

`SoraSignalingErrorCode` オブジェクト¶

class sora_sdk.SoraSignalingErrorCode¶

SoraConnection.on_disconnect のシグナリングエラーコードを提供します。

CLOSE_FAILED¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

接続のクローズに失敗したことを示すエラーコードです。

CLOSE_SUCCEEDED¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

接続のクローズが成功したことを示すエラーコードです。

ICE_FAILED¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

ICE の処理に失敗したことを示すエラーコードです。

INTERNAL_ERROR¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

内部エラーが発生したことを示すエラーコードです。

INVALID_PARAMETER¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

無効なパラメータが指定されたことを示すエラーコードです。

PEER_CONNECTION_STATE_FAILED¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

ピア接続の状態が異常なことを示すエラーコードです。

WEBSOCKET_HANDSHAKE_FAILED¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

WebSocket のハンドシェイクに失敗したことを示すエラーコードです。

WEBSOCKET_ONCLOSE¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

WebSocket の接続が閉じられたことを示すエラーコードです。

WEBSOCKET_ONERROR¶

Type:: sora_sdk.SoraSignalingErrorCode
Readonly:: True

WebSocket でエラーが発生したことを示すエラーコードです。

`SoraLoggingSeverity` オブジェクト¶

class sora_sdk.SoraLoggingSeverity¶

enable_libwebrtc_log でログレベルを指定を提供します。

VERBOSE¶

Type:: sora_sdk.SoraLoggingSeverity
Readonly:: True

最も詳細なログレベルを示す定数です。

INFO¶

Type:: sora_sdk.SoraLoggingSeverity
Readonly:: True

情報レベルのログを示す定数です。

WARNING¶

Type:: sora_sdk.SoraLoggingSeverity
Readonly:: True

警告レベルのログを示す定数です。

ERROR¶

Type:: sora_sdk.SoraLoggingSeverity
Readonly:: True

エラーレベルのログを示す定数です。

NONE¶

Type:: sora_sdk.SoraLoggingSeverity
Readonly:: True

ログ出力を無効化するための定数です。

パッケージング¶

重要

自前でのパッケージングは推奨していません。提供されているパッケージを利用してください。

パッケージは dist 以下に whl 拡張子で生成されます
Ubuntu 22.04 arm64 (NVIDIA Jetson JetPack 6) 以外はクロスコンパイルに対応していません

Windows 11 x86_64¶

$ rye sync
$ rye run python run.py windows_x86_64
$ rye run python -m build

macOS 14 arm64¶

$ rye sync
$ rye run python run.py macos_arm64
$ rye run python -m build

Ubuntu 24.04 x86_64¶

$ sudo apt install libva-dev libva-drm2 pkg-config
$ rye sync
$ rye run python run.py ubuntu-22.04_x86_64
$ rye run python -m build

Ubuntu 22.04 arm64 (NVIDIA Jetson JetPack 6)¶

Ubuntu 22.04 x86_64 でのビルド&パッケージング例です

$ sudo apt install libva-dev libva-drm2  pkg-config multistrap
$ sudo sed -e 's/Apt::Get::AllowUnauthenticated=true/Apt::Get::AllowUnauthenticated=true";\n$config_str .= " -o Acquire::AllowInsecureRepositories=true/' -i /usr/sbin/multistrap
$ rye sync
$ SORA_SDK_TARGET=ubuntu-22.04_armv8_jetson_jetpack_6 rye run python run.py
$ ./package.ubuntu-22.04_armv8_jetson_jetpack_6.sh

Sora Python SDK の開発¶

自前でビルドした sora-python-sdk でサンプルを動かす¶

rye の workspace 機能を利用することで、自前でビルドした sora-python-sdk を利用する事ができます。

pyproject.toml に以下のように workspace を設定します。

[tool.rye.workspace]
packages = ["examples"]

$ rye sync
$ rye run python run.py macos_arm64
$ cp examples/.env.template examples/.env
# .env を修正して Sora の接続情報などを設定する
$ rye run media_sendonly

rye run media_sendonly または rye run python examples/src/media/sendonly.py で実行することができるようになります。

自前でビルドした sora-python-sdk を rye add で追加してサンプルを動かす¶

$ rye sync
$ rye run python run.py macos_arm64
$ cd examples
$ rye add sora_sdk --path ..
$ rye sync
$ cp .env.template .env
# .env を修正して Sora の接続情報などを設定する
$ rye run python src/media/recvonly.py

自前でビルドしたパッケージを rye add で追加してサンプルを動かす¶

$ rye sync
$ rye run python run.py macos_arm64
$ rye run python -m build
$ cd examples
$ rye add sora_sdk --path ../dist/sora_sdk-{sdk-version}-{python-version}-{os}_{arch}.whl
$ rye sync
$ cp .env.template .env
# .env を修正して Sora の接続情報などを設定する
$ rye run python src/media/recvonly.py