""" Audio Device Manager — Enumerate and manage sound card I/O devices. Provides a singleton DeviceManager that all components use to access the currently selected input/output devices and sample rate. """ import sounddevice as sd import numpy as np import json import os from dataclasses import dataclass, field from typing import Optional @dataclass class AudioDeviceInfo: """Information about an audio device.""" index: int name: str max_input_channels: int max_output_channels: int default_samplerate: float hostapi_name: str @property def is_input(self) -> bool: return self.max_input_channels > 0 @property def is_output(self) -> bool: return self.max_output_channels > 0 def __str__(self) -> str: direction = [] if self.is_input: direction.append(f"IN:{self.max_input_channels}ch") if self.is_output: direction.append(f"OUT:{self.max_output_channels}ch") return f"[{self.index}] {self.name} ({', '.join(direction)}) [{self.hostapi_name}]" class DeviceManager: """ Manages audio device selection and configuration. Singleton pattern — use DeviceManager.instance() to get the shared instance. """ _instance: Optional["DeviceManager"] = None def __init__(self): self._input_device_id: Optional[int] = None self._output_device_id: Optional[int] = None self._sample_rate: int = 44100 self._channels: int = 2 # Stereo self._block_size: int = 1024 self._devices: list[AudioDeviceInfo] = [] self.refresh_devices() # Load saved preferences if available self._load_config() def _load_config(self): """Load settings from config.json.""" if not os.path.exists("config.json"): return try: with open("config.json", "r") as f: config = json.load(f) if "sample_rate" in config: self.sample_rate = config["sample_rate"] # Try to match saved device names to current indices if "input_device_name" in config: for dev in self.get_input_devices(): if dev.name == config["input_device_name"]: self._input_device_id = dev.index break if "output_device_name" in config: for dev in self.get_output_devices(): if dev.name == config["output_device_name"]: self._output_device_id = dev.index break except Exception as e: print(f"Failed to load config: {e}") def _save_config(self): """Save settings to config.json.""" config = { "sample_rate": self.sample_rate } if self._input_device_id is not None: dev = self.get_device_by_index(self._input_device_id) if dev: config["input_device_name"] = dev.name if self._output_device_id is not None: dev = self.get_device_by_index(self._output_device_id) if dev: config["output_device_name"] = dev.name try: with open("config.json", "w") as f: json.dump(config, f, indent=4) except Exception as e: print(f"Failed to save config: {e}") @classmethod def instance(cls) -> "DeviceManager": """Get the singleton DeviceManager instance.""" if cls._instance is None: cls._instance = DeviceManager() return cls._instance @classmethod def reset_instance(cls): """Reset the singleton (useful for testing).""" cls._instance = None def refresh_devices(self): """Re-scan available audio devices.""" self._devices = [] devices = sd.query_devices() hostapis = sd.query_hostapis() for i, dev in enumerate(devices): hostapi_name = hostapis[dev["hostapi"]]["name"] if dev["hostapi"] < len(hostapis) else "Unknown" self._devices.append(AudioDeviceInfo( index=i, name=dev["name"], max_input_channels=dev["max_input_channels"], max_output_channels=dev["max_output_channels"], default_samplerate=dev["default_samplerate"], hostapi_name=hostapi_name, )) def get_all_devices(self) -> list[AudioDeviceInfo]: """Return all available audio devices.""" return list(self._devices) def get_input_devices(self) -> list[AudioDeviceInfo]: """Return devices that support audio input.""" return [d for d in self._devices if d.is_input] def get_output_devices(self) -> list[AudioDeviceInfo]: """Return devices that support audio output.""" return [d for d in self._devices if d.is_output] def get_device_by_index(self, index: int) -> Optional[AudioDeviceInfo]: """Get device info by its index.""" for d in self._devices: if d.index == index: return d return None @property def input_device_id(self) -> Optional[int]: return self._input_device_id @input_device_id.setter def input_device_id(self, device_id: int): dev = self.get_device_by_index(device_id) if dev is None: raise ValueError(f"Device index {device_id} not found") if not dev.is_input: raise ValueError(f"Device '{dev.name}' has no input channels") self._input_device_id = device_id self._save_config() @property def output_device_id(self) -> Optional[int]: return self._output_device_id @output_device_id.setter def output_device_id(self, device_id: int): dev = self.get_device_by_index(device_id) if dev is None: raise ValueError(f"Device index {device_id} not found") if not dev.is_output: raise ValueError(f"Device '{dev.name}' has no output channels") self._output_device_id = device_id # Automatically update sample rate to device's default to prevent speed mismatch if dev.default_samplerate > 0: self._sample_rate = int(dev.default_samplerate) self._save_config() @property def sample_rate(self) -> int: return self._sample_rate @sample_rate.setter def sample_rate(self, rate: int): if rate not in (22050, 44100, 48000, 88200, 96000): raise ValueError(f"Unsupported sample rate: {rate}") self._sample_rate = rate self._save_config() @property def channels(self) -> int: return self._channels @channels.setter def channels(self, ch: int): if ch not in (1, 2): raise ValueError(f"Unsupported channel count: {ch}") self._channels = ch @property def block_size(self) -> int: return self._block_size @block_size.setter def block_size(self, size: int): if size < 64 or size > 8192: raise ValueError(f"Block size must be 64-8192, got {size}") self._block_size = size def get_sd_input_kwargs(self) -> dict: """Get sounddevice kwargs for input stream.""" kwargs = { "samplerate": self._sample_rate, "channels": self._channels, "blocksize": self._block_size, "dtype": "float32", } if self._input_device_id is not None: kwargs["device"] = self._input_device_id return kwargs def get_sd_output_kwargs(self) -> dict: """Get sounddevice kwargs for output stream.""" kwargs = { "samplerate": self._sample_rate, "channels": self._channels, "blocksize": self._block_size, "dtype": "float32", } if self._output_device_id is not None: kwargs["device"] = self._output_device_id return kwargs def auto_select_defaults(self): """Auto-select the system default input and output devices.""" defaults = sd.query_devices() default_input = sd.default.device[0] default_output = sd.default.device[1] if default_input is not None and isinstance(default_input, int): try: self.input_device_id = default_input except ValueError: pass if default_output is not None and isinstance(default_output, int): try: self.output_device_id = default_output except ValueError: pass