import os import sys import logging import gradio as gr import assemblyai as aai from translate import Translator import uuid from elevenlabs import VoiceSettings from elevenlabs.client import ElevenLabs from pathlib import Path import tempfile import shutil from pydantic import ConfigDict from typing import List, Tuple, Optional, Union gr.__dict__.update(ConfigDict(arbitrary_types_allowed=True)) sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) # Import configuration keys from config.appconfig import ASSEMBLYAI_API_KEY, ELEVENLABS_API_KEY, VOICE_ID # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(r'logs/voice_translator.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class VoiceTranslator: LANGUAGES = [ ("ar", "Arabic"), ("zh", "Chinese"), ("nl", "Dutch"), ("en", "English"), ("fr", "French"), ("de", "German"), ("hi", "Hindi"), ("it", "Italian"), ("ja", "Japanese"), ("ko", "Korean"), ("pt", "Portuguese"), ("ru", "Russian"), ("es", "Spanish"), ("sv", "Swedish"), ("tr", "Turkish"), ("uk", "Ukrainian") ] def __init__(self, assemblyai_key: str, elevenlabs_key: str, voice_id: str): """Initialize the voice translator with API keys.""" self.assemblyai_key = assemblyai_key self.elevenlabs_key = elevenlabs_key self.voice_id = voice_id self.temp_dir = Path(tempfile.mkdtemp()) aai.settings.api_key = self.assemblyai_key self.elevenlabs_client = ElevenLabs(api_key=self.elevenlabs_key) # Validate API keys self.validate_api_keys() logger.info("VoiceTranslator initialized successfully") def __del__(self): """Cleanup temporary files on deletion.""" try: shutil.rmtree(self.temp_dir) logger.info("Temporary directory cleaned up successfully") except Exception as e: logger.error(f"Error cleaning up temporary directory: {e}") def validate_api_keys(self): if not self.assemblyai_key: raise ValueError("Missing AssemblyAI API Key") if not self.elevenlabs_key: raise ValueError("Missing ElevenLabs API Key") if not self.voice_id: raise ValueError("Missing Voice ID") def handle_audio_input(self, audio_data: Union[str, tuple]) -> str: """Handle different types of audio input and return a valid file path.""" try: if isinstance(audio_data, tuple): sample_rate, audio_array = audio_data temp_path = self.temp_dir / f"input_{uuid.uuid4()}.wav" gr.Audio.write_audio(str(temp_path), audio_array, sample_rate) return str(temp_path) elif isinstance(audio_data, str): return audio_data else: raise ValueError(f"Unsupported audio input type: {type(audio_data)}") except Exception as e: logger.error(f"Error handling audio input: {e}") raise gr.Error(f"Failed to process audio input: {str(e)}") def transcribe_audio(self, audio_file: str) -> Optional[str]: """Transcribe audio file to text.""" try: logger.info(f"Starting transcription for file: {audio_file}") transcriber = aai.Transcriber() if not os.path.exists(audio_file): raise FileNotFoundError(f"Audio file not found: {audio_file}") transcript = transcriber.transcribe(audio_file) if transcript.status == aai.TranscriptStatus.error: logger.error(f"Transcription error: {transcript.error}") raise gr.Error(f"Transcription failed: {transcript.error}") logger.info("Transcription completed successfully") return transcript.text except Exception as e: logger.error(f"Transcription error: {str(e)}") raise gr.Error(f"Transcription failed: {str(e)}") def translate_text(self, text: str, selected_languages: List[Tuple[str, str]]) -> List[Tuple[str, str]]: """Translate text to selected languages.""" try: logger.info("Starting text translation") translations = [] for lang_code, lang_name in selected_languages: try: translator = Translator(from_lang="en", to_lang=lang_code) translation = translator.translate(text) translations.append((translation, lang_name)) logger.info(f"Translation to {lang_name} completed") except Exception as e: logger.error(f"Error translating to {lang_name}: {e}") translations.append((text, lang_name)) # Fallback to original text return translations except Exception as e: logger.error(f"Translation error: {str(e)}") raise gr.Error(f"Translation failed: {str(e)}") def text_to_speech(self, text: str) -> str: """Convert text to speech using ElevenLabs.""" try: logger.info("Starting text-to-speech conversion") output_path = self.temp_dir / f"{uuid.uuid4()}.mp3" response = self.elevenlabs_client.text_to_speech.convert( voice_id=self.voice_id, optimize_streaming_latency="0", output_format="mp3_22050_32", text=text[:500], # Rate limit workaround model_id="eleven_multilingual_v2", voice_settings=VoiceSettings( stability=0.5, similarity_boost=0.8, style=0.5, use_speaker_boost=True, ), ) # Write the audio file in chunks with open(output_path, "wb") as f: for chunk in response: if chunk: f.write(chunk) logger.info(f"Audio file saved successfully: {output_path}") return str(output_path) except Exception as e: logger.error(f"Text-to-speech error: {str(e)}") raise gr.Error(f"Text-to-speech conversion failed: {str(e)}") def process_voice(self, audio_input: Union[str, tuple], selected_languages: List[str]) -> List[Union[str, None]]: """Process voice input and return translated audio files and texts.""" try: # Handle audio input and get valid file path audio_file = self.handle_audio_input(audio_input) # Convert selected language names to (code, name) tuples selected_langs = [(code, name) for code, name in self.LANGUAGES if name in selected_languages] # Transcribe audio transcript = self.transcribe_audio(audio_file) if not transcript: raise gr.Error("Transcription failed") # Translate text translations = self.translate_text(transcript, selected_langs) # Generate audio for each translation outputs = [] for translation, _ in translations: try: audio_path = self.text_to_speech(translation) outputs.extend([audio_path, translation]) except Exception as e: logger.error(f"Error processing translation: {e}") outputs.extend([None, translation]) return outputs except Exception as e: logger.error(f"Voice processing error: {str(e)}") raise gr.Error(f"Voice processing failed: {str(e)}") def create_gradio_interface(translator: VoiceTranslator): """Create and configure the Gradio interface.""" with gr.Blocks(theme=gr.themes.Soft(primary_hue="blue", secondary_hue="indigo")) as demo: gr.Markdown( """ # 🌍 AI-Powered Multilingual Voice Translator ### Transform your voice into multiple languages instantly Record your message in English, select at least 4 target languages, and receive translations. """ ) with gr.Row(): with gr.Column(): audio_input = gr.Audio( sources=["microphone"], type="filepath", label="🎤 Record or Upload Audio (English)" ) lang_selection = gr.CheckboxGroup( choices=[lang_name for _, lang_name in VoiceTranslator.LANGUAGES], label="Select at least 4 Target Languages", interactive=True ) with gr.Row(): submit_btn = gr.Button("Translate", variant="primary") status = gr.Textbox(label="Status", interactive=False) # Create fixed output components output_boxes = [] with gr.Row() as output_row: for _, lang_name in VoiceTranslator.LANGUAGES: with gr.Column(visible=False) as lang_column: audio_out = gr.Audio( label=f"{lang_name} Audio", interactive=False, type="filepath" # Ensure audio files are handled as files ) text_out = gr.Textbox(label=f"{lang_name} Text", interactive=False) output_boxes.append({ "column": lang_column, "audio": audio_out, "text": text_out, "lang": lang_name }) def update_visible_outputs(selected_langs): """Update visibility of output components based on selected languages.""" if len(selected_langs) < 4: raise gr.Error("Please select at least 4 languages") visibility_updates = [] output_components = [] for box in output_boxes: is_visible = box["lang"] in selected_langs visibility_updates.append(is_visible) if is_visible: output_components.extend([box["audio"], box["text"]]) return visibility_updates + output_components def process_inputs(audio, selected_langs): """Process audio input and return translations.""" try: if not audio: raise gr.Error("Please provide an audio input") if len(selected_langs) < 4: raise gr.Error("Please select at least 4 languages") # Get translations from translator outputs = translator.process_voice(audio, selected_langs) # Calculate the total number of outputs needed total_languages = len(VoiceTranslator.LANGUAGES) # Total number of possible languages # Create visibility updates for all possible languages visibility_updates = [box["lang"] in selected_langs for box in output_boxes] # Fill remaining outputs with None for languages not selected needed_outputs = total_languages * 2 # Each language needs 2 outputs (audio and text) current_outputs = len(outputs) padding = [None] * (needed_outputs - current_outputs) return visibility_updates + ["Translation completed successfully!"] + outputs + padding except Exception as e: # Handle errors with correct number of outputs error_msg = f"Error during translation: {str(e)}" visibility_updates = [box["lang"] in selected_langs for box in output_boxes] padding = [None] * (total_languages * 2) # Full padding for all possible outputs return visibility_updates + [error_msg] + padding # Update visibility when languages are selected lang_selection.change( fn=update_visible_outputs, inputs=[lang_selection], outputs=[box["column"] for box in output_boxes] + [box["audio"] for box in output_boxes] + [box["text"] for box in output_boxes] ) # Handle translation submission submit_btn.click( fn=process_inputs, inputs=[audio_input, lang_selection], outputs=[box["column"] for box in output_boxes] + [status] + # Add status output [box["audio"] for box in output_boxes] + [box["text"] for box in output_boxes], show_progress=True ) return demo if __name__ == "__main__": translator = VoiceTranslator( assemblyai_key=ASSEMBLYAI_API_KEY, elevenlabs_key=ELEVENLABS_API_KEY, voice_id=VOICE_ID ) interface = create_gradio_interface(translator) interface.launch(share=True)