Trixy7/trixy_core/arbitration/strategies/combined.py

# -*- coding: utf-8 -*-
"""
Combined-Strategie für Arbitration.

Kombiniert mehrere Faktoren (Audio-Level, Confidence, etc.)
zu einem gewichteten Score.
"""

from dataclasses import dataclass
from typing import Any

from trixy_core.arbitration.candidate import ArbitrationCandidate
from trixy_core.arbitration.strategies.base import ArbitrationStrategy


@dataclass
class CombinedWeights:
    """Gewichtungen für die Combined-Strategie."""
    audio_level: float = 0.6    # Audio-Level Gewichtung
    confidence: float = 0.3     # Confidence Gewichtung
    recency: float = 0.1        # Bonus für frühe Erkennung

    def __post_init__(self) -> None:
        """Validierung der Gewichtungen."""
        total = self.audio_level + self.confidence + self.recency
        if abs(total - 1.0) > 0.01:
            # Normalisieren wenn Summe != 1.0
            self.audio_level /= total
            self.confidence /= total
            self.recency /= total


class CombinedStrategy(ArbitrationStrategy):
    """
    Strategie: Gewichtete Kombination mehrerer Faktoren.

    Kombiniert Audio-Level, Confidence und Reaktionszeit
    zu einem Gesamtscore. Die Gewichtungen sind konfigurierbar.

    Beispiel:
        ```python
        # Standard-Gewichtung (60% Audio, 30% Confidence, 10% Zeit)
        strategy = CombinedStrategy()

        # Benutzerdefinierte Gewichtung
        weights = CombinedWeights(audio_level=0.5, confidence=0.4, recency=0.1)
        strategy = CombinedStrategy(weights=weights)

        candidates = [
            ArbitrationCandidate(
                satellite_id="kitchen",
                audio_level=0.7,
                confidence=0.95,
            ),
            ArbitrationCandidate(
                satellite_id="living_room",
                audio_level=0.85,
                confidence=0.85,
            ),
        ]

        winner = strategy.select(candidates)
        ```
    """

    def __init__(
        self,
        weights: CombinedWeights | None = None,
        min_audio_level: float = 0.1,
        min_confidence: float = 0.5,
    ) -> None:
        """
        Initialisiert die Strategie.

        Args:
            weights: Gewichtungen für Score-Berechnung
            min_audio_level: Minimales Audio-Level
            min_confidence: Minimale Confidence
        """
        self.weights = weights or CombinedWeights()
        self.min_audio_level = min_audio_level
        self.min_confidence = min_confidence

    @property
    def name(self) -> str:
        return "combined"

    @property
    def description(self) -> str:
        return (
            f"Kombinierte Bewertung: "
            f"{self.weights.audio_level:.0%} Audio, "
            f"{self.weights.confidence:.0%} Confidence, "
            f"{self.weights.recency:.0%} Zeit"
        )

    def calculate_score(
        self,
        candidate: ArbitrationCandidate,
        context: dict[str, Any] | None = None,
    ) -> float:
        """
        Berechnet gewichteten Gesamtscore.

        Args:
            candidate: Der Kandidat
            context: Optionaler Kontext mit:
                - window_duration_ms: Fensterdauer für Recency-Berechnung

        Returns:
            Gewichteter Score (0.0-1.0)
        """
        context = context or {}

        # Audio-Level Komponente
        audio_score = candidate.audio_level

        # Confidence Komponente
        confidence_score = candidate.confidence

        # Recency Komponente (früher = besser)
        # Normalisiert auf Fensterdauer
        window_duration = context.get("window_duration_ms", 1000)
        age_normalized = min(1.0, candidate.age_ms / window_duration)
        recency_score = 1.0 - age_normalized  # Früher = höherer Score

        # Gewichteter Score
        score = (
            audio_score * self.weights.audio_level +
            confidence_score * self.weights.confidence +
            recency_score * self.weights.recency
        )

        # Score-Details speichern
        candidate.score_details = {
            "audio_level": audio_score,
            "audio_weighted": audio_score * self.weights.audio_level,
            "confidence": confidence_score,
            "confidence_weighted": confidence_score * self.weights.confidence,
            "recency": recency_score,
            "recency_weighted": recency_score * self.weights.recency,
        }

        return score

    def select(
        self,
        candidates: list[ArbitrationCandidate],
        context: dict[str, Any] | None = None,
    ) -> ArbitrationCandidate | None:
        """
        Wählt Kandidaten mit höchstem kombinierten Score.

        Args:
            candidates: Liste der Kandidaten
            context: Optionaler Kontext

        Returns:
            Kandidat mit höchstem Score oder None
        """
        valid = self.filter_valid(candidates)
        if not valid:
            return None

        # Scores berechnen
        for candidate in valid:
            candidate.score = self.calculate_score(candidate, context)

        # Höchsten Score auswählen
        winner = max(valid, key=lambda c: c.score)
        return winner

    def validate_candidate(self, candidate: ArbitrationCandidate) -> bool:
        """Prüft Audio-Level und Confidence."""
        base_valid = super().validate_candidate(candidate)
        return (
            base_valid and
            candidate.audio_level >= self.min_audio_level and
            candidate.confidence >= self.min_confidence
        )