Create app.py

app.py

@@ -0,0 +1,98 @@
+import gradio as gr
+import base64
+import numpy as np
+from scipy.io import wavfile
+from voice_processing import process_files_concurrently, get_model_names, voice_mapping
+from io import BytesIO
+import asyncio
+import aiofiles
+from aiofiles.os import stat as aio_stat
+import os
+
+# Limit concurrency to avoid overwhelming the system
+MAX_CONCURRENT_TASKS = 5
+
+async def convert_tts(model_name, text_files, selected_voice, slang_rate, use_uploaded_voice, voice_upload_files):
+    edge_tts_voice = voice_mapping.get(selected_voice)
+    if not edge_tts_voice:
+        return {"error": f"Invalid voice '{selected_voice}'."}, None
+
+    voice_upload_file_data = []
+    if use_uploaded_voice and voice_upload_files is not None:
+        for file in voice_upload_files:
+            async with aiofiles.open(file.name, 'rb') as f:
+                voice_upload_file_data.append(await f.read())
+
+    text_file_data = []
+    if not use_uploaded_voice and text_files is not None:
+        for file in text_files:
+            async with aiofiles.open(file.name, 'r') as f:
+                content = await f.read()
+                if len(content) > 5000:
+                    return {"error": "Each text file should be at most 5000 characters."}, None
+                text_file_data.append(content)
+
+    # Check total combined text length
+    if not use_uploaded_voice and sum(len(text) for text in text_file_data) > 50000:
+        return {"error": "Total combined text length should be at most 50,000 characters."}, None
+
+    # Ensure limits on the number of files
+    if len(text_files if text_files else []) > 20 or len(voice_upload_files if voice_upload_files else []) > 20:
+        return {"error": "You can upload at most 20 text or voice files."}, None
+
+    # Process the text inputs or uploaded voice files concurrently
+    if use_uploaded_voice:
+        results = await process_files_concurrently(voice_upload_file_data, model_name, "", edge_tts_voice, slang_rate, use_uploaded_voice)
+    else:
+        results = await process_files_concurrently(text_file_data, model_name, "", edge_tts_voice, slang_rate, use_uploaded_voice)
+
+    info_list = []
+    audio_uris = []
+
+    for result in results:
+        info, edge_tts_output_path, tts_output_data, edge_output_file = result
+        if tts_output_data is not None:
+            _, audio_output = tts_output_data
+
+            # Convert audio output to bytes
+            audio_bytes = None
+            if isinstance(audio_output, np.ndarray):
+                byte_io = BytesIO()
+                wavfile.write(byte_io, 40000, audio_output)
+                byte_io.seek(0)
+                audio_bytes = byte_io.read()
+            else:
+                audio_bytes = audio_output
+
+            # Create a data URI for the audio
+            audio_data_uri = f"data:audio/wav;base64,{base64.b64encode(audio_bytes).decode('utf-8')}"
+            audio_uris.append(audio_data_uri)
+
+        info_list.append(info)
+
+    return {"info": info_list}, audio_uris
+
+def get_models():
+    return get_model_names()
+
+def get_voices():
+    return list(voice_mapping.keys())
+
+iface = gr.Interface(
+    fn=convert_tts,
+    inputs=[
+        gr.Dropdown(choices=get_models(), label="Model", interactive=True),
+        gr.File(label="Text Files", type="file", file_count="multiple"),
+        gr.Dropdown(choices=get_voices(), label="Voice", interactive=True),
+        gr.Slider(minimum=0, maximum=1, step=0.01, label="Slang Rate"),
+        gr.Checkbox(label="Use Uploaded Voice"),
+        gr.File(label="Voice Files", type="file", file_count="multiple")
+    ],
+    outputs=[
+        gr.JSON(label="Info"),
+        gr.Gallery(label="Audio URIs")
+    ],
+    title="Text-to-Speech Conversion"
+)
+
+iface.launch()