""" Event handling decorators for the Trixy application. This module provides the @TrixyEvent decorator that allows methods to be registered as event handlers for specific event types. The decorator handles automatic registration with the EventHandler and provides validation. """ import functools import inspect from typing import Callable, List, Union, Any, Type, get_type_hints from .event_data import EventType, TrixyEventData def pprint(message: str) -> None: """ Debug printing function that adapts based on mode. In production: uses proper logging, in debug: uses print. """ # TODO: This should integrate with the application's logging system # For now, we'll use print but this should be replaced with proper logging print(f"[TrixyEvent] {message}") class TrixyEventRegistry: """ Registry for tracking all @TrixyEvent decorated methods. This class maintains a mapping of event types to their registered handlers and provides methods for registration and retrieval of handlers. """ def __init__(self): self._handlers: dict[str, List[Callable]] = {} self._handler_metadata: dict[Callable, dict] = {} def register_handler(self, event_types: List[str], handler: Callable, priority: int = 0, async_handler: bool = False) -> None: """ Register an event handler for the specified event types. Args: event_types: List of event type strings to handle handler: The handler function/method priority: Handler priority (higher = called first) async_handler: Whether the handler is async """ for event_type in event_types: if event_type not in self._handlers: self._handlers[event_type] = [] if handler not in self._handlers[event_type]: self._handlers[event_type].append(handler) # Store metadata self._handler_metadata[handler] = { 'event_types': event_types, 'priority': priority, 'async_handler': async_handler, 'class_name': getattr(handler, '__qualname__', handler.__name__), 'module_name': getattr(handler, '__module__', 'unknown') } # Sort handlers by priority (higher priority first) self._handlers[event_type].sort( key=lambda h: self._handler_metadata.get(h, {}).get('priority', 0), reverse=True ) pprint(f"Registered handler {handler.__name__} for events: {event_types}") def unregister_handler(self, handler: Callable) -> None: """ Unregister an event handler from all event types. Args: handler: The handler function/method to unregister """ if handler in self._handler_metadata: event_types = self._handler_metadata[handler]['event_types'] for event_type in event_types: if event_type in self._handlers and handler in self._handlers[event_type]: self._handlers[event_type].remove(handler) # Clean up empty event type lists if not self._handlers[event_type]: del self._handlers[event_type] del self._handler_metadata[handler] pprint(f"Unregistered handler {handler.__name__}") def get_handlers(self, event_type: str) -> List[Callable]: """ Get all handlers for a specific event type. Args: event_type: The event type string Returns: List of handler functions, sorted by priority """ return self._handlers.get(event_type, []).copy() def get_all_handlers(self) -> dict[str, List[Callable]]: """Get all registered handlers grouped by event type.""" return {k: v.copy() for k, v in self._handlers.items()} def get_handler_metadata(self, handler: Callable) -> dict: """Get metadata for a specific handler.""" return self._handler_metadata.get(handler, {}).copy() def get_event_types(self) -> List[str]: """Get all event types that have registered handlers.""" return list(self._handlers.keys()) def clear(self) -> None: """Clear all registered handlers.""" self._handlers.clear() self._handler_metadata.clear() pprint("Cleared all event handlers from registry") # Global registry instance _event_registry = TrixyEventRegistry() def get_event_registry() -> TrixyEventRegistry: """Get the global event registry instance.""" return _event_registry def TrixyEvent(event_types: Union[str, List[str]], priority: int = 0, async_handler: bool = False): """ Decorator for registering methods as Trixy event handlers. This decorator automatically registers the decorated method to handle the specified event types. The method will be called whenever those events are triggered through the EventHandler. Args: event_types: Single event type string or list of event type strings priority: Handler priority (0-100, higher = called first) async_handler: Set to True if the handler is an async function Example: @TrixyEvent(["wakeword_received", "text_received"]) def handle_events(self, event_name, event_data): if event_name == "wakeword_received": # Handle wakeword pass elif event_name == "text_received": # Handle text pass @TrixyEvent("system_startup", priority=10) def on_startup(self, event_name, event_data): # High priority startup handler pass """ def decorator(func: Callable) -> Callable: # Normalize event_types to list if isinstance(event_types, str): normalized_event_types = [event_types] else: normalized_event_types = list(event_types) # Validate event types for event_type in normalized_event_types: try: EventType(event_type) except ValueError: pprint(f"Warning: Unknown event type '{event_type}' in @TrixyEvent decorator") # Validate priority if not isinstance(priority, int) or priority < 0 or priority > 100: raise ValueError("Priority must be an integer between 0 and 100") # Validate function signature sig = inspect.signature(func) params = list(sig.parameters.keys()) # Check if it's a method (has self parameter) or function if len(params) < 2: raise ValueError( f"@TrixyEvent decorated function {func.__name__} must have at least " "2 parameters: (event_name, event_data) or (self, event_name, event_data)" ) # Store event information on the function for later registration if not hasattr(func, '_trixy_events'): func._trixy_events = [] func._trixy_events.append({ 'event_types': normalized_event_types, 'priority': priority, 'async_handler': async_handler }) # Mark the function as a trixy event handler func._is_trixy_event_handler = True @functools.wraps(func) def wrapper(*args, **kwargs): return func(*args, **kwargs) # Copy the trixy event metadata to the wrapper wrapper._trixy_events = func._trixy_events wrapper._is_trixy_event_handler = True return wrapper return decorator def register_event_handlers(instance: Any) -> None: """ Register all @TrixyEvent decorated methods from an instance. This function scans an object for methods decorated with @TrixyEvent and registers them with the global event registry. Args: instance: Object instance to scan for event handlers """ class_name = instance.__class__.__name__ registered_count = 0 for method_name in dir(instance): method = getattr(instance, method_name) # Check if this is a bound method with trixy event decoration if (hasattr(method, '_is_trixy_event_handler') and hasattr(method, '_trixy_events')): for event_config in method._trixy_events: _event_registry.register_handler( event_types=event_config['event_types'], handler=method, priority=event_config['priority'], async_handler=event_config['async_handler'] ) registered_count += 1 if registered_count > 0: pprint(f"Registered {registered_count} event handlers from {class_name}") def unregister_event_handlers(instance: Any) -> None: """ Unregister all @TrixyEvent decorated methods from an instance. Args: instance: Object instance to unregister handlers from """ class_name = instance.__class__.__name__ unregistered_count = 0 for method_name in dir(instance): method = getattr(instance, method_name) if (hasattr(method, '_is_trixy_event_handler') and hasattr(method, '_trixy_events')): _event_registry.unregister_handler(method) unregistered_count += 1 if unregistered_count > 0: pprint(f"Unregistered {unregistered_count} event handlers from {class_name}") def validate_event_handler(func: Callable) -> bool: """ Validate that a function has the correct signature for an event handler. Args: func: Function to validate Returns: bool: True if valid, False otherwise """ try: sig = inspect.signature(func) params = list(sig.parameters.keys()) # Should have at least 2 parameters: (event_name, event_data) or (self, event_name, event_data) if len(params) < 2: return False # Check if it has type hints for validation type_hints = get_type_hints(func) # Optional validation of type hints if len(params) >= 3: # Method with self event_name_param = params[1] event_data_param = params[2] else: # Function without self event_name_param = params[0] event_data_param = params[1] # Validate type hints if present if event_name_param in type_hints: if type_hints[event_name_param] != str: pprint(f"Warning: {func.__name__} event_name parameter should be typed as str") if event_data_param in type_hints: expected_type = type_hints[event_data_param] if not (expected_type == TrixyEventData or (hasattr(expected_type, '__origin__') and issubclass(expected_type.__origin__, TrixyEventData))): pprint(f"Warning: {func.__name__} event_data parameter should be typed as TrixyEventData") return True except Exception as e: pprint(f"Error validating event handler {func.__name__}: {e}") return False # Utility functions for debugging and introspection def list_registered_events() -> dict[str, List[str]]: """ Get a summary of all registered events and their handlers. Returns: dict: Mapping of event types to handler names """ result = {} for event_type, handlers in _event_registry.get_all_handlers().items(): result[event_type] = [ f"{h.__qualname__} (priority: {_event_registry.get_handler_metadata(h).get('priority', 0)})" for h in handlers ] return result def get_handler_info(handler: Callable) -> dict: """ Get detailed information about a specific event handler. Args: handler: The handler function to get info for Returns: dict: Handler information including events, priority, etc. """ return _event_registry.get_handler_metadata(handler)