""" Complete Conversation Workflow Integration for Trixy ML System. This module implements the complete conversation workflow from wakeword detection to response generation, coordinating all ML components and integrating with the main application to provide seamless voice interaction capabilities. Workflow Steps: 1. Continuous wakeword listening 2. Wakeword detection and speaker identification 3. Audio recording and voice activity detection 4. Voice recognition and speaker verification 5. Audio processing and network transmission 6. Response handling and audio playback 7. Return to listening state Key Features: - Complete end-to-end conversation flow - State machine-based workflow management - Audio buffer management and optimization - Real-time processing and low-latency response - Error handling and recovery mechanisms - Performance monitoring and analytics - Integration with application container - Configurable workflow parameters Usage: from trixy_core.ml import ConversationWorkflow, WorkflowConfig # Create workflow configuration config = WorkflowConfig( wakeword_sensitivity=0.7, max_recording_duration=30.0, response_timeout=10.0 ) # Initialize workflow workflow = ConversationWorkflow(config, ml_manager, app_container) # Start conversation processing await workflow.start() # Feed audio data await workflow.process_audio(audio_data) """ import asyncio import logging import time import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Any, Dict, List, Optional, Callable, Union, Tuple import numpy as np from .performance_monitor import PerformanceMonitor, MetricType class ConversationState(Enum): """Conversation workflow states.""" IDLE = "idle" LISTENING = "listening" WAKEWORD_DETECTED = "wakeword_detected" RECORDING = "recording" PROCESSING = "processing" WAITING_RESPONSE = "waiting_response" PLAYING_RESPONSE = "playing_response" ERROR = "error" TIMEOUT = "timeout" class WorkflowEvent(Enum): """Workflow events.""" START_LISTENING = "start_listening" WAKEWORD_DETECTED = "wakeword_detected" START_RECORDING = "start_recording" VOICE_ACTIVITY_DETECTED = "voice_activity_detected" SILENCE_DETECTED = "silence_detected" RECORDING_COMPLETE = "recording_complete" PROCESSING_COMPLETE = "processing_complete" RESPONSE_RECEIVED = "response_received" RESPONSE_COMPLETE = "response_complete" TIMEOUT_OCCURRED = "timeout_occurred" ERROR_OCCURRED = "error_occurred" RESET = "reset" @dataclass class WorkflowConfig: """Configuration for conversation workflow.""" # Wakeword detection settings wakeword_sensitivity: float = 0.7 wakeword_cooldown_seconds: float = 2.0 continuous_listening: bool = True # Recording settings max_recording_duration: float = 30.0 silence_timeout: float = 2.0 min_recording_duration: float = 0.5 voice_activity_threshold: float = 0.01 # Processing settings processing_timeout: float = 15.0 response_timeout: float = 10.0 enable_speaker_verification: bool = True enable_noise_reduction: bool = True # Audio settings sample_rate: int = 16000 chunk_size: int = 1024 audio_format: str = "int16" # Network settings server_timeout: float = 30.0 max_retries: int = 3 retry_delay: float = 1.0 # Performance settings enable_performance_monitoring: bool = True log_conversation_metrics: bool = True enable_audio_caching: bool = False def validate(self) -> bool: """Validate configuration.""" if not (0.0 <= self.wakeword_sensitivity <= 1.0): return False if self.max_recording_duration <= 0: return False if self.silence_timeout <= 0: return False if self.processing_timeout <= 0: return False return True @dataclass class ConversationSession: """Represents an active conversation session.""" session_id: str start_time: datetime speaker_id: Optional[str] = None speaker_name: Optional[str] = None wakeword_confidence: float = 0.0 audio_buffer: List[np.ndarray] = field(default_factory=list) processing_start_time: Optional[datetime] = None response_start_time: Optional[datetime] = None state: ConversationState = ConversationState.IDLE metadata: Dict[str, Any] = field(default_factory=dict) @property def duration(self) -> float: """Get session duration in seconds.""" return (datetime.now() - self.start_time).total_seconds() @property def audio_duration(self) -> float: """Get total audio duration in seconds.""" if not self.audio_buffer: return 0.0 total_samples = sum(len(chunk) for chunk in self.audio_buffer) return total_samples / 16000 # Assuming 16kHz sample rate class ConversationWorkflowError(Exception): """Base exception for conversation workflow errors.""" pass class ConversationWorkflow: """ Complete conversation workflow manager. Coordinates the entire conversation flow from wakeword detection to response, managing state transitions, audio processing, and integration with ML components. """ def __init__( self, config: WorkflowConfig, ml_manager: Any, app_container: Optional[Any] = None, performance_monitor: Optional[PerformanceMonitor] = None, logger: Optional[logging.Logger] = None ): """ Initialize conversation workflow. Args: config: Workflow configuration ml_manager: ML manager instance app_container: Application container performance_monitor: Performance monitor instance logger: Logger instance """ self.config = config self.ml_manager = ml_manager self.app_container = app_container self.performance_monitor = performance_monitor self.logger = logger or logging.getLogger(__name__) if not self.config.validate(): raise ConversationWorkflowError("Invalid workflow configuration") # State management self._current_state = ConversationState.IDLE self._previous_state = ConversationState.IDLE self._state_start_time = time.time() self._running = False # Session management self._current_session: Optional[ConversationSession] = None self._session_history: List[ConversationSession] = [] self._max_history_size = 100 # Audio processing self._audio_buffer = [] self._recording_start_time = 0.0 self._last_voice_activity = 0.0 self._silence_start_time = 0.0 # Performance tracking self._workflow_stats = { "total_conversations": 0, "successful_conversations": 0, "failed_conversations": 0, "average_response_time": 0.0, "total_processing_time": 0.0, "wakeword_detections": 0, "false_wakewords": 0 } # Event handlers self._event_handlers: Dict[WorkflowEvent, List[Callable]] = {} self._state_transition_handlers: Dict[Tuple[ConversationState, ConversationState], List[Callable]] = {} # Setup state machine self._setup_state_machine() self.logger.info("Conversation workflow initialized") def _setup_state_machine(self) -> None: """Setup state machine transitions and handlers.""" # Define valid state transitions self._valid_transitions = { ConversationState.IDLE: [ConversationState.LISTENING, ConversationState.ERROR], ConversationState.LISTENING: [ConversationState.WAKEWORD_DETECTED, ConversationState.IDLE, ConversationState.ERROR], ConversationState.WAKEWORD_DETECTED: [ConversationState.RECORDING, ConversationState.LISTENING, ConversationState.ERROR], ConversationState.RECORDING: [ConversationState.PROCESSING, ConversationState.LISTENING, ConversationState.TIMEOUT, ConversationState.ERROR], ConversationState.PROCESSING: [ConversationState.WAITING_RESPONSE, ConversationState.LISTENING, ConversationState.TIMEOUT, ConversationState.ERROR], ConversationState.WAITING_RESPONSE: [ConversationState.PLAYING_RESPONSE, ConversationState.LISTENING, ConversationState.TIMEOUT, ConversationState.ERROR], ConversationState.PLAYING_RESPONSE: [ConversationState.LISTENING, ConversationState.IDLE, ConversationState.ERROR], ConversationState.ERROR: [ConversationState.IDLE, ConversationState.LISTENING], ConversationState.TIMEOUT: [ConversationState.IDLE, ConversationState.LISTENING] } # Register default event handlers self._register_default_handlers() def _register_default_handlers(self) -> None: """Register default event handlers.""" # Wakeword detection handler async def handle_wakeword_detected(event_data: Dict[str, Any]) -> None: await self._transition_to_state(ConversationState.WAKEWORD_DETECTED) await self._start_new_session(event_data) self.register_event_handler(WorkflowEvent.WAKEWORD_DETECTED, handle_wakeword_detected) # Recording complete handler async def handle_recording_complete(event_data: Dict[str, Any]) -> None: await self._transition_to_state(ConversationState.PROCESSING) await self._process_recorded_audio() self.register_event_handler(WorkflowEvent.RECORDING_COMPLETE, handle_recording_complete) # Response received handler async def handle_response_received(event_data: Dict[str, Any]) -> None: await self._transition_to_state(ConversationState.PLAYING_RESPONSE) await self._play_response(event_data) self.register_event_handler(WorkflowEvent.RESPONSE_RECEIVED, handle_response_received) # Error handler async def handle_error(event_data: Dict[str, Any]) -> None: await self._handle_workflow_error(event_data) self.register_event_handler(WorkflowEvent.ERROR_OCCURRED, handle_error) # Timeout handler async def handle_timeout(event_data: Dict[str, Any]) -> None: await self._handle_workflow_timeout(event_data) self.register_event_handler(WorkflowEvent.TIMEOUT_OCCURRED, handle_timeout) async def start(self) -> None: """Start the conversation workflow.""" if self._running: self.logger.warning("Conversation workflow is already running") return try: self.logger.info("Starting conversation workflow...") # Initialize ML manager if not already done if not self.ml_manager.is_running(): await self.ml_manager.start() # Start performance monitoring if self.performance_monitor and self.config.enable_performance_monitoring: await self.performance_monitor.start_monitoring() # Set initial state self._running = True await self._transition_to_state(ConversationState.LISTENING) # Start continuous processing asyncio.create_task(self._workflow_loop()) self.logger.info("Conversation workflow started successfully") except Exception as e: self.logger.error(f"Error starting conversation workflow: {e}") await self.stop() raise ConversationWorkflowError(f"Failed to start workflow: {e}") async def stop(self) -> None: """Stop the conversation workflow.""" if not self._running: return try: self.logger.info("Stopping conversation workflow...") # Complete current session if active if self._current_session: await self._complete_session(success=False, reason="workflow_stopped") # Stop performance monitoring if self.performance_monitor and self.config.enable_performance_monitoring: await self.performance_monitor.stop_monitoring() # Set final state self._running = False await self._transition_to_state(ConversationState.IDLE) self.logger.info("Conversation workflow stopped successfully") except Exception as e: self.logger.error(f"Error stopping conversation workflow: {e}") async def process_audio(self, audio_data: np.ndarray) -> None: """ Process audio data through the conversation workflow. Args: audio_data: Audio data to process """ if not self._running: return # Record performance metrics if self.performance_monitor: timer_key = self.performance_monitor.start_timer("workflow", "process_audio") try: # Process based on current state if self._current_state == ConversationState.LISTENING: await self._process_listening_audio(audio_data) elif self._current_state == ConversationState.RECORDING: await self._process_recording_audio(audio_data) # Check for timeouts await self._check_timeouts() except Exception as e: self.logger.error(f"Error processing audio: {e}") await self._emit_event(WorkflowEvent.ERROR_OCCURRED, {"error": e}) finally: if self.performance_monitor: self.performance_monitor.end_timer(timer_key) async def _process_listening_audio(self, audio_data: np.ndarray) -> None: """Process audio in listening state (wakeword detection).""" # Feed audio to ML manager for wakeword detection await self.ml_manager.process_audio_chunk(audio_data) # Note: Wakeword detection results would be handled via ML manager callbacks # This is a simplified version - actual implementation would integrate with # wakeword detection callbacks from the ML manager async def _process_recording_audio(self, audio_data: np.ndarray) -> None: """Process audio in recording state.""" if not self._current_session: return # Add audio to session buffer self._current_session.audio_buffer.append(audio_data.copy()) # Detect voice activity has_voice = self._detect_voice_activity(audio_data) current_time = time.time() if has_voice: self._last_voice_activity = current_time self._silence_start_time = 0.0 await self._emit_event(WorkflowEvent.VOICE_ACTIVITY_DETECTED, { "session_id": self._current_session.session_id, "timestamp": current_time }) else: if self._last_voice_activity > 0 and self._silence_start_time == 0.0: self._silence_start_time = current_time await self._emit_event(WorkflowEvent.SILENCE_DETECTED, { "session_id": self._current_session.session_id, "timestamp": current_time }) # Check recording completion conditions recording_duration = current_time - self._recording_start_time silence_duration = (current_time - self._silence_start_time) if self._silence_start_time > 0 else 0 if (recording_duration >= self.config.max_recording_duration or (silence_duration >= self.config.silence_timeout and recording_duration >= self.config.min_recording_duration)): await self._emit_event(WorkflowEvent.RECORDING_COMPLETE, { "session_id": self._current_session.session_id, "duration": recording_duration, "silence_duration": silence_duration }) def _detect_voice_activity(self, audio_data: np.ndarray) -> bool: """Simple voice activity detection based on energy.""" if len(audio_data) == 0: return False # Calculate RMS energy rms = np.sqrt(np.mean(audio_data.astype(np.float64) ** 2)) return rms > self.config.voice_activity_threshold async def _workflow_loop(self) -> None: """Main workflow processing loop.""" while self._running: try: # Check for timeouts await self._check_timeouts() # Update performance metrics if self.performance_monitor and self.config.log_conversation_metrics: await self._update_performance_metrics() # Small delay to prevent excessive CPU usage await asyncio.sleep(0.1) except Exception as e: self.logger.error(f"Error in workflow loop: {e}") await asyncio.sleep(1.0) async def _check_timeouts(self) -> None: """Check for workflow timeouts.""" current_time = time.time() state_duration = current_time - self._state_start_time # Check state-specific timeouts if (self._current_state == ConversationState.PROCESSING and state_duration > self.config.processing_timeout): await self._emit_event(WorkflowEvent.TIMEOUT_OCCURRED, { "state": self._current_state.value, "duration": state_duration, "timeout_type": "processing" }) elif (self._current_state == ConversationState.WAITING_RESPONSE and state_duration > self.config.response_timeout): await self._emit_event(WorkflowEvent.TIMEOUT_OCCURRED, { "state": self._current_state.value, "duration": state_duration, "timeout_type": "response" }) async def _start_new_session(self, event_data: Dict[str, Any]) -> None: """Start a new conversation session.""" # Complete previous session if exists if self._current_session: await self._complete_session(success=False, reason="new_session_started") # Create new session session_id = str(uuid.uuid4()) self._current_session = ConversationSession( session_id=session_id, start_time=datetime.now(), speaker_id=event_data.get("speaker_id"), speaker_name=event_data.get("speaker_name"), wakeword_confidence=event_data.get("confidence", 0.0), metadata=event_data.copy() ) # Start recording await self._transition_to_state(ConversationState.RECORDING) self._recording_start_time = time.time() self._last_voice_activity = 0.0 self._silence_start_time = 0.0 # Update stats self._workflow_stats["total_conversations"] += 1 self._workflow_stats["wakeword_detections"] += 1 self.logger.info(f"Started new conversation session: {session_id}") async def _process_recorded_audio(self) -> None: """Process recorded audio from current session.""" if not self._current_session: return try: self._current_session.processing_start_time = datetime.now() # Combine audio chunks if not self._current_session.audio_buffer: await self._complete_session(success=False, reason="no_audio_recorded") return combined_audio = np.concatenate(self._current_session.audio_buffer) # Perform voice recognition and speaker identification # This would integrate with the voice recognition system recognition_result = await self._perform_voice_recognition(combined_audio) if recognition_result.get("success", False): # Send audio to server for processing await self._send_audio_to_server(combined_audio, recognition_result) else: await self._complete_session(success=False, reason="voice_recognition_failed") except Exception as e: self.logger.error(f"Error processing recorded audio: {e}") await self._complete_session(success=False, reason=f"processing_error: {e}") async def _perform_voice_recognition(self, audio_data: np.ndarray) -> Dict[str, Any]: """Perform voice recognition on recorded audio.""" # This would integrate with the voice recognition detector # For now, return a mock result return { "success": True, "speaker_id": self._current_session.speaker_id or "unknown", "confidence": 0.8, "quality_score": 0.9 } async def _send_audio_to_server(self, audio_data: np.ndarray, recognition_result: Dict[str, Any]) -> None: """Send processed audio to server.""" try: await self._transition_to_state(ConversationState.WAITING_RESPONSE) self._current_session.response_start_time = datetime.now() # This would integrate with network client to send audio to server # For now, simulate server communication await asyncio.sleep(1.0) # Simulate network delay # Mock response response_data = { "session_id": self._current_session.session_id, "response_text": "Mock response from server", "response_audio": None, # Would contain audio data "confidence": 0.9 } await self._emit_event(WorkflowEvent.RESPONSE_RECEIVED, response_data) except Exception as e: self.logger.error(f"Error sending audio to server: {e}") await self._complete_session(success=False, reason=f"server_error: {e}") async def _play_response(self, response_data: Dict[str, Any]) -> None: """Play response audio.""" try: # This would integrate with audio playback system # For now, simulate response playback await asyncio.sleep(2.0) # Simulate response duration await self._emit_event(WorkflowEvent.RESPONSE_COMPLETE, response_data) await self._complete_session(success=True, reason="response_completed") except Exception as e: self.logger.error(f"Error playing response: {e}") await self._complete_session(success=False, reason=f"playback_error: {e}") async def _complete_session(self, success: bool, reason: str) -> None: """Complete the current conversation session.""" if not self._current_session: return try: # Update session metadata self._current_session.metadata.update({ "success": success, "completion_reason": reason, "end_time": datetime.now(), "total_duration": self._current_session.duration, "audio_duration": self._current_session.audio_duration }) # Update statistics if success: self._workflow_stats["successful_conversations"] += 1 else: self._workflow_stats["failed_conversations"] += 1 # Record performance metrics if self.performance_monitor: self.performance_monitor.record_metric( "workflow", MetricType.PROCESSING_TIME, self._current_session.duration * 1000, "milliseconds" ) # Store in history self._session_history.append(self._current_session) if len(self._session_history) > self._max_history_size: self._session_history.pop(0) self.logger.info(f"Completed conversation session: {self._current_session.session_id} " f"(success={success}, reason={reason})") # Return to listening state self._current_session = None if self.config.continuous_listening: await self._transition_to_state(ConversationState.LISTENING) else: await self._transition_to_state(ConversationState.IDLE) except Exception as e: self.logger.error(f"Error completing session: {e}") async def _handle_workflow_error(self, event_data: Dict[str, Any]) -> None: """Handle workflow errors.""" error = event_data.get("error") self.logger.error(f"Workflow error: {error}") # Complete current session if active if self._current_session: await self._complete_session(success=False, reason=f"error: {error}") # Return to appropriate state if self.config.continuous_listening: await self._transition_to_state(ConversationState.LISTENING) else: await self._transition_to_state(ConversationState.IDLE) async def _handle_workflow_timeout(self, event_data: Dict[str, Any]) -> None: """Handle workflow timeouts.""" timeout_type = event_data.get("timeout_type", "unknown") self.logger.warning(f"Workflow timeout: {timeout_type}") # Complete current session if active if self._current_session: await self._complete_session(success=False, reason=f"timeout: {timeout_type}") # Return to listening state if self.config.continuous_listening: await self._transition_to_state(ConversationState.LISTENING) else: await self._transition_to_state(ConversationState.IDLE) async def _transition_to_state(self, new_state: ConversationState) -> None: """Transition to a new state.""" if new_state == self._current_state: return # Validate transition if new_state not in self._valid_transitions.get(self._current_state, []): self.logger.warning(f"Invalid state transition from {self._current_state.value} to {new_state.value}") return old_state = self._current_state self._previous_state = old_state self._current_state = new_state self._state_start_time = time.time() self.logger.debug(f"State transition: {old_state.value} -> {new_state.value}") # Call state transition handlers transition_key = (old_state, new_state) if transition_key in self._state_transition_handlers: for handler in self._state_transition_handlers[transition_key]: try: await handler(old_state, new_state) except Exception as e: self.logger.error(f"Error in state transition handler: {e}") # Record state change metrics if self.performance_monitor: self.performance_monitor.record_metric( "workflow", MetricType.PROCESSING_TIME, time.time() - self._state_start_time, "seconds", {"state_transition": f"{old_state.value}_to_{new_state.value}"} ) async def _emit_event(self, event: WorkflowEvent, data: Dict[str, Any]) -> None: """Emit a workflow event.""" try: # Add common event data event_data = { "event": event.value, "timestamp": time.time(), "current_state": self._current_state.value, "session_id": self._current_session.session_id if self._current_session else None, **data } # Call event handlers if event in self._event_handlers: for handler in self._event_handlers[event]: try: await handler(event_data) except Exception as e: self.logger.error(f"Error in event handler for {event.value}: {e}") self.logger.debug(f"Emitted event: {event.value}") except Exception as e: self.logger.error(f"Error emitting event {event.value}: {e}") async def _update_performance_metrics(self) -> None: """Update performance metrics.""" if not self.performance_monitor: return try: # Record current state duration state_duration = (time.time() - self._state_start_time) * 1000 self.performance_monitor.record_metric( "workflow", MetricType.PROCESSING_TIME, state_duration, "milliseconds", {"state": self._current_state.value} ) # Record session metrics if active if self._current_session: session_duration = self._current_session.duration * 1000 self.performance_monitor.record_metric( "workflow", MetricType.PROCESSING_TIME, session_duration, "milliseconds", {"metric_type": "session_duration"} ) except Exception as e: self.logger.error(f"Error updating performance metrics: {e}") def register_event_handler(self, event: WorkflowEvent, handler: Callable) -> None: """Register an event handler.""" if event not in self._event_handlers: self._event_handlers[event] = [] self._event_handlers[event].append(handler) def register_state_transition_handler(self, from_state: ConversationState, to_state: ConversationState, handler: Callable) -> None: """Register a state transition handler.""" transition_key = (from_state, to_state) if transition_key not in self._state_transition_handlers: self._state_transition_handlers[transition_key] = [] self._state_transition_handlers[transition_key].append(handler) def get_current_state(self) -> ConversationState: """Get current workflow state.""" return self._current_state def get_current_session(self) -> Optional[ConversationSession]: """Get current conversation session.""" return self._current_session def get_session_history(self, limit: Optional[int] = None) -> List[ConversationSession]: """Get conversation session history.""" if limit: return self._session_history[-limit:] return self._session_history.copy() def get_workflow_statistics(self) -> Dict[str, Any]: """Get workflow statistics.""" stats = self._workflow_stats.copy() stats.update({ "current_state": self._current_state.value, "state_duration": time.time() - self._state_start_time, "active_session": self._current_session.session_id if self._current_session else None, "session_history_count": len(self._session_history), "is_running": self._running }) # Calculate success rate total_conversations = stats["total_conversations"] if total_conversations > 0: stats["success_rate"] = stats["successful_conversations"] / total_conversations else: stats["success_rate"] = 0.0 return stats async def simulate_wakeword_detection(self, speaker_id: str = "test_speaker", confidence: float = 0.8) -> None: """Simulate wakeword detection for testing.""" if self._current_state != ConversationState.LISTENING: self.logger.warning("Cannot simulate wakeword - not in listening state") return await self._emit_event(WorkflowEvent.WAKEWORD_DETECTED, { "speaker_id": speaker_id, "speaker_name": f"Speaker {speaker_id}", "confidence": confidence, "wakeword_type": "hey_trixy" }) async def force_state_transition(self, new_state: ConversationState) -> None: """Force a state transition (for testing/debugging).""" self.logger.warning(f"Forcing state transition to: {new_state.value}") await self._transition_to_state(new_state) # Factory functions def create_default_workflow_config() -> WorkflowConfig: """Create default workflow configuration.""" return WorkflowConfig() def create_optimized_workflow_config() -> WorkflowConfig: """Create optimized workflow configuration for production.""" config = WorkflowConfig() config.wakeword_sensitivity = 0.8 config.max_recording_duration = 20.0 config.silence_timeout = 1.5 config.processing_timeout = 10.0 config.response_timeout = 8.0 config.enable_performance_monitoring = True config.log_conversation_metrics = True return config async def create_conversation_workflow( config: Optional[WorkflowConfig] = None, ml_manager: Optional[Any] = None, app_container: Optional[Any] = None, performance_monitor: Optional[PerformanceMonitor] = None, auto_start: bool = True ) -> ConversationWorkflow: """ Create and initialize conversation workflow. Args: config: Workflow configuration ml_manager: ML manager instance app_container: Application container performance_monitor: Performance monitor auto_start: Whether to auto-start the workflow Returns: Initialized conversation workflow """ if not config: config = create_default_workflow_config() workflow = ConversationWorkflow( config=config, ml_manager=ml_manager, app_container=app_container, performance_monitor=performance_monitor ) if auto_start: await workflow.start() return workflow