Create app.py

app.py

@@ -0,0 +1,190 @@
+# !pip install gradio==5.21.0 gTTS speechrecognition pydub deep_translator httpx==0.28.1 webrtcvad noisereduce --quiet
+
+import gradio as gr
+from gtts import gTTS
+import speech_recognition as sr
+from pydub import AudioSegment
+import os
+import tempfile
+import logging
+from deep_translator import GoogleTranslator
+import webrtcvad
+import noisereduce as nr
+import numpy as np
+from typing import Optional, Tuple
+import queue
+import threading
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+# Supported languages (expanded globally)
+SUPPORTED_LANGUAGES = {
+    "English": "en",
+    "Hindi": "hi",
+    "Tamil": "ta",
+    "Telugu": "te",
+    "Bengali": "bn",
+    "Marathi": "mr",
+    "Gujarati": "gu",
+    "Kannada": "kn",
+    "Malayalam": "ml",
+    "Punjabi": "pa",
+    "Spanish": "es",
+    "French": "fr",
+    "German": "de",
+    "Chinese (Simplified)": "zh-CN",
+    "Japanese": "ja",
+    "Arabic": "ar"
+}
+
+# Thread-safe queue for processing chunks
+task_queue = queue.Queue()
+
+# VAD setup
+vad = webrtcvad.Vad(1)  # Aggressiveness level 1 (0-3)
+
+# TTS Function with chunked support
+def text_to_speech(text: str, lang: str) -> Optional[str]:
+    try:
+        if not text or not text.strip():
+            raise ValueError("Text input cannot be empty.")
+        
+        target_lang_code = SUPPORTED_LANGUAGES.get(lang, "en")
+        logger.info(f"TTS: Text='{text}', Language='{lang}' ({target_lang_code})")
+
+        translated_text = text if target_lang_code == "en" else GoogleTranslator(source="en", target=target_lang_code).translate(text)
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file:
+            tts = gTTS(text=translated_text, lang=target_lang_code, slow=False)
+            tts.save(temp_file.name)
+            logger.info(f"TTS generated: {temp_file.name}")
+            return temp_file.name
+
+    except ValueError as e:
+        logger.error(f"Input validation failed: {str(e)}")
+        return None
+    except Exception as e:
+        logger.error(f"TTS failed: {str(e)}")
+        return None
+
+# STT Function with chunking, VAD, and noise reduction
+def speech_to_text(audio_input: str, chunk_duration_ms: int = 2000) -> Tuple[str, Optional[str]]:
+    recognizer = sr.Recognizer()
+    try:
+        if not audio_input:
+            raise ValueError("No audio provided.")
+        
+        logger.info(f"Processing STT: Audio file='{audio_input}'")
+        audio_segment = AudioSegment.from_file(audio_input)
+        
+        # Noise reduction
+        audio_np = np.array(audio_segment.get_array_of_samples(), dtype=np.float32)
+        reduced_noise = nr.reduce_noise(y=audio_np, sr=audio_segment.frame_rate)
+        audio_segment = AudioSegment(
+            reduced_noise.tobytes(),
+            frame_rate=audio_segment.frame_rate,
+            sample_width=audio_segment.sample_width,
+            channels=audio_segment.channels
+        )
+
+        # Chunk audio
+        chunk_length = chunk_duration_ms  # 2 seconds
+        chunks = [audio_segment[i:i + chunk_length] for i in range(0, len(audio_segment), chunk_length)]
+        full_text = ""
+        dubbed_audio = None
+
+        for i, chunk in enumerate(chunks):
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_wav:
+                chunk.export(temp_wav.name, format="wav")
+                with open(temp_wav.name, "rb") as f:
+                    audio_bytes = f.read()
+                
+                # VAD to detect speech
+                is_speech = vad.is_speech(audio_bytes[:30], sample_rate=chunk.frame_rate)  # Check first 30ms
+                if is_speech:
+                    with sr.AudioFile(temp_wav.name) as source:
+                        audio_data = recognizer.record(source)
+                        text = recognizer.recognize_google(audio_data, language="auto")
+                        full_text += text + " "
+                        logger.info(f"Chunk {i+1} transcribed: {text}")
+                os.remove(temp_wav.name)
+
+        return full_text.strip(), dubbed_audio
+
+    except ValueError as e:
+        logger.error(f"Input validation failed: {str(e)}")
+        return f"Error: {str(e)}", None
+    except sr.UnknownValueError:
+        logger.warning("Speech not recognized.")
+        return "Could not understand the audio.", None
+    except sr.RequestError as e:
+        logger.error(f"STT API error: {str(e)}")
+        return f"STT error: {str(e)}", None
+    except Exception as e:
+        logger.error(f"STT failed: {str(e)}")
+        return f"Error in STT: {str(e)}", None
+
+# Real-time dubbing handler
+def handle_dubbing(audio_input: str, target_lang: str) -> Tuple[str, Optional[str]]:
+    text, _ = speech_to_text(audio_input)
+    if not text.startswith("Error"):
+        dubbed_audio = text_to_speech(text, target_lang)
+        return text, dubbed_audio
+    return text, None
+
+# Professional Gradio UI
+with gr.Blocks(
+    title="World-Class Real-Time Dubbing Translator",
+    theme=gr.themes.Soft(),
+    css="""
+        .gradio-container { max-width: 900px; margin: auto; }
+        .title { font-size: 2em; text-align: center; }
+        .description { text-align: center; color: #666; }
+        .button { background-color: #4CAF50; color: white; }
+    """
+) as demo:
+    # Header
+    gr.Markdown("<h1 class='title'>Real-Time Multilingual Dubbing</h1>")
+    gr.Markdown("<p class='description'>Record your voice and hear it dubbed in another language instantly.</p>")
+
+    # Dubbing Section
+    with gr.Row():
+        audio_input = gr.Audio(
+            sources=["microphone", "upload"],
+            type="filepath",
+            label="Record or Upload Audio",
+            interactive=True
+        )
+        lang_dropdown = gr.Dropdown(
+            choices=list(SUPPORTED_LANGUAGES.keys()),
+            label="Target Language for Dubbing",
+            value="English",
+            interactive=True,
+            allow_custom_value=False,
+            filterable=True  # Adds search functionality
+        )
+    
+    dub_button = gr.Button("Dub Audio", variant="primary")
+    with gr.Row():
+        stt_output = gr.Textbox(label="Transcription", placeholder="Your audio transcription will appear here", lines=3)
+        dub_output = gr.Audio(label="Dubbed Audio", type="filepath", interactive=False)
+
+    # Event handler
+    dub_button.click(
+        fn=handle_dubbing,
+        inputs=[audio_input, lang_dropdown],
+        outputs=[stt_output, dub_output]
+    )
+
+    # Footer
+    gr.Markdown("<footer style='text-align: center; padding: 20px;'>© 2025 xAI - Powered by Advanced AI Technologies</footer>")
+
+# Launch with flexible port settings
+demo.launch(
+    share=True,
+    server_name="0.0.0.0",
+    server_port=None,
+    debug=False,
+    show_error=True
+)