tape_recorder/core/audio_device.py

"""
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