Trixy7/plugins/audio_ducking/main.py

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

Reduziert die Musik-Lautstaerke wenn das Wakeword erkannt wird
und stellt sie nach der Conversation wieder her.

Flow:
  Wakeword erkannt → Musik auf 20% (duck)
  Conversation laeuft → Musik bleibt leise
  Conversation beendet → Musik zurueck auf Original (restore)
"""

from __future__ import annotations

from typing import Any

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 Musik-Lautstaerke bei Wakeword/Conversation."""

    NAME = "audio_ducking"
    VERSION = "1.0.0"
    DESCRIPTION = "Musik leiser bei Sprachinteraktion"
    AUTHOR = "Trixy"

    def __init__(self, application, plugin_path, config=None) -> None:
        super().__init__(application, plugin_path, config)
        self._original_volume: float | None = None
        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_player(self):
        """Gibt den MusicPlayer zurueck (mehrere Wege probieren)."""
        # Weg 1: Ueber ServiceContainer
        try:
            for name, service in self.application.services.services.items():
                if "music" in name.lower() or "Music" in type(service).__name__:
                    player = getattr(service, "_player", None)
                    if player:
                        return player
        except Exception:
            pass

        # Weg 2: Direkt am Application-Objekt
        try:
            svc = getattr(self.application, "_music_player_service", None)
            if svc:
                return getattr(svc, "_player", None)
        except Exception:
            pass

        return None

    async def _duck_volume(self) -> None:
        """Reduziert die Musik-Lautstaerke."""
        player = self._get_player()
        if not player:
            pdebug("AudioDucking: Kein MusicPlayer gefunden")
            return

        if not player.is_playing:
            pdebug("AudioDucking: Keine Musik aktiv")
            return

        if self._ducked:
            return

        duck_pct = self.get_config_value("duck_volume_percent", 20)
        self._original_volume = player.volume
        duck_vol = duck_pct / 100.0

        await player.set_volume(duck_vol)
        self._ducked = True
        pinfo(f"AudioDucking: {self._original_volume:.0%} → {duck_vol:.0%}")

    async def _restore_volume(self) -> None:
        """Stellt die urspruengliche Lautstaerke wieder her."""
        if not self._ducked or self._original_volume is None:
            return

        player = self._get_player()
        if player:
            await player.set_volume(self._original_volume)
            pinfo(f"AudioDucking: Wiederhergestellt auf {self._original_volume:.0%}")

        self._original_volume = None
        self._ducked = False

    # === Events ===

    @TrixyEvent(["wakeword_detected"], priority=EventPriority.HIGHEST)
    async def on_wakeword(self, event_name: str, event_data) -> None:
        """Wakeword erkannt → Musik leiser."""
        pinfo("AudioDucking: Wakeword erkannt")
        await self._duck_volume()

    @TrixyEvent(["wakeword_manual_trigger"], priority=EventPriority.HIGHEST)
    async def on_manual_trigger(self, event_name: str, event_data) -> None:
        """Manueller Trigger → Musik leiser."""
        pinfo("AudioDucking: Manueller Trigger")
        await self._duck_volume()

    @TrixyEvent(["recording_complete"])
    async def on_recording_complete(self, event_name: str, event_data) -> None:
        """Aufnahme beendet → Musik zurueck auf normal."""
        pinfo("AudioDucking: Recording complete")
        await self._restore_volume()

    @TrixyEvent(["conversation_started"])
    async def on_conversation_started(self, event_name: str, event_data) -> None:
        """Conversation gestartet → sicherstellen dass Musik leise."""
        await self._duck_volume()