Trixy7/trixy_core/arbitration/window.py

# -*- coding: utf-8 -*-
"""
Arbitration-Zeitfenster Management.

Verwaltet das Zeitfenster, in dem Wakeword-Erkennungen gesammelt werden,
bevor eine Entscheidung getroffen wird.
"""

import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Callable, Awaitable, Any

from trixy_core.arbitration.candidate import ArbitrationCandidate, CandidateState


class WindowState(Enum):
    """Zustand des Arbitration-Fensters."""
    CLOSED = "closed"        # Kein aktives Fenster
    OPEN = "open"            # Fenster offen, sammelt Kandidaten
    DECIDING = "deciding"    # Entscheidung wird getroffen
    COMPLETED = "completed"  # Entscheidung abgeschlossen


@dataclass
class WindowConfig:
    """Konfiguration für das Arbitration-Fenster."""
    # Zeitfenster in Millisekunden
    window_duration_ms: int = 1000

    # Minimale Wartezeit bevor Entscheidung (auch wenn nur 1 Kandidat)
    min_wait_ms: int = 100

    # Maximale Anzahl Kandidaten
    max_candidates: int = 10

    # Sofortige Entscheidung wenn audio_level über diesem Wert
    immediate_threshold: float = 0.95

    # Vorzeitiges Schließen wenn Differenz zum Besten groß genug
    early_close_threshold: float = 0.3


class ArbitrationWindow:
    """
    Zeitfenster für die Wakeword-Arbitration.

    Sammelt WakewordDetected-Ereignisse von mehreren Satellites
    innerhalb eines Zeitfensters und ermöglicht dann die Auswahl.
    """

    def __init__(
        self,
        config: WindowConfig | None = None,
        on_complete: Callable[[list[ArbitrationCandidate]], Awaitable[None]] | None = None,
    ) -> None:
        """
        Initialisiert das Arbitration-Fenster.

        Args:
            config: Fenster-Konfiguration
            on_complete: Callback wenn Fenster abgeschlossen
        """
        self.config = config or WindowConfig()
        self.on_complete = on_complete

        self._state = WindowState.CLOSED
        self._candidates: list[ArbitrationCandidate] = []
        self._opened_at: datetime | None = None
        self._timer_task: asyncio.Task | None = None
        self._lock = asyncio.Lock()

    @property
    def state(self) -> WindowState:
        """Aktueller Fensterzustand."""
        return self._state

    @property
    def is_open(self) -> bool:
        """Prüft ob Fenster offen ist."""
        return self._state == WindowState.OPEN

    @property
    def is_closed(self) -> bool:
        """Prüft ob Fenster geschlossen ist."""
        return self._state == WindowState.CLOSED

    @property
    def candidates(self) -> list[ArbitrationCandidate]:
        """Liste der aktuellen Kandidaten."""
        return self._candidates.copy()

    @property
    def candidate_count(self) -> int:
        """Anzahl der Kandidaten."""
        return len(self._candidates)

    @property
    def duration_ms(self) -> float:
        """Bisherige Dauer des Fensters in Millisekunden."""
        if not self._opened_at:
            return 0.0
        delta = datetime.now() - self._opened_at
        return delta.total_seconds() * 1000

    @property
    def remaining_ms(self) -> float:
        """Verbleibende Zeit bis Fenster schließt."""
        if not self.is_open:
            return 0.0
        remaining = self.config.window_duration_ms - self.duration_ms
        return max(0.0, remaining)

    async def open(self) -> bool:
        """
        Öffnet das Arbitration-Fenster.

        Returns:
            True wenn erfolgreich geöffnet, False wenn bereits offen
        """
        async with self._lock:
            if self._state != WindowState.CLOSED:
                return False

            self._state = WindowState.OPEN
            self._candidates = []
            self._opened_at = datetime.now()

            # Timer starten
            self._timer_task = asyncio.create_task(self._window_timer())

            return True

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

        Args:
            candidate: Der hinzuzufügende Kandidat

        Returns:
            True wenn erfolgreich hinzugefügt
        """
        async with self._lock:
            if not self.is_open:
                return False

            # Prüfe ob Satellite bereits Kandidat ist
            for existing in self._candidates:
                if existing.satellite_id == candidate.satellite_id:
                    # Aktualisiere wenn neuer Wert besser
                    if candidate.audio_level > existing.audio_level:
                        existing.audio_level = candidate.audio_level
                        existing.confidence = candidate.confidence
                        existing.detected_at = candidate.detected_at
                    return True

            # Maximale Anzahl prüfen
            if len(self._candidates) >= self.config.max_candidates:
                return False

            self._candidates.append(candidate)

            # Prüfe auf sofortige Entscheidung
            # Hinweis: _close_early wird außerhalb des Locks aufgerufen
            should_close_early = candidate.audio_level >= self.config.immediate_threshold

        # Außerhalb des Locks
        if should_close_early:
            asyncio.create_task(self._close_early("immediate_threshold"))

        return True

    async def close(self, reason: str = "manual") -> list[ArbitrationCandidate]:
        """
        Schließt das Fenster und gibt Kandidaten zurück.

        Args:
            reason: Grund für das Schließen

        Returns:
            Liste der Kandidaten
        """
        # Timer außerhalb des Locks abbrechen um Deadlock zu vermeiden
        timer_task = self._timer_task
        if timer_task and not timer_task.done():
            timer_task.cancel()
            try:
                await timer_task
            except asyncio.CancelledError:
                pass

        async with self._lock:
            if self._state not in (WindowState.OPEN, WindowState.DECIDING):
                return []

            self._state = WindowState.DECIDING
            candidates = self._candidates.copy()

        # Callback außerhalb des Locks ausführen
        if self.on_complete and candidates:
            try:
                await self.on_complete(candidates)
            except Exception:
                pass

        async with self._lock:
            self._state = WindowState.COMPLETED

        return candidates

    async def reset(self) -> None:
        """Setzt das Fenster zurück für neue Arbitration."""
        async with self._lock:
            if self._timer_task and not self._timer_task.done():
                self._timer_task.cancel()
                try:
                    await self._timer_task
                except asyncio.CancelledError:
                    pass

            self._state = WindowState.CLOSED
            self._candidates = []
            self._opened_at = None
            self._timer_task = None

    def get_candidate(self, satellite_id: str) -> ArbitrationCandidate | None:
        """Holt Kandidaten nach Satellite-ID."""
        for candidate in self._candidates:
            if candidate.satellite_id == satellite_id:
                return candidate
        return None

    def has_candidate(self, satellite_id: str) -> bool:
        """Prüft ob Satellite bereits Kandidat ist."""
        return any(c.satellite_id == satellite_id for c in self._candidates)

    async def _window_timer(self) -> None:
        """Timer-Task für automatisches Schließen."""
        try:
            # Minimale Wartezeit
            await asyncio.sleep(self.config.min_wait_ms / 1000)

            # Restliche Zeit
            remaining = (self.config.window_duration_ms - self.config.min_wait_ms) / 1000
            if remaining > 0:
                await asyncio.sleep(remaining)

            # Fenster schließen
            await self.close(reason="timeout")

        except asyncio.CancelledError:
            pass

    async def _close_early(self, reason: str) -> None:
        """Schließt Fenster vorzeitig."""
        # Kurz warten für eventuelle weitere Kandidaten
        await asyncio.sleep(self.config.min_wait_ms / 1000)
        await self.close(reason=reason)

    def __repr__(self) -> str:
        return (
            f"ArbitrationWindow("
            f"state={self._state.value}, "
            f"candidates={len(self._candidates)}, "
            f"duration={self.duration_ms:.0f}ms)"
        )