--- lang: ja path: user-guide/python-sdk/quickstart labels: ユーザーガイド/Python SDK β版/クイックスタート metaTitle: クイックスタート ｜ Python SDK β版 ｜ ユーザーガイド ｜ SkyWay（スカイウェイ） --- # 🚀 クイックスタート このクイックスタートでは、SkyWay Python SDK（以下、Python SDK）を用いたシンプルなアプリケーションを作成します。 本ガイドでは、Pythonの基本的な文法を理解していることと、Pythonの実行環境がセットアップ済みであることを前提としています。 また、映像・音声の送信、データの受信には他のSDKが必要です。 あらかじめ、[他のSDKのサンプルコード](/ja/docs/sample-code/)などを参考に、映像・音声の送信（publish）、データの受信（subscribe）を行うアプリケーションを準備してください。 ## アプリケーション ID とシークレットキーを取得する SkyWay への登録がまだの方は[SkyWayコンソール](https://console.skyway.ntt.com/login/)へログインし、以下の 3 つの操作を行い、アプリケーション ID とシークレットキーを取得してください。 1. 「アプリケーションを作成」ボタンを押す 2. アプリケーション名を入力して「作成」ボタンを押す 3. アプリケーション一覧からアプリケーション ID とシークレットキーを控えておく（後の手順で利用します）。 ## Pythonプロジェクトの作成 クイックスタート用のディレクトリを作成します。 ```bash mkdir skyway-python-quickstart cd skyway-python-quickstart ``` Pythonの仮想環境を作成します。 ```bash python -m venv venv # Linux/macOS の場合 source venv/bin/activate # Windows の場合 venv\Scripts\activate ``` ## ライブラリのインストール Python SDKをインストールします。 ```bash pip install skyway-room ``` SkyWay Auth Tokenを生成するためのライブラリをインストールします。 ```bash pip install pyjwt ``` 受信した映像・音声データを処理するためのライブラリをインストールします。 ```bash pip install aiortc ``` ## 使用するクラスのインポート テキストエディタなどで `quickstart.py` という名前のファイルを作成してください。 これ以降のコードは、全て `quickstart.py` ファイルに記述していきます。 `quickstart.py` ファイルを作成したら、その中に以下のコードを記述します。 ```python #!/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 のドキュメント](/ja/docs/user-guide/authentication/skyway-auth-token/)を参照してください。 環境変数からアプリケーション ID とシークレットキーを取得し、SkyWay Auth Tokenを生成する処理を実装します。 quickstart.pyに以下のコードを追加してください。 ```python 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に以下のコードを追加してください。 ```python 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つの処理を実装します。 1. プログラムの終了シグナルをハンドリングする処理 2. プログラム終了時のクリーンアップ処理 3. コマンドライン引数からroom名を受け取る処理 quickstart.pyに以下のコードを追加してください。 ```python 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 ブロック内に記述します` の部分に追記してください。 ```python token = _generate_auth_token() context = SkyWayContext() await context.setup(token) ``` ## Roomへの参加 指定したnameのRoomに参加する処理を実装します。 Roomがすでに存在する場合はそのRoomに参加し、存在しない場合はRoomを作成してから参加します。 quickstart.pyに以下のコードを追加してください。 ```python room = await context.find_or_create_room(room_name) member = await room.join() ``` ## DataStreamのpublish Roomに参加した後、DataStreamをpublishする処理を実装します。 quickstart.pyに以下のコードを追加してください。 ```python stream = LocalDataStream() await member.publish(stream) ``` ## VideoStreamとAudioStreamのsubscribe VideoStreamとAudioStreamをsubscribeする処理を実装します。 Publicationをsubscribeするかどうかを判定するメソッドとして`should_subscribe`を作成します。 また、実際のsubscribe処理を行うメソッドとして`_subscribe`を作成します。 quickstart.pyに以下のコードを追加してください。 ```python 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に以下のコードを追加してください。 ```python @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に以下のコードを追加してください。 ```python # 1秒間隔で現在のタイムスタンプを送信する while not stop_event.is_set(): stream.write(str(time.time())) await asyncio.sleep(1) ``` これで、クイックスタートのアプリケーションが完成です。 ## アプリケーションの実行 クイックスタートを実行して動作を確認します。 最初に、SkyWayコンソールで取得したアプリケーション ID とシークレットキーを環境変数に設定します。 ```bash # 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" ``` 次に、クイックスタートを実行します。 ```bash python quickstart.py ``` `` には参加または作成する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リポジトリ](https://github.com/skyway/python-sdk-beta)の `examples/quickstart` ディレクトリで公開しています。 完成版を確認したい場合は、そちらを参照してください。 ## 次のステップ Python SDKと各種ライブラリや外部のAIサービスを組み合わせることで、より実用的なアプリケーションを実現できます。例えば、以下のような利用が可能です。 - 受信した映像・音声をマルチモーダルAI（Gemini Live APIなど）に渡し、エンドユーザーとリアルタイムに対話するエージェントを構築する - 受信した映像をOpenCVや機械学習モデルで解析し、物体検出や動作認識といったリアルタイム映像処理を行う - 受信した音声を音声認識サービスに入力し、会議の内容をリアルタイムに文字起こし・要約する ぜひ、Python SDKを活用して、さまざまなユースケースを実現してみてください。