""" Arbitration Manager for Trixy Satellite Selection This module provides the main ArbitrationManager class that implements the complete arbitration system for resolving wakeword conflicts between multiple satellites. The ArbitrationManager implements the workflow specified in CLAUDE.md: 1. Multiple satellites detect wakeword simultaneously 2. Server waits 1 second to collect all wakeword reports 3. Server selects satellite with highest volume (or other algorithm) 4. Selected satellite gets conversation session 5. Non-selected satellites are notified to ignore Key Features: - Event-driven integration with Trixy event system - 1-second collection window for wakeword reports - Multiple arbitration algorithms (volume-based primary) - Thread-safe concurrent wakeword handling - Integration with satellite and conversation managers - Comprehensive history tracking and analytics - Plugin support for custom arbitration algorithms - Real-time monitoring and debugging capabilities Usage: from trixy_core.arbitration import ArbitrationManager, ArbitrationConfig # Create arbitration manager config = ArbitrationConfig() arbitration_manager = ArbitrationManager(application, config) # Start the manager (registers event handlers) arbitration_manager.start() # The manager automatically handles wakeword_received events # and coordinates with satellite and conversation managers """ import asyncio import time import threading from typing import Dict, List, Optional, Any, Callable, Set, Union from dataclasses import dataclass, field from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, Future import uuid from trixy_core.events import ( EventHandler, TrixyEvent, WakewordReceivedEventData, SatelliteInfo, SpeakerInfo, EventType, ArbitrationStartedEventData, ArbitrationCompletedEventData, ArbitrationCancelledEventData, ArbitrationTimedOutEventData, ArbitrationReportAddedEventData, SatelliteSelectedEventData, SatelliteIgnoredEventData ) from .arbitration_config import ( ArbitrationConfig, ArbitrationAlgorithm, ArbitrationMode, ArbitrationConfigError, create_default_config ) from .arbitration_algorithms import ( ArbitrationAlgorithms, WakewordReport, SelectionResult, create_test_reports ) from .arbitration_session import ( ArbitrationSession, ArbitrationSessionManager, SessionState, create_session_manager ) def pprint(message: str) -> None: """Arbitration manager logging function.""" print(f"[ARBITRATION_MANAGER] {message}") @dataclass class ArbitrationStats: """Statistics for arbitration manager operations.""" total_arbitrations: int = 0 successful_arbitrations: int = 0 failed_arbitrations: int = 0 cancelled_arbitrations: int = 0 timed_out_arbitrations: int = 0 total_wakeword_reports: int = 0 unique_satellites_involved: Set[str] = field(default_factory=set) average_collection_time: float = 0.0 average_arbitration_time: float = 0.0 algorithm_usage: Dict[str, int] = field(default_factory=dict) satellite_selection_count: Dict[str, int] = field(default_factory=dict) room_selection_count: Dict[str, int] = field(default_factory=dict) def update_from_session(self, session: ArbitrationSession) -> None: """Update statistics from completed session.""" if session.state == SessionState.COMPLETED: self.successful_arbitrations += 1 if session.selected_satellite_id: self.satellite_selection_count[session.selected_satellite_id] = ( self.satellite_selection_count.get(session.selected_satellite_id, 0) + 1 ) if session.selection_result: algorithm = session.selection_result.algorithm_used.value self.algorithm_usage[algorithm] = self.algorithm_usage.get(algorithm, 0) + 1 elif session.state == SessionState.CANCELLED: self.cancelled_arbitrations += 1 elif session.state == SessionState.TIMED_OUT: self.timed_out_arbitrations += 1 else: self.failed_arbitrations += 1 self.total_arbitrations += 1 self.total_wakeword_reports += len(session.wakeword_reports) for report in session.wakeword_reports: self.unique_satellites_involved.add(report.satellite_id) if report.room_id: self.room_selection_count[report.room_id] = ( self.room_selection_count.get(report.room_id, 0) + 1 ) # Update timing averages if session.timing.get_collection_duration(): self.average_collection_time = self._update_average( self.average_collection_time, session.timing.get_collection_duration(), self.total_arbitrations ) if session.timing.get_arbitration_duration(): self.average_arbitration_time = self._update_average( self.average_arbitration_time, session.timing.get_arbitration_duration(), self.total_arbitrations ) def _update_average(self, current_avg: float, new_value: float, count: int) -> float: """Update running average.""" if count <= 1: return new_value return ((current_avg * (count - 1)) + new_value) / count def get_success_rate(self) -> float: """Get arbitration success rate.""" if self.total_arbitrations == 0: return 0.0 return self.successful_arbitrations / self.total_arbitrations def to_dict(self) -> Dict[str, Any]: """Convert statistics to dictionary.""" return { 'total_arbitrations': self.total_arbitrations, 'successful_arbitrations': self.successful_arbitrations, 'failed_arbitrations': self.failed_arbitrations, 'cancelled_arbitrations': self.cancelled_arbitrations, 'timed_out_arbitrations': self.timed_out_arbitrations, 'success_rate': self.get_success_rate(), 'total_wakeword_reports': self.total_wakeword_reports, 'unique_satellites_count': len(self.unique_satellites_involved), 'average_collection_time': self.average_collection_time, 'average_arbitration_time': self.average_arbitration_time, 'algorithm_usage': dict(self.algorithm_usage), 'satellite_selection_count': dict(self.satellite_selection_count), 'room_selection_count': dict(self.room_selection_count), } class ArbitrationManagerError(Exception): """Base exception for arbitration manager errors.""" pass class ArbitrationManagerNotStartedError(ArbitrationManagerError): """Raised when manager operation requires started state.""" pass class ArbitrationManagerConfigError(ArbitrationManagerError): """Raised when manager configuration is invalid.""" pass class ArbitrationManager: """ Main arbitration manager for Trixy satellite selection. This class implements the complete arbitration system as specified in CLAUDE.md, handling wakeword conflicts between multiple satellites and selecting the optimal satellite for conversation handling. The manager integrates with the Trixy event system to automatically handle wakeword_received events and coordinate with satellite and conversation managers. """ def __init__( self, application, config: Optional[ArbitrationConfig] = None, session_manager: Optional[ArbitrationSessionManager] = None ): """ Initialize arbitration manager. Args: application: Trixy application container config: Arbitration configuration (created with defaults if None) session_manager: Session manager (created if None) """ self.application = application self.config = config or create_default_config() # Initialize core components self.algorithms = ArbitrationAlgorithms(self.config) self.session_manager = session_manager or create_session_manager( self.config, self.algorithms ) # Get application components self.event_handler = application.get_event_handler() self.satellite_manager = application.get_satellite_manager() self.conversation_manager = application.get_conversation_manager() # State management self.is_started = False self.is_running = True self._lock = threading.RLock() # Statistics and monitoring self.stats = ArbitrationStats() self.history: List[Dict[str, Any]] = [] # Active arbitration tracking self.active_arbitrations: Dict[str, ArbitrationSession] = {} self.pending_reports: Dict[str, List[WakewordReport]] = {} # Threading self.executor = ThreadPoolExecutor(max_workers=self.config.thread_pool_size) # Event registry self._event_handlers_registered = False pprint("Initialized arbitration manager") def start(self) -> None: """ Start the arbitration manager. This registers event handlers and begins processing wakeword events. """ with self._lock: if self.is_started: pprint("Arbitration manager already started") return # Register event handlers self._register_event_handlers() self.is_started = True pprint("Arbitration manager started") def stop(self) -> None: """ Stop the arbitration manager. This unregisters event handlers and cancels active arbitrations. """ with self._lock: if not self.is_started: return self.is_running = False # Cancel all active arbitrations for session in list(self.active_arbitrations.values()): session.cancel("Manager shutdown") # Unregister event handlers self._unregister_event_handlers() self.is_started = False pprint("Arbitration manager stopped") def _register_event_handlers(self) -> None: """Register event handlers with the event system.""" if self._event_handlers_registered: return # Register this object's event handlers self.event_handler.register_handler_object(self) self._event_handlers_registered = True pprint("Registered arbitration event handlers") def _unregister_event_handlers(self) -> None: """Unregister event handlers from the event system.""" if not self._event_handlers_registered: return # Unregister this object's event handlers self.event_handler.unregister_handler_object(self) self._event_handlers_registered = False pprint("Unregistered arbitration event handlers") @TrixyEvent(["wakeword_received"]) def handle_wakeword_received(self, event_name: str, event_data: WakewordReceivedEventData) -> None: """ Handle wakeword_received events from satellites. This is the main entry point for arbitration when satellites detect wakewords. Implements the 1-second collection window and satellite selection process. Args: event_name: Event name ("wakeword_received") event_data: Wakeword event data from satellite """ if not self.is_running: return try: # Convert event data to wakeword report report = self._convert_event_to_report(event_data) self.logger.info(f"[SERVER] Wakeword command received from satellite {report.satellite_id} " f"(volume: {report.volume:.3f}, confidence: {report.confidence:.3f})") pprint(f"Received wakeword from {report.satellite_id} " f"(volume: {report.volume:.3f}, confidence: {report.confidence:.3f})") # Check if we should start a new arbitration session session = self._find_or_create_arbitration_session(report) if session: # Add report to session if session.add_wakeword_report(report): # Trigger report added event self._trigger_arbitration_report_added_event(session, report) pprint(f"Added wakeword report to session {session.session_id}") else: pprint(f"Failed to add wakeword report to session {session.session_id}") except Exception as e: pprint(f"Error handling wakeword event: {e}") def _convert_event_to_report(self, event_data: WakewordReceivedEventData) -> WakewordReport: """Convert wakeword event data to wakeword report.""" satellite_info = event_data.satellite_info speaker_info = event_data.speaker_info or SpeakerInfo( speaker_id="unknown", speaker_name="Unknown Speaker" ) return WakewordReport( satellite_id=satellite_info.satellite_id, wakeword_id=event_data.wakeword_id, volume=event_data.volume, confidence=event_data.confidence, timestamp=event_data.timestamp.timestamp(), room_id=satellite_info.room_id, satellite_alias=satellite_info.alias, mac_address=satellite_info.mac_address, speaker_id=speaker_info.speaker_id, speaker_name=speaker_info.speaker_name, speaker_confidence=speaker_info.confidence, audio_buffer_length=event_data.audio_buffer_length, ) def _find_or_create_arbitration_session(self, report: WakewordReport) -> Optional[ArbitrationSession]: """ Find existing arbitration session or create new one for wakeword report. Args: report: Wakeword report Returns: ArbitrationSession: Session to add report to, or None if rejected """ with self._lock: # Check if there's an existing session we can add this report to current_time = time.time() for session in list(self.active_arbitrations.values()): # Check if session is in collection phase if session.state not in [SessionState.CREATED, SessionState.COLLECTING]: continue # Check if report is within time window session_age = current_time - session.timing.created_at.timestamp() if session_age > self.config.collection_window_seconds + 0.5: # Small buffer continue # Check if same wakeword if (session.wakeword_reports and session.wakeword_reports[0].wakeword_id == report.wakeword_id): return session # Create new session session = self._create_new_arbitration_session(report) return session def _trigger_arbitration_started_event(self, session: ArbitrationSession, trigger_report: WakewordReport) -> None: """Trigger arbitration_started event.""" try: event_data = ArbitrationStartedEventData( arbitration_session_id=session.session_id, trigger_satellite_id=trigger_report.satellite_id, trigger_wakeword_id=trigger_report.wakeword_id, collection_window_seconds=self.config.collection_window_seconds, algorithm=self.config.primary_algorithm.value, expected_participants=1 ) self.event_handler.trigger_event("arbitration_started", event_data) pprint(f"Triggered arbitration_started event for session {session.session_id}") except Exception as e: pprint(f"Error triggering arbitration_started event: {e}") def _trigger_arbitration_report_added_event(self, session: ArbitrationSession, report: WakewordReport) -> None: """Trigger arbitration_report_added event.""" try: event_data = ArbitrationReportAddedEventData( arbitration_session_id=session.session_id, satellite_id=report.satellite_id, wakeword_id=report.wakeword_id, volume=report.volume, confidence=report.confidence, speaker_info=SpeakerInfo( speaker_id=report.speaker_id, speaker_name=report.speaker_name, confidence=report.speaker_confidence ) if report.speaker_id else None, satellite_info=SatelliteInfo( satellite_id=report.satellite_id, mac_address=report.mac_address, room_id=report.room_id, alias=report.satellite_alias, version="", # Would need to get from satellite manager audio_ports={} # Would need to get from satellite manager ) if report.satellite_id else None, estimated_distance=report.estimated_distance, room_id=report.room_id, total_reports_in_session=len(session.wakeword_reports) ) self.event_handler.trigger_event("arbitration_report_added", event_data) except Exception as e: pprint(f"Error triggering arbitration_report_added event: {e}") def _trigger_arbitration_completed_event(self, session: ArbitrationSession, selection_result: SelectionResult) -> None: """Trigger arbitration_completed event.""" try: collection_duration = session.timing.get_collection_duration() or 0.0 arbitration_duration = session.timing.get_arbitration_duration() or 0.0 event_data = ArbitrationCompletedEventData( arbitration_session_id=session.session_id, selected_satellite_id=selection_result.selected_satellite_id, selection_algorithm=selection_result.algorithm_used.value, selection_score=selection_result.selection_score, selection_confidence=selection_result.confidence, total_participants=len(session.wakeword_reports), collection_duration_ms=collection_duration * 1000, arbitration_duration_ms=arbitration_duration * 1000, conversation_id=session.conversation_id, selection_reason=selection_result.selection_reason, alternative_selections=selection_result.alternative_selections ) self.event_handler.trigger_event("arbitration_completed", event_data) pprint(f"Triggered arbitration_completed event for session {session.session_id}") except Exception as e: pprint(f"Error triggering arbitration_completed event: {e}") def _trigger_satellite_selected_event(self, session: ArbitrationSession, satellite_info: SatelliteInfo, selection_result: SelectionResult, conversation_id: str) -> None: """Trigger satellite_selected event.""" try: event_data = SatelliteSelectedEventData( arbitration_session_id=session.session_id, selected_satellite_id=selection_result.selected_satellite_id, satellite_info=satellite_info, selection_algorithm=selection_result.algorithm_used.value, selection_score=selection_result.selection_score, selection_confidence=selection_result.confidence, conversation_id=conversation_id, selection_metadata={ 'total_participants': len(session.wakeword_reports), 'selection_reason': selection_result.selection_reason, 'tied_candidates': selection_result.tied_candidates } ) self.event_handler.trigger_event("satellite_selected", event_data) except Exception as e: pprint(f"Error triggering satellite_selected event: {e}") def _trigger_satellite_ignored_event(self, session: ArbitrationSession, ignored_satellite_id: str, selected_satellite_id: str) -> None: """Trigger satellite_ignored event.""" try: # Find the ignored satellite's original report ignored_report = None for report in session.wakeword_reports: if report.satellite_id == ignored_satellite_id: ignored_report = report break satellite_info = None original_volume = 0.0 original_confidence = 0.0 if ignored_report: original_volume = ignored_report.volume original_confidence = ignored_report.confidence satellite_info = SatelliteInfo( satellite_id=ignored_report.satellite_id, mac_address=ignored_report.mac_address, room_id=ignored_report.room_id, alias=ignored_report.satellite_alias, version="", audio_ports={} ) event_data = SatelliteIgnoredEventData( arbitration_session_id=session.session_id, ignored_satellite_id=ignored_satellite_id, satellite_info=satellite_info, selected_satellite_id=selected_satellite_id, ignore_reason="not_selected", original_report_volume=original_volume, original_report_confidence=original_confidence ) self.event_handler.trigger_event("satellite_ignored", event_data) except Exception as e: pprint(f"Error triggering satellite_ignored event: {e}") def _trigger_arbitration_cancelled_event(self, session: ArbitrationSession, reason: str) -> None: """Trigger arbitration_cancelled event.""" try: duration = session.timing.get_total_duration() or 0.0 event_data = ArbitrationCancelledEventData( arbitration_session_id=session.session_id, cancellation_reason=reason, participants_count=len(session.wakeword_reports), duration_ms=duration * 1000, cancelled_by="system" ) self.event_handler.trigger_event("arbitration_cancelled", event_data) except Exception as e: pprint(f"Error triggering arbitration_cancelled event: {e}") def _trigger_arbitration_timed_out_event(self, session: ArbitrationSession) -> None: """Trigger arbitration_timed_out event.""" try: duration = session.timing.get_total_duration() or 0.0 event_data = ArbitrationTimedOutEventData( arbitration_session_id=session.session_id, timeout_duration_ms=duration * 1000, participants_count=len(session.wakeword_reports), reports_received=len(session.wakeword_reports), timeout_phase=session.phase.value if session.phase else "unknown" ) self.event_handler.trigger_event("arbitration_timed_out", event_data) except Exception as e: pprint(f"Error triggering arbitration_timed_out event: {e}") def _create_new_arbitration_session(self, trigger_report: WakewordReport) -> ArbitrationSession: """ Create new arbitration session for wakeword conflict. Args: trigger_report: Initial wakeword report that triggered arbitration Returns: ArbitrationSession: New arbitration session """ # Generate session ID session_id = f"arb_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}" # Create session session = self.session_manager.create_session( session_id=session_id, trigger_report=trigger_report, callback=self._session_event_callback ) # Track session self.active_arbitrations[session_id] = session # Trigger arbitration started event self._trigger_arbitration_started_event(session, trigger_report) pprint(f"Created new arbitration session {session_id} for {trigger_report.satellite_id}") return session def _session_event_callback( self, session: ArbitrationSession, event_type: str, data: Any = None ) -> None: """ Handle arbitration session events. Args: session: Session that generated the event event_type: Type of event data: Event data """ session_id = session.session_id try: if event_type == "arbitration_completed": self._handle_arbitration_completed(session, data) elif event_type == "session_completed": self._handle_session_completed(session) elif event_type == "session_cancelled": self._handle_session_cancelled(session, data) elif event_type == "session_timeout": self._handle_session_timeout(session) elif event_type == "session_error": self._handle_session_error(session, data) except Exception as e: pprint(f"Error handling session event {event_type} for {session_id}: {e}") def _handle_arbitration_completed( self, session: ArbitrationSession, selection_result: SelectionResult ) -> None: """ Handle completed arbitration by starting conversation with selected satellite. Args: session: Completed arbitration session selection_result: Satellite selection result """ session_id = session.session_id selected_satellite_id = selection_result.selected_satellite_id self.logger.info(f"[SERVER] Selected satellite {selected_satellite_id} for conversation " f"(score: {selection_result.selection_score:.3f}, " f"session: {session_id})") pprint(f"Arbitration completed for session {session_id}: " f"selected {selected_satellite_id}") # Trigger arbitration completed event self._trigger_arbitration_completed_event(session, selection_result) try: # Get selected satellite selected_satellite = self.satellite_manager.get_satellite(selected_satellite_id) if not selected_satellite: pprint(f"Warning: Selected satellite {selected_satellite_id} not found") session.cancel("Selected satellite not available") return # Create satellite info for events satellite_info = SatelliteInfo( satellite_id=selected_satellite.satellite_id, mac_address=selected_satellite.mac_address, room_id=selected_satellite.room_id, alias=selected_satellite.alias, version=selected_satellite.version, audio_ports=selected_satellite.audio_ports ) # Start conversation with selected satellite conversation_id = self._start_conversation_with_satellite( session, selected_satellite, selection_result ) if conversation_id: # Trigger satellite selected event self._trigger_satellite_selected_event( session, satellite_info, selection_result, conversation_id ) # Notify non-selected satellites to ignore self._notify_non_selected_satellites(session, selected_satellite_id) # Complete session session.complete(conversation_id) pprint(f"Started conversation {conversation_id} with {selected_satellite_id}") else: pprint(f"Failed to start conversation with {selected_satellite_id}") session.cancel("Failed to start conversation") except Exception as e: pprint(f"Error handling arbitration completion: {e}") session.cancel(f"Error: {e}") def _start_conversation_with_satellite( self, session: ArbitrationSession, selected_satellite, selection_result: SelectionResult ) -> Optional[str]: """ Start conversation with selected satellite. Args: session: Arbitration session selected_satellite: Selected satellite object selection_result: Selection result Returns: str: Conversation ID if successful, None otherwise """ try: # Get speaker information from the session speaker_info = None selected_report = None for report in session.wakeword_reports: if report.satellite_id == selection_result.selected_satellite_id: selected_report = report break if selected_report: speaker_info = SpeakerInfo( speaker_id=selected_report.speaker_id, speaker_name=selected_report.speaker_name, confidence=selected_report.speaker_confidence ) # Create satellite info satellite_info = SatelliteInfo( satellite_id=selected_satellite.satellite_id, mac_address=selected_satellite.mac_address, room_id=selected_satellite.room_id, alias=selected_satellite.alias, version=selected_satellite.version, audio_ports=selected_satellite.audio_ports ) # Start conversation conversation_session = self.conversation_manager.start_conversation( satellite_info=satellite_info, speaker_info=speaker_info, trigger_event="wakeword_received", arbitration_session_id=session.session_id ) if conversation_session: return conversation_session.conversation_id except Exception as e: pprint(f"Error starting conversation: {e}") return None def _notify_non_selected_satellites( self, session: ArbitrationSession, selected_satellite_id: str ) -> None: """ Notify non-selected satellites to ignore the wakeword. Args: session: Arbitration session selected_satellite_id: ID of selected satellite """ for report in session.wakeword_reports: if report.satellite_id != selected_satellite_id: try: satellite = self.satellite_manager.get_satellite(report.satellite_id) if satellite: # Send ignore command to satellite satellite.send_ignore_wakeword_command(session.session_id) # Trigger satellite ignored event self._trigger_satellite_ignored_event(session, report.satellite_id, selected_satellite_id) pprint(f"Notified {report.satellite_id} to ignore wakeword") except Exception as e: pprint(f"Error notifying satellite {report.satellite_id}: {e}") def _handle_session_completed(self, session: ArbitrationSession) -> None: """Handle session completion.""" self._finalize_session(session) self.stats.update_from_session(session) pprint(f"Session {session.session_id} completed successfully") def _handle_session_cancelled(self, session: ArbitrationSession, reason: str) -> None: """Handle session cancellation.""" # Trigger arbitration cancelled event self._trigger_arbitration_cancelled_event(session, reason) self._finalize_session(session) self.stats.update_from_session(session) pprint(f"Session {session.session_id} cancelled: {reason}") def _handle_session_timeout(self, session: ArbitrationSession) -> None: """Handle session timeout.""" # Trigger arbitration timed out event self._trigger_arbitration_timed_out_event(session) self._finalize_session(session) self.stats.update_from_session(session) pprint(f"Session {session.session_id} timed out") def _handle_session_error(self, session: ArbitrationSession, error: str) -> None: """Handle session error.""" self._finalize_session(session) self.stats.update_from_session(session) pprint(f"Session {session.session_id} error: {error}") def _finalize_session(self, session: ArbitrationSession) -> None: """Finalize session and update tracking.""" with self._lock: session_id = session.session_id # Remove from active tracking if session_id in self.active_arbitrations: del self.active_arbitrations[session_id] # Add to history session_summary = session.get_session_summary() self.history.append(session_summary) # Keep history within limits if len(self.history) > self.config.max_history_entries: self.history.pop(0) def get_active_arbitrations(self) -> List[ArbitrationSession]: """ Get list of currently active arbitration sessions. Returns: List[ArbitrationSession]: Active arbitration sessions """ with self._lock: return list(self.active_arbitrations.values()) def get_arbitration_statistics(self) -> Dict[str, Any]: """ Get comprehensive arbitration statistics. Returns: Dict[str, Any]: Arbitration statistics and metrics """ with self._lock: base_stats = self.stats.to_dict() session_stats = self.session_manager.get_session_statistics() return { **base_stats, 'session_manager': session_stats, 'active_arbitrations': len(self.active_arbitrations), 'history_entries': len(self.history), 'config_algorithm': self.config.primary_algorithm.value, 'config_fallback_algorithm': self.config.fallback_algorithm.value, 'config_collection_window': self.config.collection_window_seconds, } def get_arbitration_history( self, limit: Optional[int] = None, filter_by: Optional[Dict[str, Any]] = None ) -> List[Dict[str, Any]]: """ Get arbitration history with optional filtering. Args: limit: Maximum number of entries to return filter_by: Optional filters to apply Returns: List[Dict[str, Any]]: Filtered arbitration history """ with self._lock: history = self.history.copy() # Apply filters if filter_by: filtered_history = [] for entry in history: match = True for key, value in filter_by.items(): if key not in entry or entry[key] != value: match = False break if match: filtered_history.append(entry) history = filtered_history # Apply limit if limit and limit > 0: history = history[-limit:] return history def force_arbitration( self, reports: List[WakewordReport], algorithm: Optional[ArbitrationAlgorithm] = None ) -> Optional[SelectionResult]: """ Force immediate arbitration with provided reports (for testing/debugging). Args: reports: Wakeword reports to arbitrate algorithm: Algorithm to use (defaults to config) Returns: SelectionResult: Selection result if successful """ if not self.is_started: raise ArbitrationManagerNotStartedError("Manager not started") try: result = self.algorithms.select_satellite(reports, algorithm) pprint(f"Forced arbitration selected {result.selected_satellite_id} " f"with score {result.selection_score:.3f}") return result except Exception as e: pprint(f"Forced arbitration failed: {e}") return None def update_configuration(self, new_config: ArbitrationConfig) -> None: """ Update arbitration configuration. Args: new_config: New configuration to apply """ with self._lock: old_config = self.config self.config = new_config # Recreate algorithms with new config self.algorithms = ArbitrationAlgorithms(new_config) # Update session manager self.session_manager.config = new_config pprint(f"Updated arbitration configuration: " f"algorithm {old_config.primary_algorithm.value} -> {new_config.primary_algorithm.value}") def get_configuration(self) -> ArbitrationConfig: """ Get current arbitration configuration. Returns: ArbitrationConfig: Current configuration """ return self.config def enable_debug_mode(self, enabled: bool = True) -> None: """ Enable or disable debug mode. Args: enabled: Whether to enable debug mode """ self.config.enable_debug_logging = enabled pprint(f"Debug mode {'enabled' if enabled else 'disabled'}") def benchmark_algorithms( self, num_iterations: int = 100, num_reports: int = 3 ) -> Dict[str, Any]: """ Benchmark arbitration algorithms. Args: num_iterations: Number of benchmark iterations num_reports: Number of test reports per iteration Returns: Dict[str, Any]: Benchmark results """ test_reports = create_test_reports(num_reports) results = self.algorithms.benchmark_algorithms(test_reports, num_iterations) pprint(f"Completed algorithm benchmark: {num_iterations} iterations, {num_reports} reports") return results def cleanup(self) -> None: """Clean up arbitration manager resources.""" self.stop() if self.executor: self.executor.shutdown(wait=True) if self.session_manager: self.session_manager.cleanup() if self.algorithms: self.algorithms.cleanup() pprint("Arbitration manager cleaned up") def __del__(self): """Destructor to ensure cleanup.""" self.cleanup() def create_arbitration_manager( application, config: Optional[ArbitrationConfig] = None ) -> ArbitrationManager: """ Create arbitration manager with application and configuration. Args: application: Trixy application container config: Arbitration configuration (defaults created if None) Returns: ArbitrationManager: Configured arbitration manager """ return ArbitrationManager(application, config) def create_arbitration_manager_with_defaults(application) -> ArbitrationManager: """ Create arbitration manager with default configuration. Args: application: Trixy application container Returns: ArbitrationManager: Arbitration manager with defaults """ config = create_default_config() return ArbitrationManager(application, config)