`sora_sdk` モジュール¶

概要¶

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

音声と映像の扱い¶

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

libwebrtc ログ指定¶

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

sora_sdk.enable_libwebrtc_log¶

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

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

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

import sora_sdk

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

`Sora` オブジェクト¶

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

param Optional[bool] use_hardware_encoder:

ハードウェアエンコーダを使用するかどうか (デフォルトは True)

param Optional[str] openh264:

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

rtype:

None

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

create_audio_source(channels, sample_rate)¶

パラメータ:

channels (int) -- チャンネル

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

戻り値の型:

sora_sdk.SoraAudioSource

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

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

create_video_source()¶

戻り値の型:

sora_sdk.SoraVideoSource

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

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

パラメータ:

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

戻り値の型:

sora_sdk.SoraConnection

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

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

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

import time

from sora_sdk import Sora

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

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

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

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

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

import time

from sora_sdk import Sora

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

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

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

`SoraConnection` オブジェクト¶

class sora_sdk.SoraConnection¶

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

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

connect()¶

戻り値:: None

Sora に接続します。

disconnect()¶

戻り値:: None

Sora から切断します。

on_disconnect(error_code, message)¶

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

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

from sora_sdk import Sora, SoraSignalingErrorCode


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


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

    conn.connect()

send_data_channel(label, data)¶

パラメータ:

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

戻り値:

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

戻り値の型:

bool

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

on_data_channel(label)¶

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

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

on_message(label, data)¶

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

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

from sora_sdk import Sora, SoraConnection


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


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


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

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

    conn.connect()

on_notify(message)¶

Type:: Callable[[str], None]

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

import json
from typing import Any, Dict

from sora_sdk import Sora


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


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

    conn.connect()

on_push(message)¶

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

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

on_set_offer¶

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

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

on_track(track)¶

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

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

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

on_switched(message)¶

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

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

`SoraAudioSource` オブジェクト¶

class sora_sdk.SoraAudioSource¶

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

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

on_data(data, samples_per_channel, timestamp)¶

パラメータ:

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

戻り値の型:

None

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

on_data(data, samples_per_channel)¶

パラメータ:

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

戻り値の型:

None

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

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

on_data(ndarray, timestamp)¶

パラメータ:

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

戻り値の型:

None

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

on_data(ndarray)¶

パラメータ:

ndarray (ndarray[dtype=int16, shape=(*, *), order='C', device='cpu']) --

チャンネルごとのサンプル数 x チャンネル数になっている音声データ。

戻り値の型:

None

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

`SoraAudioSink` オブジェクト¶

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

パラメータ:

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

戻り値の型:

None

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

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

read(frames=0, timeout=1)¶

パラメータ:

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

戻り値:

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

戻り値の型:

tuple

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

on_data¶

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

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

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

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

on_format(sample_rate, number_of_channels)¶

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

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

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

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

from sora_sdk import SoraAudioSink

audio_sink = None


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

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

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

`SoraVideoSource` オブジェクト¶

class sora_sdk.SoraVideoSource¶

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

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

on_captured(ndarray)¶

パラメータ:

ndarray (ndarray[dtype=uint8, shape=(*, *, 3), order='C', device='cpu']) --

H x W x BGR になっているフレームデータ。

戻り値の型:

None

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

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

on_captured(ndarray, timestamp)¶

パラメータ:

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

戻り値の型:

None

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

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

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

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

on_captured(ndarray, timestamp_us)¶

パラメータ:

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

戻り値の型:

None

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

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

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

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

`SoraTrackInterface` オブジェクト¶

class sora_sdk.SoraTrackInterface¶

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

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

set_enabled(enable)¶

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

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

enabled¶

Type:: bool
Readonly:: True

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

id¶

Type:: str
Readonly:: True

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

kind¶

Type:: str
Readonly:: True

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

state¶

Type:: sora_sdk.SoraTrackState
Readonly:: True

トラックの状態。

`SoraAudioFrame` オブジェクト¶

class sora_sdk.SoraAudioFrame¶

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

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

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

data()¶

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

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

samples_per_channel¶

Type:: int

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

num_channels¶

Type:: int

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

sample_rate_hz¶

Type:: int

サンプリングレート。

absolute_capture_timestamp_ms¶

Type:: int

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

`SoraAudioStreamSink` オブジェクト¶

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

パラメータ:

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

戻り値の型:

None

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

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

on_frame¶

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

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

from sora_sdk import SoraAudioStreamSink

audio_stream_sink = None


def on_frame(frame):
    pass


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

`SoraMediaTrack` オブジェクト¶

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

class sora_sdk.SoraMediaTrack¶

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

stream_id¶

Type:: str

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

`SoraVAD` オブジェクト¶

class sora_sdk.SoraVAD¶

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

analyze(frame)¶

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

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

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

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

from sora_sdk import SoraAudioStreamSink, SoraVAD

audio_stream_sink = None
vad = SoraVAD()


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


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

`SoraVideoFrame` オブジェクト¶

class sora_sdk.SoraVideoFrame¶

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

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

data()¶

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

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

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

`SoraVideoSink` オブジェクト¶

class sora_sdk.SoraVideoSink(track)¶

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

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

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

on_frame¶

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

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

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

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

from sora_sdk import SoraVideoSink

video_sink = None


def on_frame(frame):
    pass


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