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