<!-- mikan:page doc="index" -->

# Sora Python SDK ドキュメント

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

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

# LLM 向け

LLM が読み込みやすい形式でドキュメントを提供しています。

- 目次は [llms.txt](/llms.txt) にあります
- 全文は [llms-full.txt](/llms-full.txt) にあります

## 問い合わせについて

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


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

- [リリースノート](release.html)
- [Sora Python SDK 概要](overview.html)
- [ガイド](guide.html)
- [FAQ](faq.html)
- [既知の問題](known_issues.html)

- [インストール](install.html)
- [チュートリアル](tutorial.html)
- [サンプル](examples.html)
- [映像コーデックプリファレンス](video_codec_preference.html)
- [WebRTC Encoded Transform](webrtc_encoded_transform.html)
- [NVIDIA Jetson JetPack SDK](jetson_jetpack.html)

- [音声デバイス](audio_device.html)
- [映像デバイス](video_device.html)

- [Sora Python SDK for Raspberry Pi](raspberry_pi.html)

- [sora_sdk モジュール](sora_sdk.html)

- [パッケージング](packaging.html)
- [Sora Python SDK の E2E テスト](e2e_test.html)

<!-- mikan:page doc="release" -->

# リリースノート

**CHANGE**
: 後方互換性のない変更

**UPDATE**
: 後方互換性がある変更

**ADD**
: 後方互換性がある追加

**FIX**
: バグ修正

NVIDIA Jetson JetPack SDK での利用は [NVIDIA Jetson JetPack SDK](jetson_jetpack.html) をご確認ください。


## 2025.5.2

**日付**: 2025-12-22
**対応 Sora C++ SDK**: 2025.6.2
**対応 Sora**: 2025.1.x / 2025.2.x
**対応 Python**: 3.12 / 3.13 / 3.14

- [FIX] `video_h265_params` への対応が洩れていたのを修正しました


## 2025.5.1

**日付**: 2025-12-08
**対応 Sora C++ SDK**: 2025.6.1
**対応 Sora**: 2025.1.x / 2025.2.x
**対応 Python**: 3.12 / 3.13 / 3.14

- [FIX] Sora C++ SDK のバージョンを `2025.6.1` にアップデートしました- 最新の Raspberry Pi OS で libcamera が 0.6 に上がり動作しない問題を修正しました


## 2025.5.0

**日付**: 2025-12-01
**対応 Sora C++ SDK**: 2025.6.0
**対応 Sora**: 2025.1.x / 2025.2.x
**対応 Python**: 3.12 / 3.13 / 3.14

- [ADD] Python 3.14 に対応しました
- [ADD] Python 3.11 を非対応にしました
- [ADD] Raspberry Pi OS 向けのパッケージを sora-sdk-rpi として PyPI に追加しました- `uv add sora-sdk-rpi` でインストールできます
  - Raspberry Pi OS Bookworm の arm64v8 に対応しています
- [ADD] Raspberry Pi OS 向けに libcamera を利用できる `Sora.create_libcamera_source()` を追加しました
- [ADD] Raspberry Pi OS 向けに V4L2-M2M API 経由で H.264 ハードウェアアクセラレーターを利用できる `VideoCodecImplementation.RASPI_V4L2M2M` を追加しました- Raspberry Pi 5 は H.264 ハードウェアアクセラレーターが搭載されていないため利用できません
- [ADD] サイマルキャスト機能で初期の `rid` を指定する `simulcast_request_rid` オプションを追加しました- デフォルトは未指定です- `simulcast` を `true` で指定する必要があります- `sendrecv` または `recvonly` でのみ有効です
- [UPDATE] Sora C++ SDK のバージョンを `2025.6.0` にアップデートしました- WEBRTC_BUILD_VERSION を `m143.7499.1.0` にアップデートしました
  - CMAKE_VERSION を `4.1.2` にアップデートしました


## 2025.4.0

**日付**: 2025-09-12
**対応 Sora C++ SDK**: 2025.5.0
**対応 Sora**: 2024.2.x / 2025.1.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [CHANGE] run.py をサブコマンド形式に変更しました- 従来: `python run.py <target>`- 新形式: `python run.py build <target>`
- [UPDATE] Sora C++ SDK のバージョンを `2025.5.0` にアップデートしました- WEBRTC_BUILD_VERSION を `m139.7258.3.0` にアップデートしました
  - CMAKE_VERSION を `4.1.3` にアップデートしました
  - BOOST_VERSION を `1.89.0` にアップデートしました
- [UPDATE] `.github/workflows` 内のすべてのワークフローファイルを新しい `run.py build` 形式にアップデートしました
- [UPDATE] nanobind を `2.9.2` にアップデートしました
- [UPDATE] Sora C++ SDK のバージョンを `2025.5.0` にアップデートしました- WEBRTC_BUILD_VERSION を `m139.7258.3.0` にアップデートしました- CMAKE_VERSION を `4.1.0` にアップデートしました- BOOST_VERSION を `1.89.0` にアップデートしました
- [ADD] run.py に format サブコマンドを追加しました- C++ ファイルの clang-format によるフォーマット機能- Python ファイルの ty によるタイプチェック機能
- [FIX] GitHub Actions の check_ubuntu_wheel ジョブで uv 0.8 以降の externally managed Python 環境エラーを修正しました- `uv run --with` から `uv pip install` を使用する方式に変更- checkouts せずに仮想環境を作成して wheel ファイルをテストするように変更


## 2025.3.0

**日付**: 2025-07-09
**対応 Sora C++ SDK**: 2025.4.0
**対応 Sora**: 2024.2.x / 2025.1.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [UPDATE] Sora C++ SDK のバージョンを `2025.4.0` にアップデートしました- WEBRTC_BUILD_VERSION を `m138.7204.0` にアップデートしました
  - CMAKE_VERSION を `4.0.3` にアップデートしました
- [ADD] `__version__` でバージョンを取得できるようにしました
- [ADD] Ubuntu 22.04 arm64 でのビルドを追加しました
- [ADD] WebSocket 接続時に `User-Agent` を上書きする機能を追加しました
- [ADD] `on_rpc` コールバックを追加しました- Sora 2025.2.0 から利用できる用になる予定です
- [FIX] pyi ファイルを Python のバージョン毎に生成していなかった問題を修正しました
- [FIX] Ubuntu arm64 での Python 3.11 と 3.12 に対応できていなかった問題を修正しました


## 2025.2.3

**日付**: 2025-05-23
**対応 Sora C++ SDK**: 2025.3.1
**対応 Sora**: 2024.1.x / 2024.2.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [FIX] 切断時に落ちる問題を解消するために [Sora C++ SDK](https://github.com/shiguredo/sora-cpp-sdk) を `2025.3.1` にアップデートしました


## 2025.2.2

**日付**: 2025-05-15
**対応 Sora C++ SDK**: 2025.3.0
**対応 Sora**: 2024.1.x / 2024.2.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [FIX] Python 3.13 で [nanobind](https://github.com/wjakob/nanobind) によるメモリーリークが発生している問題を修正しました


## 2025.2.1

**日付**: 2025-05-01
**対応 Sora C++ SDK**: 2025.3.0
**対応 Sora**: 2024.1.x / 2024.2.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [FIX] [PyPI](https://pypi.org/) を Organization に移行した事による Publishing ができない問題を修正しました


## 2025.2.0

**日付**: 2025-05-01
**対応 Sora C++ SDK**: 2025.3.0
**対応 Sora**: 2024.1.x / 2024.2.x
**対応 Python**: 3.11 / 3.12 / 3.13

- [UPDATE] [nanobind](https://github.com/wjakob/nanobind) を `2.5.0` にアップデートしました
- [UPDATE] [Sora C++ SDK](https://github.com/shiguredo/sora-cpp-sdk) を `2025.2.0` にアップデートしました- WEBRTC_BUILD_VERSION を `m136.7103.0.0` にアップデートしました
  - BOOST_VERSION を `1.88.0` にアップデートしました
  - CMAKE_VERSION を `4.0.1` にアップデートしました
  - OPENH264_VERSION を `v2.6.0` にアップデートしました

## 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](https://scientific-python.org/specs/spec-0000/) を参考に直近 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](https://github.com/shiguredo/sora-cpp-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](https://github.com/wjakob/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](https://www.openh264.org/) を利用した H.264 のデコードとエンコードが利用可能になりました
- [ADD] [AMD AMF](https://gpuopen.com/advanced-media-framework/) を利用したハードウェアアクセラレーター機能に対応しました
- [ADD] Sora.video_codec_preference を追加しました- コーデックのエンコード/デコードを細かく指定できるようになりました
  - 詳細は Sora.video_codec_preference をご確認ください
- [ADD] エンコード時の劣化優先順位を指定する `degradation_preference` を追加しました- `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](https://www.webrtc.org/blog/webrtc-encoded-transform) に対応しました- `SoraTransformableAudioFrame` と `SoraTransformableVideoFrame` を追加しました
  - `SoraAudioFrameTransformer` と `SoraVideoFrameTransformer` を追加しました
  - `create_connection()` の引数に `audio_frame_transformer` と `video_frame_transformer` を追加しました
  - `SoraMediaTrack` に `set_frame_transformer()` を追加しました
  - 詳細は [WebRTC Encoded Transform](webrtc_encoded_transform.html) をご確認ください
- [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](webrtc_encoded_transform.html) をご確認ください
- [ADD] macOS 向けビルドに macOS 14 arm64 を追加しました
- [ADD] `sora_sdk` に型を追加しました
- [ADD] `SoraConnection` に `get_stats` 関数を追加しました
- [ADD] Sora C++ SDK と libwebrtc のローカルビルドを利用できるようにしました
- [FIX] `SoraAudioSink.read` が `timeout` を無視して失敗を返す問題を修正しました
- [FIX] MSVC 内部コンパイルエラーにより Windows で nanobind のビルドができない問題を修正しました

## 2024.2.0

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

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

### 既知の問題

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

## 2024.1.0

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

- [CHANGE] Python フォーマッターを [Ruff](https://github.com/astral-sh/ruff) に変更しました
- [CHANGE] `SoraAudioSource.on_data` の引数名を変更しました
- [CHANGE] `SoraVideoSource.on_captured` の引数名を変更しました
- [CHANGE] `SoraVAD.analyze` の引数名を変更しました
- [CHANGE] `SoraConnection.on_track` の引数を `SoraMediaTrack` に変更しました
- [UPDATE] Python 3.12 に対応しました
- [UPDATE] [nanobind](https://github.com/wjakob/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](http://www.openh264.org/) のバイナリをダウンロードし、設定することで利用可能になりました

## 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] メッセージング機能の 送受信 / 送信 / 受信サンプルを追加しました

<!-- mikan:page doc="overview" -->

# Sora Python SDK 概要

Sora Python SDK は [株式会社時雨堂](https://shiguredo.jp) の [WebRTC SFU Sora](https://sora.shiguredo.jp) の Python 向けクライアントフレームワークです。

## 特徴

Sora Python SDK は [Sora C++ SDK](https://github.com/shiguredo/sora-cpp-sdk) をラップした Python 向けのライブラリです。
そのため、 Sora C++ SDK がサポートする機能のほとんどを利用する事ができます。

Sora C++ SDK が libwebrtc を利用している事もあり、ブラウザで利用する WebRTC と同じ機能が利用できます。

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

時雨堂が独自に様々なハードウェアアクセラレーターに対応することで、CPU 負荷を抑えて高画質な映像配信を行う事ができます。

- [Apple macOS Video Toolbox](https://developer.apple.com/documentation/videotoolbox)- H.264 / H.265
  - VP9 / AV1 のデコードには対応していません
- [NVIDIA Video Codec](https://developer.nvidia.com/video-codec-sdk)- VP8 / VP9 / H.264 / H.265
  - VP8 / VP9 はデコードのみの対応です
- [Intel VPL](https://github.com/intel/libvpl) (Intel Media SDK の後継)- VP9 / AV1 / H.264 / H.265
- [AMD AMF](https://github.com/GPUOpen-LibrariesAndSDKs/AMF)- 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 ファイルを提供しています。

[NVIDIA Jetson](https://developer.nvidia.com/embedded-computing)

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

詳細は [NVIDIA Jetson JetPack SDK](jetson_jetpack.html) をご確認ください。

## ソースコード


## サンプルソースコード

 の examples ディレクトリ

## 動作環境

### 対応 Python バージョン

- Python 3.14
- 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 Tahoe 26 arm64
- macOS Sequoia 15 arm64
- macOS Ventura 14 arm64
- Windows 11 x86_64
- Windows Server 2025 x86_64

### 対応 Sora

[リリースノート](release.html) をご確認ください

### 対応 OpenH264

OpenH264 のバージョンは 2.6.0 をサポートします。

## 対応 Python サポート方針

リリースのタイミングで、直近の 3 バージョンをサポートします。

これは Scientific Python の [SPEC 0](https://scientific-python.org/specs/spec-0000/) を参考にしてます。

例えば 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 のライセンス契約の有無に関わらず、応答時間と問題の解決を保証しませんのでご了承ください。


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

<!-- mikan:page doc="guide" -->

# ガイド

## 実装上の注意

### 必ず connect を呼び出したら disconnect を呼び出す

Sora Python SDK は C++ で実装されています。そのため、必ず connect を呼び出したら disconnect を呼び出さないとセグメンテーション違反が発生する場合があります。

できるだけ `with` を使うようにしてください。 `with` を利用する際は `__enter__` で `connect` を呼び出したけど、接続が上手くいかない場合に、例外を発生させる時は必ず `try/except` を使ってキャッチして、 `disconnect` を呼び出してください。

### その他

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

<!-- mikan:page doc="faq" -->

# FAQ

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

[Apache License, Version 2.0](https://www.apache.org/licenses/LICENSE-2.0.html) です。

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

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

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

[Python Sora SDK サンプル集](https://github.com/shiguredo/sora-python-sdk/tree/develop/examples) にて提供しています。

## Sora Python SDK のサポートする Python のバージョンはいくつですか？

Sora Python SDK のサポートする Python のバージョンは |SPEC 0 — Minimum Supported Dependencies|  に準拠しています。


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

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

## コールバックの引数

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

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

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

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

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

[Release Release Version 2.6.0 · cisco/openh264](https://github.com/cisco/openh264/releases/tag/v2.6.0)

### 参考

- [OpenH264](http://www.openh264.org)
- [cisco/openh264: Open Source H.264 Codec](https://github.com/cisco/openh264)

## バージョン

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

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

**YYYY**
: リリースした年

**MAJOR**
: メジャーバージョンのリリース回数で 1 から始まります

**FIX**
: バグフィックスのリリースの回数で 0 から始まります

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

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

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

これは Python の [PEP 440](https://peps.python.org/pep-0440/) フォーマット準拠です。
Python SDK では a / b / rc は利用せず .devN のみを採用しています。

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

<!-- mikan:page doc="known_issues" -->

# 既知の問題

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

<!-- mikan:page doc="install" -->

# インストール

ここでは [uv](https://docs.astral.sh/uv/) の例を記載しています。

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

```console
$ uv add sora_sdk
$ uv sync
```

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

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

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

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

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

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

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

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

```console
$ sudo apt install libva2 libdrm2
```

## NVIDIA Jetson JetPack SDK

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

[NVIDIA Jetson JetPack SDK](jetson_jetpack.html) をご確認ください。

<!-- mikan:page doc="tutorial" -->

# チュートリアル

このチュートリアルでは Python Sora SDK を利用して映像を配信する方法を説明します。

## 接続先の用意

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

検証目的であれば [Sora Labo](https://sora-labo.shiguredo.jp/) を利用することで、 Sora を無料で試すことが可能です。

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

必要なのは以下の 3 つです。

- シグナリング URL
- チャネル ID
- Sora Labo や Sora Cloud の場合はアクセストークン

## uv のインストール

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

[Installation | uv](https://docs.astral.sh/uv/getting-started/installation/) を参考に uv をインストールしてください。

## プロジェクトの作成

uv init でプロジェクトを作成します。

```console
$ uv init tutorial
$ cd tutorial
```

## Python Sora SDK セットアップ

uv add して sora-sdk をインストールします。

```console
$ uv add sora-sdk
```

また、映像を配信するためにはカメラデバイスを利用する必要があります。
ここでは OpenCV を利用します。

```console
$ uv add opencv-python opencv-python-headless
```

## main.py を作成

デバイスキャプチャをして、 映像を配信する main.py を作成します。今回は音声は送りません。

signaling_url や channel_id や access_token は適切なものを指定してください。

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

## main.py 起動

```console
$ uv run main.py
```

止めるときは Ctrl+C です。

## main.py からの映像を確認

[Sora-DevTools](https://github.com/shiguredo/sora-devtools) を利用して Python SDK から送られてきている映像を確認してみてください。

<!-- mikan:page doc="examples" -->

# サンプル

**Python バージョン**: 3.13 以降

 にもサンプルを公開しています。

## 音声と映像を送受信

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

## 音声と映像を送信

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

## 音声と映像を受信

```python
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 を利用して判定

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


<!-- mikan:page doc="video_codec_preference" -->

# 映像コーデックプリファレンス

## 概要

Sora Python SDK では映像コーデックのプリファレンス指定することができます。

### Raspberry Pi V4L2 M2M (VideoCore)

**URL**: 

> **重要**
>
> Raspberry Pi では [sora-sdk](https://pypi.org/project/sora-sdk/) ではなく [sora-sdk-rpi](https://pypi.org/project/sora-sdk-rpi/) をインストールしてください

- Raspberry Pi では V4L2 M2M 経由で VideoCore に搭載されている H.264 ハードウェアアクセラレーターを利用することができます
- Raspberry Pi かつ Raspberry Pi OS でのみ利用できます
- H264 のエンコードとデコードに対応しています
- `decoder` や `encoder` に `RASPI_V4L2M2M` を指定することで V4L2 M2M を利用します

> **警告**
>
> Raspberry Pi 5 には H.264 ハードウェアアクセラレーターが搭載されていません。

### Apple Video Toolbox

**URL**: 

- Apple Video Toolbox は macOS で利用できるハードウェアアクセラレーターです
- macOS でのみ利用できます
- 対応コーデックは H.264 / H.265 です
- `decoder` や `encoder` に `INTERNAL` を指定することで Apple Video Toolbox を利用します

### Intel Video Processing Library (VPL)

**URL**: 

- 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 はドライバーが不安定なことが多いため、利用は推奨していません。

**URL**: 

- 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

**URL**: 

- NVIDIA Video Codec は NVIDIA のビデオカードで利用できるハードウェアアクセラレーターです
- Linux と Windows で利用できます。対応コーデックは VP8 / VP9 / AV1 / H.264 / H.265 です
- VP8 と VP9 はデコーダーのみ利用できます
- `decoder` や `encoder` に `NVIDIA_VIDEO_CODEC_SDK` を指定することで NVIDIA Video Codec を利用します

### Cisco OpenH264

**URL**: 

- Cisco OpenH264 はオープンソースのソフトウェア H.264 エンコーダー/デコーダーです
- 対応コーデックは H.264 です
- Sora の `openh264` に OpenH264 のパスを指定することで Cisco OpenH264 が利用できるようになります
- `decoder` や `encoder` に `CISCO_OPENH264` を指定することで Cisco OpenH264 を利用します

## 利用できるコーデック一覧を取得する

**py**: attr:Sora.get_video_codec_capability を呼び出すことで利用できるコーデック一覧を取得できます。

ただし、取得した内容をそのまま Sora の `video_codec_preference` に指定することはできません。

### VP9 のエンコーダーが正常に動作しない

- Ubuntu x86_64 と Windows x86_64 では VP9 のエンコーダーが正常に動作しない問題がわかっています

### Ubuntu で Intel VPL と NVIDIA Video Codec の両方が利用できる場合

Ubuntu 環境で Intel VPL と NVIDIA Video Codec の両方が利用できる場合、
AV1 や H.264 にどちらの機能を利用するかを指定する必要があります。

ここでは Intel VPL を優先し、 Intel VPL が利用できない場合は NVIDIA Video Codec を利用する例を示します。

```python
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 が利用できない場合はエラーにする
if intel_vpl_available or nvidia_video_codec_sdk_available:
    raise RuntimeError("Intel VPL or NVIDIA Video Codec 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 を利用するようにしている例です。

```python
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 をご確認ください

### デフォルト

**py**: attr:Sora.video_codec_preference のデフォルトは None で、この場合は SoraVideoCodecPreference.Codec の全ての `decoder` と `encoder` に SoraVideoCodecImplementation.INTERNAL が指定されます。

### 指定したコーデックのみを利用する

**py**: attr:Sora.video_codec_preference に SoraVideoCodecPreference.Codec を指定した場合は、指定したコーデックのみが利用されます。もし `decoder` だけ指定した場合は `encoder` は None が指定され、利用できません。

## Raspberry Pi V4L2 M2M

Raspberry Pi V4L2 M2M は Raspberry Pi に搭載されている VideoCore を利用した H.264 ハードウェアアクセラレーターを利用することができます。

H.264 のエンコード/デコードを利用できます。

```python
from sora_sdk import (
    Sora,
    SoraVideoCodecImplementation,
    SoraVideoCodecPreference,
    SoraVideoCodecType,
)

sora = Sora(
    video_codec_preference=SoraVideoCodecPreference(
        codecs=[
            # Raspberry Pi 5 には H.264 ハードウェアアクセラレーターが搭載されていません。
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.H264,
                encoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
                decoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
            ),
        ],
    ),
)
```

## Apple Video Toolbox

Apple Video Toolbox は macOS で利用できるハードウェアアクセラレーターです。

H.264 と H.265 のエンコーダー/デコーダーを利用できます。

```python
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 のみで利用できます。

```python
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 でセットアップする

```bash
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 でセットアップする

```bash
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* 向けのグラフィックス・ドライバーの識別と検索方法](https://www.intel.co.jp/content/www/jp/ja/support/articles/000005520/graphics.html)

```bash
lspci -k | grep -EA3 'VGA|3D|Display'
```

### 参考


## AMD AMF

**AMD AMF バージョン**: v1.4.36

```python
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 をセットアップする

- セキュアブートは切っておく必要があります

`amdgpu-install_7.0.1.70001-1_all.deb` のバージョンは [AMD の ウェブサイト](https://rocm.docs.amd.com/projects/install-on-linux/en/latest/install/quick-start.html) の `AMDGPU driver installation` を確認して最新版を利用するようにしてください。

```bash
wget https://repo.radeon.com/amdgpu-install/7.0.1/ubuntu/noble/amdgpu-install_7.0.1.70001-1_all.deb
sudo apt install ./amdgpu-install_7.0.1.70001-1_all.deb
sudo apt update
sudo apt install "linux-headers-$(uname -r)" "linux-modules-extra-$(uname -r)"
sudo apt install amdgpu-dkms
sudo amdgpu-install --usecase=graphics,amf --vulkan=pro --no-32 -y --accept-eula
```

```bash
sudo usermod -aG render $USER
sudo usermod -aG video $USER

# シェルに入り直すか以下のコマンドで反映させる
newgrp render
newgrp video
```

`vainfo` をインストールする事でドライバーが正常に動作しているかを確認することができます。

```console
$ sudo apt install vainfo
$ vainfo
libva info: VA-API version 1.20.0
libva info: Trying to open /usr/lib/x86_64-linux-gnu/dri/radeonsi_drv_video.so
libva info: Found init function __vaDriverInit_1_16
libva info: va_openDriver() returns 0
vainfo: VA-API version: 1.20 (libva 2.12.0)
vainfo: Driver version: Mesa Gallium driver 25.0.0-devel for AMD Radeon Graphics (radeonsi, phoenix, LLVM 19.1.5, DRM 3.63, 6.8.0-71-generic)
vainfo: Supported profile and entrypoints
      VAProfileH264ConstrainedBaseline:	VAEntrypointVLD
      VAProfileH264ConstrainedBaseline:	VAEntrypointEncSlice
      VAProfileH264Main               :	VAEntrypointVLD
      VAProfileH264Main               :	VAEntrypointEncSlice
      VAProfileH264High               :	VAEntrypointVLD
      VAProfileH264High               :	VAEntrypointEncSlice
      VAProfileHEVCMain               :	VAEntrypointVLD
      VAProfileHEVCMain               :	VAEntrypointEncSlice
      VAProfileHEVCMain10             :	VAEntrypointVLD
      VAProfileHEVCMain10             :	VAEntrypointEncSlice
      VAProfileJPEGBaseline           :	VAEntrypointVLD
      VAProfileVP9Profile0            :	VAEntrypointVLD
      VAProfileVP9Profile2            :	VAEntrypointVLD
      VAProfileAV1Profile0            :	VAEntrypointVLD
      VAProfileAV1Profile0            :	VAEntrypointEncSlice
      VAProfileNone                   :	VAEntrypointVideoProc
```

### Windows x86_64 で AMD AMF をセットアップする

ドライバーを AMD からダウンロードしてインストールしてください。


## NVIDIA Video Codec

**NVIDIA Video Codec SDK バージョン**: 12.0
**CUDA バージョン**: 11.8.0-1

[NVIDIA Video Codec](https://developer.nvidia.com/video-codec-sdk) は [NVIDIA](https://www.nvidia.com/ja-jp/) GPU に搭載されている NVENC/NVDEC をハードウェアアクセラレーターとして利用する SDK です。

Sora Python SDK では Ubuntu または Windows の x86_64 で NVIDIA Video Codec を利用できます。

```python
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,
            ),
        ],
    ),
)
```

### 対応コーデック

対応しているコーデックはハードウェアによって異なります。

詳細は公式ページを確認してください。

Sora Python SDK では以下のコーデックに対応しています。

- VP8- デコードのみ対応
- VP9- デコードのみ対応
- AV1- エンコード/デコード対応
- H.264- エンコード/デコード対応
- H.265- エンコード/デコード対応

### NVIDIA Video Codec で AV1 や H.265 を利用したサイマルキャストの最小解像度制限

NVIDIA Video Codec で AV1 や H.265 を利用したサイマルキャストでは、
最小解像度 128x96 を下回った場合、ハードウェアによる制限でエンコードに失敗します。

そのため、NVIDIA Video Codec で AV1 または H.265 を利用したサイマルキャスト利用する場合は、
最小解像度が 128x96 以上になるよう、
`simulcast_encodings` の `scaleResolutionDownBy` または `scaleResolutionDownTo` を指定するようにしてください。

### Ubuntu x86_64 で NVIDIA Video Codec をセットアップする

```bash
# デスクトップの場合
sudo ubuntu-drivers list   

# サーバーの場合
sudo ubuntu-drivers list --gpgpu
```

```bash
# これだけでインストールできます
sudo ubuntu-drivers install
```

#### 実行の参考

```console
# インストールしたドライバーを確認できます
$ 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 を利用できます。

```python
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 をセットアップする

```bash
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 をセットアップする

```bash
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 をセットアップする

```bash
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 をセットアップする

```powershell
$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
```

<!-- mikan:page doc="webrtc_encoded_transform" -->

# WebRTC Encoded Transform

> **警告**
>
> この機能は実験的機能のため、正式版では仕様が変更される可能性があります

## 概要

[WebRTC Encoded Transform](https://www.w3.org/TR/webrtc-encoded-transform/) とは WebRTC の送信前や受信後のタイミングで、エンコードされた音声や映像フレームを直接書き換える、ブラウザ WebRTC API の拡張的な機能として提供されている仕組みです。

Sora Python SDK では [WebRTC Encoded Transform](https://www.w3.org/TR/webrtc-encoded-transform/) を SDK の一部として提供しています。

音声や映像、データを送るだけの通常の利用では必要ありません。主に音声や映像と **同時に** 何かしらのデータを送りたい場合などに利用することを目的としています。


WebRTC Encoded Transform に付いては W3C の仕様よりも MDN の記事 [Using WebRTC Encoded Transforms](https://developer.mozilla.org/en-US/docs/Web/API/WebRTC_API/Using_Encoded_Transforms) がわかりやすいです。

## 取得できるフレーム

そもそも WebRTC では音声や映像は RTP というプロトコルを利用し、それを暗号化された SRTP というプロトコルを利用します。
さらに RTP は 1200 バイト程度までしか 1 つのパケットで送れないため、より大きなサイズの場合は分割して送信します。

WebRTC Encoded Transform ではこの送信前の **分割した RTP にしていない状態** と、受信後の **RTP を結合した状態** のエンコード済みの音声や映像のフレームを取得することができます。

## 利用方法

### 送信時

送信する音声や映像のフレームを変換するには SoraAudioFrameTransformer または SoraVideoFrameTransformer を利用します。


```python
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 を利用します。

```python
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 の時のみ利用できます

```python
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)
```

<!-- mikan:page doc="jetson_jetpack" -->

# NVIDIA Jetson JetPack SDK

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

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


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

## リリースノート

### 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


### uv を利用する場合

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

```console
$ uv pin 3.10
$ uv pip install 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 サポートまでご相談ください。

<!-- mikan:page doc="audio_device" -->

# 音声デバイス

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

代わりに、  PortAudio の Python バインディングである [sounddevice](https://pypi.org/project/sounddevice) を使用することで、
音声デバイスのキャプチャや再生などを行うことができます。


## sounddevice のインストール

```console
$ uv add sounddevice
$ uv sync
```

### macOS

```console
$ brew install portaudio
```

### Ubuntu

```console
$ sudo apt install libportaudio2
```

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

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

```python
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` を利用する事で、音声デバイスから音声キャプチャして音声を表示することができます。

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

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

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

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

<!-- mikan:page doc="video_device" -->

# 映像デバイス

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

代わりに、OpenCV の Python バインディングを使用することで、
映像デバイスのキャプチャや表示などを行うことができます。

## OpenCV のインストール

```console
$ uv add opencv-python
$ uv sync
```


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

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

```python
import cv2


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

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

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

    return available_video_devices


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

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

[Capture/Webcam - FFmpeg](https://trac.ffmpeg.org/wiki/Capture/Webcam)

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

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

```python
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 に送信することができます。

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

<!-- mikan:page doc="raspberry_pi" -->

# Sora Python SDK for Raspberry Pi

Sora Python SDK は Raspberry Pi OS (64 bit) 向けに通常とは異なるパッケージを提供しています。

これは Raspberry Pi の H.264 ハードウェアアクセラレーターや libcamera を利用するためです。

## インストール

```bash
# Raspberry Pi OS (64 bit) 向けの sora-sdk をインストールする
uv add sora-sdk-rpi
```

## 利用方法

Raspberry Pi OS (64 bit) 向けの sora-sdk-rpi をインストールした後は、通常の sora-sdk と同じように利用できます。

### V4L2-M2M 経由でのハードウェアアクセラレーターの利用

> **注釈**
>
> Raspberry Pi 5 には H.264 ハードウェアアクセラレーターが搭載されていません。

Sora Python SDK for Raspberry Pi は Raspberry Pi に搭載されている H.264 ハードウェアアクセラレーターを利用することができます。

```python
video_codec_preference = SoraVideoCodecPreference(
    codecs=[
        # Raspberry Pi 5 には H.264 ハードウェアアクセラレーターが搭載されていません。
        SoraVideoCodecPreference.Codec(
            # H.264 を指定します
            type=SoraVideoCodecType.H264,
            # encoder と decoder に V4L2 M2M を指定します
            encoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
            decoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
        ),
    ],
)
```

### libcamera の利用

Raspberry Pi Camera を利用する場合は libcamera を利用する必要があります。
Sora Python SDK for Raspberry Pi では libcamera を利用するためのライブラリを提供しています。

```python
sora = Sora(
    video_codec_preference=video_codec_preference,
    # native_frame_output=True を指定する場合 force_i420_conversion=False を指定します
    force_i420_conversion=False,
)

# libcamera を利用するには create_libcamera_source
video_source = sora.create_libcamera_source(
    width=1280,
    height=720,
    fps=30,
    # native_frame_output=True を指定すると libcamera から直接 I420 フォーマットのフレームが取得できます。
    native_frame_output=True,
    # libcamera のオートフォーカスを有効にする。
    # libcamera-controls をタプル形式で指定出来ます
    controls=[("AsFocusMode", "1")],
)
```

## libcamera コントロール

libcamera は 75 以上のカメラパラメーターを提供しています。 Sora Python SDK for Raspberry Pi の `create_libcamera_source` では `controls` で libcamera コントロールを指定する事が出来ます。

### 指定方法

- `[tuple[str, str]]` の形式で指定します
- `[("AsFocusMode", "1"), ("AeEnable", "1")]` のように複数指定できます
- 値は数字やブーリアンでも文字列で指定してください
- `AsFocusMode` のように数値を指定する場合も `"1"` と文字列で指定してください

### 対応している値の形式

#### 基本型

- bool: `"0"` または `"1"`、 `"true"` または `"false"````python
  [("AeEnable", "1")]
  [("AeEnable", "true")]
  ```
- int32: 整数値```python
  [("ExposureTime", "10000")]
  ```
- int64: 長整数値```python
  [("FrameDuration", "33333")]
  ```
- float: 小数値```python
  [("AnalogueGain", "2.0")]
  [("Brightness", "-0.5")]
  ```
- enum: 文字列または数値（主要な enum のみ文字列対応）

  ```python
  [("AfMode", "Continuous")]
  # または
  [("AfMode", "2")]
  ```

  文字列対応済みの enum:

  - `AfMode`, `AfRange`, `AfSpeed`
  - `AeMeteringMode`
  - `AwbMode`
  - `ExposureTimeMode`, `AnalogueGainMode`

#### 複合型

- 配列: カンマ区切り

  ```python
  [("ColourGains", "1.5,2.0")]
  [("FrameDurationLimits", "33333,33333")]
  ```

  対応済み: `float[]`, `int32[]`, `int64[]`
- 矩形（Rectangle）: x,y,width,height```python
  [("ScalerCrop", "100,100,640,480")]
  ```
- 複数矩形: セミコロン区切り```python
  [("AfWindows", "100,100,200,200;300,300,200,200")]
  ```
  > **注釈**
  >
  > Rectangle 配列の場合、セミコロンで複数の矩形を区切ります
- マトリクス: カンマ区切り（行優先）（未対応）```python
  [("ColourCorrectionMatrix", "1.0,0,0,0,1.0,0,0,0,1.0")]
  ```

#### 未対応の型

- Size 型
- Point 型
- マトリクス型（3x3 など）
- その他の複雑な構造体

### 主要なコントロール

#### 露出制御（AE: Auto Exposure）

* - コントロール
  - 型
  - 説明
  - 値の例
* - AeEnable
  - bool
  - 自動露出の有効/無効
  - "0", "1"
* - ExposureTime
  - int32
  - 露出時間（マイクロ秒）
  - 10000 (1/100秒)
* - AnalogueGain
  - float
  - アナログゲイン（1.0以上）
  - 2.0
* - ExposureTimeMode
  - enum
  - 露出時間モード
  - Auto, Manual
* - AnalogueGainMode
  - enum
  - ゲインモード
  - Auto, Manual
* - AeMeteringMode
  - enum
  - 測光モード
  - CentreWeighted, Spot, Matrix
* - ExposureValue
  - float
  - EV補正値
  - -2.0 〜 2.0

#### オートフォーカス（AF）

* - コントロール
  - 型
  - 説明
  - 値の例
* - AfMode
  - enum
  - AFモード
  - Manual, Auto, Continuous
* - AfRange
  - enum
  - フォーカス範囲
  - Normal, Macro, Full
* - AfSpeed
  - enum
  - フォーカス速度
  - Normal, Fast
* - AfTrigger
  - enum
  - AFトリガー
  - Start, Cancel
* - AfWindows
  - Rectangle[]
  - AFエリア
  - "256,192,512,384"
* - LensPosition
  - float
  - レンズ位置（ジオプター）
  - 2.0

#### ホワイトバランス（AWB）

* - コントロール
  - 型
  - 説明
  - 値の例
* - AwbEnable
  - bool
  - AWBの有効/無効
  - "0", "1"
* - AwbMode
  - enum
  - AWBモード
  - Auto, Daylight, Cloudy, Tungsten
* - ColourTemperature
  - int32
  - 色温度（ケルビン）
  - 5500
* - ColourGains
  - float[2]
  - 赤・青ゲイン
  - 1.5,2.0

#### 画質調整

* - コントロール
  - 型
  - 説明
  - 値の例
* - Brightness
  - float
  - 明るさ
  - -1.0 〜 1.0
* - Contrast
  - float
  - コントラスト
  - 1.0（標準）
* - Saturation
  - float
  - 彩度
  - 1.0（標準）, 0.0（モノクロ）
* - Sharpness
  - float
  - シャープネス
  - 0.0 〜 10.0
* - Gamma
  - float
  - ガンマ値
  - 2.2（標準）

#### フレームレート制御

* - コントロール
  - 型
  - 説明
  - 値の例
* - FrameDurationLimits
  - int64[2]
  - フレーム時間の最小/最大（マイクロ秒）
  - 33333,33333 (30fps固定)

### enum 値一覧

#### AfMode（オートフォーカスモード）

- 手動フォーカス- `Manual` または `"0"`
- シングルAF（一度フォーカスして停止）- `Auto` または `"1"`
- コンティニュアスAF- `Continuous` または `"2"`

#### AfRange（フォーカス範囲）

- 通常範囲- `Normal` または `"0"`
- マクロ（接写）- `Macro` または `"1"`
- フルレンジ- `Full` または `"2"`

#### AfSpeed（フォーカス速度）

- 通常速度- `Normal` または `"0"`
- 高速- `Fast` または `"1"`

#### ExposureTimeMode / AnalogueGainMode

- 自動- `Auto` または `"0"`
- 手動- `Manual` または `"1"`

#### AeMeteringMode（測光モード）

- 中央重点測光- `CentreWeighted` または `"0"`
- スポット測光- `Spot` または `"1"`
- マトリックス測光- `Matrix` または `"2"`

#### AwbMode（ホワイトバランスモード）

- 自動- `Auto` または `"0"`
- 白熱灯- `Incandescent` または `"1"`
- タングステン- `Tungsten` または `"2"`
- 蛍光灯- `Fluorescent` または `"3"`
- 屋内- `Indoor` または `"4"`
- 昼光- `Daylight` または `"5"`
- 曇天- `Cloudy` または `"6"`

#### HdrMode（HDRモード）

- 無効- `Off` または `"0"`
- 複数露出（未合成）- `MultiExposureUnmerged` または `"1"`
- 複数露出（合成）- `MultiExposure` または `"2"`
- 単一露出HDR- `SingleExposure` または `"3"`
- ナイトモード- `Night` または `"4"`

### controls 指定例

#### 例1: 明るい屋外での撮影設定

```python
controls=[
    ("AeEnable", "1"),
    ("AeExposureMode", "Short"),
    ("AwbMode", "Daylight"),
    ("Contrast", "1.2"),
]
```

#### 例2: 暗所での撮影設定

```python
controls=[
    ("ExposureTimeMode", "Manual"),
    ("ExposureTime", "50000"),
    ("AnalogueGain", "8.0"),
    ("NoiseReductionMode", "HighQuality"),
]
```

#### 例3: マクロ撮影設定

```python
controls=[
    ("AfMode", "Continuous"),
    ("AfRange", "Macro"),
    ("AfSpeed", "Normal"),
    ("Sharpness", "2.0"),
]
```

#### 例4: フレームレート固定（30fps）

```python
controls=[
    ("FrameDurationLimits", "33333,33333"),
]
```

#### 例5: 手動設定での完全制御

```python
controls=[
    ("AeEnable", "0"),
    ("AwbEnable", "0"),
    ("ExposureTime", "20000"),
    ("AnalogueGain", "4.0"),
    ("ColourGains", "1.8,1.5"),
    ("AfMode", "Manual"),
    ("LensPosition", "2.0"),
]
```

## サンプル

`V4L2 M2M` で H.264 を指定し、 `libcamera` を利用したサンプルコードです。

```python
from sora_sdk import (
    Sora,
    SoraVideoCodecImplementation,
    SoraVideoCodecPreference,
    SoraVideoCodecType,
)


def main():
    signaling_urls = ["wss://sora.example.com/signaling"]
    video_codec_preference = SoraVideoCodecPreference(
        codecs=[
            # Raspberry Pi 5 には H.264 ハードウェアアクセラレーターが搭載されていません。
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.H264,
                encoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
                decoder=SoraVideoCodecImplementation.RASPI_V4L2M2M,
            ),
        ],
    )
    sora = Sora(
        video_codec_preference=video_codec_preference,
        # native_frame_output=True を指定する場合は force_i420_conversion=False を指定する必要があります。
        force_i420_conversion=False,
    )

    video_source = sora.create_libcamera_source(
        width=1280,
        height=720,
        fps=30,
        # native_frame_output=True を指定すると、libcamera から直接 I420 フォーマットのフレームが取得できます。
        native_frame_output=True,
        # libcamera のオートフォーカスを有効にする。
        controls=[("AsFocusMode", "1")],
    )

    # V4L2 M2M による H.264 と libcamera を利用して Sora に接続して映像を送信する。
    sora.create_connection(
        signaling_urls=signaling_urls,
        role="sendonly",
        channel_id="raspberry-pi",
        video=True,
        video_codec_type="H264",
        video_source=video_source,
    )
```

<!-- mikan:page doc="sora_sdk" -->

# sora_sdk モジュール


## 概要

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

## 音声と映像の扱い

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

## NDArray 型と LikeArray 型の注意

Sora Python SDK では音声や映像のデータを取り扱う際に NumPy の NDArray 型を利用します。

ただし [nanobind](https://nanobind.readthedocs.io/en/latest/) の制約により、
`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 のログ出力を指定できます。

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

.. code-block::

   import sora_sdk

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

## 利用できるコーデック一覧の取得


## SoraVideoCodecCapability オブジェクト

.. py:method:: to_json()

   :rtype: str

   JSON 形式の文字列を返します。

.. py:attribute:: engines

   :rtype: List[:py:class:`SoraVideoCodecCapability.Engine`]

   :py:class:`SoraVideoCodecCapability.Engine` のリストを返します。


.. py:class:: Codec

.. py:class:: Engine

.. py:class:: Parameters

## Sora オブジェクト

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

 :param video_codec_preference: :py:class:`SoraVideoCodecPreference` を指定します。何も指定しない場合は None で全てのコーデックは :py:class:`SoraVideoCodecImplementation.INTERNAL` を利用します。
 :type: :py:class:`SoraVideoCodecPreference` | None

 :param openh264: OpenH264 のパスを指定します。
 :type: str | None

 .. py:method:: create_audio_source(channels, sample_rate)
    
    :param int channels: チャンネル
    :param int sample_rate: サンプルレート
    :rtype: sora_sdk.SoraAudioSource

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

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

 .. py:method:: create_video_source()
    
    :rtype: sora_sdk.SoraVideoSource

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

.. py:method:: 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, simulcast_request_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)

   :param List[str] signaling_urls: シグナリング URLのリスト
   :param str role: ロール
   :param str channel_id: チャネル ID
   :param Optional[str] client_id: クライアント ID
   :param Optional[str] bundle_id: バンドル ID
   :param Optional[object] metadata: 認証メタデータ
   :param Optional[object] signaling_notify_metadata: シグナリング通知メタデータ
   :param Optional[sora_sdk.SoraTrackInterface] audio_source: 音声ソース
   :param Optional[sora_sdk.SoraTrackInterface] video_source: 映像ソース
   :param Optional[sora_sdk.SoraAudioFrameTransformer] audio_frame_transformer: 音声フレーム変換器
   :param Optional[sora_sdk.SoraVideoFrameTransformer] video_frame_transformer: 映像フレーム変換器
   :param Optional[bool] audio: 音声
   :param Optional[bool] video: 映像
   :param Optional[str] audio_codec_type: 音声コーデックタイプ
   :param Optional[str] video_codec_type: 映像コーデックタイプ
   :param Optional[int] audio_bit_rate: 音声ビットレート
   :param Optional[int] video_bit_rate: 映像ビットレート
   :param Optional[object] audio_opus_params: 音声 Opus 設定指定
   :param Optional[object] video_vp9_params: 映像 VP9 設定指定
   :param Optional[object] video_av1_params: 映像 AV1 設定指定
   :param Optional[object] video_h264_params: 映像 H.264 設定指定
   :param Optional[object] video_h265_params: 映像 H.265 設定指定
   :param Optional[bool] simulcast: サイマルキャスト
   :param Optional[bool] spotlight: スポットライト
   :param Optional[int] spotlight_number: スポットライトナンバー
   :param Optional[str] simulcast_rid: サイマルキャスト RID
   :param Optional[str] simulcast_request_rid: サイマルキャスト RID リクエスト
   :param Optional[str] spotlight_focus_rid: スポットライトフォーカス RID
   :param Optional[str] spotlight_unfocus_rid: スポットライトアンフォーカス RID
   :param Optional[object] forwarding_filter: 転送フィルター
   :param Optional[list[object]] forwarding_filters: 転送フィルターのリスト
   :param Optional[object] data_channels: データチャネル
   :param Optional[bool] data_channel_signaling: データチャネルシグナリング
   :param Optional[bool] ignore_disconnect_websocket: ウェブソケット切断無視
   :param Optional[int] data_channel_signaling_timeout: データチャネルシグナリングタイムアウト
   :param Optional[int] disconnect_wait_timeout: 切断待ちタイムアウト
   :param Optional[int] websocket_close_timeout: ウェブソケットクローズタイムアウト
   :param Optional[int] websocket_connection_timeout: ウェブソケット接続タイムアウト
   :param Optional[str] audio_streaming_language_code: 音声ストリーミング言語コード
   :param Optional[bool] insecure: サーバー証明書のチェックを無視する
   :param Optional[bytes] client_cert: クライアント証明書
   :param Optional[bytes] client_key: クライアント証明書シークレットキー
   :param Optional[bytes] ca_cert: サーバー証明書のチェックを行う CA 証明書
   :param Optional[str] proxy_url: Proxy URL
   :param Optional[str] proxy_username: Proxy ユーザ名
   :param Optional[str] proxy_password: Proxy パスワード
   :param Optional[str] proxy_agent: Proxy エージェント
   :param Optional[SoraDegradationPreference] degradation_preference: デグラデーション優先度
   :rtype: sora_sdk.SoraConnection

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

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

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

```python
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()` のように読み込んだバイナリを渡してください。

```python
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 にアクセストークンを指定してください。

```python
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 を指定してください。

```python
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 オブジェクト

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

:py:meth:`Sora.create_connection` から生成します。

.. py:method:: connect()
   
   :return: None

   Sora に接続します。

   .. important:: もしスレッドを起動して audio/video の入力ループを開始する場合は、かならず connect() を呼び出す **前に** 開始してください。

.. py:method:: disconnect()

   :return: None

   Sora から切断します。 

   .. important:: connect() を呼んだ場合は、切断時または例外が上がった場合でも disconnect() を呼んでください。

.. py:method:: get_stats()

   :return: W3C 準拠の WebRTC 統計情報。
   :rtype: str

   クライアントの WebRTC 統計情報を取得します。
   形式は `Identifiers for WebRTC's Statistics API <https://w3c.github.io/webrtc-stats/>`_ に準拠しています。

.. literalinclude:: ./sora_sdk/stats.py
   :language: python

.. py:attribute:: on_disconnect(error_code, message)
   
   :type: Callable[[sora_sdk.SoraSignalingErrorCode, str], None]

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

.. literalinclude:: ./sora_sdk/on_disconnect.py
   :language: python


.. py:method:: send_data_channel(label, data)
   
   :param str label: データチャンネルのラベル。
   :param bytes data: 送信するデータ。
   :return: 成功した場合は True 、それ以外の場合は False 。
   :rtype: bool

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

.. py:attribute:: on_data_channel(label)
   
   :type: Callable[[label: str], None]

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

.. py:attribute:: on_message(label, data)
   
   :type: Callable[[label: str, data: bytes], None]

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

.. literalinclude:: ./sora_sdk/on_message.py
   :language: python


.. py:attribute:: on_notify(message)
   
   :type: Callable[[str], None]

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

.. literalinclude:: ./sora_sdk/on_notify.py
   :language: python

.. py:attribute:: on_push(message)
   
   :type: Callable[[message: str], None]

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

.. py:attribute:: on_set_offer
   
   :type: Callable[[message: str], None]

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

.. py:attribute:: on_track(track)
   
   :type: Callable[[track: sora_sdk.SoraMediaTrack], None]

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

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


.. py:attribute:: on_switched(message)
   
   :type: Callable[[message: str], None]

   シグナリングが DataChannel に切り替わったタイミングで呼ばれるコールバックです。受信したメッセージを引数に取ります。
   主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。

.. py:attribute:: 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"`` は :py:attr:`SoraConnection.on_switched` を利用してください。

   ``"type": "ping"`` と ``"type": "pong"`` は取得できません。


.. literalinclude:: ./sora_sdk/on_signaling_message.py
   :language: python

.. py:attribute:: on_ws_close(code, reason)

   :type: Callable[[int, str], None]

   WebSocket 切断時のコールバックです。
   SDK 側で閉じた場合は reason には ``"SELF-CLOSED"`` が入ります。
   主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。

.. literalinclude:: ./sora_sdk/on_ws_close.py
   :language: python

## SoraAudioSource オブジェクト

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

:py:meth:`Sora.create_audio_source` から生成します。

.. py:method:: on_data(ndarray)
   
   :param ``ndarray[dtype=int16, shape=(*, *), order='C', device='cpu']`` ndarray: チャンネルごとのサンプル数 x チャンネル数 になっている音声データ。
   :rtype: None
   
   Sora に送る音声データを渡します。

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

.. .. py:method:: on_data(ndarray, timestamp)
..    :no-index:

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

..    .. danger:: このメソッドは非推奨です。
..    
..    Sora に送る音声データを渡します。


.. py:method:: on_data(data, samples_per_channel)
   :no-index:
   
   :param int data: 16bit PCM データのポインタ。
   :param int samples_per_channel: 送信するチャンネル毎のサンプル数。
   :rtype: None
   
   Sora に送る音声データを渡します。

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

.. .. py:method:: on_data(data, samples_per_channel, timestamp)
..    :no-index:
..    
..    :param int data: 16bit PCM データのポインタ。
..    :param int samples_per_channel: 送信するチャンネル毎のサンプル数
..    :param float timestamp: タイムスタンプ
..    :rtype: None
..    
..    .. danger:: このメソッドは非推奨です。
..    
..    Sora に送る音声データを渡します。

## SoraAudioSink オブジェクト

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

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

.. py:method:: read(frames=0, timeout=1)
   
   :param int frames: 受け取るチャンネルごとのサンプル数。0 を指定した場合には、受信済みのすべてのサンプルを返します。デフォルトは 0。
   :param float timeout: 溜まっているサンプル数が frames で指定した数を満たさない場合の待ち時間。秒単位で指定します。デフォルトは 1。
   :return: タプルでインデックス 0 には bool で成否を、成功した場合のみインデックス 1 に ``ndarray[dtype=int16, shape=(*, *)]`` で チャンネルごとのサンプル数 x チャンネル数 になっている音声データを返します。
   :rtype: tuple

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

.. py:attribute:: on_data
   
   :type: ``Callable[[numpy.ndarray[dtype=int16, shape=(*, *)]], None]``

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

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

.. py:attribute:: on_format(sample_rate, number_of_channels)
   
   :type: Callable[[int, int], None]

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

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

```python
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 を利用してください。

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

## SoraVideoSource オブジェクト

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

:py:meth:`Sora.create_video_source` から生成します。

.. py:method:: on_captured(ndarray)
   
   :param ``ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']`` ndarray: H x W x BGR になっているフレームデータ。
   :rtype: None

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

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

.. .. py:method:: on_captured(ndarray, timestamp)
..    :no-index:
..    
..    :param ``ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']`` ndarray: H x W x BGR になっているフレームデータ。
..    :param float timestamp: time.time() で取得できるエポック秒で表されるフレームのタイムスタンプ。
..    :rtype: None
..    
..    .. danger:: このメソッドは非推奨です。

..    timestamp 引数で渡されたタイムスタンプでフレームを送信します。
..    
..    フレームのタイムスタンプを指定できるようにするため用意したオーバーロードです。
..    
..    timestamp が映像になるように一定の時間差がない場合、受信側で正しく表示されない場合があります。
..    
..    表示側で音声データの timestamp と同期を取るため遅延が発生する場合があります。

.. .. py:method:: on_captured(ndarray, timestamp_us)
..    :no-index:
..    
..    :param ``ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']`` ndarray: H x W x BGR になっているフレームデータ。
..    :param int timestamp_us: マイクロ秒単位の整数で表されるフレームのタイムスタンプ。
..    :rtype: None
..    
..    .. danger:: このメソッドは非推奨です。
..    
..    timestamp_us 引数で渡されたマイクロ秒精度の整数で表されるタイムスタンプでフレームを送信します。
..    
..    libWebRTC のタイムスタンプはマイクロ秒精度のため用意したオーバーロードです。
..    
..    timestamp が映像になるように一定の時間差がない場合、受信側で正しく表示されない場合があります。
..    
..    表示側で音声データの timestamp と同期を取るため遅延が発生する場合があります。

## SoraTrackInterface オブジェクト

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

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

.. py:method:: set_enabled(enable)
   
   :param bool enable: 有効または無効を表すブール値。
   :rtype: bool
   
   トラックの有効/無効を設定します。

.. py:attribute:: enabled
   
   :type: bool
   :readonly: True

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

.. py:attribute:: id
   
   :type: str
   :readonly: True

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

.. py:attribute:: kind
   
   :type: str
   :readonly: True

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

.. py:attribute:: state
   
   :type: sora_sdk.SoraTrackState
   :readonly: True

   トラックの状態。

## SoraAudioFrame オブジェクト

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

.. py:method:: data()
   
   :return: numpy.ndarray で サンプル数 x チャンネル数 になっている音声データ。
   :rtype: ``numpy.ndarray[dtype=int16, shape=(*, *)]``

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

.. py:attribute:: samples_per_channel
   
   :type: int

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

.. py:attribute:: num_channels
   
   :type: int

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

.. py:attribute:: sample_rate_hz
   
   :type: int

   サンプリングレート。

.. py:attribute:: absolute_capture_timestamp_ms
   
   :type: int

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

## SoraAudioStreamSink オブジェクト

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

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

.. py:attribute:: on_frame
   
   :type: Callable[[track: sora_sdk.SoraAudioFrame], None]

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

```python
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 オブジェクト

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

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

.. py:attribute:: stream_id

   :type: str

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

.. py:method:: set_frame_transformer(transformer)

   :param sora_sdk.SoraFrameTransformer transformer: フレームを変換する :py:class:`SoraFrameTransformer` オブジェクト。
   :rtype: None

   フレームを変換する :py:class:`SoraFrameTransformer` オブジェクトを設定します。

## SoraVAD オブジェクト

.. py:method:: analyze(frame)
   
   :param sora_sdk.SoraAudioFrame frame: 音声である確率を求める :py:class:`SoraAudioFrame` オブジェクト。
   :return: 0 - 1 で表される音声である確率。
   :rtype: int

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

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

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

```python
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 オブジェクト

.. py:method:: data()
   
   :return: H x W x BGR になっているフレームデータ。
   :rtype: ``numpy.ndarray[dtype=uint8, shape=(*, *, 3)]``

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

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

## SoraVideoSink オブジェクト

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

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

.. py:attribute:: on_frame
   
   :type: Callable[[track: sora_sdk.SoraVideoFrame], None]

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

   **このコールバック関数内では重い処理は行わないでください。** サンプルを参考に queue を利用するなどの対応を推奨します。
   
   この関数はメインスレッドから呼び出されないため、関数内で OpenCV の cv2.imshow を実行しても macOS の場合は表示されません。

```python
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 オブジェクト

*バージョン 2025.1.0 で追加。*

映像コーデックの種類を表す列挙型です。

.. py:attribute:: VP8
   :type: SoraVideoCodecType
   :value: 1

   VP8 コーデック

.. py:attribute:: VP9 
   :type: SoraVideoCodecType
   :value: 2

   VP9 コーデック

.. py:attribute:: AV1
   :type: SoraVideoCodecType
   :value: 3

   AV1 コーデック

.. py:attribute:: H264
   :type: SoraVideoCodecType
   :value: 4

   H.264 コーデック

.. py:attribute:: H265
   :type: SoraVideoCodecType
   :value: 5

   H.265 コーデック

## SoraVideoCodecPreference オブジェクト

*バージョン 2025.1.0 で追加。*

映像コーデックのエンコード/デコードを細かく指定するためのオブジェクトです。
未指定の場合は全てのコーデックで :py:class:`SoraVideoCodecImplementation.INTERNAL` を利用します。

.. warning::
   
   SoraVideoCodecPreference を指定した場合は **指定したコーデックのみ** を利用します。
   H.264 のみを指定した場合は H.264 しか利用できなくなります。

.. py:method:: Codec(type, encoder=None, decoder=None)

   :param SoraVideoCodecType type: コーデックの種類。
   :param SoraVideoCodecImplementation encoder: エンコードに利用する実装。
   :param SoraVideoCodecImplementation decoder: デコードに利用する実装。

   コーデックの種類とエンコード/デコードに利用する実装を指定します。
   encoder や decoder を指定しない場合は None が指定され、利用できなくなります。

```python
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 を利用する
            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 オブジェクト

*バージョン 2025.1.0 で追加。*

映像コーデックの実装を表す列挙型です。

.. py:attribute:: INTERNAL
   :type: SoraVideoCodecImplementation
   :value: 0

   INTERNAL は Sora Python SDK がベースにしている ``libwebrtc`` に含まれるデコーダーやエンコーダーを利用します。
   端末のハードウェアアクセラレーターが利用できる場合、自動的に利用します。

   macOS で ``INTERNAL`` を指定し、コーデックに H.264 や H.265 を指定した場合は Apple Video Toolbox を利用します。

.. py:attribute:: CISCO_OPENH264
   :type: SoraVideoCodecImplementation
   :value: 1

   Cisco OpenH264 を利用します。OpenH264 を利用する場合は ``openh264`` に OpenH264 のパスを指定してください。

.. py:attribute:: INTEL_VPL
   :type: SoraVideoCodecImplementation
   :value: 2

   Intel VPL によるハードウェアアクセラレーターを利用します。
   利用できるコーデックは AV1 / H.264 / H.265 です。 

.. py:attribute:: NVIDIA_VIDEO_CODEC_SDK
   :type: SoraVideoCodecImplementation
   :value: 3

   NVIDIA Video Codec によるハードウェアアクセラレーターを利用します。
   利用できるコーデックは VP9 / H.264 / H.265 です。

## SoraTransformableFrame オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:attribute:: direction

   :rtype: sora_sdk.sora_sdk_ext.SoraTransformableFrameDirection

.. py:attribute:: mime_type

   :rtype: str

.. py:attribute:: payload_type

   :rtype: int

.. py:attribute:: ssrc

   :rtype: int

.. py:attribute:: rtp_timestamp

   :rtype: int

## SoraTransformableAudioFrame オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:method:: get_data()

   :rtype: numpy.ndarray[dtype=uint8, shape=(*), writable=False]

   映像データを取得します。

.. py:method:: set_data(data)

   :param numpy.ndarray[dtype=uint8, shape=(*), order='C', device='cpu', writable=False] data: 
   :rtype: None

   映像データを設定します。

.. py:attribute:: absolute_capture_timestamp

   :rtype: int | None

   Abs-Capture-Time RTP 拡張の値が含まれます。

.. py:attribute:: audio_level

   :rtype: int | None

   Audio-Level RTP 拡張の値が含まれます。

.. py:attribute:: contributing_sources

   :rtype: numpy.ndarray[dtype=uint32, shape=(*), writable=False]

   RTP CSRC のリストが含まれます。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber

.. py:attribute:: receive_time

   :rtype: int | None

.. py:attribute:: sequence_number

   :rtype: int | None

   RTP シーケンス番号が含まれます。受信する音声パケットにのみ含まれます。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber

.. py:attribute:: type

   :rtype: sora_sdk.sora_sdk_ext.SoraTransformableAudioFrameType

.. py:attribute:: direction

   :rtype: sora_sdk.sora_sdk_ext.SoraTransformableFrameDirection

.. py:attribute:: mine_type

   :rtype: str

   音声の MIME タイプが含まれます。多くの場合 `audio/opus` です。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-mimetype

.. py:attribute:: payload_type

   :rtype: int

   RTP ペイロードタイプが含まれます。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-payloadtype

.. py:attribute:: ssrc

   :rtype: int

   RTP SSRC が含まれます。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-sequencenumber

.. py:attribute:: rtp_timestamp

   :rtype: int

   RTP タイムスタンプが含まれます。

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedaudioframemetadata-rtptimestamp

## SoraTransformableVideoFrame オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:method:: get_data()

   :rtype: numpy.ndarray[dtype=uint8, shape=(*), writable=False]

   映像データを取得します。

.. py:method:: set_data(data)

   :param numpy.ndarray[dtype=uint8, shape=(*), order='C', device='cpu', writable=False] data: 
   :rtype: None

   映像データを設定します。

.. py:attribute:: contributing_sources

   :rtype: numpy.ndarray[dtype=uint32, shape=(*), writable=False]

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-contributingsources

.. py:attribute:: frame_dependencies

   :rtype: numpy.ndarray[dtype=int64, shape=(*), writable=False]

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-dependencies

.. py:attribute:: frame_id

   :rtype: int | None

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-frameid

.. py:attribute:: is_key_frame

   :rtype: bool

   キーフレームかどうかが含まれます。

.. py:attribute:: spatial_index

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-spatialindex

.. py:attribute:: temporal_index

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-temporalindex   

.. py:attribute:: height

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-height

.. py:attribute:: width

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-width

.. py:attribute:: direction

   :rtype: sora_sdk.sora_sdk_ext.SoraTransformableFrameDirection

.. py:attribute:: mine_type

   :rtype: str

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-mimetype

.. py:attribute:: payload_type

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp

.. py:attribute:: ssrc

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp

.. py:attribute:: rtp_timestamp

   :rtype: int

   https://www.w3.org/TR/webrtc-encoded-transform/#dom-rtcencodedvideoframemetadata-rtptimestamp

## SoraFrameTransformer オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:attribute:: on_transform

   :type: Callable[[sora_sdk.SoraTransformableAudioFrame], None]

.. py:method:: enqueue(frame)

   :param sora_sdk.SoraTransformableFrame frame: 変換する :py:class:`SoraTransformableFrame` オブジェクト。
   :rtype: None

.. py:method:: start_short_circuiting()

   :rtype: None

## SoraAudioFrameTransformer オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:attribute:: on_transform

   :type: Callable[[sora_sdk.SoraTransformableAudioFrame], None]

.. py:method:: enqueue(frame)

   :param sora_sdk.SoraTransformableAudioFrame frame: 変換する :py:class:`SoraTransformableAudioFrame` オブジェクト。
   :rtype: None

.. py:method:: start_short_circuiting()

   :rtype: None

## SoraVideoFrameTransformer オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:attribute:: on_transform

   :type: Callable[[sora_sdk.SoraTransformableVideoFrame], None]

.. py:method:: enqueue(frame)

   :param sora_sdk.SoraTransformableVideoFrame frame: 変換する :py:class:`SoraTransformableVideoFrame` オブジェクト。
   :rtype: None

.. py:method:: start_short_circuiting()

   :rtype: None

## SoraSignalingType オブジェクト

.. py:attribute:: WEBSOCKET

   :type: sora_sdk.SoraSignalingType
   :readonly: True

   WebSocket を利用することを示す定数です。

.. py:attribute:: DATACHANNEL

   :type: sora_sdk.SoraSignalingType
   :readonly: True

   DataChannel を利用することを示す定数です。

## SoraDegradationPreference オブジェクト

*バージョン 2025.1.0 で追加。*

.. py:attribute:: MAINTAIN_FRAMERATE

   :type: sora_sdk.SoraDegradationPreference
   :readonly: True

   フレームレートを優先することを示す定数です。

.. py:attribute:: MAINTAIN_RESOLUTION

   :type: sora_sdk.SoraDegradationPreference
   :readonly: True

   解像度を優先することを示す定数です。

.. py:attribute:: BALANCED

   :type: sora_sdk.SoraDegradationPreference
   :readonly: True

   フレームレートと解像度のバランスを優先することを示す定数です。

.. py:attribute:: DISABLED

   :type: sora_sdk.SoraDegradationPreference
   :readonly: True

   デグラデーションを無効化することを示す定数です。

## SoraSignalingDirection オブジェクト

.. py:attribute:: SENT

   :type: sora_sdk.SoraSignalingDirection
   :readonly: True

   送信したメッセージを示す定数です。

.. py:attribute:: RECEIVED

   :type: sora_sdk.SoraSignalingDirection
   :readonly: True

   受信したメッセージを示す定数です。

## SoraSignalingErrorCode オブジェクト

.. py:attribute:: CLOSE_FAILED

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: CLOSE_SUCCEEDED

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: ICE_FAILED

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: INTERNAL_ERROR

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: INVALID_PARAMETER

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: PEER_CONNECTION_STATE_FAILED

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: WEBSOCKET_HANDSHAKE_FAILED

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: WEBSOCKET_ONCLOSE

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

.. py:attribute:: WEBSOCKET_ONERROR

   :type: sora_sdk.SoraSignalingErrorCode
   :readonly: True

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

## SoraLoggingSeverity オブジェクト

.. py:attribute:: VERBOSE

   :type: sora_sdk.SoraLoggingSeverity
   :readonly: True

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

.. py:attribute:: INFO

   :type: sora_sdk.SoraLoggingSeverity
   :readonly: True

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

.. py:attribute:: WARNING

   :type: sora_sdk.SoraLoggingSeverity
   :readonly: True

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

.. py:attribute:: ERROR

   :type: sora_sdk.SoraLoggingSeverity
   :readonly: True

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

.. py:attribute:: NONE

   :type: sora_sdk.SoraLoggingSeverity
   :readonly: True

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

<!-- mikan:page doc="packaging" -->

# パッケージング

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

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

## Windows 11 x86_64

```console
uv sync
uv run python run.py windows_x86_64
uv run python -m build
```

## macOS arm64

```console
uv sync
uv run python run.py macos_arm64
uv run python -m build
```

## Ubuntu 24.04 x86_64

```console
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 でビルドとパッケージができます。

```console
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

```console
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 でビルドとパッケージができます。

```console
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
```

<!-- mikan:page doc="e2e_test" -->

# Sora Python SDK の E2E テスト

## uv のインストール

```bash
# uv をインストールする
curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.local/bin/env
```

## Sora Python SDK のビルド

```bash
# 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
```

## 環境変数の設定

```bash
# .env を修正して Sora の接続情報などを設定する
$ cp .env.template .env
```

```bash
# 設定例
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 テストの実行

```bash
uv run pytest tests/
```

## Intel VPL の挙動の E2E テストを Ubuntu 24.04 で行う

Intel VPL の挙動を確認するための E2E テストを用意しています。

```bash
# Intel VPL のテストを実行する
LIBVA_MESSAGING_LEVEL=0 INTEL_VPL=true uv run pytest tests/test_intel_vpl.py -s
```