petter2025 commited on
Commit
47e1a25
·
verified ·
1 Parent(s): 5dbe9c3

Delete voice_narrator.py

Browse files
Files changed (1) hide show
  1. voice_narrator.py +0 -179
voice_narrator.py DELETED
@@ -1,179 +0,0 @@
1
- """
2
- Voice Narration for Incident Alerts
3
- Supports multiple TTS providers with graceful fallback
4
- """
5
-
6
- import os
7
- import requests
8
- import logging
9
- from typing import Optional, Dict, Any
10
-
11
- logger = logging.getLogger(__name__)
12
-
13
-
14
- class VoiceNarrator:
15
- """
16
- Narrate critical incidents using TTS API
17
-
18
- Supports:
19
- - Hathora Voice API (if available)
20
- - Generic TTS APIs
21
- - Graceful fallback (silent fail)
22
- """
23
-
24
- def __init__(self):
25
- # Check for API key (set in HF Secrets)
26
- self.api_key = os.environ.get("HATHORA_VOICE_API_KEY", "") or \
27
- os.environ.get("VOICE_API_KEY", "")
28
-
29
- # API endpoint (update when you find the correct Hathora voice endpoint)
30
- self.api_endpoint = os.environ.get(
31
- "VOICE_API_ENDPOINT",
32
- "https://api.hathora.dev/v1/tts" # PLACEHOLDER - update this!
33
- )
34
-
35
- self.enabled = bool(self.api_key)
36
-
37
- if self.enabled:
38
- logger.info("✅ Voice narrator initialized with API key")
39
- else:
40
- logger.warning("⚠️ Voice narrator disabled (no API key found)")
41
-
42
- def narrate_incident(
43
- self,
44
- component: str,
45
- severity: str,
46
- latency: float,
47
- error_rate: float,
48
- root_cause: str = "Unknown",
49
- recovery_action: str = "Investigating"
50
- ) -> Optional[str]:
51
- """
52
- Generate voice narration for a critical incident
53
-
54
- Returns:
55
- Audio URL (str) if successful, None if failed
56
- """
57
-
58
- if not self.enabled:
59
- logger.debug("Voice narration skipped (disabled)")
60
- return None
61
-
62
- # Only narrate HIGH and CRITICAL incidents
63
- if severity not in ["HIGH", "CRITICAL"]:
64
- return None
65
-
66
- try:
67
- # Build dramatic narration text (30-60 seconds when spoken)
68
- narration_text = self._build_narration(
69
- component, severity, latency, error_rate, root_cause, recovery_action
70
- )
71
-
72
- # Call TTS API
73
- audio_url = self._call_tts_api(narration_text)
74
-
75
- if audio_url:
76
- logger.info(f"✅ Generated voice narration for {component}")
77
- return audio_url
78
- else:
79
- logger.warning("Voice API returned no audio URL")
80
- return None
81
-
82
- except Exception as e:
83
- # Silent fail - don't break the app
84
- logger.error(f"Voice narration failed: {e}", exc_info=True)
85
- return None
86
-
87
- def _build_narration(
88
- self,
89
- component: str,
90
- severity: str,
91
- latency: float,
92
- error_rate: float,
93
- root_cause: str,
94
- recovery_action: str
95
- ) -> str:
96
- """Build dramatic narration text"""
97
-
98
- # Format component name (remove dashes, capitalize)
99
- component_spoken = component.replace("-", " ").title()
100
-
101
- # Severity-specific intro
102
- if severity == "CRITICAL":
103
- intro = f"Critical alert. {component_spoken} is experiencing severe failure."
104
- else:
105
- intro = f"High priority alert. {component_spoken} degradation detected."
106
-
107
- # Metrics
108
- metrics = f"Latency: {int(latency)} milliseconds. Error rate: {error_rate*100:.0f} percent."
109
-
110
- # Root cause (if available)
111
- if root_cause and root_cause != "Unknown":
112
- cause = f"Root cause: {root_cause}."
113
- else:
114
- cause = "Root cause under investigation."
115
-
116
- # Recovery action
117
- action = f"Recovery action: {recovery_action}."
118
-
119
- # Combine into ~30-60 second narration
120
- full_text = f"{intro} {metrics} {cause} {action} Immediate attention required."
121
-
122
- logger.debug(f"Narration text: {full_text[:100]}...")
123
- return full_text
124
-
125
- def _call_tts_api(self, text: str) -> Optional[str]:
126
- """
127
- Call TTS API to generate audio
128
-
129
- This is a GENERIC implementation - adapt to actual Hathora Voice API format
130
- """
131
-
132
- try:
133
- # PLACEHOLDER REQUEST FORMAT
134
- # Update this based on actual Hathora Voice API docs
135
- response = requests.post(
136
- self.api_endpoint,
137
- headers={
138
- "Authorization": f"Bearer {self.api_key}",
139
- "Content-Type": "application/json"
140
- },
141
- json={
142
- "text": text,
143
- "voice": "en-US-neural", # Adjust based on available voices
144
- "speed": 1.0,
145
- "format": "mp3"
146
- },
147
- timeout=10.0
148
- )
149
-
150
- if response.status_code == 200:
151
- data = response.json()
152
-
153
- # Extract audio URL (adjust key based on API response)
154
- audio_url = data.get("audio_url") or \
155
- data.get("url") or \
156
- data.get("audioUrl")
157
-
158
- return audio_url
159
- else:
160
- logger.error(f"TTS API error: {response.status_code} - {response.text[:200]}")
161
- return None
162
-
163
- except requests.exceptions.Timeout:
164
- logger.warning("TTS API timeout")
165
- return None
166
- except Exception as e:
167
- logger.error(f"TTS API call failed: {e}")
168
- return None
169
-
170
-
171
- # Singleton instance
172
- _narrator = None
173
-
174
- def get_narrator() -> VoiceNarrator:
175
- """Get global narrator instance"""
176
- global _narrator
177
- if _narrator is None:
178
- _narrator = VoiceNarrator()
179
- return _narrator