Trixy7/trixy_core/arbitration/strategies/base.py

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