# -*- 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)" )