Trixy7/trixy_core/audio/volume.py

# -*- coding: utf-8 -*-
"""
ALSA Volume Handler.

Steuert die System-Lautstaerke ueber amixer.
Reagiert auf volume_change und mute_toggle Events
von HID-Geraeten oder anderen Quellen.

Laeuft auf Client und Standalone — regelt die
Hardware-Lautstaerke fuer alle Audio-Ausgaben
(TTS, Musik, Sounds).
"""

from __future__ import annotations

import asyncio
import shutil
from typing import TYPE_CHECKING

from trixy_core.utils.debug import pinfo, pdebug, perror

if TYPE_CHECKING:
    from trixy_core.application import BaseApplication


class VolumeHandler:
    """
    ALSA Volume Handler.

    Registriert sich auf volume_change und mute_toggle Events
    und steuert den System-Mixer via amixer.

    Beispiel:
        ```python
        handler = VolumeHandler(app, mixer_control="Master")
        await handler.start()

        # Wird automatisch bei HID-Events aufgerufen:
        # volume_change {direction: "up", step: 0.05}
        # mute_toggle {type: "speaker"}
        ```
    """

    def __init__(
        self,
        application: "BaseApplication",
        mixer_control: str = "",
    ) -> None:
        """
        Args:
            application: Anwendungsreferenz
            mixer_control: ALSA Mixer Control (z.B. "Master", "PCM")
                          Leer = automatisch erkennen
        """
        self._application = application
        self._mixer_control = mixer_control
        self._muted = False
        self._available = False

    @property
    def available(self) -> bool:
        """True wenn amixer verfuegbar und Mixer-Control gefunden."""
        return self._available

    @property
    def mixer_control(self) -> str:
        """Name des aktiven ALSA Mixer Controls."""
        return self._mixer_control

    @property
    def is_muted(self) -> bool:
        """Ob der Mixer stummgeschaltet ist."""
        return self._muted

    async def start(self) -> None:
        """Startet den Volume Handler und registriert Event-Listener."""
        if not shutil.which("amixer"):
            pdebug("amixer nicht gefunden — Volume Handler deaktiviert")
            return

        # Mixer-Control erkennen
        if not self._mixer_control:
            self._mixer_control = await self._detect_mixer_control()

        if not self._mixer_control:
            pdebug("Kein ALSA Mixer Control gefunden")
            return

        self._available = True

        # Aktuellen Mute-Status lesen
        self._muted = await self._get_mute_state()

        # Events registrieren
        events = getattr(self._application, "events", None)
        if events:
            events.register("volume_change", self._on_volume_change)
            events.register("mute_toggle", self._on_mute_toggle)

        volume = await self.get_volume()
        pinfo(
            f"Volume Handler gestartet: {self._mixer_control} "
            f"({volume}%{', MUTE' if self._muted else ''})"
        )

    async def stop(self) -> None:
        """Stoppt den Volume Handler."""
        events = getattr(self._application, "events", None)
        if events:
            events.unregister("volume_change", self._on_volume_change)
            events.unregister("mute_toggle", self._on_mute_toggle)

    # === Lautstaerke-Steuerung ===

    async def get_volume(self) -> int:
        """Gibt die aktuelle Lautstaerke in Prozent zurueck (0-100)."""
        if not self._available:
            return 0

        try:
            proc = await asyncio.create_subprocess_exec(
                "amixer", "get", self._mixer_control,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            stdout, _ = await proc.communicate()
            output = stdout.decode()

            # "[75%]" aus der Ausgabe parsen
            for line in output.splitlines():
                if "%" in line:
                    start = line.index("[") + 1
                    end = line.index("%")
                    return int(line[start:end])
        except Exception:
            pass
        return 0

    async def set_volume(self, percent: int) -> None:
        """Setzt die Lautstaerke in Prozent (0-100)."""
        if not self._available:
            return

        percent = max(0, min(100, percent))
        await self._amixer("set", self._mixer_control, f"{percent}%")
        pdebug(f"Volume: {percent}%")

        # Event emittieren
        events = getattr(self._application, "events", None)
        if events:
            await events.emit("audio_level_changed", {
                "volume": percent,
                "source": "alsa",
            })

    async def volume_up(self, step_percent: int = 5) -> None:
        """Erhoeht die Lautstaerke."""
        await self._amixer("set", self._mixer_control, f"{step_percent}%+")
        volume = await self.get_volume()
        pdebug(f"Volume: +{step_percent}% → {volume}%")

    async def volume_down(self, step_percent: int = 5) -> None:
        """Verringert die Lautstaerke."""
        await self._amixer("set", self._mixer_control, f"{step_percent}%-")
        volume = await self.get_volume()
        pdebug(f"Volume: -{step_percent}% → {volume}%")

    async def mute(self) -> None:
        """Schaltet stumm."""
        await self._amixer("set", self._mixer_control, "mute")
        self._muted = True
        pdebug("Volume: MUTE")

    async def unmute(self) -> None:
        """Hebt Stummschaltung auf."""
        await self._amixer("set", self._mixer_control, "unmute")
        self._muted = False
        pdebug("Volume: UNMUTE")

    async def toggle_mute(self) -> None:
        """Schaltet Stummschaltung um."""
        if self._muted:
            await self.unmute()
        else:
            await self.mute()

    # === Event-Handler ===

    async def _on_volume_change(self, event_name: str, data: dict) -> None:
        """Handler fuer volume_change Event (von HID oder anderen Quellen)."""
        direction = data.get("direction", "")
        step = data.get("step", 0.05)
        step_pct = max(1, int(step * 100))

        if direction == "up":
            await self.volume_up(step_pct)
        elif direction == "down":
            await self.volume_down(step_pct)

    async def _on_mute_toggle(self, event_name: str, data: dict) -> None:
        """Handler fuer mute_toggle Event."""
        mute_type = data.get("type", "speaker")

        # Nur auf Lautsprecher-Mute reagieren, nicht auf Mikrofon
        if mute_type == "speaker":
            await self.toggle_mute()

    # === Hilfsmethoden ===

    async def _amixer(self, *args: str) -> str:
        """Fuehrt einen amixer-Befehl aus."""
        try:
            proc = await asyncio.create_subprocess_exec(
                "amixer", *args,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            stdout, _ = await proc.communicate()
            return stdout.decode()
        except Exception as e:
            perror(f"amixer Fehler: {e}")
            return ""

    async def _detect_mixer_control(self) -> str:
        """Erkennt den passenden ALSA Mixer Control."""
        # Reihenfolge: Master > PCM > Speaker > Headphone
        candidates = ["Master", "PCM", "Speaker", "Headphone"]

        output = await self._amixer("scontrols")
        for name in candidates:
            if f"'{name}'" in output:
                pdebug(f"ALSA Mixer Control erkannt: {name}")
                return name

        # Fallback: ersten Control nehmen
        for line in output.splitlines():
            if "Simple mixer control" in line:
                # "Simple mixer control 'Master',0" → "Master"
                start = line.index("'") + 1
                end = line.index("'", start)
                name = line[start:end]
                pdebug(f"ALSA Mixer Control (Fallback): {name}")
                return name

        return ""

    async def _get_mute_state(self) -> bool:
        """Liest den aktuellen Mute-Status."""
        output = await self._amixer("get", self._mixer_control)
        return "[off]" in output