Trixy7/trixy_core/arbitration/arbitrator.py

# -*- coding: utf-8 -*-
"""
Wakeword-Arbitrator Service.

Zentraler Service für die Wakeword-Arbitration.
Koordiniert das Zeitfenster, die Strategieauswahl und
die Kommunikation mit Satellites.
"""

import asyncio
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, Awaitable, TYPE_CHECKING

from trixy_core.arbitration.candidate import ArbitrationCandidate, CandidateState
from trixy_core.arbitration.window import ArbitrationWindow, WindowConfig, WindowState
from trixy_core.arbitration.strategies import (
    ArbitrationStrategy,
    LoudestStrategy,
    CombinedStrategy,
)
from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

if TYPE_CHECKING:
    from trixy_core.satellite import Satellite
    from trixy_core.application import IApplication


@dataclass
class ArbitrationConfig:
    """Konfiguration für den Arbitrator."""
    # Zeitfenster
    window_duration_ms: int = 1000
    min_wait_ms: int = 100
    max_candidates: int = 10

    # Schwellenwerte
    immediate_threshold: float = 0.95
    min_audio_level: float = 0.1
    min_confidence: float = 0.5

    # Strategie
    strategy: str = "combined"  # "loudest", "confidence", "combined", "proximity"

    # Combined-Strategie Gewichtungen
    audio_weight: float = 0.6
    confidence_weight: float = 0.3
    recency_weight: float = 0.1

    # Verhalten
    auto_start_conversation: bool = True
    notify_rejected: bool = True


@dataclass
class ArbitrationResult:
    """Ergebnis einer Arbitration."""
    # Ergebnis
    success: bool = False
    winner: ArbitrationCandidate | None = None
    rejected: list[ArbitrationCandidate] = field(default_factory=list)

    # Session
    conversation_id: str = ""
    arbitration_id: str = ""

    # Timing
    started_at: datetime = field(default_factory=datetime.now)
    decided_at: datetime | None = None
    duration_ms: float = 0.0

    # Details
    candidate_count: int = 0
    strategy_used: str = ""
    close_reason: str = ""

    def __post_init__(self) -> None:
        """Berechnet Dauer wenn decided_at gesetzt."""
        if self.decided_at:
            delta = self.decided_at - self.started_at
            self.duration_ms = delta.total_seconds() * 1000

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "success": self.success,
            "winner": self.winner.to_dict() if self.winner else None,
            "rejected_count": len(self.rejected),
            "conversation_id": self.conversation_id,
            "arbitration_id": self.arbitration_id,
            "duration_ms": self.duration_ms,
            "candidate_count": self.candidate_count,
            "strategy_used": self.strategy_used,
            "close_reason": self.close_reason,
        }


class WakewordArbitrator:
    """
    Wakeword-Arbitrator für Multi-Satellite-Entscheidung.

    Koordiniert die Auswahl eines Satellites wenn mehrere
    gleichzeitig ein Wakeword erkennen.

    Ablauf:
    1. Erstes WakewordDetected öffnet Arbitration-Fenster
    2. Weitere Erkennungen werden als Kandidaten hinzugefügt
    3. Nach Ablauf des Fensters wird Strategie angewendet
    4. Gewinner erhält WakewordSelected
    5. Verlierer erhalten WakewordAbort

    Kann als eigenständige Komponente oder als Service verwendet werden.

    Beispiel:
        ```python
        # Standalone-Verwendung
        arbitrator = WakewordArbitrator(config)

        # Bei WakewordDetected
        candidate = ArbitrationCandidate(
            satellite_id="living_room",
            audio_level=0.85,
            confidence=0.9,
        )
        await arbitrator.add_candidate(candidate)

        # Callback für Entscheidung
        @arbitrator.on_decision
        async def handle_decision(result):
            print(f"Gewinner: {result.winner.satellite_id}")
        ```
    """

    # Service-Metadaten (für optionale Service-Integration)
    NAME = "wakeword_arbitrator"
    PRIORITY = 30  # MANAGER
    GROUP = "manager"
    DEPENDENCIES = ["event_manager", "satellite_manager"]

    def __init__(
        self,
        config: ArbitrationConfig | None = None,
        application: "IApplication | None" = None,
    ) -> None:
        """
        Initialisiert den Arbitrator.

        Args:
            config: Arbitrator-Konfiguration
            application: Anwendungsreferenz (optional)
        """
        self.config = config or ArbitrationConfig()
        self.application = application

        # Strategie erstellen
        self._strategy = self._create_strategy()

        # Aktuelles Fenster
        self._window: ArbitrationWindow | None = None
        self._current_arbitration_id: str = ""

        # Callbacks
        self._on_decision_callbacks: list[
            Callable[[ArbitrationResult], Awaitable[None]]
        ] = []

        # Statistiken
        self._stats = {
            "total_arbitrations": 0,
            "successful": 0,
            "single_candidate": 0,
            "multiple_candidates": 0,
            "average_candidates": 0.0,
            "average_duration_ms": 0.0,
        }

        # Lock für Thread-Sicherheit
        self._lock = asyncio.Lock()

    def _create_strategy(self) -> ArbitrationStrategy:
        """Erstellt die konfigurierte Strategie."""
        strategy_name = self.config.strategy.lower()

        if strategy_name == "loudest":
            return LoudestStrategy(
                min_audio_level=self.config.min_audio_level,
            )
        elif strategy_name == "confidence":
            from trixy_core.arbitration.strategies import ConfidenceStrategy
            return ConfidenceStrategy(
                min_confidence=self.config.min_confidence,
            )
        elif strategy_name == "proximity":
            from trixy_core.arbitration.strategies import ProximityStrategy
            return ProximityStrategy(
                min_audio_level=self.config.min_audio_level,
                min_confidence=self.config.min_confidence,
            )
        else:  # combined (default)
            from trixy_core.arbitration.strategies.combined import CombinedWeights
            weights = CombinedWeights(
                audio_level=self.config.audio_weight,
                confidence=self.config.confidence_weight,
                recency=self.config.recency_weight,
            )
            return CombinedStrategy(
                weights=weights,
                min_audio_level=self.config.min_audio_level,
                min_confidence=self.config.min_confidence,
            )

    @property
    def strategy(self) -> ArbitrationStrategy:
        """Aktuelle Strategie."""
        return self._strategy

    @strategy.setter
    def strategy(self, strategy: ArbitrationStrategy) -> None:
        """Setzt neue Strategie."""
        self._strategy = strategy

    @property
    def is_arbitrating(self) -> bool:
        """Prüft ob Arbitration aktiv ist."""
        return self._window is not None and self._window.is_open

    @property
    def current_candidates(self) -> list[ArbitrationCandidate]:
        """Aktuelle Kandidaten."""
        if self._window:
            return self._window.candidates
        return []

    @property
    def stats(self) -> dict[str, Any]:
        """Statistiken."""
        return self._stats.copy()

    async def start(self) -> None:
        """
        Startet den Arbitrator.

        Registriert Event-Handler wenn Application verfügbar.
        """
        pinfo(
            f"WakewordArbitrator gestartet mit Strategie: {self._strategy.name}"
        )

        # Event-Handler registrieren wenn Application vorhanden
        if self.application and hasattr(self.application, "events"):
            self.application.events.register(
                "wakeword_detected",
                self._on_wakeword_detected,
            )

    async def stop(self) -> None:
        """
        Stoppt den Arbitrator.

        Bricht laufende Arbitration ab und entfernt Event-Handler.
        """
        # Laufende Arbitration abbrechen
        if self._window and self._window.is_open:
            await self._window.close(reason="service_shutdown")
            await self._window.reset()

        # Event-Handler entfernen
        if self.application and hasattr(self.application, "events"):
            self.application.events.unregister(
                "wakeword_detected",
                self._on_wakeword_detected,
            )

        pinfo("WakewordArbitrator gestoppt")

    async def _on_wakeword_detected(
        self,
        event_name: str,
        event_data: dict[str, Any],
    ) -> None:
        """
        Handler für wakeword_detected Event.

        Wird automatisch aufgerufen wenn ein Satellite
        ein Wakeword erkennt.
        """
        # Satellite-Daten extrahieren
        satellite_id = event_data.get("satellite_id", "")
        if not satellite_id:
            return

        # Kandidat erstellen
        candidate = ArbitrationCandidate(
            satellite_id=satellite_id,
            room_id=event_data.get("room_id", ""),
            alias=event_data.get("alias", ""),
            wakeword_model=event_data.get("wakeword_model", ""),
            confidence=event_data.get("confidence", 0.0),
            audio_level=event_data.get("audio_level", 0.0),
            buffer_samples=event_data.get("buffer_samples", 0),
            forced=event_data.get("forced", False),
        )

        # Zur Arbitration hinzufügen
        await self.add_candidate(candidate)

    async def add_candidate(
        self,
        candidate: ArbitrationCandidate,
    ) -> bool:
        """
        Fügt einen Kandidaten zur Arbitration hinzu.

        Öffnet bei Bedarf ein neues Arbitration-Fenster.

        Args:
            candidate: Der hinzuzufügende Kandidat

        Returns:
            True wenn erfolgreich hinzugefügt
        """
        async with self._lock:
            # Neues Fenster öffnen wenn nötig
            if not self._window or not self._window.is_open:
                self._window = ArbitrationWindow(
                    config=WindowConfig(
                        window_duration_ms=self.config.window_duration_ms,
                        min_wait_ms=self.config.min_wait_ms,
                        max_candidates=self.config.max_candidates,
                        immediate_threshold=self.config.immediate_threshold,
                    ),
                    on_complete=self._on_window_complete,
                )
                self._current_arbitration_id = f"arb_{uuid.uuid4().hex[:8]}"
                await self._window.open()

                pdebug(
                    f"Arbitration-Fenster geöffnet: {self._current_arbitration_id}"
                )

            # Kandidat hinzufügen
            success = await self._window.add_candidate(candidate)

            if success:
                pdebug(
                    f"Kandidat hinzugefügt: {candidate.satellite_id} "
                    f"(audio={candidate.audio_level:.2f}, "
                    f"conf={candidate.confidence:.2f}"
                    f"{', FORCED' if candidate.forced else ''})"
                )

            # Bei forciertem Kandidaten: Fenster sofort schliessen
            # (keine Wartezeit fuer andere Satellites)
            if success and candidate.forced and self._window and self._window.is_open:
                pinfo(
                    f"Forcierte Arbitration: {candidate.satellite_id} "
                    f"— Fenster sofort geschlossen"
                )
                await self._window.close(reason="forced")

            return success

    async def _on_window_complete(
        self,
        candidates: list[ArbitrationCandidate],
    ) -> None:
        """
        Callback wenn Arbitration-Fenster abgeschlossen.

        Führt Strategieauswahl durch und benachrichtigt Satellites.
        """
        if not candidates:
            return

        started_at = datetime.now()

        # Strategie anwenden
        context = await self._build_context()
        winner = self._strategy.select(candidates, context)

        # Ergebnis erstellen
        result = ArbitrationResult(
            success=winner is not None,
            winner=winner,
            rejected=[c for c in candidates if c != winner],
            arbitration_id=self._current_arbitration_id,
            started_at=started_at,
            decided_at=datetime.now(),
            candidate_count=len(candidates),
            strategy_used=self._strategy.name,
            close_reason=self._window.state.value if self._window else "unknown",
        )

        if winner:
            winner.select()
            result.conversation_id = f"conv_{uuid.uuid4().hex[:8]}"

            # Abgelehnte markieren
            for candidate in result.rejected:
                candidate.reject(reason="Anderer Satellite ausgewählt")

        # Statistiken aktualisieren
        self._update_stats(result)

        # Satellites benachrichtigen
        await self._notify_satellites(result)

        # Callbacks ausführen
        for callback in self._on_decision_callbacks:
            try:
                await callback(result)
            except Exception as e:
                perror(f"Callback-Fehler: {e}")

        # Conversation starten
        if result.success and self.config.auto_start_conversation:
            await self._start_conversation(result)

        # Fenster zurücksetzen
        if self._window:
            await self._window.reset()

        pinfo(
            f"Arbitration abgeschlossen: {result.arbitration_id} - "
            f"Gewinner: {winner.satellite_id if winner else 'keiner'} "
            f"({result.candidate_count} Kandidaten, {result.duration_ms:.0f}ms)"
        )

    async def _build_context(self) -> dict[str, Any]:
        """Baut Kontext für Strategieauswahl."""
        context: dict[str, Any] = {
            "window_duration_ms": self.config.window_duration_ms,
        }

        # Letzte aktive Conversation hinzufügen wenn verfügbar
        if self.application and hasattr(self.application, "conversation_manager"):
            conv_manager = self.application.conversation_manager
            last_conv = getattr(conv_manager, "last_conversation", None)
            if last_conv:
                context["last_active_room"] = getattr(last_conv, "room_id", "")
                context["last_active_satellite"] = getattr(
                    last_conv, "satellite_id", ""
                )

        return context

    async def _notify_satellites(self, result: ArbitrationResult) -> None:
        """Benachrichtigt Satellites über Arbitration-Ergebnis."""
        if not self.application or not hasattr(self.application, "satellites"):
            return

        satellite_manager = self.application.satellites

        # Gewinner benachrichtigen
        if result.winner:
            satellite = satellite_manager.get(result.winner.satellite_id)
            if satellite:
                await self._send_selected(satellite, result)

        # Verlierer benachrichtigen
        if self.config.notify_rejected:
            for candidate in result.rejected:
                satellite = satellite_manager.get(candidate.satellite_id)
                if satellite:
                    await self._send_abort(satellite, result)

    async def _send_selected(
        self,
        satellite: "Satellite",
        result: ArbitrationResult,
    ) -> None:
        """Sendet WakewordSelected an Gewinner."""
        try:
            from trixy_core.network.cmd import WakewordSelected

            msg = WakewordSelected(
                conversation_id=result.conversation_id,
                start_streaming=True,
            )
            await satellite.send_command(msg)
            pdebug(
                f"WakewordSelected gesendet an {satellite.id}"
            )
        except Exception as e:
            perror(
                f"Fehler beim Senden von WakewordSelected: {e}"
            )

    async def _send_abort(
        self,
        satellite: "Satellite",
        result: ArbitrationResult,
    ) -> None:
        """Sendet WakewordAbort an Verlierer."""
        try:
            from trixy_core.network.cmd import WakewordAbort

            msg = WakewordAbort(
                reason="Anderer Satellite ausgewählt",
                resume_detection=True,
            )
            await satellite.send_command(msg)
            pdebug(
                f"WakewordAbort gesendet an {satellite.id}"
            )
        except Exception as e:
            perror(
                f"Fehler beim Senden von WakewordAbort: {e}"
            )

    async def _start_conversation(self, result: ArbitrationResult) -> None:
        """Startet Conversation-Session für Gewinner."""
        if not result.winner:
            return

        # Event auslösen
        if self.application and hasattr(self.application, "events"):
            await self.application.events.emit(
                "conversation_started",
                {
                    "satellite_id": result.winner.satellite_id,
                    "conversation_id": result.conversation_id,
                    "arbitration_id": result.arbitration_id,
                },
            )

    def _update_stats(self, result: ArbitrationResult) -> None:
        """Aktualisiert Statistiken."""
        self._stats["total_arbitrations"] += 1

        if result.success:
            self._stats["successful"] += 1

        if result.candidate_count == 1:
            self._stats["single_candidate"] += 1
        else:
            self._stats["multiple_candidates"] += 1

        # Durchschnitte aktualisieren
        total = self._stats["total_arbitrations"]
        self._stats["average_candidates"] = (
            (self._stats["average_candidates"] * (total - 1) + result.candidate_count)
            / total
        )
        self._stats["average_duration_ms"] = (
            (self._stats["average_duration_ms"] * (total - 1) + result.duration_ms)
            / total
        )

    def on_decision(
        self,
        callback: Callable[[ArbitrationResult], Awaitable[None]],
    ) -> None:
        """
        Registriert Callback für Arbitration-Entscheidungen.

        Args:
            callback: Async Callback-Funktion
        """
        self._on_decision_callbacks.append(callback)

    def remove_decision_callback(
        self,
        callback: Callable[[ArbitrationResult], Awaitable[None]],
    ) -> bool:
        """
        Entfernt Callback.

        Args:
            callback: Zu entfernender Callback

        Returns:
            True wenn entfernt
        """
        try:
            self._on_decision_callbacks.remove(callback)
            return True
        except ValueError:
            return False

    async def cancel_current(self) -> bool:
        """
        Bricht aktuelle Arbitration ab.

        Returns:
            True wenn abgebrochen
        """
        if not self._window or not self._window.is_open:
            return False

        await self._window.close(reason="cancelled")
        await self._window.reset()
        return True

    def get_window_status(self) -> dict[str, Any]:
        """Gibt Status des aktuellen Fensters zurück."""
        if not self._window:
            return {
                "state": "closed",
                "candidates": 0,
            }

        return {
            "state": self._window.state.value,
            "candidates": self._window.candidate_count,
            "duration_ms": self._window.duration_ms,
            "remaining_ms": self._window.remaining_ms,
            "arbitration_id": self._current_arbitration_id,
        }