""" Event System Integration for Trixy ML Components. This module provides comprehensive event system integration for the complete ML workflow in the Trixy voice assistant. It handles event routing, workflow orchestration, and integration with the main application event system. Key Features: - Complete workflow event integration (wakeword -> voice recognition -> response) - Event routing and filtering - Workflow state management - Event analytics and monitoring - Custom event handlers and middleware - Integration with application container - Network protocol event bridging - Performance tracking and metrics Usage: from trixy_core.ml import MLEventManager, EventWorkflow # Create event manager event_manager = MLEventManager(app_container) # Register ML components event_manager.register_ml_manager(ml_manager) event_manager.register_audio_pipeline(audio_pipeline) # Start event processing await event_manager.start() """ import asyncio import logging import time import threading from typing import Dict, Any, Optional, List, Callable, Union, Type from dataclasses import dataclass, field from enum import Enum import uuid from datetime import datetime import json from collections import defaultdict, deque # ML component imports from .ml_manager import MLManager, WakewordEvent, AudioProcessingResult, MLProcessingState from .audio_pipeline import AudioPipeline, AudioChunk, AudioQualityMetrics from .wakeword import WakewordType, DetectionResult from .voice_recognition import IdentificationResult class EventType(Enum): """Event types in the ML workflow.""" # System events SYSTEM_STARTED = "system_started" SYSTEM_STOPPED = "system_stopped" SYSTEM_ERROR = "system_error" # Audio events AUDIO_INPUT_RECEIVED = "audio_input_received" AUDIO_QUALITY_ALERT = "audio_quality_alert" AUDIO_PIPELINE_ERROR = "audio_pipeline_error" SILENCE_DETECTED = "silence_detected" SILENCE_TIMEOUT = "silence_timeout" # Wakeword events WAKEWORD_RECEIVED = "wakeword_received" WAKEWORD_VALIDATED = "wakeword_validated" WAKEWORD_REJECTED = "wakeword_rejected" # Voice recognition events SPEAKER_IDENTIFIED = "speaker_identified" SPEAKER_UNKNOWN = "speaker_unknown" RAW_AUDIO_INPUT_RECEIVED = "raw_audio_input_received" # Workflow events CONVERSATION_STARTED = "conversation_started" CONVERSATION_ENDED = "conversation_ended" RECORDING_STARTED = "recording_started" RECORDING_COMPLETED = "recording_completed" # Network events SERVER_CONNECTED = "server_connected" SERVER_DISCONNECTED = "server_disconnected" AUDIO_STREAM_STARTED = "audio_stream_started" AUDIO_STREAM_STOPPED = "audio_stream_stopped" # Performance events PERFORMANCE_ALERT = "performance_alert" MEMORY_THRESHOLD_EXCEEDED = "memory_threshold_exceeded" PROCESSING_LATENCY_HIGH = "processing_latency_high" class EventPriority(Enum): """Event priority levels.""" LOW = 1 NORMAL = 2 HIGH = 3 CRITICAL = 4 class WorkflowState(Enum): """Workflow state enumeration.""" IDLE = "idle" LISTENING = "listening" WAKEWORD_DETECTED = "wakeword_detected" RECORDING = "recording" PROCESSING = "processing" WAITING_FOR_RESPONSE = "waiting_for_response" ERROR = "error" @dataclass class MLEvent: """ML event data structure.""" event_id: str event_type: EventType timestamp: float source: str priority: EventPriority = EventPriority.NORMAL data: Dict[str, Any] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) conversation_id: Optional[str] = None workflow_state: Optional[WorkflowState] = None def to_dict(self) -> Dict[str, Any]: """Convert event to dictionary.""" return { "event_id": self.event_id, "event_type": self.event_type.value, "timestamp": self.timestamp, "source": self.source, "priority": self.priority.value, "data": self.data, "metadata": self.metadata, "conversation_id": self.conversation_id, "workflow_state": self.workflow_state.value if self.workflow_state else None } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'MLEvent': """Create event from dictionary.""" return cls( event_id=data["event_id"], event_type=EventType(data["event_type"]), timestamp=data["timestamp"], source=data["source"], priority=EventPriority(data["priority"]), data=data.get("data", {}), metadata=data.get("metadata", {}), conversation_id=data.get("conversation_id"), workflow_state=WorkflowState(data["workflow_state"]) if data.get("workflow_state") else None ) @dataclass class ConversationContext: """Context for a conversation session.""" conversation_id: str start_time: float end_time: Optional[float] = None wakeword_event: Optional[WakewordEvent] = None audio_result: Optional[AudioProcessingResult] = None workflow_state: WorkflowState = WorkflowState.IDLE events: List[MLEvent] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) def duration(self) -> float: """Get conversation duration.""" end = self.end_time or time.time() return end - self.start_time def add_event(self, event: MLEvent) -> None: """Add event to conversation.""" event.conversation_id = self.conversation_id event.workflow_state = self.workflow_state self.events.append(event) class EventHandler: """Base event handler interface.""" def __init__(self, name: str): self.name = name self.enabled = True async def handle_event(self, event: MLEvent) -> bool: """ Handle an event. Args: event: Event to handle Returns: True if event was handled successfully """ raise NotImplementedError class EventFilter: """Event filter for selective processing.""" def __init__(self, name: str): self.name = name self.enabled = True def should_process(self, event: MLEvent) -> bool: """ Check if event should be processed. Args: event: Event to check Returns: True if event should be processed """ raise NotImplementedError class WorkflowEventHandler(EventHandler): """Handler for workflow state transitions.""" def __init__(self, event_manager: 'MLEventManager'): super().__init__("workflow_handler") self.event_manager = event_manager async def handle_event(self, event: MLEvent) -> bool: """Handle workflow events.""" if event.event_type == EventType.WAKEWORD_RECEIVED: await self._handle_wakeword_event(event) elif event.event_type == EventType.RAW_AUDIO_INPUT_RECEIVED: await self._handle_audio_input_event(event) elif event.event_type == EventType.SILENCE_TIMEOUT: await self._handle_silence_timeout(event) elif event.event_type == EventType.SYSTEM_ERROR: await self._handle_system_error(event) return True async def _handle_wakeword_event(self, event: MLEvent) -> None: """Handle wakeword detection event.""" # Start new conversation conversation_id = str(uuid.uuid4()) # Create conversation context context = ConversationContext( conversation_id=conversation_id, start_time=time.time(), workflow_state=WorkflowState.WAKEWORD_DETECTED ) # Store conversation self.event_manager._conversations[conversation_id] = context self.event_manager._current_conversation = conversation_id # Trigger conversation started event await self.event_manager.emit_event( EventType.CONVERSATION_STARTED, source="workflow_handler", data={ "conversation_id": conversation_id, "wakeword_type": event.data.get("wakeword_type"), "confidence": event.data.get("confidence") }, conversation_id=conversation_id ) # Update workflow state context.workflow_state = WorkflowState.RECORDING # Trigger recording started event await self.event_manager.emit_event( EventType.RECORDING_STARTED, source="workflow_handler", conversation_id=conversation_id ) async def _handle_audio_input_event(self, event: MLEvent) -> None: """Handle raw audio input event.""" conversation_id = event.conversation_id or self.event_manager._current_conversation if conversation_id and conversation_id in self.event_manager._conversations: context = self.event_manager._conversations[conversation_id] context.workflow_state = WorkflowState.PROCESSING # Trigger recording completed event await self.event_manager.emit_event( EventType.RECORDING_COMPLETED, source="workflow_handler", data={ "speaker_id": event.data.get("speaker_id"), "speaker_name": event.data.get("speaker_name"), "confidence": event.data.get("confidence"), "duration": event.data.get("duration") }, conversation_id=conversation_id ) async def _handle_silence_timeout(self, event: MLEvent) -> None: """Handle silence timeout event.""" conversation_id = event.conversation_id or self.event_manager._current_conversation if conversation_id and conversation_id in self.event_manager._conversations: await self._end_conversation(conversation_id, "silence_timeout") async def _handle_system_error(self, event: MLEvent) -> None: """Handle system error event.""" conversation_id = event.conversation_id or self.event_manager._current_conversation if conversation_id and conversation_id in self.event_manager._conversations: context = self.event_manager._conversations[conversation_id] context.workflow_state = WorkflowState.ERROR await self._end_conversation(conversation_id, "system_error") async def _end_conversation(self, conversation_id: str, reason: str) -> None: """End a conversation.""" if conversation_id in self.event_manager._conversations: context = self.event_manager._conversations[conversation_id] context.end_time = time.time() context.workflow_state = WorkflowState.IDLE # Trigger conversation ended event await self.event_manager.emit_event( EventType.CONVERSATION_ENDED, source="workflow_handler", data={ "conversation_id": conversation_id, "reason": reason, "duration": context.duration() }, conversation_id=conversation_id ) # Clear current conversation if this was it if self.event_manager._current_conversation == conversation_id: self.event_manager._current_conversation = None class NetworkEventHandler(EventHandler): """Handler for network protocol events.""" def __init__(self, network_client: Optional[Any] = None): super().__init__("network_handler") self.network_client = network_client async def handle_event(self, event: MLEvent) -> bool: """Handle network events.""" if not self.network_client: return True try: if event.event_type in [ EventType.WAKEWORD_RECEIVED, EventType.RAW_AUDIO_INPUT_RECEIVED, EventType.CONVERSATION_STARTED, EventType.CONVERSATION_ENDED ]: await self._send_event_to_server(event) return True except Exception as e: logging.error(f"Error handling network event: {e}") return False async def _send_event_to_server(self, event: MLEvent) -> None: """Send event to server.""" # This would integrate with the actual network client logging.debug(f"Sending event to server: {event.event_type.value}") class PerformanceEventHandler(EventHandler): """Handler for performance monitoring events.""" def __init__(self): super().__init__("performance_handler") self.metrics = { "events_processed": 0, "average_processing_time": 0.0, "error_count": 0, "conversation_count": 0 } async def handle_event(self, event: MLEvent) -> bool: """Handle performance events.""" start_time = time.time() try: # Update metrics self.metrics["events_processed"] += 1 if event.event_type == EventType.SYSTEM_ERROR: self.metrics["error_count"] += 1 elif event.event_type == EventType.CONVERSATION_STARTED: self.metrics["conversation_count"] += 1 # Track processing time processing_time = time.time() - start_time self.metrics["average_processing_time"] = ( (self.metrics["average_processing_time"] * (self.metrics["events_processed"] - 1) + processing_time) / self.metrics["events_processed"] ) return True except Exception as e: logging.error(f"Error in performance handler: {e}") self.metrics["error_count"] += 1 return False class PriorityEventFilter(EventFilter): """Filter events by priority.""" def __init__(self, min_priority: EventPriority): super().__init__("priority_filter") self.min_priority = min_priority def should_process(self, event: MLEvent) -> bool: """Check if event priority meets minimum threshold.""" return event.priority.value >= self.min_priority.value class EventTypeFilter(EventFilter): """Filter events by type.""" def __init__(self, allowed_types: List[EventType]): super().__init__("type_filter") self.allowed_types = set(allowed_types) def should_process(self, event: MLEvent) -> bool: """Check if event type is allowed.""" return event.event_type in self.allowed_types class MLEventManagerError(Exception): """ML event manager exception.""" pass class MLEventManager: """ ML Event Manager for comprehensive event system integration. This class manages the complete event workflow for ML components, handling event routing, filtering, and integration with the application event system. """ def __init__( self, app_container: Optional[Any] = None, logger: Optional[logging.Logger] = None ): """ Initialize ML event manager. Args: app_container: Application container for dependency injection logger: Logger instance """ self.app_container = app_container self.logger = logger or logging.getLogger(__name__) # Event management self._event_queue = asyncio.Queue() self._event_handlers: List[EventHandler] = [] self._event_filters: List[EventFilter] = [] self._running = False self._processing_task: Optional[asyncio.Task] = None # Workflow management self._conversations: Dict[str, ConversationContext] = {} self._current_conversation: Optional[str] = None self._workflow_state = WorkflowState.IDLE # Component references self._ml_manager: Optional[MLManager] = None self._audio_pipeline: Optional[AudioPipeline] = None # Event analytics self._event_history = deque(maxlen=1000) self._event_counts = defaultdict(int) # Initialize default handlers self._init_default_handlers() self.logger.info("ML Event Manager initialized") def _init_default_handlers(self) -> None: """Initialize default event handlers.""" # Add workflow handler workflow_handler = WorkflowEventHandler(self) self.add_event_handler(workflow_handler) # Add performance handler performance_handler = PerformanceEventHandler() self.add_event_handler(performance_handler) # Add network handler if network client is available network_client = getattr(self.app_container, 'network_client', None) if self.app_container else None if network_client: network_handler = NetworkEventHandler(network_client) self.add_event_handler(network_handler) async def start(self) -> None: """Start event processing.""" if self._running: self.logger.warning("Event manager already running") return self.logger.info("Starting ML event manager...") self._running = True # Start event processing task self._processing_task = asyncio.create_task(self._event_processing_loop()) # Emit system started event await self.emit_event( EventType.SYSTEM_STARTED, source="event_manager", data={"timestamp": time.time()} ) self.logger.info("ML event manager started") async def stop(self) -> None: """Stop event processing.""" if not self._running: return self.logger.info("Stopping ML event manager...") # Emit system stopped event await self.emit_event( EventType.SYSTEM_STOPPED, source="event_manager", data={"timestamp": time.time()} ) self._running = False # Cancel processing task if self._processing_task: self._processing_task.cancel() try: await self._processing_task except asyncio.CancelledError: pass # End any active conversations for conversation_id in list(self._conversations.keys()): if self._conversations[conversation_id].end_time is None: await self._end_conversation(conversation_id, "system_shutdown") self.logger.info("ML event manager stopped") def register_ml_manager(self, ml_manager: MLManager) -> None: """Register ML manager for event integration.""" self._ml_manager = ml_manager # Set up ML manager event callbacks ml_manager.event_handler = self self.logger.info("ML manager registered with event system") def register_audio_pipeline(self, audio_pipeline: AudioPipeline) -> None: """Register audio pipeline for event integration.""" self._audio_pipeline = audio_pipeline # Set up audio pipeline callbacks audio_pipeline.on_audio_processed = self._handle_audio_processed audio_pipeline.on_silence_detected = self._handle_silence_detected audio_pipeline.on_quality_alert = self._handle_quality_alert audio_pipeline.on_error = self._handle_audio_error self.logger.info("Audio pipeline registered with event system") def add_event_handler(self, handler: EventHandler) -> None: """Add event handler.""" self._event_handlers.append(handler) self.logger.info(f"Added event handler: {handler.name}") def remove_event_handler(self, handler: EventHandler) -> None: """Remove event handler.""" if handler in self._event_handlers: self._event_handlers.remove(handler) self.logger.info(f"Removed event handler: {handler.name}") def add_event_filter(self, event_filter: EventFilter) -> None: """Add event filter.""" self._event_filters.append(event_filter) self.logger.info(f"Added event filter: {event_filter.name}") def remove_event_filter(self, event_filter: EventFilter) -> None: """Remove event filter.""" if event_filter in self._event_filters: self._event_filters.remove(event_filter) self.logger.info(f"Removed event filter: {event_filter.name}") async def emit_event( self, event_type: EventType, source: str, data: Optional[Dict[str, Any]] = None, priority: EventPriority = EventPriority.NORMAL, conversation_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None ) -> str: """ Emit an event. Args: event_type: Type of event source: Event source identifier data: Event data priority: Event priority conversation_id: Associated conversation ID metadata: Event metadata Returns: Event ID """ event = MLEvent( event_id=str(uuid.uuid4()), event_type=event_type, timestamp=time.time(), source=source, priority=priority, data=data or {}, metadata=metadata or {}, conversation_id=conversation_id, workflow_state=self._workflow_state ) # Add to queue for processing await self._event_queue.put(event) return event.event_id async def _event_processing_loop(self) -> None: """Main event processing loop.""" self.logger.info("Event processing loop started") while self._running: try: # Get next event event = await asyncio.wait_for(self._event_queue.get(), timeout=1.0) # Process event await self._process_event(event) except asyncio.TimeoutError: continue except Exception as e: self.logger.error(f"Error in event processing loop: {e}") await asyncio.sleep(0.1) # Brief pause to prevent tight error loop self.logger.info("Event processing loop stopped") async def _process_event(self, event: MLEvent) -> None: """Process a single event.""" try: # Apply filters for event_filter in self._event_filters: if event_filter.enabled and not event_filter.should_process(event): return # Add to history and update counts self._event_history.append(event) self._event_counts[event.event_type] += 1 # Add to conversation if applicable if event.conversation_id and event.conversation_id in self._conversations: self._conversations[event.conversation_id].add_event(event) # Process with handlers for handler in self._event_handlers: if handler.enabled: try: await handler.handle_event(event) except Exception as e: self.logger.error(f"Error in event handler {handler.name}: {e}") # Integration with application event system if self.app_container and hasattr(self.app_container, 'event_system'): await self._integrate_with_app_events(event) except Exception as e: self.logger.error(f"Error processing event {event.event_id}: {e}") # Emit error event await self.emit_event( EventType.SYSTEM_ERROR, source="event_manager", data={"error": str(e), "original_event": event.event_id}, priority=EventPriority.HIGH ) async def _integrate_with_app_events(self, event: MLEvent) -> None: """Integrate with application event system.""" try: # Convert ML event to application event format app_event_data = { "type": f"ml_{event.event_type.value}", "source": "ml_system", "data": event.to_dict() } # Emit to application event system # This would integrate with the actual application event system self.logger.debug(f"Integrating event with app system: {event.event_type.value}") except Exception as e: self.logger.error(f"Error integrating with app events: {e}") # Event callback methods for ML components async def _handle_audio_processed(self, audio_chunk: AudioChunk) -> None: """Handle audio processed callback.""" await self.emit_event( EventType.AUDIO_INPUT_RECEIVED, source="audio_pipeline", data={ "timestamp": audio_chunk.timestamp, "sequence_number": audio_chunk.sequence_number, "volume": audio_chunk.volume, "is_silence": audio_chunk.is_silence } ) async def _handle_silence_detected(self, audio_chunk: AudioChunk) -> None: """Handle silence detected callback.""" await self.emit_event( EventType.SILENCE_DETECTED, source="audio_pipeline", data={ "timestamp": audio_chunk.timestamp, "sequence_number": audio_chunk.sequence_number } ) async def _handle_quality_alert(self, metrics: AudioQualityMetrics, alerts: List[str]) -> None: """Handle audio quality alert callback.""" await self.emit_event( EventType.AUDIO_QUALITY_ALERT, source="audio_pipeline", data={ "alerts": alerts, "snr_db": metrics.snr_estimate, "peak_level": metrics.peak_level, "silence_ratio": metrics.silence_ratio }, priority=EventPriority.HIGH ) async def _handle_audio_error(self, error_message: str) -> None: """Handle audio pipeline error callback.""" await self.emit_event( EventType.AUDIO_PIPELINE_ERROR, source="audio_pipeline", data={"error": error_message}, priority=EventPriority.HIGH ) def get_conversation_context(self, conversation_id: str) -> Optional[ConversationContext]: """Get conversation context by ID.""" return self._conversations.get(conversation_id) def get_current_conversation(self) -> Optional[ConversationContext]: """Get current active conversation.""" if self._current_conversation: return self._conversations.get(self._current_conversation) return None def get_event_statistics(self) -> Dict[str, Any]: """Get event processing statistics.""" total_events = sum(self._event_counts.values()) return { "total_events": total_events, "event_counts": dict(self._event_counts), "active_conversations": len([c for c in self._conversations.values() if c.end_time is None]), "total_conversations": len(self._conversations), "handlers_count": len(self._event_handlers), "filters_count": len(self._event_filters), "queue_size": self._event_queue.qsize(), "workflow_state": self._workflow_state.value } async def _end_conversation(self, conversation_id: str, reason: str) -> None: """End a conversation (used by workflow handler).""" if conversation_id in self._conversations: context = self._conversations[conversation_id] context.end_time = time.time() context.workflow_state = WorkflowState.IDLE # Clear current conversation if this was it if self._current_conversation == conversation_id: self._current_conversation = None def create_ml_event_manager( app_container: Optional[Any] = None, logger: Optional[logging.Logger] = None ) -> MLEventManager: """ Create ML event manager. Args: app_container: Application container logger: Logger instance Returns: ML event manager instance """ return MLEventManager(app_container, logger) def create_priority_filter(min_priority: EventPriority) -> PriorityEventFilter: """ Create priority event filter. Args: min_priority: Minimum priority level Returns: Priority event filter """ return PriorityEventFilter(min_priority) def create_type_filter(allowed_types: List[EventType]) -> EventTypeFilter: """ Create event type filter. Args: allowed_types: List of allowed event types Returns: Event type filter """ return EventTypeFilter(allowed_types)