Trixy7/trixy_core/arbitration/candidate.py

# -*- coding: utf-8 -*-
"""
Arbitration-Kandidat Datenklasse.

Repräsentiert einen Satellite, der ein Wakeword erkannt hat
und für die Konversation in Betracht gezogen wird.
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class CandidateState(Enum):
    """Zustand eines Arbitration-Kandidaten."""
    PENDING = "pending"      # Wartet auf Entscheidung
    SELECTED = "selected"    # Wurde ausgewählt
    REJECTED = "rejected"    # Wurde abgelehnt
    TIMEOUT = "timeout"      # Zeitüberschreitung


@dataclass
class ArbitrationCandidate:
    """
    Ein Kandidat für die Wakeword-Arbitration.

    Repräsentiert einen Satellite, der ein Wakeword erkannt hat.
    Enthält alle relevanten Daten für die Auswahlentscheidung.
    """
    # Identifikation
    satellite_id: str
    room_id: str = ""
    alias: str = ""

    # Wakeword-Daten
    wakeword_model: str = ""
    confidence: float = 0.0
    audio_level: float = 0.0
    buffer_samples: int = 0

    # Zeitstempel
    detected_at: datetime = field(default_factory=datetime.now)

    # Zustand
    state: CandidateState = CandidateState.PENDING

    # Forcierte Auswahl (z.B. durch HID-Taste — ueberspringt Arbitration-Fenster)
    forced: bool = False

    # Zusätzliche Metadaten für erweiterte Strategien
    metadata: dict[str, Any] = field(default_factory=dict)

    # Berechneter Score (wird von Strategy gesetzt)
    score: float = 0.0
    score_details: dict[str, float] = field(default_factory=dict)

    def __post_init__(self) -> None:
        """Validierung nach Initialisierung."""
        if not self.satellite_id:
            raise ValueError("satellite_id ist erforderlich")

        # Normalisiere audio_level und confidence auf 0.0-1.0
        self.audio_level = max(0.0, min(1.0, self.audio_level))
        self.confidence = max(0.0, min(1.0, self.confidence))

    @property
    def is_pending(self) -> bool:
        """Prüft ob Kandidat noch auf Entscheidung wartet."""
        return self.state == CandidateState.PENDING

    @property
    def is_selected(self) -> bool:
        """Prüft ob Kandidat ausgewählt wurde."""
        return self.state == CandidateState.SELECTED

    @property
    def is_rejected(self) -> bool:
        """Prüft ob Kandidat abgelehnt wurde."""
        return self.state == CandidateState.REJECTED

    @property
    def age_ms(self) -> float:
        """Alter des Kandidaten in Millisekunden."""
        delta = datetime.now() - self.detected_at
        return delta.total_seconds() * 1000

    def select(self) -> None:
        """Markiert Kandidaten als ausgewählt."""
        self.state = CandidateState.SELECTED

    def reject(self, reason: str = "") -> None:
        """Markiert Kandidaten als abgelehnt."""
        self.state = CandidateState.REJECTED
        if reason:
            self.metadata["reject_reason"] = reason

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "satellite_id": self.satellite_id,
            "room_id": self.room_id,
            "alias": self.alias,
            "wakeword_model": self.wakeword_model,
            "confidence": self.confidence,
            "audio_level": self.audio_level,
            "buffer_samples": self.buffer_samples,
            "detected_at": self.detected_at.isoformat(),
            "state": self.state.value,
            "score": self.score,
            "score_details": self.score_details,
            "metadata": self.metadata,
        }

    def __repr__(self) -> str:
        return (
            f"ArbitrationCandidate("
            f"satellite_id={self.satellite_id!r}, "
            f"audio_level={self.audio_level:.2f}, "
            f"confidence={self.confidence:.2f}, "
            f"score={self.score:.2f}, "
            f"state={self.state.value})"
        )