Sora Python SDK ドキュメント¶
このドキュメントは Sora Python SDK バージョン 2025.1.0 に対応しています。
Sora 自体の製品お問い合わせは sora at shiguredo dot jp までお願い致します。 (このメールアドレスへの特定電子メールの送信を拒否いたします)
問い合わせについて¶
Sora Python SDK の質問などについては Discord の #sora-sdk-faq をご利用ください。
ただし、 Sora のライセンス契約の有無に関わらず、応答時間と問題の解決を保証しませんのでご了承ください。
Sora Python SDK に対する有償のサポートについては提供しておりません。
リリースノート¶
- CHANGE
後方互換性のない変更
- UPDATE
後方互換性がある変更
- ADD
後方互換性がある追加
- FIX
バグ修正
NVIDIA Jetson JetPack SDK での利用は NVIDIA Jetson JetPack SDK をご確認ください。
2025.1.0¶
- 日付:
2025-03-19
- 対応 Sora C++ SDK:
2025.2.0
- 対応 Sora:
2024.1.x / 2024.2.x
- 対応 Python:
3.11 / 3.12 / 3.13
[CHANGE] macOS Sonoma 13 のサポートを終了しました
今後は優先実装でのサポートとなります
[CHANGE] Python 3.10 のサポートを終了しました
今後は優先実装でのサポートとなります
SPEC 0 — Minimum Supported Dependencies を参考に直近 3 バージョンのサポートに変更しました
[CHANGE]
Sora()からuse_hardware_encoderを廃止しました今後は
Sora.video_codec_preferenceを利用してください
[CHANGE] mTLS で利用する
client_certとclient_keyをファイルパスではなくファイルの中身を指定するように変更しましたstr型からbytes型に変更しましたopen("cert-key.pem", “rb”).read() などで読み込んだバイナリを渡すようにしてください
[CHANGE] シグナリング接続時の
"type": "connect"時にmultistream項目を送らないように変更しました今回の変更により Sora 2022.1.0 以前には接続できなくなります
[UPDATE] Sora C++ SDK を
2025.2.0にアップデートしましたWEBRTC_BUILD_VERSION を
m132.6834.5.8にアップデートしましたBOOST_VERSION を
1.87.0にアップデートしましたCMAKE_VERSION を
3.31.6にアップデートしましたOPENH264_VERSION を
2.6.0にアップデートしました
[UPDATE] nanobind を
2.5.0にアップデートしました[ADD] マルチ転送フィルター用の
forwarding_filtersを追加しました将来的に
forwarding_filterは廃止する予定ですので、forwarding_filtersを利用してください
[ADD] Ubuntu 24.04 arm64v8 に対応しました
Python 3.12 でのみ対応しています
[ADD] Ubuntu 24.04 arm64 上で arm64v8 向けのビルドを行えるようにしました
[ADD] Windows x86_64で OpenH264 を利用した H.264 のデコードとエンコードが利用可能になりました
[ADD] AMD AMF を利用したハードウェアアクセラレーター機能に対応しました
[ADD]
Sora.video_codec_preferenceを追加しましたコーデックのエンコード/デコードを細かく指定できるようになりました
詳細は
Sora.video_codec_preferenceをご確認ください
[ADD] エンコード時の劣化優先順位を指定する
degradation_preferenceを追加しましたhttps://w3c.github.io/mst-content-hint/#degradation-preference-when-encoding
Sora.create_connection()の引数にdegradation_preferenceを追加しましたEnum 型
SoraDegradationPreferenceを追加しましたMAINTAIN_RESOLUTIONは解像度を優先MAINTAIN_FRAMERATEはフレームレートを優先BALANCEDはバランスを優先DISABLEDは無効
[ADD]
Sora.create_connection()の引数にaudio_opus_paramsを追加しましたOpus のパラメーターを指定できるようになりました
[ADD] WebRTC Encoded Transform に対応しました
SoraTransformableAudioFrameとSoraTransformableVideoFrameを追加しましたSoraAudioFrameTransformerとSoraVideoFrameTransformerを追加しましたcreate_connection()の引数にaudio_frame_transformerとvideo_frame_transformerを追加しましたSoraMediaTrackにset_frame_transformer()を追加しました詳細は WebRTC Encoded Transform をご確認ください
[ADD] Python 3.13 に対応しました
[ADD] CA 証明書を指定できる
ca_certを追加しましたopen("ca.pem", “rb”).read() などで読み込んだバイナリを渡すようにしてください
[ADD] 受信したシグナリングメッセージを取得できる
on_signaling_messageコールバックを追加しましたconnect/redirect/offer/answer/re-offer/re-answer/disconnectメッセージが取得できますswitchedメッセージはSoraConnection.on_switchedを利用してくださいpingとpongメッセージは取得できません詳細は
SoraConnection.on_signaling_messageをご確認ください
[ADD] シグナリングの WebSocket 終了時のコードと理由が取得できる
on_ws_closeコールバックを追加しましたSDK 側で WebSocket が切断された際には
codeは1000、reasonは"SELF-CLOSED"が返ります詳細は
SoraConnection.on_ws_closeをご確認ください
[ADD] 転送フィルターを複数指定できるマルチ転送フィルターに対応しました
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] メッセージング機能の
headerに対応しました[ADD] WebRTC Encoded Transform に対応しました
SoraTransformableAudioFrameとSoraTransformableVideoFrameを追加しましたSoraAudioFrameTransformerとSoraVideoFrameTransformerを追加しましたcreate_connection()の引数にaudio_frame_transformerとvideo_frame_transformerを追加しましたSoraMediaTrackにset_frame_transformer()を追加しました詳細は WebRTC Encoded Transform をご確認ください
[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_bitrateaudio_codec_lyra_usedtxcheck_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] 音声と映像の送受信 / 送信 / 受信サンプルを追加しました
[ADD] メッセージング機能の 送受信 / 送信 / 受信サンプルを追加しました
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 負荷を抑えて高画質な映像配信を行う事ができます。
-
H.264 / H.265
VP9 / AV1 のデコードには対応していません
-
VP8 / VP9 / H.264 / H.265
VP8 / VP9 はデコードのみの対応です
Intel VPL (Intel Media SDK の後継)
AV1 / H.264 / H.265
VP9 エンコードには対応していません
-
VP8 / VP9 /AV1 / H.264 / H.265
AV1 のデコードは Ubuntu x86_64 のみ対応です
VP8 / VP9 はデコードのみの対応です
ソフトウェアコーデック¶
libwebrtc に含まれている VP8 / VP9 / AV1 に対応しています。
また、 OpenH264 を利用する事でハードウェアアクセラレーターが利用できない環境でも H.264 を利用する事ができます。
OpenH264 は Ubuntu x86_64 と macOS arm64 環境で利用できます。
注釈
OpenH264 は Baseline Profile のみに対応しています
NVIDIA Jetson JetPack SDK 対応¶
NVIDIA Jetson JetPack SDK で利用できる wheel ファイルを提供しています。
Ubuntu 22.04 arm64
NVIDIA Jetson JetPack 6 に対応
Python 3.10 のみ対応
詳細は NVIDIA Jetson JetPack SDK をご確認ください。
ソースコード¶
サンプルソースコード¶
https://github.com/shiguredo/sora-python-sdk の examples ディレクトリ
動作環境¶
対応 Python バージョン¶
Python 3.13
Python 3.12
Python 3.11
対応プラットフォーム¶
Ubuntu 24.04 LTS x86_64
Ubuntu 24.04 LTS arm64
Ubuntu 22.04 LTS x86_64
Ubuntu 22.04 LTS arm64 (NVIDIA Jetson JetPack 6) - PyPI 経由でのインストールには対応していません
macOS Sequoia 15 arm64
macOS Ventura 14 arm64
Windows 11 x86_64
Windows Server 2022 x86_64
対応 Sora¶
リリースノート をご確認ください
対応 OpenH264¶
OpenH264 のバージョンは 2.6.0 をサポートします。
対応 Python サポート方針¶
リリースのタイミングで、直近の 3 バージョンをサポートします。
これは Scientific Python の SPEC 0 を参考にしてます。
例えば 2024 年 12 月にリリースした場合は、2024 年 10 月に Python 3.13 がリリースされているため、 Python 3.11, 3.12, 3.13 をサポートします。
古い Python バージョンのサポートについて¶
サポート終了後も優先実装にて対応が可能ですので、 Sora サポートまでお問い合わせください。
対応プラットフォームサポート方針¶
Windows サポート方針¶
最新の 1 バージョンのみをサポートします。
Windows 11
Windows 12 リリース後、12 ヶ月以内に通常サポート終了します
Windows Server 2022
Windows Server 2025 リリース後、12 ヶ月以内に通常サポート終了します
macOS サポート方針¶
最新の 2 バージョンのみをサポートします。
macOS 15
macOS 17 リリース後、12 ヶ月以内に通常サポート終了します
macOS 14
macOS 16 リリース後、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 ヶ月以内に通常サポート終了します
古い OS バージョンのサポートについて¶
サポート終了後も優先実装にて対応が可能ですので、 Sora サポートまでお問い合わせください。
問い合わせについて¶
Sora Python SDK の質問などについては Discord の #sora-sdk-faq チャンネルにお願いします。
ただし、 Sora のライセンス契約の有無に関わらず、応答時間と問題の解決を保証しませんのでご了承ください。
またビルドやパッケージングに関する質問に対しては、コミュニティ管理者は回答は行いません。
ガイド¶
実装上の注意¶
必ず connect を呼び出したら disconnect を呼び出す¶
Sora Python SDK は C++ で実装されています。そのため、必ず connect を呼び出したら disconnect を呼び出さないとセグメンテーション違反が発生する場合があります。
できるだけ with を使うようにしてください。 with を利用する際は __enter__ で connect を呼び出したけど、接続が上手くいかない場合に、例外を発生させる時は必ず try/except を使ってキャッチして、 disconnect を呼び出してください。
その他¶
Sora Python SDK のコールバックメソッドは、Python ランタイムのスレッドではなく、 C++ で実装された処理を実行するために別に立てたスレッドから呼び出されるため、以下の点に注意する必要があります:
コールバックの中で例外を使う場合には、必ずコールバック内でキャッチして外に漏らしてはいけません （例外が外に漏れると Python プログラムが異常終了します）
コールバック処理の中にブロックする処理を記述してはいけません （コールバック時呼び出しスレッド上では WebRTC 通信を実現する諸々の処理も走っているので、ブロックするとそれらの実行を阻害してしまう）
一度切断された Sora インスタンスを使い回して、新しい接続を始めることはできません
FAQ¶
Sora Python SDK のライセンスはなんですか？¶
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 や uv で --pre を指定することで利用可能です。
既知の問題¶
Sora Python SDK では現在既知の問題はありません。
インストール¶
ここでは pip と Rye と uv の例を記載しています。
PyPI 経由でインストールする¶
pip を利用している場合¶
$ pip install sora_sdk
uv を利用している場合¶
$ uv add sora_sdk
$ uv sync
Whl パッケージ経由でインストールする¶
警告
非推奨です。PyPI 経由でインストールしてください。
$ uv add sora_sdk --path <パッケージファイル PATH>
$ uv sync
URL 経由でインストールする¶
警告
非推奨です。PyPI 経由でインストールしてください。
$ uv add sora_sdk --url <パッケージファイル URL>
$ uv 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 をご確認ください。
チュートリアル¶
このチュートリアルでは Python Sora SDK を利用して映像を配信する方法を説明します。
接続先の用意¶
接続先は時雨堂が開発、販売している WebRTC SFU Sora を利用します。
検証目的であれば Sora Labo を利用することで、 Sora を無料で試すことが可能です。
GitHub アカウントを用意して Sora Labo のドキュメント を読んだ後、 https://sora-labo.shiguredo.jp/ にサインアップしてください。
必要なのは以下の 3 つです。
シグナリング URL
チャネル ID
Sora Labo や Sora Cloud の場合はアクセストークン
uv のインストール¶
Sora Python SDK の例では uv を利用しています。
Installation | uv を参考に uv をインストールしてください。
プロジェクトの作成¶
uv init でプロジェクトを作成します。
$ uv init tutorial
$ cd tutorial
Python Sora SDK セットアップ¶
uv add して sora-sdk をインストールします。
$ uv add sora-sdk
また、映像を配信するためにはカメラデバイスを利用する必要があります。 ここでは OpenCV を利用します。
$ uv add opencv-python opencv-python-headless
main.py を作成¶
デバイスキャプチャをして、 映像を配信する main.py を作成します。今回は音声は送りません。
signaling_url や channel_id や access_token は適切なものを指定してください。
import platform
import cv2 # type: ignore
from sora_sdk import (
Sora,
)
def get_video_capture(fps=30) -> cv2.VideoCapture:
"""
ここは Python SDK 関係なくカメラデバイスをキャプチャする部分です。
"""
video_capture = None
# ここからカメラの設定
if platform.system() == "Windows":
# CAP_DSHOW を設定しないと、カメラの起動が遅くなる
video_capture = cv2.VideoCapture(0, cv2.CAP_DSHOW)
else:
video_capture = cv2.VideoCapture(0)
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
video_capture.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG"))
video_capture.set(cv2.CAP_PROP_FPS, fps)
# Ubuntu → FOURCC を設定すると FPS が初期化される
# Windows → FPS を設定すると FOURCC が初期化される
# ので、両方に対応するため２回設定する
fourcc = cv2.VideoWriter_fourcc(*"MJPG")
target_fourcc = video_capture.get(cv2.CAP_PROP_FOURCC)
if fourcc != target_fourcc:
video_capture.set(cv2.CAP_PROP_FOURCC, fourcc)
if fps != int(video_capture.get(cv2.CAP_PROP_FPS)):
video_capture.set(cv2.CAP_PROP_FPS, fps)
return video_capture
def main() -> None:
signaling_url = "ws://192.0.2.1:5000/signaling"
channel_id = "sora"
access_token = "access_token"
sora: Sora = Sora()
video_source = sora.create_video_source()
connection = sora.create_connection(
signaling_urls=[
signaling_url,
],
role="sendonly",
channel_id=channel_id,
metadata={"access_token": access_token},
video_source=video_source,
)
# カメラデバイス確保
video_capture = get_video_capture()
# 接続
connection.connect()
try:
while True:
# カメラデバイスからの映像を取得
success, frame = video_capture.read()
if not success:
continue
# カメラデバイスからの映像を Python SDK に渡す
video_source.on_captured(frame)
except KeyboardInterrupt:
pass
finally:
# 切断
connection.disconnect()
# カメラデバイスを解放
video_capture.release()
if __name__ == "__main__":
main()
hell.py 起動¶
$ uv run main.py
止めるときは Ctrl+C です。
main.py からの映像を確認¶
Sora-DevTools を利用して Python SDK から送られてきている映像を確認してみてください。
サンプル¶
- Python バージョン:
3.13 以降
https://github.com/shiguredo/sora-python-sdk-examples にもサンプルを公開しています。
音声と映像を送受信¶
import json
import os
import threading
import time
from threading import Event
from typing import Any, 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 __enter__(self) -> "Sendrecv":
return self.connect()
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.disconnect()
def connect(self) -> "Sendrecv":
"""
Sora へ接続する
:return: 接続が成功した場合、自身のインスタンス
:raises AssertionError: 接続に失敗した場合
"""
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()
try:
# Sora へ接続
self._connection.connect()
# 接続完了まで待機
assert self._connected.wait(10), "接続に失敗しました"
# 統計情報の定期取得用スレッド
self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
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 へ接続
with sample.connect():
time.sleep(5)
# 接続の維持する場合は sample.connect().run() を呼ぶ
# sample.connect().run()
if __name__ == "__main__":
main()
音声と映像を送信¶
import json
import os
import threading
import time
from threading import Event
from typing import Any, 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 __enter__(self) -> "Sendonly":
return self.connect()
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.disconnect()
def connect(self) -> "Sendonly":
"""
Sora へ接続する
:return: 接続が成功した場合、自身のインスタンス
:raises AssertionError: 接続に失敗した場合
"""
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()
try:
# Sora へ接続
self._connection.connect()
# 接続が成功するまで待つ
assert self._connected.wait(10), "接続に失敗しました"
# 統計情報の定期取得用スレッド
self._stats_thread = threading.Thread(target=self._stats_loop, daemon=True)
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
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 へ接続
with sample.connect():
time.sleep(5)
# 接続の維持する場合は sample.connect().run() を呼ぶ
# sample.connect().run()
if __name__ == "__main__":
main()
音声と映像を受信¶
import json
import os
import threading
import time
from threading import Event
from typing import Any, 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, 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 __enter__(self) -> "RecvonlyVAD":
return self.connect()
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.disconnect()
def connect(self) -> "RecvonlyVAD":
"""
Sora に接続する
"""
try:
self._connection.connect()
# 接続が成功するまで待つ
assert self._connected.wait(10), "接続に失敗しました"
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
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()
映像コーデックプリファレンス¶
概要¶
Sora Python SDK では映像コーデックのプリファレンス指定することができます。
Apple Video Toolbox¶
Apple Video Toolbox は macOS で利用できるハードウェアアクセラレーターです
macOS でのみ利用できます。対応コーデックは H.264 / H.265 です
decoderやencoderにINTERNALを指定することで Apple Video Toolbox を利用します
Intel Video Processing Library (VPL)¶
Intel VPL は Intel 組込 GPU や Arc で利用できるハードウェアアクセラレーターです
Linux と Windows で利用できます。対応コーデックは VP9 / AV1 / H.264 / H.265 です
VP9 はデコーダーのみ利用できます
decoderやencoderにINTEL_VPLを指定することで Intel VPL を利用します
AMD Advanced Media Framework (AMF) SDK¶
AMD AMF は AMD 組込 GPU や AMD のビデオカードで利用できるハードウェアアクセラレーターです
Linux と Windows で利用できます。対応コーデックは VP9 / AV1 / H.264 / H.265 です
AV1 デコーダーは Windows x86_64 でのみ利用できます
decoderやencoderにAMD_AMFを指定することで AMD AMF を利用します
NVIDIA Video Codec SDK¶
NVIDIA Video Codec SDK は NVIDIA のビデオカードで利用できるハードウェアアクセラレーターです
Linux と Windows で利用できます。対応コーデックは VP8 / VP9 / AV1 / H.264 / H.265 です
VP8 と VP9 はデコーダーのみ利用できます
decoderやencoderにNVIDIA_VIDEO_CODEC_SDKを指定することで NVIDIA Video Codec SDK を利用します
Cisco OpenH264¶
Cisco OpenH264 はオープンソースのソフトウェア H.264 エンコーダー/デコーダーです
対応コーデックは H.264 です
Soraのopenh264に OpenH264 のパスを指定することで Cisco OpenH264 が利用できるようになりますdecoderやencoderにCISCO_OPENH264を指定することで Cisco OpenH264 を利用します
利用できるコーデック一覧を取得する¶
Sora.get_video_codec_capability を呼び出すことで利用できるコーデック一覧を取得できます。
ただし、取得した内容をそのまま Sora の video_codec_preference に指定することはできません。
VP9 のエンコーダーが正常に動作しない¶
Ubuntu x86_64 と Windows x86_64 では VP9 のエンコーダーが正常に動作しない問題がわかっています
Ubuntu で Intel VPL と NVIDIA Video Codec SDK の両方が利用できる場合¶
Ubuntu 環境で Intel VPL と NVIDIA Video Codec SDK の両方が利用できる場合、 AV1 や H.264 にどちらの機能を利用するかを指定する必要があります。
ここでは Intel VPL を優先し、 Intel VPL が利用できない場合は NVIDIA Video Codec SDK を利用する例を示します。
import sys
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
get_video_codec_capability,
)
if sys.platform != "linux":
raise RuntimeError("This script must be run in a Linux environment")
capability = get_video_codec_capability()
intel_vpl_available = None
nvidia_video_codec_sdk_available = None
# 利用できるエンジンを確認する
for engine in capability.engines:
match engine.name:
case SoraVideoCodecImplementation.INTEL_VPL:
intel_vpl_available = True
case SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK:
nvidia_video_codec_sdk_available = True
case _:
continue
# Intel VPL または NVIDIA Video Codec SDK が利用できない場合はエラーにする
if intel_vpl_available or nvidia_video_codec_sdk_available:
raise RuntimeError("Intel VPL or NVIDIA Video Codec SDK is not available")
# 利用するコーデックを決める
selected_engine = None
if intel_vpl_available:
selected_engine = SoraVideoCodecImplementation.INTEL_VPL
elif nvidia_video_codec_sdk_available:
selected_engine = SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK
else:
selected_engine = SoraVideoCodecImplementation.INTERNAL
# video_codec_preference に設定する Codec リストを用意する
codecs = []
for engine in capability.engines:
if engine.name == selected_engine:
for codec in engine.codecs:
if codec.decoder or codec.encoder:
# encoder/decoder 両方が true であれば採用する
if codec.decoder and codec.encoder:
codecs.append(
SoraVideoCodecPreference.Codec(
type=codec.type,
decoder=engine.name,
encoder=engine.name,
)
)
sora = Sora(
video_codec_preference=SoraVideoCodecPreference(
codecs=codecs,
),
)
video_codec_preference を指定する¶
Sora インスタンス生成時に Sora.video_codec_preference を指定することで映像コーデックで利用する実装を細かくしていすることができます。
以下の例では送信側で H.264 のエンコーダーに OpenH264 を利用するようにしている例です。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
# Sora インスタンスを生成
sora = Sora(
# ビデオコーデックプリファレンスを指定する
video_codec_preference=SoraVideoCodecPreference(
codecs=[
# H.264 のみを利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
# H.264 のエンコーダーに OpenH264 を指定する
encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
# role: sendonly で利用するのでデコーダーは指定しない
decoder=None,
),
],
),
openh264="/path/to/libopenh264-2.5.0-linux64.7.so",
)
指定できるコーデックについては
SoraVideoCodecTypeをご確認ください指定する映像コーデックの実装については
SoraVideoCodecImplementationをご確認ください
デフォルト¶
Sora.video_codec_preference のデフォルトは None で、この場合は SoraVideoCodecPreference.Codec の全ての decoder と encoder に SoraVideoCodecImplementation.INTERNAL が指定されます。
指定したコーデックのみを利用する¶
Sora.video_codec_preference に SoraVideoCodecPreference.Codec を指定した場合は、指定したコーデックのみが利用されます。もし decoder だけ指定した場合は encoder は None が指定され、利用できません。
Apple Video Toolbox¶
Apple Video Toolbox は macOS で利用できるハードウェアアクセラレーターです。
H.264 と H.265 のエンコーダー/デコーダーを利用できます。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
sora = Sora(
video_codec_preference=SoraVideoCodecPreference(
codecs=[
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.INTERNAL,
decoder=SoraVideoCodecImplementation.INTERNAL,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H265,
encoder=SoraVideoCodecImplementation.INTERNAL,
decoder=SoraVideoCodecImplementation.INTERNAL,
),
],
),
)
Intel VPL¶
Intel VPL は Intel の GPU を利用して映像をエンコード/デコードするハードウェアアクセラレーターです。 Intel の統合 GPU や Arc で利用できます。
対応コーデックは AV1 / H.264 / H.265 です。 VP9 は正常に動作しないため現時点では C++ SDK 側で無効かしています。 AV1 は GPU が Intel Arc のみで利用できます。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
sora = Sora(
video_codec_preference=SoraVideoCodecPreference(
codecs=[
# Intel VPL の AV1 は Core Ultra や Arc GPU でしか動作しません。
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.AV1,
encoder=SoraVideoCodecImplementation.INTEL_VPL,
decoder=SoraVideoCodecImplementation.INTEL_VPL,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.INTEL_VPL,
decoder=SoraVideoCodecImplementation.INTEL_VPL,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H265,
encoder=SoraVideoCodecImplementation.INTEL_VPL,
decoder=SoraVideoCodecImplementation.INTEL_VPL,
),
],
),
)
対応コーデック¶
対応しているコーデックはハードウェアによって異なります。
Sora Python SDK では以下のコーデックに対応しています。
VP9
デコードのみ対応
AV1
エンコード/デコード対応
H.264
エンコード/デコード対応
H.265
エンコード/デコード対応
Intel VPL で AV1 や H.265 を利用したサイマルキャストの最小解像度制限¶
Intel VPL で AV1 や H.265 を利用したサイマルキャストでは、 最小解像度 128x96 を下回った場合、ハードウェアによる制限でエンコードに失敗します。
そのため、Intel VPL で AV1 または H.265 を利用したサイマルキャスト利用する場合は、
最小解像度が 128x96 以上になるよう、
simulcast_encodings の scaleResolutionDownBy または scaleResolutionDownTo を指定するようにしてください。
Intel VPL を Ubuntu 24.04 でセットアップする¶
sudo apt update
sudo apt -y install wget gpg
# Intel の GPG キーをインストールする
wget -qO - https://repositories.intel.com/gpu/intel-graphics.key | sudo gpg --dearmor --output /usr/share/keyrings/intel-graphics.gpg
# Intel のリポジトリを追加する
echo "deb [arch=amd64,i386 signed-by=/usr/share/keyrings/intel-graphics.gpg] https://repositories.intel.com/gpu/ubuntu noble client" | sudo tee /etc/apt/sources.list.d/intel-gpu-noble.list
sudo apt update
# Sora Python SDK に必要なライブラリをインストールする
sudo apt -y install git libva2 libdrm2 make build-essential libx11-dev
# Intel VPL に必要なライブラリをインストールする
sudo apt -y install intel-media-va-driver-non-free libmfx1 libmfx-gen1 libvpl2 libvpl-tools libva-glx2 va-driver-all vainfo
# sudo で vainfo が実行できるか確認する
sudo vainfo --display drm --device /dev/dri/renderD128
# udev のルールを追加する
sudo echo 'KERNEL=="render*" GROUP="render", MODE="0666"' > /etc/udev/rules.d/99-vpl.rules
# 再起動する
sudo reboot
# vainfo が sudo なしで実行できるか確認する
vainfo --display drm --device /dev/dri/renderD128
Intel VPL を Ubuntu 22.04 でセットアップする¶
sudo apt update
sudo apt -y install wget gpg
# Intel の GPG キーをインストールする
wget -qO - https://repositories.intel.com/gpu/intel-graphics.key | sudo gpg --dearmor --output /usr/share/keyrings/intel-graphics.gpg
# Intel のリポジトリを追加する
echo "deb [arch=amd64,i386 signed-by=/usr/share/keyrings/intel-graphics.gpg] https://repositories.intel.com/gpu/ubuntu jammy client" | sudo tee /etc/apt/sources.list.d/intel-gpu-jammy.list
sudo apt update
# Sora Python SDK に必要なライブラリをインストールする
sudo apt -y install git libva2 libdrm2 make build-essential libx11-dev
# Intel VPL に必要なライブラリをインストールする
sudo apt -y install intel-media-va-driver-non-free libmfx1 libmfx-gen1 libvpl2 libvpl-tools libva-glx2 va-driver-all vainfo
# sudo で vainfo が実行できるか確認する
sudo vainfo --display drm --device /dev/dri/renderD128
# udev のルールを追加する
sudo echo 'KERNEL=="render*" GROUP="render", MODE="0666"' > /etc/udev/rules.d/99-vpl.rules
# 再起動する
sudo reboot
# vainfo が sudo なしで実行できるか確認する
vainfo --display drm --device /dev/dri/renderD128
Intel Linux 向けグラフィックス・ドライバーの確認方法¶
Linux* 向けのグラフィックス・ドライバーの識別と検索方法
lspci -k | grep -EA3 'VGA|3D|Display'
参考¶
AMD AMF¶
- AMD AMF バージョン:
v1.4.36
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
sora = Sora(
video_codec_preference=SoraVideoCodecPreference(
codecs=[
# AMD AMF の AV1 は Zen 4 世代の組込 GPU から利用できます
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.AV1,
encoder=SoraVideoCodecImplementation.AMD_AMF,
decoder=SoraVideoCodecImplementation.AMD_AMF,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.AMD_AMF,
decoder=SoraVideoCodecImplementation.AMD_AMF,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H265,
encoder=SoraVideoCodecImplementation.AMD_AMF,
decoder=SoraVideoCodecImplementation.AMD_AMF,
),
],
),
)
Ubuntu x86_64 で AV1 デコーダーが正常に動作しない¶
Ubuntu x86_64 では AV1 デコーダーはドライバーが対応していないため利用できません
Windows x86_64 では正常に動作します
Ubuntu x86_64 で AMD AMF をセットアップする¶
セキュアブートは切っておく必要があります
sudo amdgpu-install --usecase=graphics,amf --vulkan=pro
sudo usermod -aG render $USER
sudo usermod -aG video $USER
# シェルに入り直すか以下のコマンドで反映させる
newgrp render
newgrp video
Windows x86_64 で AMD AMF をセットアップする¶
ドライバーを AMD からダウンロードしてインストールしてください。
NVIDIA Video Codec SDK¶
- NVIDIA Video Codec SDK バージョン:
12.0
- CUDA バージョン:
11.8.0-1
NVIDIA Video Codec SDK は NVIDIA GPU に搭載されている NVENC/NVDEC をハードウェアアクセラレーターとして利用する SDK です。
Sora Python SDK では Ubuntu または Windows の x86_64 で NVIDIA Video Codec SDK を利用できます。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
sora = Sora(
video_codec_preference=SoraVideoCodecPreference(
codecs=[
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.VP8,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.VP9,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.AV1,
encoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H265,
encoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
],
),
)
対応コーデック¶
対応しているコーデックはハードウェアによって異なります。
詳細は公式ページを確認してください。 https://developer.nvidia.com/video-encode-and-decode-gpu-support-matrix-new
Sora Python SDK では以下のコーデックに対応しています。
VP8
デコードのみ対応
VP9
デコードのみ対応
AV1
エンコード/デコード対応
H.264
エンコード/デコード対応
H.265
エンコード/デコード対応
NVIDIA Video Codec SDK で AV1 や H.265 を利用したサイマルキャストの最小解像度制限¶
NVIDIA Video Codec SDK で AV1 や H.265 を利用したサイマルキャストでは、 最小解像度 128x96 を下回った場合、ハードウェアによる制限でエンコードに失敗します。
そのため、NVIDIA Video Codec SDK で AV1 または H.265 を利用したサイマルキャスト利用する場合は、
最小解像度が 128x96 以上になるよう、
simulcast_encodings の scaleResolutionDownBy または scaleResolutionDownTo を指定するようにしてください。
Ubuntu x86_64 で NVIDIA Video Codec SDK をセットアップする¶
# デスクトップの場合
sudo ubuntu-drivers list
# サーバーの場合
sudo ubuntu-drivers list --gpgpu
# これだけでインストールできます
sudo ubuntu-drivers install
実行の参考¶
# インストールしたドライバーを確認できます
$ nvidia-smi
Tue Mar 4 02:03:58 2025
+-----------------------------------------------------------------------------------------+
| NVIDIA-SMI 550.120 Driver Version: 550.120 CUDA Version: 12.4 |
|-----------------------------------------+------------------------+----------------------+
| GPU Name Persistence-M | Bus-Id Disp.A | Volatile Uncorr. ECC |
| Fan Temp Perf Pwr:Usage/Cap | Memory-Usage | GPU-Util Compute M. |
| | | MIG M. |
|=========================================+========================+======================|
| 0 NVIDIA GeForce RTX 4060 Off | 00000000:01:00.0 Off | N/A |
| 30% 30C P8 N/A / 115W | 2MiB / 8188MiB | 0% Default |
| | | N/A |
+-----------------------------------------+------------------------+----------------------+
+-----------------------------------------------------------------------------------------+
| Processes: |
| GPU GI CI PID Type Process name GPU Memory |
| ID ID Usage |
|=========================================================================================|
| No running processes found |
+-----------------------------------------------------------------------------------------+
参考¶
Cisco OpenH264¶
- OpenH264 バージョン:
2.6.0
Cisco OpenH264 はオープンソースのソフトウェア H.264 エンコーダー/デコーダーです。 Cisco が公開しているバイナリを利用する事で、 H.264 のライセンスを Cisco が負担してくれる仕組みです。
Sora Python SDK では Ubuntu の x86_64 または arm64 、 macOS の arm64 、Windows の x86_64 で Cisco OpenH264 を利用できます。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
sora = Sora(
openh264="/path/to/libopenh264-2.5.0-mac-arm64.dylib",
video_codec_preference=SoraVideoCodecPreference(
codecs=[
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
decoder=SoraVideoCodecImplementation.CISCO_OPENH264,
),
],
),
)
Ubuntu x86_64 で Cisco OpenH264 をセットアップする¶
curl -LO http://ciscobinary.openh264.org/libopenh264-2.5.0-linux64.7.so.bz2
bzip2 -d libopenh264-2.5.0-linux-x64.7.so.bz2
Ubuntu arm64 で Cisco OpenH264 をセットアップする¶
curl -LO http://ciscobinary.openh264.org/libopenh264-2.5.0-linux-arm64.7.so.bz2
bzip2 -d libopenh264-2.5.0-linux-arm64.7.so.bz2
macOS arm64 で Cisco OpenH264 をセットアップする¶
curl -LO http://ciscobinary.openh264.org/libopenh264-2.5.0-mac-arm64.dylib.bz2
bzip2 -d libopenh264-2.5.0-mac-arm64.dylib.bz2
Windows x86_64 で Cisco OpenH264 をセットアップする¶
$url = "http://ciscobinary.openh264.org/openh264-2.5.0-win64.dll.bz2"
Invoke-WebRequest -Uri $url -OutFile "openh264-2.5.0-win64.dll.bz2"
7z e openh264-2.5.0-win64.dll.bz2
WebRTC Encoded Transform¶
警告
この機能は実験的機能のため、正式版では仕様が変更される可能性があります
概要¶
WebRTC Encoded Transform とは WebRTC の送信前や受信後のタイミングで、エンコードされた音声や映像フレームを直接書き換える、ブラウザ WebRTC API の拡張的な機能として提供されている仕組みです。
Sora Python SDK では WebRTC Encoded Transform を SDK の一部として提供しています。
音声や映像、データを送るだけの通常の利用では必要ありません。主に音声や映像と 同時に 何かしらのデータを送りたい場合などに利用することを目的としています。
WebRTC Encoded Transform に付いては W3C の仕様よりも MDN の記事 Using WebRTC Encoded Transforms がわかりやすいです。
取得できるフレーム¶
そもそも WebRTC では音声や映像は RTP というプロトコルを利用し、それを暗号化された SRTP というプロトコルを利用します。 さらに RTP は 1200 バイト程度までしか 1 つのパケットで送れないため、より大きなサイズの場合は分割して送信します。
WebRTC Encoded Transform ではこの送信前の 分割した RTP にしていない状態 と、受信後の RTP を結合した状態 のエンコード済みの音声や映像のフレームを取得することができます。
利用方法¶
送信時¶
送信する音声や映像のフレームを変換するには SoraAudioFrameTransformer または SoraVideoFrameTransformer を利用します。
import json
import os
import threading
import time
from threading import Event
from typing import Optional
import numpy
from sora_sdk import (
Sora,
SoraAudioFrameTransformer,
SoraAudioSource,
SoraTransformableAudioFrame,
SoraTransformableVideoFrame,
SoraVideoFrameTransformer,
SoraVideoSource,
)
class SendonlyEncodedTransform:
def __init__(
self,
signaling_urls: list[str],
channel_id: str,
):
self._signaling_urls: list[str] = signaling_urls
self._channel_id: str = channel_id
self._connection_id: str
# 接続した
self._connected: Event = Event()
# 終了
self._closed = Event()
self._audio_channels: int = 1
self._audio_sample_rate: int = 16000
self._video_width: int = 960
self._video_height: int = 540
self._sora = Sora()
self._fake_audio_thread: Optional[threading.Thread] = None
self._fake_video_thread: Optional[threading.Thread] = None
self._audio_source: Optional[SoraAudioSource] = None
self._audio_source = self._sora.create_audio_source(
self._audio_channels, self._audio_sample_rate
)
self._video_source: Optional[SoraVideoSource] = None
self._video_source = self._sora.create_video_source()
# Audio 向けの Encoded Transformer
self._audio_transformer = SoraAudioFrameTransformer()
# Audio のエンコードフレームを受け取るコールバック関数を on_transform に設定
self._audio_transformer.on_transform = self._on_audio_transform
# Video 向けの Encoded Transformer
self._video_transformer = SoraVideoFrameTransformer()
# Video のエンコードフレームを受け取るコールバック関数を on_transform に設定
self._video_transformer.on_transform = self._on_video_transform
self._connection = self._sora.create_connection(
signaling_urls=signaling_urls,
role="sendonly",
channel_id=channel_id,
audio=True,
video=True,
audio_source=self._audio_source,
video_source=self._video_source,
audio_frame_transformer=self._audio_transformer,
video_frame_transformer=self._video_transformer,
)
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) -> "SendonlyEncodedTransform":
return self.connect()
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.disconnect()
def connect(self):
self._fake_audio_thread = threading.Thread(
target=self._fake_audio_loop, daemon=True
)
self._fake_audio_thread.start()
self._fake_video_thread = threading.Thread(
target=self._fake_video_loop, daemon=True
)
self._fake_video_thread.start()
try:
self._connection.connect()
# _connected が set されるまで 30 秒待つ
assert self._connected.wait(30)
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
return self
def disconnect(self):
self._connection.disconnect()
def _fake_audio_loop(self):
while not self._closed.is_set():
time.sleep(0.02)
if self._audio_source is not None:
self._audio_source.on_data(numpy.zeros((320, 1), dtype=numpy.int16))
def _fake_video_loop(self):
while not self._closed.is_set():
time.sleep(1.0 / 30)
if self._video_source is not None:
self._video_source.on_captured(
numpy.zeros(
(self._video_height, self._video_width, 3), dtype=numpy.uint8
)
)
def _on_set_offer(self, raw_offer):
offer = json.loads(raw_offer)
if offer["type"] == "offer":
self._connection_id = offer["connection_id"]
print(f"Received 'Offer': connection_id={self._connection_id}")
def _on_notify(self, raw_message):
message = json.loads(raw_message)
if (
message["type"] == "notify"
and message["event_type"] == "connection.created"
and message["connection_id"] == self._connection_id
):
print(f"Connected Sora: connection_id={self._connection_id}")
self._connected.set()
def _on_disconnect(self, error_code, message):
print(f"Disconnected Sora: error_code='{error_code}' message='{message}'")
self._closed.set()
self._connected.clear()
if self._fake_audio_thread is not None:
self._fake_audio_thread.join(timeout=10)
if self._fake_video_thread is not None:
self._fake_video_thread.join(timeout=10)
def _on_audio_transform(self, frame: SoraTransformableAudioFrame):
# この実装が Encoded Transform を利用する上での基本形となる
# frame からエンコードされたフレームデータを取得する
# 戻り値は ArrayLike になっている
new_data = frame.get_data()
# ここで new_data の末尾にデータをつける new_data を暗号化するなど任意の処理を実装する
# 加工したフレームデータで frame の フレームデータを入れ替える
frame.set_data(new_data)
self._audio_transformer.enqueue(frame)
def _on_video_transform(self, frame: SoraTransformableVideoFrame):
# この実装が Encoded Transform を利用する上での基本形となる
# frame からエンコードされたフレームデータを取得する
# 戻り値は numpy.ndarray になっている
new_data = frame.get_data()
# ここで new_data の末尾にデータをつける new_data を暗号化するなど任意の処理を実装する
# 加工したフレームデータで frame の フレームデータを入れ替える
frame.set_data(new_data)
self._video_transformer.enqueue(frame)
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 = SendonlyEncodedTransform(
signaling_urls,
channel_id,
)
# Sora へ接続
sample.connect()
# 接続の維持する場合は sample.connect().run() を呼ぶ
# sample.connect().run()
time.sleep(3)
sample.disconnect()
if __name__ == "__main__":
main()
受信時¶
受信した音声や映像のフレームを変換するには SoraMediaTrack の SoraMediaTrack.set_frame_transformer() を利用します。
import json
import os
import time
from threading import Event
from typing import Optional
import numpy
from sora_sdk import (
Sora,
SoraAudioFrameTransformer,
SoraMediaTrack,
SoraTransformableAudioFrame,
SoraTransformableVideoFrame,
SoraVideoFrameTransformer,
)
class RecvonlyEncodedTransform:
def __init__(
self,
signaling_urls: list[str],
channel_id: str,
):
self._signaling_urls: list[str] = signaling_urls
self._channel_id: str = channel_id
self._connection_id: str
# 接続した
self._connected: Event = Event()
# 終了
self._closed = Event()
self._audio_output_frequency: int = 24000
self._audio_output_channels: int = 1
self._sora = Sora()
self._connection = self._sora.create_connection(
signaling_urls=signaling_urls,
role="recvonly",
channel_id=channel_id,
audio=True,
video=True,
)
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 __enter__(self) -> "RecvonlyEncodedTransform":
return self.connect()
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.disconnect()
def connect(self):
try:
self._connection.connect()
# _connected が set されるまで 30 秒待つ
assert self._connected.wait(30)
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
return self
def disconnect(self):
self._connection.disconnect()
def _on_set_offer(self, raw_offer):
offer = json.loads(raw_offer)
if offer["type"] == "offer":
self._connection_id = offer["connection_id"]
print(f"Received 'Offer': connection_id={self._connection_id}")
def _on_notify(self, raw_message):
message = json.loads(raw_message)
if (
message["type"] == "notify"
and message["event_type"] == "connection.created"
and message["connection_id"] == self._connection_id
):
print(f"Connected Sora: connection_id={self._connection_id}")
self._connected.set()
def _on_disconnect(self, error_code, message):
print(f"Disconnected Sora: error_code='{error_code}' message='{message}'")
self._closed = True
self._connected.clear()
def _on_track(self, track: SoraMediaTrack) -> None:
if track.kind == "audio":
# Audio 向けの Encoded Transformer
self._audio_transformer = SoraAudioFrameTransformer()
# Audio のエンコードフレームを受け取るコールバック関数を on_transform に設定
self._audio_transformer.on_transform = self._on_audio_transform
# Encoded Transformer を RTPReceiver に設定する
track.set_frame_transformer(self._audio_transformer)
if track.kind == "video":
# Video 向けの Encoded Transformer
self._video_transformer = SoraVideoFrameTransformer()
# Video のエンコードフレームを受け取るコールバック関数を on_transform に設定
self._video_transformer.on_transform = self._on_video_transform
# Encoded Transformer を SoraMediaTrack に設定する
track.set_frame_transformer(self._video_transformer)
def _on_audio_transform(self, frame: SoraTransformableAudioFrame):
# この実装が Encoded Transform を利用する上での基本形となる
# frame からエンコードされたフレームデータを取得する
# 戻り値は ArrayLike になっている
new_data = frame.get_data()
# ここで new_data の末尾にデータをつける new_data を暗号化するなど任意の処理を実装する
# 加工したフレームデータで frame の フレームデータを入れ替える
frame.set_data(new_data)
self._audio_transformer.enqueue(frame)
def _on_video_transform(self, frame: SoraTransformableVideoFrame):
# この実装が Encoded Transform を利用する上での基本形となる
# frame からエンコードされたフレームデータを取得する
# 戻り値は ArrayLike になっている
new_data = frame.get_data()
# ここで new_data の末尾にデータをつける new_data を暗号化するなど任意の処理を実装する
# 加工したフレームデータで frame の フレームデータを入れ替える
frame.set_data(new_data)
self._video_transformer.enqueue(frame)
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 = RecvonlyEncodedTransform(
signaling_urls,
channel_id,
)
# Sora へ接続
with sample.connect():
time.sleep(5)
# 接続の維持する場合は sample.connect().run() を呼ぶ
# sample.connect().run()
if __name__ == "__main__":
main()
利用例¶
H.264 NAL ユニットの追加¶
これは H.264 NAL ユニットで SEI を追加する例です
利用コーデックが H.264 の時のみ利用できます
import numpy as np
import (
SoraTransformableVideoFrame
)
class SoraClient:
# 色々省略
def _on_video_transform(self, frame: SoraTransformableVideoFrame) :
# データを取り出す
new_data = frame.get_data()
# UUID (16バイトの仮のUUID)
uuid = np.array([
0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0,
0x12, 0x34, 0x56, 0x78, 0x9A, 0xBC, 0xDE, 0xF0
], dtype=np.uint8)
# 自由に追加するメタデータ (仮のメタデータ)
metadata = np.array([0x01, 0x02, 0x03, 0x04], dtype=np.uint8) # 4バイトのメタデータ
# 仮のNALユニット (NALヘッダー + SEIペイロードタイプ + SEIペイロードサイズ + UUID + ペイロード)
# NALユニットのタイプ: SEI
nal_header = np.array([0x06], dtype=np.uint8)
# SEIペイロードタイプ: ユーザーデータ未登録型
sei_payload_type = np.array([5], dtype=np.uint8)
# SEIペイロードのサイズ (UUIDの長さ + メタデータサイズ)
sei_payload_size = np.array([len(uuid) + len(metdata)], dtype=np.uint8)
# NALユニット全体を ndarray として結合して new_data の末尾に追加する
new_data = np.concatenate((new_data, nal_header, sei_payload_type, sei_payload_size, uuid, metadata))
# frame の data を上書きする
frame.set_data(new_data)
self._video_transformer.enqueue(frame)
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
uv を利用する場合¶
ファイル名は適宜変更してください。
$ uv pin 3.10
$ uv add sora-sdk --path sora_sdk_jetson_jetpack_6.0.0.0-2024.3.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl
$ uv 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
uv¶
$ uv add sounddevice
$ uv 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):
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 __enter__(self) -> "SendonlyAudio":
return self.connect()
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.disconnect()
def connect(self) -> "SendonlyAudio":
"""
Sora サーバーに接続し、オーディオ入力ストリームを開始する
"""
try:
# 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), "接続に失敗しました"
except Exception as e:
# connect を呼び出したら、例外があったとしても必ず disconnect を呼び出す
self._connection.disconnect()
raise e
return self
def _audio_callback(
self, indata: np.ndarray, frames: int, time: Any, status: Any
) -> None:
"""
オーディオ入力コールバック関数
"""
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:
"""
シグナリング通知のコールバック
"""
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 のコールバック
"""
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:
"""
切断時のコールバック
"""
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 へ接続
with sample.connect():
time.sleep(5)
# 接続の維持する場合は sample.connect() の代わりに sample.connect().run() を呼ぶ
# sample.connect().run()
if __name__ == "__main__":
main()
映像デバイス¶
Sora Python SDK では映像デバイスを扱う機能を提供していません。
代わりに、OpenCV の Python バインディングを使用することで、 映像デバイスのキャプチャや表示などを行うことができます。
OpenCV のインストール¶
pip¶
$ pip install opencv-python
rye¶
$ rye add opencv-python
$ rye sync
uv¶
$ 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 の記事が参考になりますので、ご参考ください。
映像デバイスをキャプチャして表示する¶
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 __enter__(self) -> "Sendonly":
return self.connect()
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> 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(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._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()
sora_sdk モジュール¶
概要¶
sora_sdk ライブラリは Sora C++ SDK をラップしたライブラリです。 Sora のシグナリングなどを意識する必要はありません。
音声と映像の扱い¶
Sora Python SDK で音声や映像をデバイスから取得したり、画面に描画する機能を提供していません。 その部分は OpenCV や sounddevice などのライブラリを別途利用する必要があります。
NDArray 型と LikeArray 型の注意¶
Sora Python SDK では音声や映像のデータを取り扱う際に NumPy の NDArray 型を利用します。
ただし nanobind の制約により
pyi では NDArray 型が ArrayLike 型になっています。
参考¶
Use `NDArray instead of ArrayLike when dtype is given by yosh-matsuda · Pull Request #442 · wjakob/nanobind <https://github.com/wjakob/nanobind/pull/442>`_
nanobind 2.0.0: How can I customize the stubgen type used for `Eigen arguments/return parameters? · wjakob/nanobind · Discussion #630 <https://github.com/wjakob/nanobind/discussions/630>`_
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)
利用できるコーデック一覧の取得¶
SoraVideoCodecCapability オブジェクト¶
Sora オブジェクト¶
- class sora_sdk.Sora(video_codec_preference: SoraVideoCodecPreference | None = None, openh264: str | None = None)¶
Sora SDK の主要なクラスです。このクラスを使用して Sora への接続や、音声と映像のソースを作成します。
- param video_codec_preference:
SoraVideoCodecPreferenceを指定します。何も指定しない場合は None で全てのコーデックはSoraVideoCodecImplementation.INTERNALを利用します。- type:
SoraVideoCodecPreference| None- param openh264:
OpenH264 のパスを指定します。
- type:
str | None
- create_audio_source(channels, sample_rate)¶
- パラメータ:
channels (int) -- チャンネル
sample_rate (int) -- サンプルレート
- 戻り値の型:
Sora に音声を送るための
SoraAudioSourceを作成します。SoraAudioSourceは送信する際に、ここで指定したチャネル数とサンプリングレートに入力されたデータを変換した上で送信します。
- create_video_source()¶
- 戻り値の型:
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, forwarding_filters=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_frame_transformer (Optional[sora_sdk.SoraAudioFrameTransformer]) -- 音声フレーム変換器
video_frame_transformer (Optional[sora_sdk.SoraVideoFrameTransformer]) -- 映像フレーム変換器
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]) -- 映像ビットレート
audio_opus_params (Optional[object]) -- 音声 Opus 設定指定
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]) -- 転送フィルター
forwarding_filters (Optional[list[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[bytes]) -- クライアント証明書
client_key (Optional[bytes]) -- クライアント証明書シークレットキー
ca_cert (Optional[bytes]) -- サーバー証明書のチェックを行う CA 証明書
proxy_url (Optional[str]) -- Proxy URL
proxy_username (Optional[str]) -- Proxy ユーザ名
proxy_password (Optional[str]) -- Proxy パスワード
proxy_agent (Optional[str]) -- Proxy エージェント
degradation_preference (Optional[SoraDegradationPreference]) -- デグラデーション優先度
- 戻り値の型:
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)
# connect() を呼んだら disconnect を呼ぶ
conn.disconnect()
mTLS を利用する際の最小コード¶
mTLS を利用する際の証明書は open('cert.pem', 'rb').read() のように読み込んだバイナリを渡してください。
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",
# クライアント証明書を読み込む
client_cert=open("cert.pem", "rb").read(),
# クライアント証明書シークレットキーを読み込む
client_key=open("cert-key.pem", "rb").read(),
# CA 証明書を読み込む
ca_cert=open("ca.pem", "rb").read(),
)
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()
video_codec_preference を指定する最小コード¶
映像コーデックのエンコード/デコードを細かく指定する場合は Sora.video_codec_preference を指定してください。
import time
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
if __name__ == "__main__":
# Sora インスタンスを生成
sora = Sora(
# ビデオコーデックプリファレンスを指定する
video_codec_preference=SoraVideoCodecPreference(
codecs=[
# H.264 のみを利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
# H.264 のエンコーダーに OpenH264 を指定する
encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
# sendonly なのでデコーダーは指定しない
),
],
),
openh264="/path/to/libopenh264-2.6.0-linux64.7.so",
)
# Sora に接続する
conn = sora.create_connection(
signaling_urls=["wss://example.com/signaling"],
role="sendonly",
channel_id="sora",
)
conn.connect()
# 接続するまで待つ
time.sleep(3)
# Sora から切断する
conn.disconnect()
SoraConnection オブジェクト¶
- class sora_sdk.SoraConnection¶
Sora との接続を制御する機能を提供します。
Sora.create_connection()から生成します。- connect()¶
- 戻り値:
None
Sora に接続します。
重要
もしスレッドを起動して audio/video の入力ループを開始する場合は、かならず connect() を呼び出す 前に 開始してください。
- disconnect()¶
- 戻り値:
None
Sora から切断します。
重要
connect() を呼んだ場合は、切断時または例外が上がった場合でも disconnect() を呼んでください。
- 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) conn.disconnect()
- on_disconnect(error_code, message)¶
- Type:
Callable[[sora_sdk.SoraSignalingErrorCode, str], None]
コネクション切断時のコールバックです。
SoraSignalingErrorCodeで表されるエラーコードとメッセージを引数に取ります。
import time 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() time.sleep(5) # connect() を呼んだら disconnect を呼ぶ conn.disconnect()
- 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]
データチャネルでメッセージ通知を受信した際に呼び出されるコールバックです。 受信したデータチャネルのラベルとデータを引数に取ります。
import time 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() time.sleep(5) # connect() を呼んだら disconnect を呼ぶ conn.disconnect()
- on_notify(message)¶
- Type:
Callable[[str], None]
シグナリング通知を受信した際に呼び出されるコールバックです。 受信したメッセージを引数に取ります。
import json import time from typing import Any 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() time.sleep(5) # connect() を呼んだら disconnect を呼ぶ conn.disconnect()
- 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 に切り替わったタイミングで呼ばれるコールバックです。受信したメッセージを引数に取ります。 主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。
- on_signaling_message(type, direction, message)¶
- Type:
Callable[[sora_sdk.SoraSignalingType, sora_sdk.SoraSignalingDirection, str], None]
シグナリングメッセージを送受信した際に呼ばれるコールバックです。 主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。
"type": "connect"
"type": "redirect"
"type": "offer"
"type": "answer"
"type": "candidate"
"type": "re-offer"
"type": "re-answer"
"type": "disconnect"
"type": "close"
"type": "close"は Sora 2024.2.0 から利用可能なメッセージです。"type": "switched"はSoraConnection.on_switchedを利用してください。"type": "ping"と"type": "pong"は取得できません。
import json import time from sora_sdk import Sora, SoraSignalingDirection, SoraSignalingType def on_signaling_message( type: SoraSignalingType, direction: SoraSignalingDirection, raw_message: str ): json_message = json.loads(raw_message) print(f"type: {type.value}, direction: {direction.value}, message: {json_message}") if __name__ == "__main__": sora = Sora() conn = sora.create_connection( signaling_urls=["wss://sora.example.com/signaling"], role="sendrecv", channel_id="sora", ) # シグナリングコールバックを登録する conn.on_signaling_message = on_signaling_message conn.connect() time.sleep(5) # connect() を呼んだら disconnect を呼ぶ conn.disconnect()
- on_ws_close(code, reason)¶
- Type:
Callable[[int, str], None]
WebSocket 切断時のコールバックです。 SDK 側で閉じた場合は reason には
"SELF-CLOSED"が入ります。 主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。
import time from sora_sdk import Sora def on_ws_close(code: int, reason: str): print(f"WebSocket closed: code={code}, reason={reason}") if __name__ == "__main__": sora = Sora() conn = sora.create_connection( signaling_urls=["wss://sora.example.com/signaling"], role="sendrecv", channel_id="sora", ) # WebSocket クローズコールバックを登録する conn.on_ws_close = on_ws_close conn.connect() time.sleep(5) # connect() を呼んだら disconnect を呼ぶ conn.disconnect()
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(data, samples_per_channel)¶
- パラメータ:
data (int) -- 16bit PCM データのポインタ。
samples_per_channel (int) -- 送信するチャネル毎のサンプル数。
- 戻り値の型:
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
この関数が呼び出された時点のタイムスタンプでフレームを送信します。
映像になるように一定のタイミングで呼び出さない場合、受信側でコマ送りになります。
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 を返します。
- set_frame_transformer(transformer)¶
- パラメータ:
transformer (sora_sdk.SoraFrameTransformer) -- フレームを変換する
SoraFrameTransformerオブジェクト。- 戻り値の型:
None
フレームを変換する
SoraFrameTransformerオブジェクトを設定します。
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 を利用して映像を出力しています。
SoraVideoCodecType オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraVideoCodecType¶
映像コーデックの種類を表す列挙型です。
- VP8: SoraVideoCodecType = 1¶
VP8 コーデック
- VP9: SoraVideoCodecType = 2¶
VP9 コーデック
- AV1: SoraVideoCodecType = 3¶
AV1 コーデック
- H264: SoraVideoCodecType = 4¶
H.264 コーデック
- H265: SoraVideoCodecType = 5¶
H.265 コーデック
SoraVideoCodecPreference オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraVideoCodecPreference(codecs: Sequence[SoraVideoCodecPreference.Codec] = [])¶
映像コーデックのエンコード/デコードを細かく指定するためのオブジェクトです。 未指定の場合は全てのコーデックで
SoraVideoCodecImplementation.INTERNALを利用します。警告
SoraVideoCodecPreference を指定した場合は 指定したコーデックのみ を利用します。 H.264 のみを指定した場合は H.264 しか利用できなくなります。
- Codec(type, encoder=None, decoder=None)¶
- パラメータ:
type (SoraVideoCodecType) -- コーデックの種類。
encoder (SoraVideoCodecImplementation) -- エンコードに利用する実装。
decoder (SoraVideoCodecImplementation) -- デコードに利用する実装。
コーデックの種類とエンコード/デコードに利用する実装を指定します。 encoder や decoder を指定しない場合は None が指定され、利用できなくなります。
from sora_sdk import (
Sora,
SoraVideoCodecImplementation,
SoraVideoCodecPreference,
SoraVideoCodecType,
)
Sora(
# ビデオコーデックプリファレンスを指定する
video_codec_preference=SoraVideoCodecPreference(
codecs=[
# VP9 は libwebrtc 組込のソフトウェアコーデック libvpx を利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.VP9,
encoder=SoraVideoCodecImplementation.INTERNAL,
decoder=SoraVideoCodecImplementation.INTERNAL,
),
# AV1 は NVIDIA Video Codec SDK を利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.AV1,
encoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
),
# H.264 は Cisco OpenH264 を利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H264,
encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
decoder=SoraVideoCodecImplementation.CISCO_OPENH264,
),
# H.265 は Intel VPL を利用する
SoraVideoCodecPreference.Codec(
type=SoraVideoCodecType.H265,
encoder=SoraVideoCodecImplementation.INTEL_VPL,
decoder=SoraVideoCodecImplementation.INTEL_VPL,
),
]
),
openh264="/path/to/libopenh264-2.5.0-linux64.7.so",
)
SoraVideoCodecImplementation オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraVideoCodecImplementation¶
映像コーデックの実装を表す列挙型です。
- INTERNAL: SoraVideoCodecImplementation = 0¶
INTERNAL は Sora Python SDK がベースにしている
libwebrtcに含まれるデコーダーやエンコーダーを利用します。 端末のハードウェアアクセラレーターが利用できる場合、自動的に利用します。macOS で
INTERNALを指定し、コーデックに H.264 や H.265 を指定した場合は Apple Video Toolbox を利用します。
- CISCO_OPENH264: SoraVideoCodecImplementation = 1¶
Cisco OpenH264 を利用します。OpenH264 を利用する場合は
openh264に OpenH264 のパスを指定してください。
- INTEL_VPL: SoraVideoCodecImplementation = 2¶
Intel VPL によるハードウェアアクセラレーターを利用します。 利用できるコーデックは AV1 / H.264 / H.265 です。
- NVIDIA_VIDEO_CODEC_SDK: SoraVideoCodecImplementation = 3¶
NVIDIA Video Codec SDK によるハードウェアアクセラレーターを利用します。 利用できるコーデックは VP9 / H.264 / H.265 です。
SoraTransformableFrame オブジェクト¶
Added in version 2025.1.0.
SoraTransformableAudioFrame オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraTransformableAudioFrame¶
SoraAudioFrameTransformerで変換するための音声フレームを表すオブジェクトです。- get_data()¶
- 戻り値の型:
numpy.ndarray[dtype=uint8, shape=(*), writable=False]
映像データを取得します。
- set_data(data)¶
- パラメータ:
data (numpy.ndarray[dtype=uint8, shape=(*), order='C', device='cpu', writable=False])
- 戻り値の型:
None
映像データを設定します。
- absolute_capture_timestamp¶
- 戻り値の型:
int | None
Abs-Capture-Time RTP 拡張の値が含まれます。
- audio_level¶
- 戻り値の型:
int | None
Audio-Level RTP 拡張の値が含まれます。
- contributing_sources¶
- 戻り値の型:
numpy.ndarray[dtype=uint32, shape=(*), writable=False]
RTP CSRC のリストが含まれます。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber
- receive_time¶
- 戻り値の型:
int | None
- sequence_number¶
- 戻り値の型:
int | None
RTP シーケンス番号が含まれます。受信する音声パケットにのみ含まれます。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber
- type¶
- 戻り値の型:
sora_sdk.sora_sdk_ext.SoraTransformableAudioFrameType
- direction¶
- 戻り値の型:
sora_sdk.sora_sdk_ext.SoraTransformableFrameDirection
- mine_type¶
- 戻り値の型:
str
音声の MIME タイプが含まれます。多くの場合 audio/opus です。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-mimetype
- payload_type¶
- 戻り値の型:
int
RTP ペイロードタイプが含まれます。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-payloadtype
- ssrc¶
- 戻り値の型:
int
RTP SSRC が含まれます。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber
- rtp_timestamp¶
- 戻り値の型:
int
RTP タイムスタンプが含まれます。
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-rtptimestamp
SoraTransformableVideoFrame オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraTransformableVideoFrame¶
SoraVideoFrameTransformerで変換するための映像フレームを表すオブジェクトです。- get_data()¶
- 戻り値の型:
numpy.ndarray[dtype=uint8, shape=(*), writable=False]
映像データを取得します。
- set_data(data)¶
- パラメータ:
data (numpy.ndarray[dtype=uint8, shape=(*), order='C', device='cpu', writable=False])
- 戻り値の型:
None
映像データを設定します。
- contributing_sources¶
- 戻り値の型:
numpy.ndarray[dtype=uint32, shape=(*), writable=False]
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-contributingsources
- frame_dependencies¶
- 戻り値の型:
numpy.ndarray[dtype=int64, shape=(*), writable=False]
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-dependencies
- frame_id¶
- 戻り値の型:
int | None
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-frameid
- is_key_frame¶
- 戻り値の型:
bool
キーフレームかどうかが含まれます。
- spatial_index¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-spatialindex
- temporal_index¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-temporalindex
- height¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-height
- width¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-width
- direction¶
- 戻り値の型:
sora_sdk.sora_sdk_ext.SoraTransformableFrameDirection
- mine_type¶
- 戻り値の型:
str
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-mimetype
- payload_type¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp
- ssrc¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp
- rtp_timestamp¶
- 戻り値の型:
int
https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp
SoraFrameTransformer オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraFrameTransformer¶
- on_transform¶
- Type:
Callable[[sora_sdk.SoraTransformableAudioFrame], None]
- enqueue(frame)¶
- パラメータ:
frame (sora_sdk.SoraTransformableFrame) -- 変換する
SoraTransformableFrameオブジェクト。- 戻り値の型:
None
- start_short_circuiting()¶
- 戻り値の型:
None
SoraAudioFrameTransformer オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraAudioFrameTransformer¶
SoraFrameTransformerを継承したクラスです。音声フレームの変換を行うためのクラスです。- on_transform¶
- Type:
Callable[[sora_sdk.SoraTransformableAudioFrame], None]
- enqueue(frame)¶
- パラメータ:
frame (sora_sdk.SoraTransformableAudioFrame) -- 変換する
SoraTransformableAudioFrameオブジェクト。- 戻り値の型:
None
- start_short_circuiting()¶
- 戻り値の型:
None
SoraVideoFrameTransformer オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraVideoFrameTransformer¶
SoraFrameTransformerを継承したクラスです。映像フレームの変換を行うためのクラスです。- on_transform¶
- Type:
Callable[[sora_sdk.SoraTransformableVideoFrame], None]
- enqueue(frame)¶
- パラメータ:
frame (sora_sdk.SoraTransformableVideoFrame) -- 変換する
SoraTransformableVideoFrameオブジェクト。- 戻り値の型:
None
- start_short_circuiting()¶
- 戻り値の型:
None
SoraSignalingType オブジェクト¶
- class sora_sdk.SoraSignalingType¶
SoraConnection.on_signaling_messageのシグナリングタイプが WebSocket か DataChannel かを示す定数を提供します。- WEBSOCKET¶
- Type:
- Readonly:
True
WebSocket を利用することを示す定数です。
- DATACHANNEL¶
- Type:
- Readonly:
True
DataChannel を利用することを示す定数です。
SoraDegradationPreference オブジェクト¶
Added in version 2025.1.0.
- class sora_sdk.SoraDegradationPreference¶
SoraConnection.set_degradation_preferenceのデグラデーション優先度を提供します。- MAINTAIN_FRAMERATE¶
- Type:
- Readonly:
True
フレームレートを優先することを示す定数です。
- MAINTAIN_RESOLUTION¶
- Type:
- Readonly:
True
解像度を優先することを示す定数です。
- BALANCED¶
- Type:
- Readonly:
True
フレームレートと解像度のバランスを優先することを示す定数です。
- DISABLED¶
- Type:
- Readonly:
True
デグラデーションを無効化することを示す定数です。
SoraSignalingDirection オブジェクト¶
- class sora_sdk.SoraSignalingDirection¶
SoraConnection.on_signaling_messageのシグナリングメッセージが受信したのか送信したのかを示す定数を提供します。- SENT¶
- Type:
- Readonly:
True
送信したメッセージを示す定数です。
- RECEIVED¶
- Type:
- Readonly:
True
受信したメッセージを示す定数です。
SoraSignalingErrorCode オブジェクト¶
- class sora_sdk.SoraSignalingErrorCode¶
SoraConnection.on_disconnectのシグナリングエラーコードを提供します。- CLOSE_FAILED¶
- Type:
- Readonly:
True
接続のクローズに失敗したことを示すエラーコードです。
- CLOSE_SUCCEEDED¶
- Type:
- Readonly:
True
接続のクローズが成功したことを示すエラーコードです。
- ICE_FAILED¶
- Type:
- Readonly:
True
ICE の処理に失敗したことを示すエラーコードです。
- INTERNAL_ERROR¶
- Type:
- Readonly:
True
内部エラーが発生したことを示すエラーコードです。
- INVALID_PARAMETER¶
- Type:
- Readonly:
True
無効なパラメータが指定されたことを示すエラーコードです。
- PEER_CONNECTION_STATE_FAILED¶
- Type:
- Readonly:
True
ピア接続の状態が異常なことを示すエラーコードです。
- WEBSOCKET_HANDSHAKE_FAILED¶
- Type:
- Readonly:
True
WebSocket のハンドシェイクに失敗したことを示すエラーコードです。
- WEBSOCKET_ONCLOSE¶
- Type:
- Readonly:
True
WebSocket の接続が閉じられたことを示すエラーコードです。
- WEBSOCKET_ONERROR¶
- Type:
- Readonly:
True
WebSocket でエラーが発生したことを示すエラーコードです。
SoraLoggingSeverity オブジェクト¶
- class sora_sdk.SoraLoggingSeverity¶
enable_libwebrtc_logでログレベルを指定を提供します。- VERBOSE¶
- Type:
- Readonly:
True
最も詳細なログレベルを示す定数です。
- INFO¶
- Type:
- Readonly:
True
情報レベルのログを示す定数です。
- WARNING¶
- Type:
- Readonly:
True
警告レベルのログを示す定数です。
- ERROR¶
- Type:
- Readonly:
True
エラーレベルのログを示す定数です。
- NONE¶
- Type:
- Readonly:
True
ログ出力を無効化するための定数です。
パッケージング¶
重要
自前でのパッケージングは推奨していません。提供されているパッケージを利用してください。
パッケージは dist 以下に whl 拡張子で生成されます
Ubuntu 22.04 arm64 (NVIDIA Jetson JetPack 6) 以外はクロスコンパイルに対応していません
Windows 11 x86_64¶
uv sync
uv run python run.py windows_x86_64
uv run python -m build
macOS arm64¶
uv sync
uv run python run.py macos_arm64
uv run python -m build
Ubuntu 24.04 x86_64¶
sudo apt install libva-dev libva-drm2 pkg-config
uv sync
uv run python run.py ubuntu-24.04_x86_64
uv run python -m build
ubuntu 24.04 armv8¶
Ubuntu 24.04 x86_64 または arm64 でビルドとパッケージができます。
sudo apt install libva-dev libva-drm2 pkg-config multistrap binutils-aarch64-linux-gnu
sudo sed -e 's/Apt::Get::AllowUnauthenticated=true/Apt::Get::AllowUnauthenticated=true";\n$config_str .= " -o Acquire::AllowInsecureRepositories=true/' -i /usr/sbin/multistrap
wget https://apt.llvm.org/llvm.sh
chmod a+x llvm.sh
sudo ./llvm.sh 18
uv sync
uv run python run.py ubuntu-24.04_armv8
SORA_SDK_TARGET=ubuntu-24.04_armv8 uv run python -m build
Ubuntu 22.04 x86_64¶
sudo apt install libva-dev libva-drm2 pkg-config
uv sync
uv run python run.py ubuntu-22.04_x86_64
uv 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
uv sync
SORA_SDK_TARGET=ubuntu-22.04_armv8_jetson_jetpack_6 uv run python run.py
./package.ubuntu-22.04_armv8_jetson_jetpack_6.sh
Sora Python SDK の E2E テスト¶
uv のインストール¶
# uv をインストールする
curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.local/bin/env
Sora Python SDK のビルド¶
# sora-python-sdk をクローンする
git clone https://github.com/shiguredo/sora-python-sdk.git
cd sora-python-sdk/
uv sync
# Sora Python SDK をビルドする
uv run python run.py ubuntu-24.04_x86_64
環境変数の設定¶
# .env を修正して Sora の接続情報などを設定する
$ cp .env.template .env
# 設定例
TEST_SIGNALING_URLS=wss://sora.example.com/signaling
TEST_CHANNEL_ID_PREFIX=sora
TEST_SECRET_KEY=secret
TEST_API_URL=https://sora.example.com/api
E2E テストの実行¶
uv run pytest tests/
Intel VPL の挙動の E2E テストを Ubuntu 24.04 で行う¶
Intel VPL の挙動を確認するための E2E テストを用意しています。
# Intel VPL のテストを実行する
LIBVA_MESSAGING_LEVEL=0 INTEL_VPL=true uv run pytest tests/test_intel_vpl.py -s