import gradio as gr import requests import json import os from pathlib import Path import uuid import fcntl import time # gr.NO_RELOAD = False # API Base URL BASE_URL = os.environ.get("BASE_URL", "") # Counter persistence file COUNTER_FILE = Path("generation_counter.json") # Example texts EXAMPLE_TEXT_ENGLISH = "Welcome to Ringg TTS! This is a text to speech system that can convert your text into natural-sounding audio. Try it out with your own content!" EXAMPLE_TEXT_HINDI = "नमस्ते! मैं रिंग टीटीएस हूँ। मैं आपके टेक्स्ट को प्राकृतिक आवाज़ में बदल सकता हूँ। कृपया अपना टेक्स्ट यहाँ लिखें और सुनें।" EXAMPLE_TEXT_MIXED = "Hello दोस्तों! Welcome to Ringg TTS. यह एक बहुत ही शानदार text to speech system है जो Hindi और English दोनों languages को support करता है।" def load_counter(): """Load universal generation counter from file (thread-safe)""" try: if COUNTER_FILE.exists(): with open(COUNTER_FILE, "r") as f: # Try to acquire shared lock for reading try: fcntl.flock(f.fileno(), fcntl.LOCK_SH) data = json.load(f) fcntl.flock(f.fileno(), fcntl.LOCK_UN) return data.get("count", 0) except: # If locking fails, just read without lock f.seek(0) data = json.load(f) return data.get("count", 0) except Exception as e: print(f"Error loading counter: {e}") return 0 def save_counter(count): """Save universal generation counter to file (thread-safe)""" try: # Use file locking to prevent race conditions with multiple users with open(COUNTER_FILE, "w") as f: try: fcntl.flock(f.fileno(), fcntl.LOCK_EX) json.dump({"count": count, "last_updated": time.time()}, f) f.flush() os.fsync(f.fileno()) fcntl.flock(f.fileno(), fcntl.LOCK_UN) except: # If locking fails, just write without lock json.dump({"count": count, "last_updated": time.time()}, f) f.flush() except Exception as e: print(f"Error saving counter: {e}") def increment_counter(): """Atomically increment and return the new counter value""" try: # Read current value, increment, and save atomically with open(COUNTER_FILE, "r+" if COUNTER_FILE.exists() else "w+") as f: try: fcntl.flock(f.fileno(), fcntl.LOCK_EX) # Read current count f.seek(0) try: data = json.load(f) current_count = data.get("count", 0) except: current_count = 0 # Increment new_count = current_count + 1 # Write back f.seek(0) f.truncate() json.dump({"count": new_count, "last_updated": time.time()}, f) f.flush() os.fsync(f.fileno()) fcntl.flock(f.fileno(), fcntl.LOCK_UN) return new_count except: # Fallback without locking f.seek(0) try: data = json.load(f) current_count = data.get("count", 0) except: current_count = 0 new_count = current_count + 1 f.seek(0) f.truncate() json.dump({"count": new_count, "last_updated": time.time()}, f) f.flush() return new_count except Exception as e: print(f"Error incrementing counter: {e}") return 0 def get_voices(): """Fetch available voices from API""" try: response = requests.get(f"{BASE_URL}/voices", timeout=10) if response.status_code == 200: voices_data = response.json().get("voices", {}) # Create a list of tuples (display_name, voice_id) voices = [] for voice_id, voice_info in voices_data.items(): name = voice_info.get("name", "Unknown") gender = voice_info.get("gender", "N/A") display_name = f"{name} ({gender})" voices.append((display_name, voice_id)) return sorted(voices, key=lambda x: x[0]) return [] except Exception as e: print(f"Error fetching voices: {e}") return [] def synthesize_speech(text, voice_id): """Synthesize speech from text""" if not text or not text.strip(): return None, "⚠️ Please enter some text", "", "", "", "", "", "" if not voice_id: return None, "⚠️ Please select a voice", "", "", "", "", "", "" # Print input text length text_length = len(text) print(f"Input text length: {text_length} characters") try: payload = {"text": text, "voice_id": voice_id} response = requests.post( f"{BASE_URL}/synthesize", headers={"Content-Type": "application/json"}, json=payload, timeout=30, ) if response.status_code == 200: result = response.json() if result.get("success"): audio_url = result.get("audio_url", "") metrics = result.get("metrics", {}) # Format metrics total_time = f"{metrics.get('t', 0):.3f}s" rtf = f"{metrics.get('rtf', 0):.4f}" wav_duration = f"{metrics.get('wav_seconds', 0):.2f}s" vocoder_time = f"{metrics.get('t_vocoder', 0):.3f}s" no_vocoder_time = f"{metrics.get('t_no_vocoder', 0):.3f}s" rtf_no_vocoder = f"{metrics.get('rtf_no_vocoder', 0):.4f}" status_msg = "✅ Audio generated successfully!" return ( audio_url, status_msg, total_time, rtf, wav_duration, vocoder_time, no_vocoder_time, rtf_no_vocoder, ) else: error_msg = result.get("message", "Unknown error") return None, f"❌ Synthesis failed: {error_msg}", "", "", "", "", "", "" else: return ( None, f"❌ API returned status code: {response.status_code}", "", "", "", "", "", "", ) except Exception as e: return None, f"❌ Error: {str(e)}", "", "", "", "", "", "" # Load initial counter value initial_counter = load_counter() # Create Gradio interface with gr.Blocks( theme=gr.themes.Base( font=[gr.themes.GoogleFont("Source Sans Pro"), "Arial", "sans-serif"] ), css=".gradio-container {max-width: none !important;}", ) as demo: # Title with Health Status with gr.Row(): with gr.Column(scale=4): audio_image = gr.HTML( value="""
Logo

Ringg Squirrel TTS v1.0 🐿️

""" ) with gr.Column(scale=1): generation_counter = gr.Markdown( f"**🌍 Generations:** {initial_counter}", elem_id="counter" ) # Best Practices Section gr.Markdown(""" ### 📝 Best Practices for Best Results - **Supported Languages:** Hindi and English only - **Check spelling carefully:** Misspelled words may be mispronounced - **Punctuation matters:** Use proper punctuation for natural pauses and intonation - **Technical terms:** Extremely rare or specialized technical terms might be mispronounced - **Numbers & dates:** Write numbers as words for better pronunciation (e.g., "twenty-five" instead of "25") """) # Text Input text_input = gr.Textbox( label="Text (max 500 characters)", placeholder="Type or paste your text here (max 500 characters)...", lines=6, max_lines=10, max_length=500, ) # Character count display char_count = gr.Markdown("**Character count:** 0 / 500") with gr.Row(): with gr.Column(scale=1): # Voice Selection voices = get_voices() voice_choices = {display: vid for display, vid in voices} voice_dropdown = gr.Dropdown( choices=list(voice_choices.keys()), label="Choose a voice style", info=f"{len(voices)} voices available", value=list(voice_choices.keys())[0] if voices else None, ) with gr.Column(scale=1): audio_output = gr.Audio(label="Listen to your audio", type="filepath") metrics_header = gr.Markdown("### 📊 Generation Metrics", visible=False) metrics_output = gr.Code( label="Metrics", language="json", interactive=False, visible=False ) generate_btn = gr.Button("🎬 Generate Speech", variant="primary", size="lg") gr.Markdown("#### 🎯 Try these examples:") with gr.Row(): example_btn1 = gr.Button("English Example", size="sm") example_btn2 = gr.Button("Hindi Example", size="sm") example_btn3 = gr.Button("Mixed Example", size="sm") # Footer gr.Markdown("---") gr.Markdown("# 🙏 Acknowledgements") # gr.Markdown("- Based on [ZipVoice](https://github.com/k2-fsa/ZipVoice)") gr.Markdown( "- Special thanks to [@jeremylee12](https://huggingface.co/jeremylee12) for his contributions" ) # Event Handlers def update_char_count(text): """Update character count as user types""" count = len(text) if text else 0 return f"**Character count:** {count} / 500" def load_example_text(example_text): """Load example text and update character count""" count = len(example_text) return example_text, f"**Character count:** {count} / 500" def clear_text(): """Clear text input""" return "", "**Character count:** 0 / 500" def on_generate(text, voice_display): voice_id = voice_choices.get(voice_display) audio_url, _status, t_time, rtf, wav_dur, voc_time, no_voc_time, rtf_no_voc = ( synthesize_speech(text, voice_id) ) # Download audio if URL is available audio_file = None # Always get fresh counter from file new_count = load_counter() if audio_url: try: audio_response = requests.get(audio_url, timeout=30) if audio_response.status_code == 200: # Save to temporary file audio_file = f"/tmp/ringg_{str(uuid.uuid4())}.wav" with open(audio_file, "wb") as f: f.write(audio_response.content) # Atomically increment the UNIVERSAL counter new_count = increment_counter() except Exception as e: _status = f"⚠️ Audio generated but download failed: {str(e)}" # Format metrics as JSON string (only if available) has_metrics = any([t_time, rtf, wav_dur, voc_time, no_voc_time, rtf_no_voc]) metrics_json = "" if has_metrics: metrics_json = json.dumps( { "total_time": t_time, "rtf": rtf, "audio_duration": wav_dur, "vocoder_time": voc_time, "no_vocoder_time": no_voc_time, "rtf_no_vocoder": rtf_no_voc, }, indent=2, ) return ( audio_file, gr.update(visible=has_metrics), gr.update(value=metrics_json, visible=has_metrics), f"**🌍 Generations:** {new_count}", ) def refresh_counter_on_load(): """Refresh the universal generation counter when the UI loads/reloads""" return f"**🌍 Generations:** {load_counter()}" # Update character count on text input change text_input.change(fn=update_char_count, inputs=[text_input], outputs=[char_count]) # Example button clicks example_btn1.click( fn=lambda: load_example_text(EXAMPLE_TEXT_ENGLISH), inputs=None, outputs=[text_input, char_count], ) example_btn2.click( fn=lambda: load_example_text(EXAMPLE_TEXT_HINDI), inputs=None, outputs=[text_input, char_count], ) example_btn3.click( fn=lambda: load_example_text(EXAMPLE_TEXT_MIXED), inputs=None, outputs=[text_input, char_count], ) generate_btn.click( fn=on_generate, inputs=[text_input, voice_dropdown], outputs=[ audio_output, # status_output, metrics_header, metrics_output, generation_counter, ], concurrency_limit=2, concurrency_id="synthesis", ) # Refresh global generation counter on page load/refresh demo.load(fn=refresh_counter_on_load, inputs=None, outputs=[generation_counter]) if __name__ == "__main__": demo.queue(default_concurrency_limit=2, max_size=20) demo.launch(share=False, server_name="0.0.0.0", server_port=7860, debug=True)