""" Simplified Groq ASR Service using HTTP requests Works around Python client compatibility issues while providing superior transcription """ import asyncio import logging import tempfile import os import aiohttp import base64 from typing import Optional, Dict, Any from pathlib import Path from config import ( ENABLE_VOICE_FEATURES, TTS_PROVIDER, VOICE_LANGUAGE, DEFAULT_VOICE_SPEED, GROQ_API_KEY ) logger = logging.getLogger("voicebot") class SimpleGroqASRService: def __init__(self): self.voice_enabled = ENABLE_VOICE_FEATURES self.tts_provider = TTS_PROVIDER self.asr_provider = "groq" self.language = VOICE_LANGUAGE self.voice_speed = DEFAULT_VOICE_SPEED self.groq_api_key = GROQ_API_KEY # Groq API endpoint self.groq_audio_url = "https://api.groq.com/openai/v1/audio/transcriptions" if self.groq_api_key: logger.info("✅ Simple Groq ASR service initialized") self.asr_available = True else: logger.error("❌ GROQ_API_KEY not found") self.asr_available = False # Initialize TTS service if self.voice_enabled: self._init_tts_service() logger.info(f"🎤 Simple Groq ASR Service ready - ASR: Groq HTTP, TTS: {self.tts_provider}") def _init_tts_service(self): """Initialize Text-to-Speech service""" try: if self.tts_provider == "edge-tts": import edge_tts self.tts_available = True logger.info("✅ Edge TTS initialized") elif self.tts_provider == "murf": self.tts_available = True logger.info("✅ Murf AI TTS initialized") else: self.tts_available = False logger.warning(f"⚠️ Unknown TTS provider: {self.tts_provider}") except ImportError as e: self.tts_available = False logger.warning(f"⚠️ TTS dependencies not available: {e}") def _get_default_voice(self) -> str: """Get default voice based on language setting""" language_voices = { 'hi-IN': 'hi-IN-SwaraNeural', 'en-IN': 'en-IN-NeerjaNeural', 'en-US': 'en-US-AriaNeural', } return language_voices.get(self.language, 'en-US-AriaNeural') async def groq_asr_bytes(self, audio_bytes: bytes, user_language: str = None) -> Optional[str]: """ Transcribe audio using Groq API with HTTP requests Superior accuracy compared to local Whisper """ if not self.asr_available: logger.error("❌ Groq ASR not available - missing API key") return None try: # Create temporary file for API upload with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file: temp_file.write(audio_bytes) temp_file_path = temp_file.name try: # Prepare form data for Groq API headers = { "Authorization": f"Bearer {self.groq_api_key}" } language_code = self._get_groq_language_code(user_language) async with aiohttp.ClientSession() as session: with open(temp_file_path, 'rb') as audio_file: form_data = aiohttp.FormData() form_data.add_field('file', audio_file, filename='audio.wav', content_type='audio/wav') form_data.add_field('model', 'whisper-large-v3') form_data.add_field('language', language_code) form_data.add_field('temperature', '0.0') form_data.add_field('response_format', 'json') logger.info(f"🎤 Sending audio to Groq ASR (language: {language_code})") async with session.post(self.groq_audio_url, headers=headers, data=form_data) as response: if response.status == 200: result = await response.json() transcribed_text = result.get('text', '').strip() logger.info(f"✅ Groq ASR result: '{transcribed_text}'") return transcribed_text else: error_text = await response.text() logger.error(f"❌ Groq API error {response.status}: {error_text}") return None finally: # Clean up temp file try: os.unlink(temp_file_path) except Exception as e: logger.warning(f"⚠️ Failed to cleanup temp file: {e}") except Exception as e: logger.error(f"❌ Groq ASR error: {e}") return None def _get_groq_language_code(self, user_language: str = None) -> str: """Convert user language to Groq language code""" if not user_language: return self.language.split('-')[0] if self.language else 'en' user_lang_lower = user_language.lower() language_mapping = { 'english': 'en', 'hindi': 'hi', 'hinglish': 'hi', 'en': 'en', 'hi': 'hi', 'en-in': 'en', 'hi-in': 'hi', 'en-us': 'en' } if '-' in user_lang_lower: base_lang = user_lang_lower.split('-')[0] return language_mapping.get(base_lang, 'en') return language_mapping.get(user_lang_lower, 'en') async def text_to_speech(self, text: str, voice: str = None) -> Optional[bytes]: """Convert text to speech audio""" if not self.voice_enabled or not self.tts_available: return None if voice is None: voice = self._get_default_voice() try: if self.tts_provider == "edge-tts": import edge_tts communicate = edge_tts.Communicate(text, voice, rate=f"{int((self.voice_speed - 1) * 100):+d}%") audio_data = b"" async for chunk in communicate.stream(): if chunk["type"] == "audio": audio_data += chunk["data"] return audio_data elif self.tts_provider == "murf": return await self._murf_tts(text, voice) except Exception as e: logger.error(f"❌ TTS Error: {e}") return None async def _murf_tts(self, text: str, voice: str = None) -> Optional[bytes]: """Murf TTS implementation""" murf_api_key = os.environ.get("MURF_API_KEY") if not murf_api_key: return None murf_url = "https://api.murf.ai/v1/speech/generate" payload = { "text": text, "voice": voice or "en-US-1", "format": "mp3" } headers = { "Authorization": f"Bearer {murf_api_key}", "Content-Type": "application/json" } try: async with aiohttp.ClientSession() as session: async with session.post(murf_url, json=payload, headers=headers) as resp: if resp.status == 200: result = await resp.json() audio_url = result.get("audio_url") if audio_url: async with session.get(audio_url) as audio_resp: if audio_resp.status == 200: return await audio_resp.read() except Exception as e: logger.error(f"❌ Murf TTS error: {e}") return None async def speech_to_text(self, audio_file_path: str, user_language: str = None) -> Optional[str]: """Convert speech file to text using Groq ASR""" if not self.voice_enabled or not self.asr_available: return None try: with open(audio_file_path, 'rb') as audio_file: audio_bytes = audio_file.read() return await self.groq_asr_bytes(audio_bytes, user_language) except Exception as e: logger.error(f"❌ Speech to text error: {e}") return None def get_voice_status(self) -> Dict[str, Any]: """Get current voice service status""" return { "voice_enabled": self.voice_enabled, "tts_available": getattr(self, 'tts_available', False), "asr_available": self.asr_available, "tts_provider": self.tts_provider, "asr_provider": "groq-http", "language": self.language, "voice_speed": self.voice_speed, "groq_available": bool(self.groq_api_key) } def is_voice_enabled(self) -> bool: """Check if voice features are enabled""" return self.voice_enabled # Global instance simple_groq_asr_service = SimpleGroqASRService()