voice_narrator.py

"""
Voice Handler for Hathora Text-to-Speech Integration
Converts metrics into natural, conversational speech for audio synthesis
"""

import os
import logging
from typing import Dict, Any, Optional
import hathora

logger = logging.getLogger(__name__)


class VoiceHandler:
    """
    Handles conversion of metrics to natural speech and Hathora TTS integration
    """
    
    def __init__(self):
        """Initialize the VoiceHandler with Hathora API configuration"""
        self.api_key = os.getenv(
            "HATHORA_API_KEY",
            "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6IkVFcDg4bkpRU25idmVtMFIyTWZTRyJ9.eyJlbWFpbCI6InNhbWlzb2NpYWw3MkBnbWFpbC5jb20iLCJpc3MiOiJodHRwczovL2F1dGguaGF0aG9yYS5jb20vIiwic3ViIjoiZ29vZ2xlLW9hdXRoMnwxMTc1OTY2NDkzODcyNjUzNDQxOTgiLCJhdWQiOlsiaHR0cHM6Ly9jbG91ZC5oYXRob3JhLmNvbSIsImh0dHBzOi8vZGV2LXRjaHA2aW45LnVzLmF1dGgwLmNvbS91c2VyaW5mbyJdLCJpYXQiOjE3MzM4OTIxODYsImV4cCI6MTczMzk3ODU4Niwic2NvcGUiOiJvcGVuaWQgcHJvZmlsZSBlbWFpbCBvZmZsaW5lX2FjY2VzcyIsImF6cCI6IkROdnJ0RnY2NHJVdEExM1dIQjdvZ1BUMEI0SmtZb1AxIn0.VQwUFdDPOLLI2jOE0qbeRrZzp6oA5svMCaL2bmaMYiXZgOZHT5f5kpJSE8D5KBUwCJv9OyrGmyHCinxTtAthBLeliUL3OPUfORnbA6NfJnZ5Y-kM1IWIVI8a-in3HcWvxV5XUK2Rfq2QvMpQO_w9BnLh2JaMdwb-tsG6IeokH5owqo5sivQKJ8AiLb1ypuGFkcMENEOj0n0upnrd6A6-NDuUGms_x3Gx6d6k9Ih9iH9NAUKcm97AVwxXxBLM15nkZF1BDKztMM_A3YlI_aLfn3Oj7QLPRMsnHxz-tyWaVWLV0-wcYAzRVsrzCj-t6LEyRGBZ3fO19-KIEFvol12Wsw"
        )
        
        if not self.api_key:
            logger.warning("HATHORA_API_KEY not set. Voice synthesis will not work.")
            self.client = None
        else:
            try:
                self.client = hathora.Hathora(api_key=self.api_key)
                logger.info("Hathora client initialized successfully with kokoro model")
            except Exception as e:
                logger.error(f"Failed to initialize Hathora client: {e}")
                self.client = None
    
    def speak_metrics(self, metrics: Dict[str, Any]) -> Optional[str]:
        """
        Convert metrics dictionary into natural conversational speech and synthesize with Hathora
        
        Args:
            metrics: Dictionary containing:
                - confidence: str (e.g., "24.9%")
                - forecast: str (e.g., "1 critical risks")
                - insights: str (e.g., "High error rate")
                - actions: str (e.g., "traffic_shift")
        
        Returns:
            Path to the generated MP3 audio file, or None if synthesis failed
        
        Example input:
            {
                "confidence": "24.9%",
                "forecast": "1 critical risks",
                "insights": "High error rate",
                "actions": "traffic_shift"
            }
        
        Example output sentence:
            "Anomaly detected with a confidence of 24.9% and a high error rate. 
             We are currently forecasting 1 critical risk, so we have initiated a traffic shift."
        """
        try:
            # Extract metrics with defaults
            confidence = metrics.get("confidence", "unknown confidence")
            forecast = metrics.get("forecast", "no forecast available")
            insights = metrics.get("insights", "no specific insights")
            actions = metrics.get("actions", "no action")
            
            # Convert action code to natural language
            action_text = self._format_action(actions)
            
            # Construct natural conversational sentence
            speech_text = (
                f"Anomaly detected with a confidence of {confidence} and {insights.lower()}. "
                f"We are currently forecasting {forecast}, so we have {action_text}."
            )
            
            logger.info(f"Generated speech text: {speech_text}")
            
            # Send to Hathora for synthesis
            audio_path = self._synthesize_with_hathora(speech_text)
            
            return audio_path
            
        except Exception as e:
            logger.error(f"Error in speak_metrics: {e}", exc_info=True)
            return None
    
    def _format_action(self, action: str) -> str:
        """
        Convert action code into natural language
        
        Args:
            action: Action code (e.g., "traffic_shift", "scale_up", "restart")
        
        Returns:
            Natural language description of the action
        """
        action_map = {
            "traffic_shift": "initiated a traffic shift",
            "scale_up": "scaled up the infrastructure",
            "scale_down": "scaled down the infrastructure",
            "restart": "restarted the affected services",
            "rollback": "performed a rollback",
            "alert": "sent alerts to the team",
            "no_action": "determined no immediate action is required",
            "investigate": "flagged this for investigation"
        }
        
        return action_map.get(action, f"taken the action: {action}")
    
    def _synthesize_with_hathora(self, text: str, output_filename: str = "alert_audio.mp3") -> Optional[str]:
        """
        Send text to Hathora TTS API for speech synthesis using the kokoro model
        
        Args:
            text: The text to convert to speech
            output_filename: Name of the output MP3 file
        
        Returns:
            Path to the generated MP3 file, or None if synthesis failed
        """
        if not self.client:
            logger.error("Cannot synthesize: Hathora client not initialized")
            return None
        
        try:
            logger.info(f"Synthesizing speech with Hathora kokoro model...")
            
            # Use the kokoro model with af_sarah voice (same as your working test)
            response = self.client.text_to_speech.convert(
                "kokoro",
                text,
                voice="af_sarah",  # Professional female voice
                speed=1.0
            )
            
            # Save the audio file
            response.save(output_filename)
            logger.info(f"Successfully generated audio: {output_filename}")
            
            return output_filename
                
        except Exception as e:
            logger.error(f"Error during Hathora synthesis: {e}", exc_info=True)
            return None
    
    def speak_alert(self, diagnosis: str, action: str, output_filename: str = "alert.mp3") -> Optional[str]:
        """
        Generate alert voice for incidents (simplified version for direct use)
        
        Args:
            diagnosis: The diagnosis from Claude (e.g., "Payment gateway timeout")
            action: The action being taken (e.g., "REROUTE", "RESTART")
            output_filename: Name of output file
        
        Returns:
            Path to the generated MP3 file, or None if synthesis failed
        """
        if not self.client:
            logger.warning("Voice synthesis disabled: Hathora client not available")
            return None
        
        try:
            # Create natural speech text
            action_lower = action.lower().replace("_", " ")
            speech_text = f"Critical alert detected. {diagnosis}. Initiating {action_lower}."
            
            logger.info(f"Generating alert audio: {speech_text}")
            
            # Synthesize
            return self._synthesize_with_hathora(speech_text, output_filename)
            
        except Exception as e:
            logger.error(f"Error generating alert audio: {e}", exc_info=True)
            return None