# -*- coding: utf-8 -*- """ Basis-Klasse für Arbitration-Strategien. Definiert das Interface für alle Strategien zur Satellite-Auswahl. """ from abc import ABC, abstractmethod from typing import Any from trixy_core.arbitration.candidate import ArbitrationCandidate class ArbitrationStrategy(ABC): """ Abstrakte Basisklasse für Arbitration-Strategien. Eine Strategie entscheidet, welcher Satellite bei gleichzeitiger Wakeword-Erkennung die Konversation führen soll. Beispiel: ```python class MyStrategy(ArbitrationStrategy): @property def name(self) -> str: return "my_strategy" def calculate_score(self, candidate: ArbitrationCandidate) -> float: return candidate.audio_level * 0.5 + candidate.confidence * 0.5 def select(self, candidates: list[ArbitrationCandidate]) -> ArbitrationCandidate: return max(candidates, key=lambda c: self.calculate_score(c)) ``` """ @property @abstractmethod def name(self) -> str: """ Eindeutiger Name der Strategie. Returns: Name der Strategie (z.B. "loudest", "combined") """ pass @property def description(self) -> str: """ Beschreibung der Strategie. Returns: Menschenlesbare Beschreibung """ return f"Arbitration-Strategie: {self.name}" @abstractmethod def calculate_score( self, candidate: ArbitrationCandidate, context: dict[str, Any] | None = None, ) -> float: """ Berechnet den Score für einen Kandidaten. Args: candidate: Der zu bewertende Kandidat context: Optionaler Kontext (z.B. Raum-Informationen) Returns: Score zwischen 0.0 und 1.0 (höher = besser) """ pass @abstractmethod def select( self, candidates: list[ArbitrationCandidate], context: dict[str, Any] | None = None, ) -> ArbitrationCandidate | None: """ Wählt den besten Kandidaten aus. Args: candidates: Liste der Kandidaten context: Optionaler Kontext Returns: Ausgewählter Kandidat oder None wenn Liste leer """ pass def rank( self, candidates: list[ArbitrationCandidate], context: dict[str, Any] | None = None, ) -> list[ArbitrationCandidate]: """ Sortiert Kandidaten nach Score (absteigend). Args: candidates: Liste der Kandidaten context: Optionaler Kontext Returns: Sortierte Liste (bester zuerst) """ if not candidates: return [] # Scores berechnen for candidate in candidates: candidate.score = self.calculate_score(candidate, context) # Nach Score sortieren (absteigend) return sorted(candidates, key=lambda c: c.score, reverse=True) def validate_candidate(self, candidate: ArbitrationCandidate) -> bool: """ Prüft ob Kandidat gültig für Auswahl ist. Kann von Subklassen überschrieben werden für zusätzliche Validierungslogik. Args: candidate: Zu prüfender Kandidat Returns: True wenn gültig """ return ( candidate.is_pending and candidate.audio_level >= 0.0 and candidate.confidence >= 0.0 ) def filter_valid( self, candidates: list[ArbitrationCandidate], ) -> list[ArbitrationCandidate]: """ Filtert ungültige Kandidaten heraus. Args: candidates: Liste der Kandidaten Returns: Gefilterte Liste """ return [c for c in candidates if self.validate_candidate(c)] def __repr__(self) -> str: return f"{self.__class__.__name__}(name={self.name!r})"