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