Trixy7/trixy_core/audio/volume.py

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

Steuert die System-Lautstaerke ueber amixer fuer
Playback (Lautsprecher) und Capture (Mikrofon).

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-Ein-/Ausgaben
(TTS, Musik, Sounds, Wakeword-Erkennung).

Quellen-Konzept:
    Aenderungen kennen eine "source" (z.B. "user", "duck",
    "hid"). Nur Aenderungen mit source="user" oder "hid"
    werden in client_config.json persistiert. Ducking
    (source="duck") aendert nur die ALSA-Lautstaerke,
    nicht die gespeicherten Werte.
"""

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


# Quellen die in die Config persistiert werden
PERSIST_SOURCES = {"user", "hid"}


class _MixerChannel:
    """Ein einzelner ALSA-Mixer-Kanal (Playback ODER Capture)."""

    def __init__(
        self,
        kind: str,
        control: str = "",
        capture: bool = False,
    ) -> None:
        self.kind = kind  # "output" | "input"
        self.control = control
        self.capture = capture
        self.available = False
        self.muted = False

    async def detect(self) -> None:
        """Erkennt einen passenden Mixer-Control falls keiner gesetzt ist."""
        if self.control:
            self.available = await _control_exists(self.control)
            return
        if self.capture:
            candidates = ["Capture", "Mic", "Microphone", "Front Mic", "Internal Mic"]
        else:
            candidates = ["Master", "PCM", "Speaker", "Headphone"]

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

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

    async def get_volume(self) -> int:
        if not self.available:
            return 0
        try:
            output = await _amixer("get", self.control)
            for line in output.splitlines():
                if "%" in line:
                    start = line.index("[") + 1
                    end = line.index("%", start)
                    return int(line[start:end])
        except Exception:
            pass
        return 0

    async def set_volume(self, percent: int) -> None:
        if not self.available:
            return
        percent = max(0, min(100, percent))
        await _amixer("set", self.control, f"{percent}%")
        pdebug(f"ALSA {self.kind} Volume: {percent}%")

    async def set_mute(self, muted: bool) -> None:
        if not self.available:
            return
        await _amixer("set", self.control, "mute" if muted else "unmute")
        self.muted = muted
        pdebug(f"ALSA {self.kind} {'MUTE' if muted else 'UNMUTE'}")

    async def read_mute(self) -> bool:
        if not self.available:
            return False
        output = await _amixer("get", self.control)
        self.muted = "[off]" in output
        return self.muted


async def _amixer(*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 _control_exists(name: str) -> bool:
    output = await _amixer("scontrols")
    return f"'{name}'" in output


class VolumeHandler:
    """
    ALSA Volume Handler fuer Playback und Capture.

    Persistiert Aenderungen mit source in PERSIST_SOURCES
    in client_config.json. Ducking-Aenderungen (source="duck")
    werden NICHT persistiert.
    """

    def __init__(
        self,
        application: "BaseApplication",
        output_control: str = "",
        input_control: str = "",
        config_name: str = "client",
    ) -> None:
        self._application = application
        self._output = _MixerChannel("output", output_control, capture=False)
        self._input = _MixerChannel("input", input_control, capture=True)
        self._available = False
        self._config_name = config_name

    @property
    def available(self) -> bool:
        return self._available

    @property
    def output(self) -> _MixerChannel:
        return self._output

    @property
    def input(self) -> _MixerChannel:
        return self._input

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

        await self._output.detect()
        await self._input.detect()

        if not (self._output.available or self._input.available):
            pdebug("Kein ALSA Mixer Control gefunden")
            return

        self._available = True

        # Config-Werte beim Start anwenden
        await self._apply_config()

        events = getattr(self._application, "events", None)
        if events:
            events.register("volume_change", self._on_volume_change)
            events.register("mute_toggle", self._on_mute_toggle)
            events.register("audio_volume_set", self._on_volume_set)
            events.register("audio_mute_set", self._on_mute_set)

        out_v = await self._output.get_volume()
        in_v = await self._input.get_volume()
        pinfo(
            f"Volume Handler gestartet — "
            f"Out: {self._output.control} ({out_v}%{', MUTE' if self._output.muted else ''}) "
            f"In: {self._input.control} ({in_v}%{', MUTE' if self._input.muted else ''})"
        )

    async def stop(self) -> None:
        events = getattr(self._application, "events", None)
        if events:
            events.unregister("volume_change", self._on_volume_change)
            events.unregister("mute_toggle", self._on_mute_toggle)
            events.unregister("audio_volume_set", self._on_volume_set)
            events.unregister("audio_mute_set", self._on_mute_set)

    # === Config-Anbindung ===

    def _audio_config(self):
        cfg = getattr(self._application, "_client_config", None)
        if cfg is None:
            cfg = getattr(self._application, "client_config", None)
        return getattr(cfg, "audio", None) if cfg else None

    async def _apply_config(self) -> None:
        """Wendet die in client_config.json gespeicherten Werte auf ALSA an."""
        ac = self._audio_config()
        if ac is None:
            return
        if self._output.available:
            await self._output.set_volume(int(ac.output_volume))
            await self._output.set_mute(bool(ac.output_muted))
        if self._input.available:
            await self._input.set_volume(int(ac.input_volume))
            await self._input.set_mute(bool(ac.input_muted))

    def _persist(self, **changes) -> None:
        """Schreibt Aenderungen in client_config.json."""
        ac = self._audio_config()
        if ac is None:
            return
        for key, value in changes.items():
            if hasattr(ac, key):
                setattr(ac, key, value)
        cm = getattr(self._application, "config_manager", None)
        if cm is None:
            return
        try:
            cm.save(self._config_name)
        except Exception as e:
            perror(f"Volume-Persistenz fehlgeschlagen: {e}")

    # === Public API ===

    async def set_output_volume(self, percent: int, source: str = "user") -> None:
        if not self._output.available:
            return
        await self._output.set_volume(percent)
        if source in PERSIST_SOURCES:
            self._persist(output_volume=max(0, min(100, percent)))
        await self._emit_changed("output", percent, source)

    async def set_input_volume(self, percent: int, source: str = "user") -> None:
        if not self._input.available:
            return
        await self._input.set_volume(percent)
        if source in PERSIST_SOURCES:
            self._persist(input_volume=max(0, min(100, percent)))
        await self._emit_changed("input", percent, source)

    async def set_output_mute(self, muted: bool, source: str = "user") -> None:
        if not self._output.available:
            return
        await self._output.set_mute(muted)
        if source in PERSIST_SOURCES:
            self._persist(output_muted=muted)

    async def set_input_mute(self, muted: bool, source: str = "user") -> None:
        if not self._input.available:
            return
        await self._input.set_mute(muted)
        if source in PERSIST_SOURCES:
            self._persist(input_muted=muted)

    async def restore_from_config(self, channel: str = "output") -> None:
        """Stellt die in der Config gespeicherte Lautstaerke wieder her.

        Wird nach Ducking aufgerufen — nutzt config-Werte statt
        temporaerer Variablen (solider).
        """
        ac = self._audio_config()
        if ac is None:
            return
        if channel in ("output", "all") and self._output.available:
            await self._output.set_volume(int(ac.output_volume))
            await self._output.set_mute(bool(ac.output_muted))
        if channel in ("input", "all") and self._input.available:
            await self._input.set_volume(int(ac.input_volume))
            await self._input.set_mute(bool(ac.input_muted))

    async def get_status(self) -> dict:
        """Gibt aktuellen Status zurueck — fuer F6 Audio View."""
        return {
            "output": {
                "available": self._output.available,
                "control": self._output.control,
                "volume": await self._output.get_volume() if self._output.available else 0,
                "muted": await self._output.read_mute() if self._output.available else False,
            },
            "input": {
                "available": self._input.available,
                "control": self._input.control,
                "volume": await self._input.get_volume() if self._input.available else 0,
                "muted": await self._input.read_mute() if self._input.available else False,
            },
        }

    # === Event-Handler ===

    async def _on_volume_change(self, event_name: str, data: dict) -> None:
        """Handler fuer relative Aenderungen (HID Hotkeys)."""
        direction = data.get("direction", "")
        step = data.get("step", 0.05)
        target = data.get("target", "output")
        step_pct = max(1, int(step * 100))

        channel = self._output if target == "output" else self._input
        if not channel.available:
            return
        current = await channel.get_volume()
        new_value = current + step_pct if direction == "up" else current - step_pct
        if target == "output":
            await self.set_output_volume(new_value, source="hid")
        else:
            await self.set_input_volume(new_value, source="hid")

    async def _on_mute_toggle(self, event_name: str, data: dict) -> None:
        mute_type = data.get("type", "speaker")
        if mute_type == "speaker":
            await self.set_output_mute(not self._output.muted, source="hid")
        elif mute_type in ("mic", "microphone"):
            await self.set_input_mute(not self._input.muted, source="hid")

    async def _on_volume_set(self, event_name: str, data: dict) -> None:
        """Handler fuer absolute Volume-Aenderungen (z.B. F6 Audio View, Ducking)."""
        target = data.get("target", "output")
        percent = int(data.get("volume", 0))
        source = data.get("source", "user")
        if target == "output":
            await self.set_output_volume(percent, source=source)
        else:
            await self.set_input_volume(percent, source=source)

    async def _on_mute_set(self, event_name: str, data: dict) -> None:
        target = data.get("target", "output")
        muted = bool(data.get("muted", False))
        source = data.get("source", "user")
        if target == "output":
            await self.set_output_mute(muted, source=source)
        else:
            await self.set_input_mute(muted, source=source)

    async def _emit_changed(self, target: str, percent: int, source: str) -> None:
        events = getattr(self._application, "events", None)
        if events:
            await events.emit("audio_level_changed", {
                "target": target,
                "volume": percent,
                "source": source,
            })

    # === Legacy-Kompatibilitaet ===

    @property
    def mixer_control(self) -> str:
        return self._output.control

    @property
    def is_muted(self) -> bool:
        return self._output.muted

    async def get_volume(self) -> int:
        return await self._output.get_volume()

    async def set_volume(self, percent: int) -> None:
        await self.set_output_volume(percent, source="user")

    async def volume_up(self, step_percent: int = 5) -> None:
        current = await self._output.get_volume()
        await self.set_output_volume(current + step_percent, source="hid")

    async def volume_down(self, step_percent: int = 5) -> None:
        current = await self._output.get_volume()
        await self.set_output_volume(current - step_percent, source="hid")

    async def mute(self) -> None:
        await self.set_output_mute(True, source="user")

    async def unmute(self) -> None:
        await self.set_output_mute(False, source="user")

    async def toggle_mute(self) -> None:
        await self.set_output_mute(not self._output.muted, source="user")