import base64 import io import os import wave import struct 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" 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 ಆಸ್ಪತ್ರೆಗೆ ಸುಸ್ವಾಗತ. ಇಂದು ನಿಮ್ಮ ಆರೋಗ್ಯದ ತೊಂದರೆ ಬಗ್ಗೆ ತಿಳಿಸಬಹುದೇ?" 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 v2 for Native Indian Languages (Ultra-fast ~0.9s) 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=2.5) as client: files = { "file": (filename, audio_bytes, "audio/webm") } data = { "model": "saaras:v2", "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() ) # 1. Use Sarvam AI TTS for Indian Languages (Native Indian Accent) 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", } 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) 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