""" Example usage of the Trixy Event System. This file demonstrates how to use the event system in practice, including defining event handlers, registering components, and triggering events. It also serves as a test of the implementation. """ import time from datetime import datetime from typing import Any import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) from trixy_core.events import ( EventHandler, TrixyEvent, EventType, SatelliteInfo, SpeakerInfo, WakewordReceivedEventData, TextReceivedEventData, SystemStartupEventData, PluginLoadedEventData ) class ExamplePlugin: """Example plugin demonstrating event handling.""" def __init__(self, name: str): self.name = name self.wakeword_count = 0 self.text_count = 0 @TrixyEvent(["wakeword_received", "text_received"], priority=10) def handle_voice_events(self, event_name: str, event_data: Any): """Handle voice-related events with high priority.""" if event_name == "wakeword_received": self.wakeword_count += 1 speaker_name = event_data.speaker_info.speaker_name satellite_room = event_data.satellite_info.room_id print(f"[{self.name}] Wakeword from {speaker_name} in {satellite_room} " f"(total: {self.wakeword_count})") elif event_name == "text_received": self.text_count += 1 text = event_data.text confidence = event_data.confidence print(f"[{self.name}] Text: '{text}' (confidence: {confidence:.2f}) " f"(total: {self.text_count})") @TrixyEvent("system_startup", priority=5) def on_startup(self, event_name: str, event_data: SystemStartupEventData): """Handle system startup.""" print(f"[{self.name}] System started in {event_data.mode} mode, " f"version {event_data.version}") @TrixyEvent(["plugin_loaded", "plugin_unloaded"]) def on_plugin_events(self, event_name: str, event_data: Any): """Handle plugin events.""" plugin_name = event_data.plugin_name if event_name == "plugin_loaded": print(f"[{self.name}] Plugin '{plugin_name}' was loaded") else: print(f"[{self.name}] Plugin '{plugin_name}' was unloaded") class ExampleSatelliteManager: """Example satellite manager component.""" def __init__(self): self.connected_satellites = {} @TrixyEvent(["satellite_connected", "satellite_disconnected"], priority=20) def manage_satellites(self, event_name: str, event_data: Any): """Manage satellite connections.""" satellite_info = event_data.satellite_info satellite_id = satellite_info.satellite_id if event_name == "satellite_connected": self.connected_satellites[satellite_id] = satellite_info print(f"[SatelliteManager] Connected: {satellite_info.alias} " f"in {satellite_info.room_id} ({len(self.connected_satellites)} total)") elif event_name == "satellite_disconnected": if satellite_id in self.connected_satellites: del self.connected_satellites[satellite_id] print(f"[SatelliteManager] Disconnected: {satellite_id} " f"({len(self.connected_satellites)} remaining)") def demonstrate_event_system(): """Demonstrate the complete event system functionality.""" print("=== Trixy Event System Demonstration ===\n") # Create event handler print("1. Creating EventHandler...") event_handler = EventHandler(max_history_size=50) event_handler.enable_debug_mode(True) # Create example components print("\n2. Creating components...") voice_plugin = ExamplePlugin("VoicePlugin") nlp_plugin = ExamplePlugin("NLPPlugin") satellite_manager = ExampleSatelliteManager() # Register components print("\n3. Registering event handlers...") event_handler.register_handler_object(voice_plugin) event_handler.register_handler_object(nlp_plugin) event_handler.register_handler_object(satellite_manager) # Display registered handlers print("\n4. Registered handlers:") handlers = event_handler.get_registered_handlers() for event_type, handler_list in handlers.items(): print(f" {event_type}: {', '.join(handler_list)}") print("\n5. Triggering events...\n") # Trigger system startup print("--- System Startup ---") event_handler.trigger_event( EventType.SYSTEM_STARTUP, mode="server", version="1.0.0", config_file="server_config.json", debug_mode=True ) # Create sample satellite and speaker info satellite_info = SatelliteInfo( satellite_id="sat_001", mac_address="AA:BB:CC:DD:EE:FF", room_id="kitchen", alias="Kitchen Speaker", version="1.0.0", audio_ports={"input": 2102, "output": 2103} ) speaker_info = SpeakerInfo( speaker_id="spk_001", speaker_name="Alice", confidence=0.95 ) time.sleep(0.1) # Small delay for readability # Trigger satellite connection print("\n--- Satellite Connection ---") event_handler.trigger_event( EventType.SATELLITE_CONNECTED, satellite_info=satellite_info, server_assigned_ports={"input": 2102, "output": 2103, "music": 2104} ) time.sleep(0.1) # Trigger wakeword detection print("\n--- Wakeword Detection ---") event_handler.trigger_event( EventType.WAKEWORD_RECEIVED, wakeword_id="trixy", speaker_info=speaker_info, satellite_info=satellite_info, volume=0.8, confidence=0.92 ) time.sleep(0.1) # Trigger text processing print("\n--- Text Processing ---") event_handler.trigger_event( EventType.TEXT_RECEIVED, conversation_id="conv_001", text="What's the weather like today?", confidence=0.89, speaker_info=speaker_info, satellite_info=satellite_info ) time.sleep(0.1) # Trigger plugin events print("\n--- Plugin Events ---") event_handler.trigger_event( EventType.PLUGIN_LOADED, plugin_name="weather_plugin", plugin_path="/plugins/weather_plugin", plugin_version="2.1.0", load_time_ms=150.5 ) time.sleep(0.1) # Trigger more events to show statistics print("\n--- Additional Events for Statistics ---") for i in range(3): event_handler.trigger_event( EventType.WAKEWORD_RECEIVED, wakeword_id="trixy", speaker_info=speaker_info, satellite_info=satellite_info, volume=0.7 + i * 0.1, confidence=0.85 + i * 0.05 ) # Show event statistics print("\n6. Event Statistics:") stats = event_handler.get_event_statistics() for event_type, type_stats in stats.items(): print(f" {event_type}:") print(f" Triggered: {type_stats['triggered']}") print(f" Completed: {type_stats['completed']}") print(f" Failed: {type_stats['failed']}") print(f" Success Rate: {type_stats['success_rate_percent']}%") print(f" Avg Execution Time: {type_stats['avg_execution_time_ms']:.2f}ms") print(f" Handlers: {type_stats['handlers_registered']}") # Show recent event history print("\n7. Recent Event History:") history = event_handler.get_event_history(limit=5) for entry in history: print(f" {entry.timestamp.strftime('%H:%M:%S.%f')[:-3]} - " f"{entry.event_type} - {entry.status.value} - " f"{entry.execution_time_ms:.1f}ms") # Test async events print("\n8. Testing Async Events...") async_event_ids = [] for i in range(3): event_id = event_handler.trigger_event_async( EventType.TEXT_RECEIVED, conversation_id=f"async_conv_{i}", text=f"Async message {i + 1}", confidence=0.9, speaker_info=speaker_info ) async_event_ids.append(event_id) # Wait for async events to complete print(" Waiting for async events to complete...") for event_id in async_event_ids: result = event_handler.wait_for_event(event_id, timeout=5.0) if result: print(f" Async event {event_id} completed in {result.execution_time_ms:.1f}ms") else: print(f" Async event {event_id} timed out") # Test event filtering print("\n9. Testing Event Filtering...") print(" Disabling wakeword_received events...") event_handler.disable_event_type("wakeword_received") # This should be ignored event_handler.trigger_event( EventType.WAKEWORD_RECEIVED, wakeword_id="trixy", speaker_info=speaker_info, satellite_info=satellite_info, volume=0.5 ) print(" Re-enabling wakeword_received events...") event_handler.enable_event_type("wakeword_received") # This should work event_handler.trigger_event( EventType.WAKEWORD_RECEIVED, wakeword_id="trixy", speaker_info=speaker_info, satellite_info=satellite_info, volume=0.9 ) # Final statistics print("\n10. Final Statistics:") final_stats = event_handler.get_event_statistics() for event_type, type_stats in final_stats.items(): print(f" {event_type}: {type_stats['triggered']} triggered, " f"{type_stats['completed']} completed") # Cleanup print("\n11. Shutting down event handler...") event_handler.shutdown() print("\n=== Demonstration Complete ===") if __name__ == "__main__": demonstrate_event_system()