Trixy3/trixy_core/ml/event_integration.py

"""
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)