# Trixy Event System A comprehensive, thread-safe event handling system for the Trixy voice assistant application. This system serves as the central communication hub between all components, including plugins, satellites, and core systems. ## Features - **Event-Driven Architecture**: All system components communicate through events - **Thread-Safe Operations**: Supports multi-satellite environments with concurrent event processing - **Type-Safe Event Data**: Structured event data classes with full type hints - **Event History & Debugging**: Comprehensive logging and debugging capabilities - **Event Filtering**: Enable/disable specific event types dynamically - **Async Support**: Both synchronous and asynchronous event processing - **Priority System**: Event handlers can be prioritized for execution order - **Auto-Registration**: Decorator-based event handler registration ## Quick Start ```python from trixy_core.events import EventHandler, TrixyEvent, EventType # Create event handler event_handler = EventHandler() # Define a component with event handlers class MyComponent: @TrixyEvent(["wakeword_received", "text_received"]) def handle_voice_events(self, event_name, event_data): if event_name == "wakeword_received": print(f"Wakeword from {event_data.speaker_info.speaker_name}") elif event_name == "text_received": print(f"Text: {event_data.text}") # Register the component component = MyComponent() event_handler.register_handler_object(component) # Trigger events event_handler.trigger_event( EventType.SYSTEM_STARTUP, mode="server", version="1.0.0" ) ``` ## Core Components ### 1. EventHandler Class The main event processing engine that manages event registration, triggering, and history. ```python event_handler = EventHandler( max_history_size=1000, # Maximum events in history max_worker_threads=10 # Thread pool size for async events ) # Configuration event_handler.enable_debug_mode(True) event_handler.enable_event_type("wakeword_received") event_handler.disable_event_type("debug_events") # Event triggering sync_event_id = event_handler.trigger_event(EventType.WAKEWORD_RECEIVED, ...) async_event_id = event_handler.trigger_event_async(EventType.TEXT_RECEIVED, ...) # Monitoring history = event_handler.get_event_history(limit=10) stats = event_handler.get_event_statistics() ``` ### 2. @TrixyEvent Decorator Decorator for automatic event handler registration with priority support. ```python class Plugin: @TrixyEvent("wakeword_received", priority=10) def high_priority_handler(self, event_name, event_data): # This runs first (higher priority) pass @TrixyEvent(["text_received", "intent_received"], priority=5) def multi_event_handler(self, event_name, event_data): # Handle multiple event types pass @TrixyEvent("training_completed", async_handler=True) async def async_handler(self, event_name, event_data): # Async event handler await some_async_operation() ``` ### 3. Event Data Classes Structured, type-safe data containers for event information. ```python from trixy_core.events import ( WakewordReceivedEventData, SatelliteInfo, SpeakerInfo ) # Create event data satellite_info = SatelliteInfo( satellite_id="kitchen_01", mac_address="AA:BB:CC:DD:EE:FF", room_id="kitchen", alias="Kitchen Speaker", version="1.0.0" ) speaker_info = SpeakerInfo( speaker_id="user_001", speaker_name="Alice", confidence=0.95 ) # Use with events event_handler.trigger_event( EventType.WAKEWORD_RECEIVED, wakeword_id="trixy", speaker_info=speaker_info, satellite_info=satellite_info, volume=0.8, confidence=0.92 ) ``` ## Supported Events ### Satellite Management Events - `satellite_connected` - Satellite establishes connection - `satellite_disconnected` - Satellite loses connection - `satellite_registered` - New satellite registration ### Wakeword & Audio Events - `wakeword_received` - Wakeword detection from satellite - `raw_audio_input_received` - Audio recording completion ### Processing Events - `text_received` - STT (Speech-to-Text) conversion - `intent_received` - NLP intent extraction - `tts_received` - TTS (Text-to-Speech) generation ### System Events - `system_startup` - System initialization complete - `system_shutdown` - System shutting down - `plugin_loaded` - Plugin successfully loaded - `plugin_unloaded` - Plugin unloaded - `training_started` - ML training begins - `training_completed` - ML training finishes - `schedule_triggered` - Scheduled event fires ## Advanced Usage ### Event Filtering ```python # Disable specific events during maintenance event_handler.disable_event_type("wakeword_received") # Re-enable later event_handler.enable_event_type("wakeword_received") # Check if enabled if event_handler.is_event_type_enabled("text_received"): # Process normally pass ``` ### Event Listeners ```python def event_listener(history_entry): print(f"Event {history_entry.event_type} completed in {history_entry.execution_time_ms}ms") event_handler.add_event_listener(event_listener) ``` ### Waiting for Events ```python # Trigger async event event_id = event_handler.trigger_event_async(EventType.TRAINING_STARTED, ...) # Wait for completion result = event_handler.wait_for_event(event_id, timeout=30.0) if result and result.status == EventStatus.COMPLETED: print("Training started successfully!") ``` ### Event History Analysis ```python # Get recent events recent = event_handler.get_event_history(limit=50) # Get events by type wakeword_events = event_handler.get_event_history( event_type="wakeword_received", limit=10 ) # Get events since timestamp from datetime import datetime, timedelta since_hour_ago = datetime.now() - timedelta(hours=1) recent_events = event_handler.get_event_history(since=since_hour_ago) ``` ### Statistics and Monitoring ```python stats = event_handler.get_event_statistics() for event_type, type_stats in stats.items(): print(f"{event_type}:") print(f" Success Rate: {type_stats['success_rate_percent']}%") print(f" Avg Execution: {type_stats['avg_execution_time_ms']}ms") print(f" Total Triggered: {type_stats['triggered']}") ``` ## Plugin Integration The event system integrates seamlessly with the Trixy plugin system: ```python from trixy_core.plugins import TrixyPlugin from trixy_core.events import TrixyEvent class WeatherPlugin(TrixyPlugin): @TrixyEvent("intent_received", priority=5) def handle_weather_intent(self, event_name, event_data): if event_data.intent == "weather_query": # Process weather request self.get_weather_info(event_data.entities) @TrixyEvent("system_startup") def on_startup(self, event_name, event_data): # Initialize weather service self.initialize_weather_api() ``` ## Error Handling The event system provides comprehensive error handling: ```python # Events with errors are logged @TrixyEvent("text_received") def potentially_failing_handler(self, event_name, event_data): if some_error_condition: raise ValueError("Processing failed") # Error is caught, logged, and doesn't stop other handlers # Check for errors in history history = event_handler.get_event_history(limit=1) if history[0].errors: print(f"Event had errors: {history[0].errors}") ``` ## Performance Considerations - **Thread Pool**: Async events use a configurable thread pool - **History Limits**: Event history is automatically truncated to prevent memory issues - **Event Filtering**: Disabled events are filtered early to minimize overhead - **Priority Ordering**: Handlers are sorted by priority once during registration ## Production Deployment ```python # Production configuration event_handler = EventHandler( max_history_size=5000, # Larger history for production max_worker_threads=20 # More threads for busy systems ) # Disable debug mode for performance event_handler.enable_debug_mode(False) # Add monitoring def production_monitor(history_entry): if history_entry.execution_time_ms > 1000: # Log slow events logger.warning(f"Slow event: {history_entry.event_type} took {history_entry.execution_time_ms}ms") event_handler.add_event_listener(production_monitor) ``` ## Files Structure ``` trixy_core/events/ ├── __init__.py # Public API exports ├── event_handler.py # Main EventHandler class ├── decorators.py # @TrixyEvent decorator and registry ├── event_data.py # Event data structures and factory ├── example_usage.py # Usage examples and demonstrations ├── test_basic.py # Basic functionality tests └── README.md # This documentation ``` ## Integration with Trixy Core The event system is designed to be the central communication hub: ```python # In main application from trixy_core.events import setup_event_system # Initialize event system event_handler = setup_event_system(debug_mode=True) # Register with application container application.register_component("event_handler", event_handler) # Other components get access satellite_manager = application.get_satellite_manager() plugin_system = application.get_plugin_system() # Components automatically register their event handlers event_handler.register_handler_object(satellite_manager) event_handler.register_handler_object(plugin_system) ``` ## Best Practices 1. **Use Type Hints**: Always type your event handlers for better IDE support 2. **Handle Errors**: Wrap handler logic in try-catch for robustness 3. **Minimize Handler Time**: Keep event handlers fast to avoid blocking 4. **Use Priority**: Set appropriate priorities for handler execution order 5. **Event Filtering**: Use event filtering for maintenance and debugging 6. **Monitor Performance**: Use event listeners to monitor system performance ## Troubleshooting ### Common Issues 1. **Handler Not Called**: Ensure the component is registered with `register_handler_object()` 2. **Import Errors**: Use absolute imports: `from trixy_core.events import ...` 3. **Performance Issues**: Check event statistics for slow handlers 4. **Memory Usage**: Adjust `max_history_size` for your system requirements ### Debug Mode Enable debug mode for verbose logging: ```python event_handler.enable_debug_mode(True) # Now all event processing is logged with detailed information ``` For more examples, see `example_usage.py` and `test_basic.py` in this directory.