Create voice_narrator.py

voice_narrator.py

@@ -0,0 +1,179 @@
+"""
+Voice Narration for Incident Alerts
+Supports multiple TTS providers with graceful fallback
+"""
+
+import os
+import requests
+import logging
+from typing import Optional, Dict, Any
+
+logger = logging.getLogger(__name__)
+
+
+class VoiceNarrator:
+    """
+    Narrate critical incidents using TTS API
+    
+    Supports:
+    - Hathora Voice API (if available)
+    - Generic TTS APIs
+    - Graceful fallback (silent fail)
+    """
+    
+    def __init__(self):
+        # Check for API key (set in HF Secrets)
+        self.api_key = os.environ.get("HATHORA_VOICE_API_KEY", "") or \
+                       os.environ.get("VOICE_API_KEY", "")
+        
+        # API endpoint (update when you find the correct Hathora voice endpoint)
+        self.api_endpoint = os.environ.get(
+            "VOICE_API_ENDPOINT",
+            "https://api.hathora.dev/v1/tts"  # PLACEHOLDER - update this!
+        )
+        
+        self.enabled = bool(self.api_key)
+        
+        if self.enabled:
+            logger.info("✅ Voice narrator initialized with API key")
+        else:
+            logger.warning("⚠️ Voice narrator disabled (no API key found)")
+    
+    def narrate_incident(
+        self,
+        component: str,
+        severity: str,
+        latency: float,
+        error_rate: float,
+        root_cause: str = "Unknown",
+        recovery_action: str = "Investigating"
+    ) -> Optional[str]:
+        """
+        Generate voice narration for a critical incident
+        
+        Returns:
+            Audio URL (str) if successful, None if failed
+        """
+        
+        if not self.enabled:
+            logger.debug("Voice narration skipped (disabled)")
+            return None
+        
+        # Only narrate HIGH and CRITICAL incidents
+        if severity not in ["HIGH", "CRITICAL"]:
+            return None
+        
+        try:
+            # Build dramatic narration text (30-60 seconds when spoken)
+            narration_text = self._build_narration(
+                component, severity, latency, error_rate, root_cause, recovery_action
+            )
+            
+            # Call TTS API
+            audio_url = self._call_tts_api(narration_text)
+            
+            if audio_url:
+                logger.info(f"✅ Generated voice narration for {component}")
+                return audio_url
+            else:
+                logger.warning("Voice API returned no audio URL")
+                return None
+                
+        except Exception as e:
+            # Silent fail - don't break the app
+            logger.error(f"Voice narration failed: {e}", exc_info=True)
+            return None
+    
+    def _build_narration(
+        self,
+        component: str,
+        severity: str,
+        latency: float,
+        error_rate: float,
+        root_cause: str,
+        recovery_action: str
+    ) -> str:
+        """Build dramatic narration text"""
+        
+        # Format component name (remove dashes, capitalize)
+        component_spoken = component.replace("-", " ").title()
+        
+        # Severity-specific intro
+        if severity == "CRITICAL":
+            intro = f"Critical alert. {component_spoken} is experiencing severe failure."
+        else:
+            intro = f"High priority alert. {component_spoken} degradation detected."
+        
+        # Metrics
+        metrics = f"Latency: {int(latency)} milliseconds. Error rate: {error_rate*100:.0f} percent."
+        
+        # Root cause (if available)
+        if root_cause and root_cause != "Unknown":
+            cause = f"Root cause: {root_cause}."
+        else:
+            cause = "Root cause under investigation."
+        
+        # Recovery action
+        action = f"Recovery action: {recovery_action}."
+        
+        # Combine into ~30-60 second narration
+        full_text = f"{intro} {metrics} {cause} {action} Immediate attention required."
+        
+        logger.debug(f"Narration text: {full_text[:100]}...")
+        return full_text
+    
+    def _call_tts_api(self, text: str) -> Optional[str]:
+        """
+        Call TTS API to generate audio
+        
+        This is a GENERIC implementation - adapt to actual Hathora Voice API format
+        """
+        
+        try:
+            # PLACEHOLDER REQUEST FORMAT
+            # Update this based on actual Hathora Voice API docs
+            response = requests.post(
+                self.api_endpoint,
+                headers={
+                    "Authorization": f"Bearer {self.api_key}",
+                    "Content-Type": "application/json"
+                },
+                json={
+                    "text": text,
+                    "voice": "en-US-neural",  # Adjust based on available voices
+                    "speed": 1.0,
+                    "format": "mp3"
+                },
+                timeout=10.0
+            )
+            
+            if response.status_code == 200:
+                data = response.json()
+                
+                # Extract audio URL (adjust key based on API response)
+                audio_url = data.get("audio_url") or \
+                           data.get("url") or \
+                           data.get("audioUrl")
+                
+                return audio_url
+            else:
+                logger.error(f"TTS API error: {response.status_code} - {response.text[:200]}")
+                return None
+                
+        except requests.exceptions.Timeout:
+            logger.warning("TTS API timeout")
+            return None
+        except Exception as e:
+            logger.error(f"TTS API call failed: {e}")
+            return None
+
+
+# Singleton instance
+_narrator = None
+
+def get_narrator() -> VoiceNarrator:
+    """Get global narrator instance"""
+    global _narrator
+    if _narrator is None:
+        _narrator = VoiceNarrator()
+    return _narrator