""" Trixy Arbitration System This package provides comprehensive arbitration functionality for resolving wakeword conflicts between multiple satellites in the Trixy voice assistant system. The arbitration system 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 Components: - ArbitrationManager: Main arbitration coordinator - ArbitrationConfig: Configuration management - ArbitrationAlgorithms: Satellite selection algorithms - ArbitrationSession: Session tracking and management - Event integration: Full integration with Trixy event system Features: - Multiple algorithms: volume-based (primary), distance-based, priority-based, round-robin, hybrid - 1-second collection window for wakeword reports - Thread-safe concurrent processing - Comprehensive analytics and history tracking - Plugin support for custom algorithms - Real-time monitoring and debugging Usage: from trixy_core.arbitration import ( ArbitrationManager, ArbitrationConfig, ArbitrationAlgorithm ) # Create configuration config = ArbitrationConfig() config.primary_algorithm = ArbitrationAlgorithm.VOLUME_BASED config.collection_window_seconds = 1.0 # Create and start arbitration manager arbitration_manager = ArbitrationManager(application, config) arbitration_manager.start() # The manager automatically handles wakeword_received events # and coordinates satellite selection Event Integration: The arbitration system automatically handles these events: - wakeword_received: Triggers arbitration process And generates these events: - arbitration_started: When session begins - arbitration_completed: When satellite is selected - satellite_selected: For selected satellite - satellite_ignored: For non-selected satellites """ # Core arbitration classes from .arbitration_manager import ( ArbitrationManager, ArbitrationManagerError, ArbitrationManagerNotStartedError, ArbitrationManagerConfigError, ArbitrationStats, create_arbitration_manager, create_arbitration_manager_with_defaults ) from .arbitration_config import ( ArbitrationConfig, ArbitrationAlgorithm, ArbitrationMode, SatellitePriority, RoomSettings, AlgorithmWeights, ArbitrationConfigError, InvalidConfigurationError, ConfigurationFileError, create_default_config, load_config_from_file, create_config_with_overrides ) from .arbitration_algorithms import ( ArbitrationAlgorithms, WakewordReport, SelectionResult, SelectionCriteria, BaseAlgorithm, VolumeBasedAlgorithm, DistanceBasedAlgorithm, PriorityBasedAlgorithm, RoundRobinAlgorithm, HybridAlgorithm, create_test_reports ) from .arbitration_session import ( ArbitrationSession, ArbitrationSessionManager, SessionState, SessionPhase, SessionTiming, SessionMetrics, SessionError, SessionTimeoutError, SessionStateError, SessionConfigurationError, create_session_manager ) # Version information __version__ = "1.0.0" __author__ = "Trixy Development Team" # Public API __all__ = [ # Main arbitration manager "ArbitrationManager", "ArbitrationManagerError", "ArbitrationManagerNotStartedError", "ArbitrationManagerConfigError", "ArbitrationStats", "create_arbitration_manager", "create_arbitration_manager_with_defaults", # Configuration management "ArbitrationConfig", "ArbitrationAlgorithm", "ArbitrationMode", "SatellitePriority", "RoomSettings", "AlgorithmWeights", "ArbitrationConfigError", "InvalidConfigurationError", "ConfigurationFileError", "create_default_config", "load_config_from_file", "create_config_with_overrides", # Arbitration algorithms "ArbitrationAlgorithms", "WakewordReport", "SelectionResult", "SelectionCriteria", "BaseAlgorithm", "VolumeBasedAlgorithm", "DistanceBasedAlgorithm", "PriorityBasedAlgorithm", "RoundRobinAlgorithm", "HybridAlgorithm", "create_test_reports", # Session management "ArbitrationSession", "ArbitrationSessionManager", "SessionState", "SessionPhase", "SessionTiming", "SessionMetrics", "SessionError", "SessionTimeoutError", "SessionStateError", "SessionConfigurationError", "create_session_manager", ] def pprint(message: str) -> None: """ Arbitration system logging function. Adapts based on debug mode as specified in CLAUDE.md. """ print(f"[ARBITRATION] {message}") def get_arbitration_documentation() -> dict: """ Get comprehensive documentation for the arbitration system. Returns: dict: Documentation mapping with descriptions of components """ return { "overview": "Multi-satellite wakeword conflict resolution system", "workflow": [ "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" ], "algorithms": { "volume_based": "Select satellite with highest wakeword volume (primary)", "distance_based": "Select satellite closest to speaker (estimated)", "priority_based": "Select based on room/satellite priority settings", "round_robin": "Fair rotation among available satellites", "hybrid": "Weighted combination of multiple algorithms", "custom": "Plugin-based custom arbitration algorithms" }, "features": [ "1-second collection window for wakeword reports", "Thread-safe concurrent processing", "Comprehensive analytics and history tracking", "Plugin support for custom algorithms", "Real-time monitoring and debugging capabilities", "Full integration with Trixy event system", "Configurable timing and algorithm parameters" ], "events": { "handled": ["wakeword_received"], "generated": [ "arbitration_started", "arbitration_completed", "arbitration_cancelled", "arbitration_timed_out", "arbitration_report_added", "satellite_selected", "satellite_ignored" ] } } def get_default_arbitration_config() -> ArbitrationConfig: """ Get default arbitration configuration for quick setup. Returns: ArbitrationConfig: Default configuration instance """ return create_default_config() def setup_arbitration_system( application, algorithm: ArbitrationAlgorithm = ArbitrationAlgorithm.VOLUME_BASED, collection_window: float = 1.0, enable_analytics: bool = True, debug_mode: bool = False ) -> ArbitrationManager: """ Set up a complete arbitration system with specified parameters. Args: application: Trixy application container algorithm: Primary arbitration algorithm to use collection_window: Collection window in seconds enable_analytics: Enable analytics and history tracking debug_mode: Enable debug logging Returns: ArbitrationManager: Configured and ready-to-use arbitration manager """ config = ArbitrationConfig() config.primary_algorithm = algorithm config.collection_window_seconds = collection_window config.enable_analytics = enable_analytics config.enable_debug_logging = debug_mode manager = ArbitrationManager(application, config) pprint(f"Set up arbitration system: algorithm={algorithm.value}, " f"window={collection_window}s, analytics={enable_analytics}") return manager def validate_arbitration_setup(application) -> list: """ Validate that all required components are available for arbitration. Args: application: Trixy application container Returns: list: List of validation issues (empty if all valid) """ issues = [] try: # Check for event handler if not hasattr(application, 'get_event_handler'): issues.append("Application missing get_event_handler method") else: event_handler = application.get_event_handler() if not event_handler: issues.append("Event handler not initialized") # Check for satellite manager if not hasattr(application, 'get_satellite_manager'): issues.append("Application missing get_satellite_manager method") else: satellite_manager = application.get_satellite_manager() if not satellite_manager: issues.append("Satellite manager not initialized") # Check for conversation manager if not hasattr(application, 'get_conversation_manager'): issues.append("Application missing get_conversation_manager method") else: conversation_manager = application.get_conversation_manager() if not conversation_manager: issues.append("Conversation manager not initialized") except Exception as e: issues.append(f"Error during validation: {e}") return issues def get_supported_algorithms() -> list: """ Get list of all supported arbitration algorithms. Returns: list: List of ArbitrationAlgorithm enum values """ return list(ArbitrationAlgorithm) def get_algorithm_descriptions() -> dict: """ Get descriptions of all arbitration algorithms. Returns: dict: Mapping of algorithm names to descriptions """ return { ArbitrationAlgorithm.VOLUME_BASED.value: "Select satellite with highest wakeword detection volume", ArbitrationAlgorithm.DISTANCE_BASED.value: "Select satellite closest to speaker based on audio analysis", ArbitrationAlgorithm.PRIORITY_BASED.value: "Select satellite based on configured room/satellite priorities", ArbitrationAlgorithm.ROUND_ROBIN.value: "Fair rotation among available satellites for equal opportunity", ArbitrationAlgorithm.HYBRID.value: "Weighted combination of multiple algorithms for sophisticated selection", ArbitrationAlgorithm.CUSTOM.value: "Plugin-based custom arbitration algorithm" } # Module-level configuration defaults ARBITRATION_DEFAULTS = { "algorithm": ArbitrationAlgorithm.VOLUME_BASED, "fallback_algorithm": ArbitrationAlgorithm.PRIORITY_BASED, "collection_window_seconds": 1.0, "arbitration_timeout_seconds": 2.0, "volume_threshold": 0.1, "enable_analytics": True, "enable_debug_logging": False, "max_concurrent_arbitrations": 10, "thread_pool_size": 4, } def get_arbitration_defaults() -> dict: """ Get default configuration values for arbitration system. Returns: dict: Default configuration values """ return ARBITRATION_DEFAULTS.copy() # Initialize logging for the arbitration package def _init_arbitration_logging(): """Initialize logging for the arbitration system.""" # This will be enhanced when integrated with proper logging system pass # Initialize the package _init_arbitration_logging() # Export additional utility functions __all__.extend([ "pprint", "get_arbitration_documentation", "get_default_arbitration_config", "setup_arbitration_system", "validate_arbitration_setup", "get_supported_algorithms", "get_algorithm_descriptions", "get_arbitration_defaults", ]) # Package-level convenience function for common use case def create_simple_arbitration_manager(application) -> ArbitrationManager: """ Create arbitration manager with sensible defaults for most use cases. Args: application: Trixy application container Returns: ArbitrationManager: Ready-to-use arbitration manager """ return setup_arbitration_system( application, algorithm=ArbitrationAlgorithm.VOLUME_BASED, collection_window=1.0, enable_analytics=True, debug_mode=False ) # Add convenience function to exports __all__.append("create_simple_arbitration_manager")