Create app.py

app.py

@@ -0,0 +1,38 @@
+# app.py
+
+import gradio as gr
+import requests
+
+from model_handler import call_model_with_timeout
+
+
+def synthesize_speech(text, ref_audio, ref_text):
+    if not ref_audio:
+        return "⛔ Please upload a reference WAV file", None
+
+    # ref_audio is already a file path
+    status, output_path = call_model_with_timeout(text, ref_audio, ref_text)
+    return status, output_path
+
+
+ref_audio_url = "https://github.com/AI4Bharat/IndicF5/raw/refs/heads/main/prompts/PAN_F_HAPPY_00002.wav"
+ref_audio_path = "PAN_F_HAPPY_00002.wav"
+with open(ref_audio_path, "wb") as f:
+    f.write(requests.get(ref_audio_url).content)
+iface = gr.Interface(
+    fn=synthesize_speech,
+    inputs=[
+        gr.Textbox(label="Input Text", value="यह एक परीक्षण वाक्य है।"),
+        gr.Audio(type="filepath", label="Reference Audio", value="PAN_F_HAPPY_00002.wav"),
+        gr.Textbox(label="Reference Text",
+                   value="ਇੱਕ ਗ੍ਰਾਹਕ ਨੇ ਸਾਡੀ ਬੇਮਿਸਾਲ ਸੇਵਾ ਬਾਰੇ ਦਿਲੋਂਗਵਾਹੀ ਦਿੱਤੀ ਜਿਸ ਨਾਲ ਸਾਨੂੰ ਅਨੰਦ ਮਹਿਸੂਸ ਹੋਇਆ।"),
+    ],
+    outputs=[
+        gr.Textbox(label="Status"),
+        gr.Audio(label="Synthesized Output")
+    ],
+    title="Speech Synthesis Demo",
+    description="Provide input text, a reference audio sample, and its corresponding text to synthesize new speech."
+)
+
+iface.launch()

model_handler.py

@@ -0,0 +1,43 @@
+# model_handler.py
+
+from gradio_client import Client, handle_file
+import threading
+import requests
+
+client = Client("ai4bharat/IndicF5")
+
+def call_model_with_timeout(text, ref_audio_path, ref_text="", timeout=120):
+    result = {"status": None, "audio_url": None}
+
+    def run():
+        try:
+            audio_url = client.predict(
+                text,
+                handle_file(ref_audio_path),
+                ref_text,
+                api_name="/synthesize_speech"
+            )
+            result["status"] = "success"
+            result["audio_url"] = audio_url
+        except Exception as e:
+            result["status"] = f"error: {e}"
+
+    thread = threading.Thread(target=run)
+    thread.start()
+    thread.join(timeout)
+
+    if thread.is_alive():
+        return "⛔ Timed out!", None
+
+    if result["status"] == "success":
+        audio_url = result["audio_url"]
+        try:
+            output_path = "output.wav"
+            audio_response = requests.get(audio_url)
+            with open(output_path, "wb") as f:
+                f.write(audio_response.content)
+            return "✅ Success", output_path
+        except Exception as e:
+            return f"⛔ Download error: {e}", None
+    else:
+        return result["status"], None