Python SDK β版

🚀 クイックスタート

このクイックスタートでは、SkyWay Python SDK（以下、Python SDK）を用いたシンプルなアプリケーションを作成します。

本ガイドでは、Pythonの基本的な文法を理解していることと、Pythonの実行環境がセットアップ済みであることを前提としています。

また、映像・音声の送信、データの受信には他のSDKが必要です。あらかじめ、他のSDKのサンプルコードなどを参考に、映像・音声の送信（publish）、データの受信（subscribe）を行うアプリケーションを準備してください。

アプリケーション ID とシークレットキーを取得する

SkyWay への登録がまだの方はSkyWayコンソールへログインし、以下の 3 つの操作を行い、アプリケーション ID とシークレットキーを取得してください。

「アプリケーションを作成」ボタンを押す
アプリケーション名を入力して「作成」ボタンを押す
アプリケーション一覧からアプリケーション ID とシークレットキーを控えておく（後の手順で利用します）。

Pythonプロジェクトの作成

クイックスタート用のディレクトリを作成します。

mkdir skyway-python-quickstart
cd skyway-python-quickstart

Pythonの仮想環境を作成します。

python -m venv venv
# Linux/macOS の場合
source venv/bin/activate
# Windows の場合
venv\Scripts\activate

ライブラリのインストール

Python SDKをインストールします。

pip install skyway-room

SkyWay Auth Tokenを生成するためのライブラリをインストールします。

pip install pyjwt

受信した映像・音声データを処理するためのライブラリをインストールします。

pip install aiortc

使用するクラスのインポート

テキストエディタなどで quickstart.py という名前のファイルを作成してください。これ以降のコードは、全て quickstart.py ファイルに記述していきます。

quickstart.py ファイルを作成したら、その中に以下のコードを記述します。

#!/usr/bin/env python3

import argparse
import asyncio
import os
import signal
import sys
import time
import uuid

import jwt

from aiortc.mediastreams import MediaStreamTrack, MediaStreamError

from skyway.room import (
    SkyWayContext,
    StreamPublishedEvent,
    Publication,
    ContentType,
    LocalDataStream,
    RemoteMediaStream,
)

SkyWay Auth Tokenの生成用関数の追加

このガイドでは検証用として、アプリケーション内でSkyWay Auth Tokenを生成します。エンドユーザーの環境でアプリケーションを実行する場合など、シークレットキーをエンドユーザーに知られないようにする必要がある場合は、アプリケーションサーバーを構築し、そちらでSkyWay Auth Tokenを生成してクライアントアプリに払い出すことを推奨します。

SkyWay Auth Token の詳細については、SkyWay Auth Token のドキュメントを参照してください。

環境変数からアプリケーション ID とシークレットキーを取得し、SkyWay Auth Tokenを生成する処理を実装します。

quickstart.pyに以下のコードを追加してください。

TOKEN_EXPIRY_SECONDS: int = 60 * 60

def _generate_auth_token() -> str:
    app_id = os.environ.get("SKYWAY_APP_ID", "")
    secret_key = os.environ.get("SKYWAY_SECRET_KEY", "")
    if not app_id or not secret_key:
        print("エラー: SKYWAY_APP_ID と SKYWAY_SECRET_KEY を設定してください。")
        sys.exit(1)
    issued_at = int(time.time())
    payload = {
        "jti": str(uuid.uuid4()),
        "iat": issued_at,
        "exp": issued_at + TOKEN_EXPIRY_SECONDS,
        "version": 3,
        "scope": {
            "appId": app_id,
            "rooms": [
                {
                    "name": "*",
                    "methods": ["create"],
                    "member": {
                        "name": "*",
                        "methods": ["publish", "subscribe"],
                    },
                },
            ],
        },
    }
    return jwt.encode(payload, secret_key, algorithm="HS256")

映像・音声データ受信用関数の追加

映像と音声のデータを受信するための処理を実装します。

ここでは動作確認のために、受信したフレーム数をカウントして100フレームごとにログを出力しています。

quickstart.pyに以下のコードを追加してください。

async def _receive_frames(track: MediaStreamTrack) -> None:
    frame_count = 0
    while True:
        try:
            frame = await track.recv()
            frame_count += 1
            if frame_count % 100 == 0:
                print(
                    f"[{track.kind}] Received {frame_count} frames from track {track.id}"
                )
        except MediaStreamError:
            return

main関数の作成

SkyWayを利用する一連のコードを記述するためのmain関数を作成します。

ここでは、以下の3つの処理を実装します。

プログラムの終了シグナルをハンドリングする処理
プログラム終了時のクリーンアップ処理
コマンドライン引数からroom名を受け取る処理

quickstart.pyに以下のコードを追加してください。

async def main(room_name: str) -> None:
    # プログラムの終了シグナルをハンドリングする処理
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    try:
        loop.add_signal_handler(signal.SIGINT, stop_event.set)
        loop.add_signal_handler(signal.SIGTERM, stop_event.set)
    except (NotImplementedError, RuntimeError):
        pass

    receive_tasks: list[asyncio.Task[None]] = []
    context: SkyWayContext | None = None
    try:
        # 「SkyWay Contextの作成」以後の処理はこの try ブロック内に記述します
    finally: # プログラム終了時のクリーンアップ処理
        for task in receive_tasks:
            task.cancel()
        await asyncio.gather(*receive_tasks, return_exceptions=True)
        if context is not None:
            await context.dispose()

# コマンドライン引数からroom名を受け取る処理
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="SkyWay Python SDK クイックスタート")
    parser.add_argument(
        "room_name",
        nargs="?",
        default="quickstart-room",
        help="参加または作成する room 名（デフォルト: quickstart-room）",
    )
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    asyncio.run(main(args.room_name))

SkyWay Contextの作成

SkyWayを利用する際の起点となるオブジェクトのSkyWay Contextを作成する処理を実装します。

本セクション以降で説明するコードは、全てquickstart.pyの # 「SkyWay Contextの作成」以後の処理はこの try ブロック内に記述します の部分に追記してください。

        token = _generate_auth_token()

        context = SkyWayContext()
        await context.setup(token)

Roomへの参加

指定したnameのRoomに参加する処理を実装します。 Roomがすでに存在する場合はそのRoomに参加し、存在しない場合はRoomを作成してから参加します。

quickstart.pyに以下のコードを追加してください。

        room = await context.find_or_create_room(room_name)
        member = await room.join()

DataStreamのpublish

Roomに参加した後、DataStreamをpublishする処理を実装します。

quickstart.pyに以下のコードを追加してください。

        stream = LocalDataStream()
        await member.publish(stream)

VideoStreamとAudioStreamのsubscribe

VideoStreamとAudioStreamをsubscribeする処理を実装します。

Publicationをsubscribeするかどうかを判定するメソッドとしてshould_subscribeを作成します。

また、実際のsubscribe処理を行うメソッドとして_subscribeを作成します。

quickstart.pyに以下のコードを追加してください。

        def should_subscribe(pub: Publication) -> bool:
            return (
                pub.content_type in (ContentType.AUDIO, ContentType.VIDEO)
                and pub.publisher_id is not None
                and pub.publisher_id != member.id
            )

        async def _subscribe(publication_id: str) -> None:
            subscription = await member.subscribe(publication_id)
            if not isinstance(subscription.stream, RemoteMediaStream):
                return
            # 受信した映像・音声データには subscription.stream.track を通じてアクセスできます
            await _receive_frames(subscription.stream.track)

次に、今後publishされるPublicationと、すでにpublishされているPublicationをそれぞれsubscribeする処理を実装します。

quickstart.pyに以下のコードを追加してください。

        @room.on("stream_published")
        def on_stream_published(event: StreamPublishedEvent) -> None:
            publication = room.publications.get(event.publication_id)
            if publication is None or not should_subscribe(publication):
                return
            receive_task = asyncio.create_task(_subscribe(publication.id))
            receive_tasks.append(receive_task)

        for pub in list(room.publications.values()):
            if should_subscribe(pub):
                receive_task = asyncio.create_task(_subscribe(pub.id))
                receive_tasks.append(receive_task)

データの送信処理

定期的にDataStreamでデータを送信する処理を実装します。

quickstart.pyに以下のコードを追加してください。

        # 1秒間隔で現在のタイムスタンプを送信する
        while not stop_event.is_set():
            stream.write(str(time.time()))
            await asyncio.sleep(1)

これで、クイックスタートのアプリケーションが完成です。

アプリケーションの実行

クイックスタートを実行して動作を確認します。

最初に、SkyWayコンソールで取得したアプリケーション ID とシークレットキーを環境変数に設定します。

# Linux/macOS の場合
export SKYWAY_APP_ID="YOUR_APP_ID"
export SKYWAY_SECRET_KEY="YOUR_SECRET_KEY"
# Windows の場合
$env:SKYWAY_APP_ID="YOUR_APP_ID"
$env:SKYWAY_SECRET_KEY="YOUR_SECRET_KEY"

次に、クイックスタートを実行します。

python quickstart.py <room_name>

<room_name> には参加または作成するRoomの名前を指定します。指定しない場合はデフォルトで quickstart-room が使用されます。

クイックスタートが正常に起動すると、ターミナルに以下のようなログが出力されます。

[WebSocket] (RtcApiClient) is connected.
[SkyWayContext] SkyWay is set up
...

続いて、あらかじめ用意した、映像と音声のpublish、およびデータのsubscribeを行うアプリケーションを実行します。 room名は、クイックスタートで指定したroom名と同じにしてください。

送信側で映像と音声のpublishが行われると、Python SDKのクイックスタートを実行しているターミナルに以下のようなログが表示されます。このログは、映像と音声を受信していることを示しています。

[audio] Received 100 frames from track 52fc60b6-64e9-46d3-b270-c8ab39b9185a
[video] Received 100 frames from track 4dcf721d-f909-4ab4-9974-ade63264ebb8
[audio] Received 200 frames from track 52fc60b6-64e9-46d3-b270-c8ab39b9185a
[video] Received 200 frames from track 4dcf721d-f909-4ab4-9974-ade63264ebb8

また、クイックスタートから定期的に送信されるタイムスタンプが、映像と音声のpublishを行っているアプリケーションで受信できることを確認してください。

動作することが確認できたら、ターミナルで Ctrl + C を押してクイックスタートを終了してください。

以上で、クイックスタートのアプリケーションの動作確認は完了です。

完成版のソースコードは、Python SDKのGitHubリポジトリの examples/quickstart ディレクトリで公開しています。完成版を確認したい場合は、そちらを参照してください。

次のステップ

Python SDKと各種ライブラリや外部のAIサービスを組み合わせることで、より実用的なアプリケーションを実現できます。例えば、以下のような利用が可能です。

受信した映像・音声をマルチモーダルAI（Gemini Live APIなど）に渡し、エンドユーザーとリアルタイムに対話するエージェントを構築する
受信した映像をOpenCVや機械学習モデルで解析し、物体検出や動作認識といったリアルタイム映像処理を行う
受信した音声を音声認識サービスに入力し、会議の内容をリアルタイムに文字起こし・要約する

ぜひ、Python SDKを活用して、さまざまなユースケースを実現してみてください。

概要

SFU