Trixy7/plugins/audio_ducking/main.py

# -*- coding: utf-8 -*-
"""
Audio-Ducking Plugin.

Reduziert die System-Lautstaerke (ALSA/PulseAudio) wenn das
Wakeword erkannt wird und stellt sie nach der Conversation
wieder her.

Flow:
  Wakeword erkannt → ALSA Output auf duck_volume_percent (duck)
  Conversation laeuft → Lautstaerke bleibt reduziert
  Conversation beendet → Restore aus client_config.json

Wichtig:
  - Ducking aendert NUR die ALSA-Lautstaerke, NICHT den
    in client_config.json gespeicherten output_volume Wert.
  - Restore liest die gespeicherten Werte aus der Config
    statt einer temporaer gespeicherten Variable. Das ist
    solider und kommt auch nach Crashes wieder zurecht.
  - Im Server-Mode wird das Ducking spaeter ueber eine
    Satellite-Audio-Update Nachricht an den aktiven
    Satellite weitergereicht. Aktuell unterstuetzt das
    Plugin nur Standalone/Client-Mode (lokaler VolumeHandler).
"""

from __future__ import annotations

from trixy_core.events.decorators import TrixyEvent
from trixy_core.events.enums import EventPriority
from trixy_core.plugins.trixy_plugin import TrixyPlugin
from trixy_core.utils.debug import pdebug, pinfo


class AudioDuckingPlugin(TrixyPlugin):
    """Reduziert System-Lautstaerke bei Sprachinteraktion."""

    NAME = "audio_ducking"
    VERSION = "2.0.0"
    DESCRIPTION = "ALSA-Lautstaerke leiser bei Sprachinteraktion"
    AUTHOR = "Trixy"

    def __init__(self, application, plugin_path, config=None) -> None:
        super().__init__(application, plugin_path, config)
        self._ducked = False

    async def on_load(self) -> None:
        pinfo(f"AudioDucking: Lade v{self.VERSION}")

    async def on_unload(self) -> None:
        if self._ducked:
            await self._restore_volume()

    def _get_volume_handler(self):
        """Gibt den lokalen VolumeHandler zurueck (falls vorhanden)."""
        return getattr(self.application, "_volume_handler", None)

    async def _duck_volume(self) -> None:
        """Reduziert die ALSA-Lautstaerke ohne Persistenz."""
        if self._ducked:
            return

        handler = self._get_volume_handler()
        if handler is None or not handler.available:
            pdebug("AudioDucking: Kein VolumeHandler verfuegbar")
            return

        duck_pct = int(self.get_config_value("duck_volume_percent", 20))
        # source="duck" → Volume wird NICHT in Config persistiert
        await handler.set_output_volume(duck_pct, source="duck")
        self._ducked = True
        pinfo(f"AudioDucking: ALSA Output → {duck_pct}%")

    async def _restore_volume(self) -> None:
        """Stellt die in der Config gespeicherte Lautstaerke wieder her."""
        if not self._ducked:
            return

        handler = self._get_volume_handler()
        if handler is not None and handler.available:
            # Liest output_volume aus client_config.json — solider als
            # eine temporaer gespeicherte Variable, ueberlebt Crashes.
            await handler.restore_from_config("output")
            pinfo("AudioDucking: ALSA Output aus Config wiederhergestellt")

        self._ducked = False

    # === Events ===

    @TrixyEvent(["wakeword_detected"], priority=EventPriority.HIGHEST)
    async def on_wakeword(self, event_name: str, event_data) -> None:
        await self._duck_volume()

    @TrixyEvent(["wakeword_manual_trigger"], priority=EventPriority.HIGHEST)
    async def on_manual_trigger(self, event_name: str, event_data) -> None:
        await self._duck_volume()

    @TrixyEvent(["recording_complete"])
    async def on_recording_complete(self, event_name: str, event_data) -> None:
        await self._restore_volume()

    @TrixyEvent(["conversation_started"])
    async def on_conversation_started(self, event_name: str, event_data) -> None:
        await self._duck_volume()

    @TrixyEvent(["conversation_ended"])
    async def on_conversation_ended(self, event_name: str, event_data) -> None:
        await self._restore_volume()