Trixy3/trixy_core/events

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

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.

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.

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.

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

# 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

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

# 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

# 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

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:

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:

# 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

# 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:

# 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:

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.