e-ink-dash/server/audio_mixer.py

"""Audio mixer abstraction with pluggable backends.

Supports Windows Core Audio (default output) and Voicemeeter as backends.
Both expose the same interface so the client sees a unified audio view.
"""

from __future__ import annotations

import asyncio
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Literal

logger = logging.getLogger("paperdash.audio")


# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------


@dataclass
class AudioChannel:
    """A single audio channel/strip."""

    name: str
    volume: float = 0.0  # 0.0 - 1.0
    muted: bool = False
    peak_left: float = 0.0
    peak_right: float = 0.0


@dataclass
class AudioMixerState:
    """Unified audio mixer state from any backend."""

    backend: Literal["windows", "voicemeeter"]
    master_volume: float = 0.0
    master_muted: bool = False
    channels: list[AudioChannel] = field(default_factory=list)


# ---------------------------------------------------------------------------
# Backend Interface
# ---------------------------------------------------------------------------


class AudioMixerBackend(ABC):
    """Abstract interface for audio mixer backends."""

    @property
    @abstractmethod
    def name(self) -> str: ...

    @abstractmethod
    async def get_state(self) -> AudioMixerState: ...

    @abstractmethod
    async def set_volume(self, channel: str, value: float) -> None: ...

    @abstractmethod
    async def set_mute(self, channel: str, muted: bool) -> None: ...


# ---------------------------------------------------------------------------
# Windows Core Audio Backend
# ---------------------------------------------------------------------------


class WindowsAudioBackend(AudioMixerBackend):
    """Backend for the default Windows audio output device."""

    @property
    def name(self) -> str:
        return "windows"

    async def get_state(self) -> AudioMixerState:
        """Read master volume and mute from Windows Core Audio."""
        try:
            import pycaw  # type: ignore
            from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume  # type: ignore

            devices = AudioUtilities.GetSpeakers()
            interface = devices.Activate(IAudioEndpointVolume._iid_, 0, None)
            volume = interface.QueryInterface(IAudioEndpointVolume)

            master_volume = volume.MasterVolumeLevelScalar
            master_muted = volume.Mute

            return AudioMixerState(
                backend="windows",
                master_volume=master_volume,
                master_muted=master_muted,
                channels=[
                    AudioChannel(
                        name="Master",
                        volume=master_volume,
                        muted=master_muted,
                    )
                ],
            )
        except ImportError:
            logger.warning("pycaw not installed — Windows audio unavailable")
            return AudioMixerState(backend="windows")
        except Exception as e:
            logger.error(f"Windows audio error: {e}")
            return AudioMixerState(backend="windows")

    async def set_volume(self, channel: str, value: float) -> None:
        try:
            from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume

            devices = AudioUtilities.GetSpeakers()
            interface = devices.Activate(IAudioEndpointVolume._iid_, 0, None)
            volume = interface.QueryInterface(IAudioEndpointVolume)
            volume.MasterVolumeLevelScalar = max(0.0, min(1.0, value))
        except Exception as e:
            logger.error(f"Failed to set Windows volume: {e}")

    async def set_mute(self, channel: str, muted: bool) -> None:
        try:
            from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume

            devices = AudioUtilities.GetSpeakers()
            interface = devices.Activate(IAudioEndpointVolume._iid_, 0, None)
            volume = interface.QueryInterface(IAudioEndpointVolume)
            volume.Mute = muted
        except Exception as e:
            logger.error(f"Failed to set Windows mute: {e}")


# ---------------------------------------------------------------------------
# Voicemeeter Backend
# ---------------------------------------------------------------------------


class VoicemeeterBackend(AudioMixerBackend):
    """Backend for Voicemeeter Remote API."""

    @property
    def name(self) -> str:
        return "voicemeeter"

    async def get_state(self) -> AudioMixerState:
        """Read Voicemeeter strip states."""
        try:
            import voicemeeter_api  # type: ignore

            vm = voicemeeter_api.VoicemeeterRemote()
            vm.connect()

            channels = []
            # Read hardware inputs (A1, A2, A3)
            for i in range(1, 4):
                gain = vm.get_input_gain(i)
                mute = bool(vm.get_input_mute(i))
                channels.append(
                    AudioChannel(
                        name=f"A{i}",
                        volume=max(0.0, min(1.0, (gain + 40) / 40)),
                        muted=mute,
                    )
                )

            # Read virtual outputs (B1, B2, B3, B4, B5)
            for i in range(1, 6):
                gain = vm.get_output_gain(i)
                mute = bool(vm.get_output_mute(i))
                channels.append(
                    AudioChannel(
                        name=f"B{i}",
                        volume=max(0.0, min(1.0, (gain + 40) / 40)),
                        muted=mute,
                    )
                )

            vm.disconnect()

            return AudioMixerState(
                backend="voicemeeter",
                master_volume=sum(c.volume for c in channels) / max(len(channels), 1),
                master_muted=all(c.muted for c in channels) if channels else False,
                channels=channels,
            )
        except ImportError:
            logger.warning("voicemeeter-api not installed — Voicemeeter unavailable")
            return AudioMixerState(backend="voicemeeter")
        except Exception as e:
            logger.error(f"Voicemeeter error: {e}")
            return AudioMixerState(backend="voicemeeter")

    async def set_volume(self, channel: str, value: float) -> None:
        try:
            import voicemeeter_api

            vm = voicemeeter_api.VoicemeeterRemote()
            vm.connect()

            # Map channel names to Voicemeeter strip indices
            if channel.startswith("A"):
                strip = int(channel[1])
                # Convert 0-1 to Voicemeeter gain (-40 to 0 dB)
                gain = (value * -40) + 40
                vm.set_input_gain(strip, gain)
            elif channel.startswith("B"):
                strip = int(channel[1])
                gain = (value * -40) + 40
                vm.set_output_gain(strip, gain)

            vm.disconnect()
        except Exception as e:
            logger.error(f"Failed to set Voicemeeter volume for {channel}: {e}")

    async def set_mute(self, channel: str, muted: bool) -> None:
        try:
            import voicemeeter_api

            vm = voicemeeter_api.VoicemeeterRemote()
            vm.connect()

            if channel.startswith("A"):
                strip = int(channel[1])
                vm.set_input_mute(strip, int(muted))
            elif channel.startswith("B"):
                strip = int(channel[1])
                vm.set_output_mute(strip, int(muted))

            vm.disconnect()
        except Exception as e:
            logger.error(f"Failed to set Voicemeeter mute for {channel}: {e}")


# ---------------------------------------------------------------------------
# Manager (orchestrates backends and polling)
# ---------------------------------------------------------------------------


class AudioMixerManager:
    """Manages audio mixer backends and periodic state polling."""

    def __init__(self) -> None:
        self._backends: list[AudioMixerBackend] = []
        self._current_state: AudioMixerState = AudioMixerState(backend="windows")
        self._poll_interval: float = 2.0

        # Initialize available backends
        self._init_backends()

    def _init_backends(self) -> None:
        """Try to initialize backends in priority order."""
        # Voicemeeter first (primary focus)
        try:
            self._backends.append(VoicemeeterBackend())
            logger.info("Voicemeeter backend initialized")
        except Exception:
            logger.debug("Voicemeeter not available")

        # Windows Core Audio as fallback
        try:
            self._backends.append(WindowsAudioBackend())
            logger.info("Windows audio backend initialized")
        except Exception:
            logger.debug("Windows audio not available")

        if not self._backends:
            logger.warning("No audio backends available")

    async def run(self) -> None:
        """Periodically poll audio state."""
        while True:
            try:
                self._current_state = await self._get_best_state()
            except Exception as e:
                logger.error(f"Audio poll error: {e}")
            await asyncio.sleep(self._poll_interval)

    async def _get_best_state(self) -> AudioMixerState:
        """Get state from the best available backend."""
        for backend in self._backends:
            try:
                state = await backend.get_state()
                if state.channels or state.master_volume > 0 or state.master_muted:
                    return state
            except Exception:
                continue
        return self._current_state

    def get_state(self) -> dict:
        """Return current state as a serializable dict."""
        return {
            "backend": self._current_state.backend,
            "master_volume": self._current_state.master_volume,
            "master_muted": self._current_state.master_muted,
            "channels": [
                {
                    "name": ch.name,
                    "volume": ch.volume,
                    "muted": ch.muted,
                    "peak_left": ch.peak_left,
                    "peak_right": ch.peak_right,
                }
                for ch in self._current_state.channels
            ],
        }

    async def set_volume(self, bus: str, value: float) -> None:
        """Set volume on the active backend."""
        for backend in self._backends:
            try:
                await backend.set_volume(bus, value)
                return
            except Exception:
                continue

    async def set_mute(self, bus: str, muted: bool) -> None:
        """Set mute on the active backend."""
        for backend in self._backends:
            try:
                await backend.set_mute(bus, muted)
                return
            except Exception:
                continue