src/voice_dialogue/audio/devices.py

"""
音频设备枚举工具。

提供列出系统可用输入/输出设备（包括外置麦克风阵列、外置扬声器）的能力，
供 CLI、API 以及前端进行设备选择。
"""
from typing import List, Optional, TypedDict

import pyaudio

from voice_dialogue.utils.logger import logger


class InputDeviceInfo(TypedDict):
    """输入设备信息。"""
    index: int
    name: str
    max_input_channels: int
    default_sample_rate: int
    is_default: bool


class OutputDeviceInfo(TypedDict):
    """输出设备信息。"""
    index: int
    name: str
    max_output_channels: int
    default_sample_rate: int
    is_default: bool


def _get_default_input_index(p: pyaudio.PyAudio) -> Optional[int]:
    """获取系统默认输入设备索引，失败时返回 None。"""
    try:
        return int(p.get_default_input_device_info().get("index"))
    except Exception:
        return None


def list_input_devices() -> List[InputDeviceInfo]:
    """
    列出所有可用的音频输入设备。

    Returns:
        List[InputDeviceInfo]: 输入设备列表（仅包含 maxInputChannels > 0 的设备）。
    """
    devices: List[InputDeviceInfo] = []
    p = pyaudio.PyAudio()
    try:
        default_index = _get_default_input_index(p)
        for i in range(p.get_device_count()):
            try:
                info = p.get_device_info_by_index(i)
            except Exception as e:
                logger.warning(f"读取音频设备 {i} 信息失败: {e}")
                continue

            max_input_channels = int(info.get("maxInputChannels", 0))
            if max_input_channels <= 0:
                continue

            devices.append(
                InputDeviceInfo(
                    index=int(info.get("index", i)),
                    name=str(info.get("name", f"device-{i}")),
                    max_input_channels=max_input_channels,
                    default_sample_rate=int(info.get("defaultSampleRate", 16000)),
                    is_default=(int(info.get("index", i)) == default_index),
                )
            )
    finally:
        p.terminate()

    return devices


def get_default_input_device_index() -> Optional[int]:
    """获取系统默认输入设备索引。"""
    p = pyaudio.PyAudio()
    try:
        return _get_default_input_index(p)
    finally:
        p.terminate()


def is_valid_input_device(index: Optional[int]) -> bool:
    """
    校验给定索引是否为有效的输入设备。

    Args:
        index: 设备索引；None 表示使用系统默认设备，视为有效。

    Returns:
        bool: 是否有效。
    """
    if index is None:
        return True
    return any(d["index"] == index for d in list_input_devices())


def _get_default_output_index(p: pyaudio.PyAudio) -> Optional[int]:
    """获取系统默认输出设备索引，失败时返回 None。"""
    try:
        return int(p.get_default_output_device_info().get("index"))
    except Exception:
        return None


def list_output_devices() -> List[OutputDeviceInfo]:
    """
    列出所有可用的音频输出设备。

    Returns:
        List[OutputDeviceInfo]: 输出设备列表（仅包含 maxOutputChannels > 0 的设备）。
    """
    devices: List[OutputDeviceInfo] = []
    p = pyaudio.PyAudio()
    try:
        default_index = _get_default_output_index(p)
        for i in range(p.get_device_count()):
            try:
                info = p.get_device_info_by_index(i)
            except Exception as e:
                logger.warning(f"读取音频设备 {i} 信息失败: {e}")
                continue

            max_output_channels = int(info.get("maxOutputChannels", 0))
            if max_output_channels <= 0:
                continue

            devices.append(
                OutputDeviceInfo(
                    index=int(info.get("index", i)),
                    name=str(info.get("name", f"device-{i}")),
                    max_output_channels=max_output_channels,
                    default_sample_rate=int(info.get("defaultSampleRate", 48000)),
                    is_default=(int(info.get("index", i)) == default_index),
                )
            )
    finally:
        p.terminate()

    return devices


def get_default_output_device_index() -> Optional[int]:
    """获取系统默认输出设备索引。"""
    p = pyaudio.PyAudio()
    try:
        return _get_default_output_index(p)
    finally:
        p.terminate()


def is_valid_output_device(index: Optional[int]) -> bool:
    """
    校验给定索引是否为有效的输出设备。

    Args:
        index: 设备索引；None 表示使用系统默认设备，视为有效。

    Returns:
        bool: 是否有效。
    """
    if index is None:
        return True
    return any(d["index"] == index for d in list_output_devices())