""" Trixy Event System This module provides a comprehensive event handling system for the Trixy application. It includes event data structures, decorators for event handler registration, and a thread-safe event handler with history tracking and debugging capabilities. The event system is the central component that enables communication between all parts of the Trixy application, including plugins, satellites, and core systems. Usage Example: # Import the event system from trixy_core.events import EventHandler, TrixyEvent, EventType # Create an event handler event_handler = EventHandler() # Define an event handler method class MyComponent: @TrixyEvent(["wakeword_received", "text_received"]) def handle_events(self, event_name, event_data): if event_name == "wakeword_received": print(f"Wakeword detected from {event_data.speaker_info.speaker_name}") elif event_name == "text_received": print(f"Text received: {event_data.text}") # Register the component component = MyComponent() event_handler.register_handler_object(component) # Trigger events event_handler.trigger_event("system_startup", mode="server", version="1.0.0") """ from .event_data import ( # Core event data classes TrixyEventData, EventType, EventDataFactory, # Data structures SatelliteInfo, SpeakerInfo, # Event data classes - Satellite Management SatelliteConnectedEventData, SatelliteDisconnectedEventData, SatelliteRegisteredEventData, # Event data classes - Wakeword & Audio WakewordReceivedEventData, RawAudioInputReceivedEventData, # Event data classes - Processing TextReceivedEventData, IntentReceivedEventData, TTSReceivedEventData, # Event data classes - System SystemStartupEventData, SystemShutdownEventData, PluginEventData, PluginLoadedEventData, PluginUnloadedEventData, TrainingStartedEventData, TrainingCompletedEventData, ScheduleTriggeredEventData, # Event data classes - Arbitration ArbitrationStartedEventData, ArbitrationCompletedEventData, ArbitrationCancelledEventData, ArbitrationTimedOutEventData, ArbitrationReportAddedEventData, SatelliteSelectedEventData, SatelliteIgnoredEventData, ) from .decorators import ( # Main decorator TrixyEvent, # Registry functions register_event_handlers, unregister_event_handlers, get_event_registry, TrixyEventRegistry, # Utility functions validate_event_handler, list_registered_events, get_handler_info, ) from .event_handler import ( # Main event handler class EventHandler, # Supporting classes EventStatus, EventHistoryEntry, ) # Version information __version__ = "1.0.0" __author__ = "Trixy Development Team" # Public API __all__ = [ # Core classes "EventHandler", "TrixyEvent", "EventType", "TrixyEventData", "EventDataFactory", # Event status and history "EventStatus", "EventHistoryEntry", # Data structures "SatelliteInfo", "SpeakerInfo", # Event data classes - Satellite Management "SatelliteConnectedEventData", "SatelliteDisconnectedEventData", "SatelliteRegisteredEventData", # Event data classes - Wakeword & Audio "WakewordReceivedEventData", "RawAudioInputReceivedEventData", # Event data classes - Processing "TextReceivedEventData", "IntentReceivedEventData", "TTSReceivedEventData", # Event data classes - System "SystemStartupEventData", "SystemShutdownEventData", "PluginEventData", "PluginLoadedEventData", "PluginUnloadedEventData", "TrainingStartedEventData", "TrainingCompletedEventData", "ScheduleTriggeredEventData", # Event data classes - Arbitration "ArbitrationStartedEventData", "ArbitrationCompletedEventData", "ArbitrationCancelledEventData", "ArbitrationTimedOutEventData", "ArbitrationReportAddedEventData", "SatelliteSelectedEventData", "SatelliteIgnoredEventData", # Registry and utilities "TrixyEventRegistry", "register_event_handlers", "unregister_event_handlers", "get_event_registry", "validate_event_handler", "list_registered_events", "get_handler_info", ] def create_default_event_handler(**kwargs) -> EventHandler: """ Create a default EventHandler instance with reasonable defaults. Args: **kwargs: Additional arguments to pass to EventHandler constructor Returns: EventHandler: Configured event handler instance """ return EventHandler(**kwargs) def get_supported_event_types(): """ Get a list of all supported event types. Returns: List[EventType]: All supported event types """ return list(EventType) def get_event_documentation(): """ Get documentation for all event types and their data structures. Returns: Dict[str, str]: Mapping of event types to their descriptions """ return { # Satellite Management Events "satellite_connected": "Triggered when a satellite establishes connection to the server", "satellite_disconnected": "Triggered when a satellite loses connection to the server", "satellite_registered": "Triggered when a new satellite is registered with the system", # Wakeword & Audio Events "wakeword_received": "Triggered when a satellite detects the wakeword", "raw_audio_input_received": "Triggered when audio recording from satellite completes", # Processing Events "text_received": "Triggered when STT (Speech-to-Text) converts audio to text", "intent_received": "Triggered when NLP extracts intent from text", "tts_received": "Triggered when TTS (Text-to-Speech) generates response audio", # System Events "system_startup": "Triggered when the system completes initialization", "system_shutdown": "Triggered when the system is shutting down", "plugin_loaded": "Triggered when a plugin is successfully loaded", "plugin_unloaded": "Triggered when a plugin is unloaded", "training_started": "Triggered when ML model training begins", "training_completed": "Triggered when ML model training finishes", "schedule_triggered": "Triggered when a scheduled event fires", # Arbitration Events "arbitration_started": "Triggered when wakeword arbitration session begins", "arbitration_completed": "Triggered when satellite selection is completed", "arbitration_cancelled": "Triggered when arbitration session is cancelled", "arbitration_timed_out": "Triggered when arbitration session times out", "arbitration_report_added": "Triggered when wakeword report is added to session", "satellite_selected": "Triggered when satellite is selected for conversation", "satellite_ignored": "Triggered when satellite is told to ignore wakeword", } # Initialize logging function that will be replaced by proper logging system def _init_logging(): """Initialize logging for the events module.""" # This will be replaced with proper logging integration pass # Initialize the module _init_logging() # Module-level debug function for consistent logging def pprint(message: str) -> None: """ Module-level debug printing function. This should be replaced with proper logging integration. """ print(f"[trixy_core.events] {message}") # Convenience function for quick event handler setup def setup_event_system(debug_mode: bool = False, max_history: int = 1000) -> EventHandler: """ Set up a complete event system with reasonable defaults. Args: debug_mode: Enable debug mode for verbose logging max_history: Maximum number of events to keep in history Returns: EventHandler: Configured and ready-to-use event handler """ event_handler = EventHandler(max_history_size=max_history) event_handler.enable_debug_mode(debug_mode) pprint(f"Event system initialized (debug: {debug_mode}, max_history: {max_history})") return event_handler