`sora_sdk` モジュール¶

概要¶

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

音声と映像の扱い¶

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

NDArray 型と LikeArray 型の注意¶

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

ただし nanobind の制約により pyi では NDArray 型が ArrayLike 型になっています。

参考¶

Use `NDArray instead of ArrayLike when dtype is given by yosh-matsuda · Pull Request #442 · wjakob/nanobind <https://github.com/wjakob/nanobind/pull/442>`_
nanobind 2.0.0: How can I customize the stubgen type used for `Eigen arguments/return parameters? · wjakob/nanobind · Discussion #630 <https://github.com/wjakob/nanobind/discussions/630>`_

libwebrtc ログ指定¶

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

sora_sdk.enable_libwebrtc_log¶

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

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

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

import sora_sdk

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

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

`SoraVideoCodecCapability` オブジェクト¶

class sora_sdk.SoraVideoCodecCapability¶

to_json()¶

戻り値の型:: str

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

engines¶

戻り値の型:: List[SoraVideoCodecCapability.Engine]

SoraVideoCodecCapability.Engine のリストを返します。

class Codec¶

class Engine¶

class Parameters¶

`Sora` オブジェクト¶

class sora_sdk.Sora(video_codec_preference: SoraVideoCodecPreference | None = None, openh264: str | None = None)¶

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

param video_codec_preference:

SoraVideoCodecPreference を指定します。何も指定しない場合は None で全てのコーデックは SoraVideoCodecImplementation.INTERNAL を利用します。

type:

SoraVideoCodecPreference | None

param openh264:

OpenH264 のパスを指定します。

type:

str | None

create_audio_source(channels, sample_rate)¶

パラメータ:

channels (int) -- チャンネル

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

戻り値の型:

sora_sdk.SoraAudioSource

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

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

create_video_source()¶

戻り値の型:

sora_sdk.SoraVideoSource

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

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

パラメータ:

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

戻り値の型:

sora_sdk.SoraConnection

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

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

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

import time

from sora_sdk import Sora

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

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

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

mTLS を利用する際の最小コード¶

mTLS を利用する際の証明書は open('cert.pem', 'rb').read() のように読み込んだバイナリを渡してください。

import time

from sora_sdk import Sora

if __name__ == "__main__":
    # Sora インスタンスを生成
    sora = Sora()
    # Sora に接続する
    conn = sora.create_connection(
        signaling_urls=["wss://example.com/signaling"],
        role="sendonly",
        channel_id="sora",
        # クライアント証明書を読み込む
        client_cert=open("cert.pem", "rb").read(),
        # クライアント証明書シークレットキーを読み込む
        client_key=open("cert-key.pem", "rb").read(),
        # CA 証明書を読み込む
        ca_cert=open("ca.pem", "rb").read(),
    )
    conn.connect()

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

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

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

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

import time

from sora_sdk import Sora

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

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

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

video_codec_preference を指定する最小コード¶

映像コーデックのエンコード/デコードを細かく指定する場合は Sora.video_codec_preference を指定してください。

import time

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

if __name__ == "__main__":
    # Sora インスタンスを生成
    sora = Sora(
        # ビデオコーデックプリファレンスを指定する
        video_codec_preference=SoraVideoCodecPreference(
            codecs=[
                # H.264 のみを利用する
                SoraVideoCodecPreference.Codec(
                    type=SoraVideoCodecType.H264,
                    # H.264 のエンコーダーに OpenH264 を指定する
                    encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
                    # sendonly なのでデコーダーは指定しない
                ),
            ],
        ),
        openh264="/path/to/libopenh264-2.6.0-linux64.7.so",
    )
    # Sora に接続する
    conn = sora.create_connection(
        signaling_urls=["wss://example.com/signaling"],
        role="sendonly",
        channel_id="sora",
    )
    conn.connect()

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

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

`SoraConnection` オブジェクト¶

class sora_sdk.SoraConnection¶

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

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

connect()¶

戻り値:: None

Sora に接続します。

重要

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

disconnect()¶

戻り値:: None

Sora から切断します。

重要

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

get_stats()¶

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

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

import json
import time

from sora_sdk import Sora

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

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

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

    conn.disconnect()

on_disconnect(error_code, message)¶

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

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

import time

from sora_sdk import Sora, SoraSignalingErrorCode


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


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

    conn.connect()

    time.sleep(5)

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

send_data_channel(label, data)¶

パラメータ:

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

戻り値:

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

戻り値の型:

bool

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

on_data_channel(label)¶

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

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

on_message(label, data)¶

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

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

import time

from sora_sdk import Sora, SoraConnection


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


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


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

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

    conn.connect()

    time.sleep(5)

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

on_notify(message)¶

Type:: Callable[[str], None]

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

import json
import time
from typing import Any

from sora_sdk import Sora


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


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

    conn.connect()

    time.sleep(5)

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

on_push(message)¶

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

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

on_set_offer¶

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

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

on_track(track)¶

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

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

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

on_switched(message)¶

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

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

on_signaling_message(type, direction, message)¶

Type:: Callable[[sora_sdk.SoraSignalingType, sora_sdk.SoraSignalingDirection, str], None]

シグナリングメッセージを送受信した際に呼ばれるコールバックです。主にログ出力などに利用する事を想定しており、戻り値を返すことはできません。

"type": "connect"
"type": "redirect"
"type": "offer"
"type": "answer"
"type": "candidate"
"type": "re-offer"
"type": "re-answer"
"type": "disconnect"
"type": "close"

"type": "close" は Sora 2024.2.0 から利用可能なメッセージです。

"type": "switched" は SoraConnection.on_switched を利用してください。

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

import json
import time

from sora_sdk import Sora, SoraSignalingDirection, SoraSignalingType


def on_signaling_message(
    type: SoraSignalingType, direction: SoraSignalingDirection, raw_message: str
):
    json_message = json.loads(raw_message)
    print(f"type: {type.value}, direction: {direction.value}, message: {json_message}")


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

    conn.connect()

    time.sleep(5)

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

on_ws_close(code, reason)¶

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

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

import time

from sora_sdk import Sora


def on_ws_close(code: int, reason: str):
    print(f"WebSocket closed: code={code}, reason={reason}")


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

    conn.connect()

    time.sleep(5)

    # connect() を呼んだら disconnect を呼ぶ
    conn.disconnect()

`SoraAudioSource` オブジェクト¶

class sora_sdk.SoraAudioSource¶

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

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

on_data(ndarray)¶

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

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

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

on_data(data, samples_per_channel)¶

パラメータ:

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

戻り値の型:

None

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

`SoraAudioSink` オブジェクト¶

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

パラメータ:

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

戻り値の型:

None

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

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

read(frames=0, timeout=1)¶

パラメータ:

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

戻り値:

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

戻り値の型:

tuple

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

on_data¶

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

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

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

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

on_format(sample_rate, number_of_channels)¶

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

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

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

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

from sora_sdk import SoraAudioSink

audio_sink = None


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

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

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

`SoraVideoSource` オブジェクト¶

class sora_sdk.SoraVideoSource¶

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

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

on_captured(ndarray)¶

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

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

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

`SoraTrackInterface` オブジェクト¶

class sora_sdk.SoraTrackInterface¶

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

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

set_enabled(enable)¶

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

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

enabled¶

Type:: bool
Readonly:: True

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

id¶

Type:: str
Readonly:: True

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

kind¶

Type:: str
Readonly:: True

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

state¶

Type:: sora_sdk.SoraTrackState
Readonly:: True

トラックの状態。

`SoraAudioFrame` オブジェクト¶

class sora_sdk.SoraAudioFrame¶

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

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

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

data()¶

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

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

samples_per_channel¶

Type:: int

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

num_channels¶

Type:: int

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

sample_rate_hz¶

Type:: int

サンプリングレート。

absolute_capture_timestamp_ms¶

Type:: int

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

`SoraAudioStreamSink` オブジェクト¶

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

パラメータ:

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

戻り値の型:

None

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

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

on_frame¶

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

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

from sora_sdk import SoraAudioStreamSink

audio_stream_sink = None


def on_frame(frame):
    pass


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

`SoraMediaTrack` オブジェクト¶

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

class sora_sdk.SoraMediaTrack¶

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

stream_id¶

Type:: str

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

set_frame_transformer(transformer)¶

パラメータ:: transformer (sora_sdk.SoraFrameTransformer) -- フレームを変換する SoraFrameTransformer オブジェクト。
戻り値の型:: None

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

`SoraVAD` オブジェクト¶

class sora_sdk.SoraVAD¶

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

analyze(frame)¶

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

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

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

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

from sora_sdk import SoraAudioStreamSink, SoraVAD

audio_stream_sink = None
vad = SoraVAD()


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


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

`SoraVideoFrame` オブジェクト¶

class sora_sdk.SoraVideoFrame¶

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

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

data()¶

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

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

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

`SoraVideoSink` オブジェクト¶

class sora_sdk.SoraVideoSink(track)¶

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

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

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

on_frame¶

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

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

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

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

from sora_sdk import SoraVideoSink

video_sink = None


def on_frame(frame):
    pass


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

OpenCV¶

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

`SoraVideoCodecType` オブジェクト¶

Added in version 2025.1.0.

class sora_sdk.SoraVideoCodecType¶

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

VP8: SoraVideoCodecType = 1¶: VP8 コーデック

VP9: SoraVideoCodecType = 2¶: VP9 コーデック

AV1: SoraVideoCodecType = 3¶: AV1 コーデック

H264: SoraVideoCodecType = 4¶: H.264 コーデック

H265: SoraVideoCodecType = 5¶: H.265 コーデック

`SoraVideoCodecPreference` オブジェクト¶

Added in version 2025.1.0.

class sora_sdk.SoraVideoCodecPreference(codecs: Sequence[SoraVideoCodecPreference.Codec] = [])¶

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

警告

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

Codec(type, encoder=None, decoder=None)¶

パラメータ:

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

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

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

Sora(
    # ビデオコーデックプリファレンスを指定する
    video_codec_preference=SoraVideoCodecPreference(
        codecs=[
            # VP9 は libwebrtc 組込のソフトウェアコーデック libvpx を利用する
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.VP9,
                encoder=SoraVideoCodecImplementation.INTERNAL,
                decoder=SoraVideoCodecImplementation.INTERNAL,
            ),
            # AV1 は NVIDIA Video Codec SDK を利用する
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.AV1,
                encoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
                decoder=SoraVideoCodecImplementation.NVIDIA_VIDEO_CODEC_SDK,
            ),
            # H.264 は Cisco OpenH264 を利用する
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.H264,
                encoder=SoraVideoCodecImplementation.CISCO_OPENH264,
                decoder=SoraVideoCodecImplementation.CISCO_OPENH264,
            ),
            # H.265 は Intel VPL を利用する
            SoraVideoCodecPreference.Codec(
                type=SoraVideoCodecType.H265,
                encoder=SoraVideoCodecImplementation.INTEL_VPL,
                decoder=SoraVideoCodecImplementation.INTEL_VPL,
            ),
        ]
    ),
    openh264="/path/to/libopenh264-2.5.0-linux64.7.so",
)