音声デバイス¶

sounddevice のインストール¶

$ uv add sounddevice
$ uv sync

$ brew install portaudio

$ sudo apt install libportaudio2

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

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

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

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

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