import base64 import io import os import wave import struct import hashlib from pathlib import Path from uuid import uuid4 import httpx2 from openai import AsyncOpenAI from app.config import OPENAI_API_KEY, SARVAM_API_KEY, GEMINI_API_KEY, GEMINI_MODEL from app.services.medical_speech_normalizer import normalize_medical_speech BASE_DIR = Path(__file__).resolve().parent.parent.parent AUDIO_DIR = BASE_DIR / "data" / "audio" CACHE_DIR = AUDIO_DIR / "cache" CACHE_DIR.mkdir(parents=True, exist_ok=True) AUDIO_DIR.mkdir(parents=True, exist_ok=True) class VoiceService: def __init__(self): self.openai_key = OPENAI_API_KEY self.sarvam_key = SARVAM_API_KEY self.gemini_key = GEMINI_API_KEY self.gemini_model = GEMINI_MODEL def get_initial_greeting(self, patient_name: str | None, language: str) -> str: name = patient_name or "அவர்களே" lang_lower = language.lower() if lang_lower.startswith("ta"): return f"வணக்கம் {name}, CureSelect மருத்துவமனைக்கு உங்களை அன்போடு வரவேற்கிறோம். இன்று உங்களுக்கு என்ன மாதிரி உடம்பு முடியலை, என்ன அறிகுறிகள் இருக்குன்னு சொல்ல முடியுமா?" elif lang_lower.startswith("hi"): return f"नमस्ते {name} जी, CureSelect अस्पताल में आपका स्वागत है। आज आपको क्या तकलीफ या लक्षण महसूस हो रहे हैं, कृपया बताएंगे?" elif lang_lower.startswith("te"): return f"నమస్కారం {name} గారు, CureSelect హాస్పిటల్‌కు స్వాగతం. ఈరోజు మీకున్న ఆరోగ్య సమస్యలు లేదా లక్షణాల గురించి చెప్పగలరా?" elif lang_lower.startswith("kn"): return f"ನಮಸ್ಕಾರ {name} ಅವರೇ, CureSelect ಆಸ್ಪತ್ರೆಗೆ ಸುಸ್ವಾಗತ. ಇಂದು ನಿಮ್ಮ ಆರೋಗ್ಯದಲ್ಲಿ ಏನು ತೊಂದರೆ ಇದೆ ಎಂದು ತಿಳಿಸಬಹುದೇ?" elif lang_lower.startswith("ml"): return f"നമസ്കാരം {name}, CureSelect ആശുപത്രിയിലേക്ക് സ്വാഗതം. ഇന്ന് താങ്കൾക്ക് എന്തെല്ലാം ആരോഗ്യ ബുദ്ധിമുട്ടുകളാണ് ഉള്ളത് എന്ന് പറയാമോ?" elif lang_lower.startswith("bn"): return f"নমস্কার {name}, CureSelect হাসপাতালে আপনাকে স্বাগতম। আজ আপনার কি ধরনের শারীরিক অসুবিধা হচ্ছে বলবেন কি?" elif lang_lower.startswith("mr"): return f"नमस्कार {name} जी, CureSelect रुग्णालयात आपले स्वागत आहे। आज आपल्याला काय त्रास होत आहे, ते सांगू शकाल का?" elif lang_lower.startswith("gu"): return f"નમસ્તે {name} જી, CureSelect હોસ્પિટલમાં આપનું સ્વાગત છે। આજે આપને શું તકલીફ કે લક્ષણો છે, જરા જણાવશો?" elif lang_lower.startswith("pa"): return f"ਸਤਿ ਸ਼੍ਰੀ ਅਕਾਲ {name} ਜੀ, CureSelect ਹਸਪਤਾਲ ਵਿੱਚ ਤੁਹਾਡਾ ਸਵਾਗਤ ਹੈ। ਅੱਜ ਤੁਹਾਨੂੰ ਕੀ ਤਕਲੀਫ਼ ਹੈ, ਕਿਰਪਾ ਕਰਕੇ ਦੱਸੋਗੇ?" else: return f"Hello {name}, welcome to CureSelect Healthcare. Could you please share what health symptoms you are experiencing today?" async def speech_to_text( self, audio_bytes: bytes, filename: str = "audio.webm", language_code: str = "en-IN", specialty_code: str = "general", patient_context: dict | None = None, ) -> str: """ High-precision multi-model Indian speech transcription pipeline: 1. Sarvam AI Saaras v2 (Specialized Indic Speech Engine) 2. Google Gemini Multimodal Audio (Multilingual Code-Switching & Acoustic Context) 3. OpenAI Whisper-1 (Medical Specialty Context Boosted) 4. Clinical & Phonetic Post-Processing Normalizer """ lang_prefix = language_code.lower()[:2] raw_transcript = "" # 1. Primary: Sarvam AI Saaras v3 for Native Indian Languages (Ultra-fast Indic STT) if ( self.sarvam_key and self.sarvam_key.strip() and self.sarvam_key not in ["your_sarvam_api_key", "your_sarvam_key", ""] and lang_prefix in ["ta", "hi", "te", "kn", "bn", "mr", "gu", "ml", "pa", "od", "en"] ): try: async with httpx2.AsyncClient(timeout=5.0) as client: files = { "file": (filename, audio_bytes, "audio/webm") } data = { "model": "saaras:v3", "language_code": language_code, "with_diacritics": "true", } headers = { "api-subscription-key": self.sarvam_key } res = await client.post( "https://api.sarvam.ai/speech-to-text", files=files, data=data, headers=headers, ) if res.status_code == 200: res_json = res.json() transcript = res_json.get("transcript", "").strip() if transcript: raw_transcript = transcript except Exception as e: print(f"[Sarvam STT Error]: {e}") # Apply Medical Speech Normalizer to eliminate phonetic mishearings normalized = normalize_medical_speech(raw_transcript, language_code=language_code) return normalized async def text_to_speech( self, text: str, language_code: str = "en-IN", ) -> tuple[str, bytes]: lang_prefix = language_code.lower()[:2] audio_id = f"aud_{uuid4().hex[:12]}" # Clean punctuation and brackets before synthesis cleaned_text = ( text.replace("(", "") .replace(")", "") .replace("*", "") .replace("#", "") .replace("- ", "") .strip() ) # Instant Disk Cache Lookup (< 0.001s response) cache_key = hashlib.md5(f"{language_code}:{cleaned_text}".encode("utf-8")).hexdigest() cache_path = CACHE_DIR / f"{cache_key}.wav" if cache_path.exists(): cached_data = cache_path.read_bytes() # Also save to current session audio_id for endpoint streaming (AUDIO_DIR / f"{audio_id}.wav").write_bytes(cached_data) return audio_id, cached_data # 1. Use Sarvam AI TTS for Indian Languages (Native Indian Accent with 1.15x Human Pace) if ( self.sarvam_key and self.sarvam_key.strip() and self.sarvam_key not in ["your_sarvam_api_key", "your_sarvam_key", ""] and lang_prefix in ["ta", "hi", "te", "kn", "bn", "mr", "gu", "ml", "pa", "od", "en"] ): try: speaker = "kavya" if lang_prefix in ["ta", "hi", "te", "kn"] else "gokul" async with httpx2.AsyncClient(timeout=5.0) as client: headers = { "api-subscription-key": self.sarvam_key, "Content-Type": "application/json", } payload = { "inputs": [cleaned_text], "target_language_code": language_code, "speaker": speaker, "model": "bulbul:v3", "pace": 1.15, } res = await client.post( "https://api.sarvam.ai/text-to-speech", json=payload, headers=headers, ) if res.status_code == 200: res_json = res.json() audios = res_json.get("audios", []) if audios: audio_data = base64.b64decode(audios[0]) file_path = AUDIO_DIR / f"{audio_id}.wav" file_path.write_bytes(audio_data) cache_path.write_bytes(audio_data) return audio_id, audio_data else: print(f"[Sarvam TTS HTTP Error]: {res.status_code} - {res.text}") except Exception as e: print(f"[Sarvam TTS Error]: {e}") # 2. Use OpenAI TTS if self.openai_key and self.openai_key.strip() not in ["your_openai_api_key", "your_openai_key", ""]: client = AsyncOpenAI(api_key=self.openai_key, timeout=5.0) try: response = await client.audio.speech.create( model="tts-1", voice="shimmer", input=cleaned_text, speed=1.0, ) audio_data = response.content file_path = AUDIO_DIR / f"{audio_id}.mp3" file_path.write_bytes(audio_data) return audio_id, audio_data except Exception as e: print(f"[OpenAI TTS Error]: {e}") finally: await client.close() # 3. Fallback audio_data = self._generate_fallback_wav() file_path = AUDIO_DIR / f"{audio_id}.wav" file_path.write_bytes(audio_data) return audio_id, audio_data def _generate_fallback_wav(self) -> bytes: buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) data = [int(1000 * (i % 32 < 16)) for i in range(4000)] wf.writeframes(struct.pack(f"<{len(data)}h", *data)) return buf.getvalue() def get_audio_path(self, audio_id: str) -> Path | None: for ext in [".wav", ".mp3", ".webm", ".m4a"]: p = AUDIO_DIR / f"{audio_id}{ext}" if p.exists(): return p return None