import os import html import math import re import struct import wave import numpy as np import gradio as gr import time from pathlib import Path from tts import ( cancel_pregeneration, generate_audio_stream, get_pregeneration_status, pregenerate_story_audio, split_into_chunks, ) from inference import transcribe_audio, answer_story_question from inference_lfm import answer_question_audio from qa_flow import build_qa_response, claim_audio_submission, release_audio_submission from runtime_config import SAMPLE_SOUNDS_DIR, gradio_allowed_paths from voice_clone import load_default_profile, list_saved_profiles # Generate relaxing audio chime waveforms procedurally # This runs fully client/server-side with standard Python, eliminating mock limits or Torch wait times def generate_chimes_wav(filename, duration=120, melody_type="lullaby"): sample_rate = 16000 n_samples = int(duration * sample_rate) with wave.open(str(filename), 'wb') as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(sample_rate) # Pentatonic cozy scales if melody_type == "lullaby": notes = [261.63, 293.66, 329.63, 392.00, 440.00] # C4, D4, E4, G4, A4 elif melody_type == "adventure": notes = [196.00, 246.94, 293.63, 329.63, 392.00] # G3, B3, D4, E4, G4 else: notes = [220.00, 261.63, 293.66, 329.63, 392.00] # A3, C4, D4, E4, G4 for i in range(n_samples): # cycle notes based on melody speed note_speed = sample_rate * 1.5 if melody_type == "lullaby" else sample_rate * 1.2 note_idx = int((i / note_speed) % len(notes)) freq = notes[note_idx] # envelope to prevent crackles sample_in_note = i % int(note_speed) envelope = 1.0 if sample_in_note < 1200: # attack envelope = sample_in_note / 1200 else: # decay decay_length = note_speed - 1200 envelope = max(0.0, 1.0 - (sample_in_note - 1200) / decay_length) # Synthesize voice-harmonic chime val = math.sin(2 * math.pi * freq * i / sample_rate) val += 0.45 * math.sin(2 * math.pi * (freq * 1.5) * i / sample_rate) val += 0.25 * math.sin(2 * math.pi * (freq * 2.0) * i / sample_rate) val = val / 1.7 * envelope packed_val = struct.pack(' str: p = COVERS_DIR / f"{slug}.png" return f"/gradio_api/file={p.resolve().as_posix()}" def load_books_from_stories(stories_dir="stories"): books = [] story_files = sorted( f for f in os.listdir(stories_dir) if f.endswith(".txt") ) if os.path.isdir(stories_dir) else [] for i, filename in enumerate(story_files): title = filename.replace(".txt", "").replace("_", " ") slug = filename.replace(".txt", "") synopsis = "" try: with open(os.path.join(stories_dir, filename), encoding="utf-8") as f: lines = [l.strip() for l in f.readlines() if l.strip()] synopsis = " ".join(lines[1:4]) if len(lines) > 1 else lines[0] if lines else "" except Exception: pass books.append({ "id": str(i + 1), "title": title, "author": "Public Domain", "cover_url": cover_path(slug), "voice_name": "Mom's Voice", "duration": 0, "elapsed_time": 0, "synopsis": synopsis, "category": "Children's Story", "is_cloned": False, "percentage": 0, "audio_path": None, "story_path": os.path.join(stories_dir, filename), }) return books mock_books = load_books_from_stories() mock_voices = [ { "id": "v1", "name": "Mom's Voice", "avatar_url": "πŸ‘©β€πŸ¦³", "description": "Warm, soft, with a natural soothing cadence. Best for bedtime stories.", "gender": "Female", "created_date": "2026-06-01", "status": "ready" }, { "id": "v2", "name": "Grandpa Joseph", "avatar_url": "πŸ‘΄", "description": "Deep, resonant, with an occasional comforting chuckle. Perfect for classic adventure stories.", "gender": "Male", "created_date": "2026-06-03", "status": "ready" }, { "id": "v3", "name": "Aunt Sarah", "avatar_url": "πŸ‘©", "description": "Bubbly, expressive, brilliant at acting out funny character voices.", "gender": "Female", "created_date": "2026-06-06", "status": "processing" } ] with open(os.path.join(os.path.dirname(__file__), "static", "style.css"), encoding="utf-8") as _f: css_code = _f.read() def generate_dashboard_html(books_list): if not books_list: return "
No stories found. Add .txt files to the stories/ folder.
" featured = books_list[0] others = books_list[1:] featured_js_title = featured['title'].replace("'", "\\'").replace('"', '\\"') html = f"""
Welcome Back, Sarah

Explore Your Warm Audiobook Shelves

Featured Clone

{featured['title']}

by {featured['author']}

{featured['synopsis']}

⏱️ 20 Min Remaining πŸŽ™οΈ Narrator: {featured['voice_name']} ⭐ Editorial Spotlight

Continue Reading with Channeled Voices

""" for bk in books_list: js_safe_title = bk['title'].replace("'", "\\'").replace('"', '\\"') html += f"""
{bk['category']} {bk['percentage']}%
{bk['title']}

by {bk['author']}

πŸŽ™οΈ {bk['voice_name']}
""" html += """
""" return html def generate_library_html(books_list, query="", category_filt="All", empty_mode=False): if empty_mode: return """
πŸ“š πŸŽ™οΈ

Your Bookshelf is Waiting to be Voiced

Choose a comforting companion, clone their warm voice, and hear any classical or modern tale come to life immediately.

Use 'Clone Voice' tab to craft a companion
""" filtered = [] for bk in books_list: match_query = (query.lower() in bk["title"].lower()) or (query.lower() in bk["author"].lower()) or (query.lower() in bk["voice_name"].lower()) match_cat = True if category_filt == "In Progress": match_cat = 0 < bk["percentage"] < 100 elif category_filt == "Completed": match_cat = bk["percentage"] == 100 if match_query and match_cat: filtered.append(bk) if len(filtered) == 0: return """
πŸ”

No audiobooks match your search criteria.

""" html_out = """
""" for bk in filtered: safe_title = html.escape(bk['title']) js_safe_title = bk['title'].replace("'", "\\'").replace('"', '\\"') html_out += f"""
{bk['percentage']}%
{bk['category']}
{safe_title}
Narrator: 🌟 {bk['voice_name']}
▢️ Tap to Play
""" html_out += "
" return html_out def load_paragraphs(story_path: str) -> list: try: with open(story_path, encoding="utf-8") as f: raw = f.read() paras = [p.strip() for p in raw.split("\n\n") if p.strip()] return paras except Exception: return [] def render_story_text(paragraphs: list, current_idx: int) -> str: if not paragraphs: return "" result = """
""" for i, para in enumerate(paragraphs): safe_para = html.escape(para) if i == current_idx: result += f"""

{safe_para}

""" else: result += f"""

{safe_para}

""" result += "
" return result def render_cloned_voices_html(voices_list): markup = """
""" for v in voices_list: status_color = "#16a34a" if v["status"] == "ready" else "#d97706" status_bg = "#f0fdf4" if v["status"] == "ready" else "#fef3c7" safe_name = html.escape(v['name']) safe_desc = html.escape(v['description']) voice_id = html.escape(v.get('profile_id', v['id'])) onclick = f"""onclick=" const sel = document.querySelector('#voice_selector textarea, #voice_selector input'); if (sel) {{ sel.value = '{voice_id}'; sel.dispatchEvent(new Event('input', {{bubbles: true}})); sel.dispatchEvent(new Event('change', {{bubbles: true}})); }} document.querySelectorAll('.voice-card-selected').forEach(c => c.classList.remove('voice-card-selected')); this.classList.add('voice-card-selected'); " """ markup += f"""
{v['avatar_url']}
{safe_name}
{v['status']}

{safe_desc}

Gender: {v['gender']} Created: {v['created_date']}
""" markup += "
" return markup # Load default voice profile from Voice_Profile/ (if any saved from previous sessions) _default_profile_id = load_default_profile() _saved_profiles = list_saved_profiles() if _default_profile_id: _default_voice_name = next( (p["voice_name"] for p in _saved_profiles if p["profile_id"] == _default_profile_id), "Cloned Voice" ) print(f"[MomsVoice] Default voice profile loaded: {_default_profile_id} ({_default_voice_name})") else: _default_voice_name = None print("[MomsVoice] No saved voice profile β€” using stock voice as default.") # Preload models at startup for fast response print("[MomsVoice] Preloading Qwen3-TTS 1.7B Base model (cloning + stock voice)...") from voice_clone import get_qwen_tts, _ensure_stock_voice_profile get_qwen_tts() print("[MomsVoice] TTS model ready.") # Pre-create stock voice profile so Play works immediately print("[MomsVoice] Creating stock voice profile (vivian)...") _ensure_stock_voice_profile() print("[MomsVoice] Stock voice profile ready.") # Load LFM Q&A model asynchronously β€” users typically start with Play tab, # so LFM can load in the background while they browse stories. import threading def _load_lfm_background(): from inference_lfm import get_lfm_model, warmup_lfm print("[MomsVoice] (background) Loading LFM2.5-Audio-1.5B Q&A model...") get_lfm_model() print("[MomsVoice] (background) LFM loaded. Running warmup generation...") warmup_lfm() print("[MomsVoice] (background) LFM Q&A model ready (compiled + warmed up).") _lfm_thread = threading.Thread(target=_load_lfm_background, daemon=True) _lfm_thread.start() print("[MomsVoice] LFM loading in background thread. App starting now.") # Gradio Application Core setup with gr.Blocks(title="MomsVoice", css=css_code) as demo: books_state = gr.State(mock_books) voices_state = gr.State(mock_voices) paragraphs_state = gr.State([]) tts_chunks_state = gr.State([]) voice_profile_state = gr.State(_default_profile_id) gr.HTML("""
πŸ’Ώ

VoiceBook

AI Bedtime Story Narrator
Hugging Face Space Node: ● Active
""") with gr.Tabs() as main_tabs: # TAB 1: Explore Dashboard with gr.TabItem("πŸ›οΈ Explore Space") as explore_tab: dashboard_container = gr.HTML(value=generate_dashboard_html(mock_books)) gr.HTML("""
🎧 Head over to My Library and tap any book to start listening!
""") # TAB 2: Library + Player (톡합) with gr.TabItem("πŸ“š My Library") as library_tab: with gr.Row(): # Left: book grid with gr.Column(scale=3): with gr.Row(): with gr.Column(scale=3): search_input = gr.Textbox(placeholder="Search by title, author, or voice...", label="Filter", show_label=False) with gr.Column(scale=1): category_filter = gr.Radio(["All", "In Progress", "Completed"], value="All", label="Progress", show_label=False) shelf_grid = gr.HTML(value=generate_library_html(mock_books)) # Hidden radio β€” book cards click this via JS onclick book_titles_list = [b["title"] for b in mock_books] book_selector = gr.Radio( choices=book_titles_list, value=None, label="", show_label=False, elem_classes="library-book-selector", visible=False, ) # Right: player panel with gr.Column(scale=2, elem_classes="dark-card-container"): gr.HTML("""
Now Playing SYNCHRONIZED PITCH
""") player_title_info = gr.HTML("""
πŸ“–

Select a story

Tap any book on the left to begin

""") player_audio_control = gr.Audio( visible=False, interactive=False, label="Audio Stream", type="numpy", streaming=True, autoplay=True, ) player_status_bar = gr.HTML("""
SELECT A STORY
""") with gr.Row(): play_btn = gr.Button("▢️ Play", variant="primary", scale=1, visible=False) pause_btn = gr.Button("⏸ Pause", variant="secondary", scale=1, visible=False) resume_btn = gr.Button("↩️ Resume Story", variant="secondary", scale=2, visible=False) chunk_status = gr.HTML("""
Chunk 0 / 0
""") story_finished_panel = gr.HTML("""
πŸŽ‰

Story Complete!

Tap another book to start a new story.

""", visible=False) # Story text display + slider story_text_display = gr.HTML(visible=False) timeline_slider = gr.Slider(minimum=0, maximum=100, step=1, value=0, label="πŸ“ Story position (paragraph)", visible=False) # Q&A Panel β€” always visible qa_panel = gr.HTML("""

❓ Ask About the Story

""") with gr.Group() as qa_input_group: question_audio = gr.Audio( sources=["microphone"], type="filepath", label="🎀 Ask your question β€” answer generates when you stop recording", show_label=True, streaming=False, format="wav", elem_id="qa_audio_input", ) question_text = gr.Textbox( placeholder="…or type a question here", label="Question (text fallback)", ) submit_question_btn = gr.Button("πŸ” Get Answer", variant="primary", elem_id="qa_submit_btn") answer_display = gr.HTML() answer_audio = gr.Audio(label="Answer Audio", interactive=False, autoplay=True, show_download_button=False) with gr.Row(): ask_btn = gr.Button("❓ Ask a Question", variant="primary", scale=2, visible=False) # TAB 3: Clone Voice Studio with gr.TabItem("πŸŽ™οΈ Clone Voice Studio") as clone_tab: with gr.Row(): with gr.Column(scale=1, elem_classes="card-container"): gr.HTML("""

πŸ§™ Voice Details Form

Provide a comforting nickname and details below to prepare vocal weights matching.

""") new_voice_name = gr.Textbox(placeholder="Enter a descriptive nickname (e.g., Grandma Judy, Dad)...", label="Voice Nickname") new_voice_gender = gr.Radio(["Female", "Male"], value="Female", label="Voice Gender Avatar") gr.HTML("""
Story Script to read aloud:

"Deep within the ancient forest, the giant oak tree stood. Its roots whispered stories of forgotten eras. 'Close your eyes,' the wind hummed softly, 'for our beautiful journey begins tonight.'"

""") mic_recorder = gr.Audio(sources=["microphone"], type="filepath", label="Record speaking story prompt (Min 10 seconds sample)") extract_btn = gr.Button("πŸͺ„ Analyze & Synthesize Customized Vocal Profile", variant="primary") with gr.Column(scale=1, elem_classes="card-container"): gr.HTML("""

Status Pipeline Visualizer

Track extraction, de-noising, and pitch tuning.

""") cloning_progress_msg = gr.HTML("""
πŸŽ™οΈ

Fill parameters on the left and read the story script aloud to begin extraction.

""") voice_cloning_success_panel = gr.HTML(visible=False) voice_sample_preview_widget = gr.Audio(value=str(_preview_chime_path), visible=False, label="Pre-listening synthesized vocal pitch weights match preview", interactive=False) add_to_library_btn = gr.Button("✨ Synced successfully! Bind Companion Voice to Inventory", visible=False) gr.HTML("""

Currently Available Voices

Click a voice card to select it as the narrator.

""") cloned_voices_list_grid = gr.HTML(value=render_cloned_voices_html(mock_voices)) voice_selector = gr.Textbox(visible=False, elem_id="voice_selector") # TAB 4: Profile with gr.TabItem("βš™οΈ Profile") as profile_tab: with gr.Row(): with gr.Column(scale=1, elem_classes="card-container"): gr.HTML("""
πŸ‘©β€πŸ’»

Your Profile

ReadBookMom β€” Hugging Face Hackathon 2026

""") gr.HTML("""
πŸ“š 10 Stories
🧠 2 Models
πŸŽ™οΈ Base 1.7B TTS Engine
""") with gr.Column(scale=1, elem_classes="card-container"): gr.HTML("""

Model Pipeline

Active AI models powering the app.

""") gr.HTML("""
πŸŽ™οΈ TTS + Cloning: Qwen3-TTS-1.7B-Base (single model)
πŸ—£οΈ Stock Voice: Vivian (via reference audio)
🧠 Q&A (end-to-end): LFM2.5-Audio-1.5B (LiquidAI)
""") # REACTIVE LOGIC # 1. Search / filter shelf def filter_books_shelf(query, category, current_books): return generate_library_html(current_books, query=query, category_filt=category) search_input.change(filter_books_shelf, inputs=[search_input, category_filter, books_state], outputs=shelf_grid) category_filter.change(filter_books_shelf, inputs=[search_input, category_filter, books_state], outputs=shelf_grid) # 2. Book card click β†’ load into player def handle_book_select(title_chosen, current_inventory, profile_id): if not title_chosen: return [gr.update()] * 12 selected = next((b for b in current_inventory if b["title"] == title_chosen), current_inventory[0]) # Show cloned voice name if profile exists from voice_clone import has_profile, list_saved_profiles as _list_profiles if profile_id and has_profile(profile_id): saved = _list_profiles() voice_label = next( (p["voice_name"] for p in saved if p["profile_id"] == profile_id), "Cloned Voice" ) narrator_label = f"πŸŽ™οΈ {voice_label}" narrator_color = "#4ade80" else: narrator_label = f"πŸ”Š {selected['voice_name']} (stock)" narrator_color = "#ffd3a6" safe_title = html.escape(selected['title']) safe_author = html.escape(selected['author']) player_html = f"""

{safe_title}

by {safe_author}

Active Narrator: {narrator_label}
""" paras = load_paragraphs(selected["story_path"]) total = max(len(paras) - 1, 0) tts_chunks = split_into_chunks("\n\n".join(paras)) # Pre-generate only the first few chunks in background for fast start. cancel_pregeneration() pregenerate_story_audio(tts_chunks, voice_profile_id=profile_id) pregen = get_pregeneration_status(tts_chunks, voice_profile_id=profile_id) warmed = int(pregen["cached"]) total_chunks = int(pregen["total"]) target_chunks = int(pregen.get("target", min(total_chunks, 3))) errors = int(pregen.get("errors", 0)) if pregen["complete"]: chunk_text = f"{warmed} / {total_chunks} chunks cached β€” ready" chunk_color = "#4ade80" status_label = "READY" status_color = "#94a3b8" elif warmed >= target_chunks and errors == 0: chunk_text = f"First {target_chunks} chunks cached β€” tap Play" chunk_color = "#4ade80" status_label = "READY β€” FIRST CHUNKS CACHED" status_color = "#4ade80" elif errors: chunk_text = f"Warmup hit {errors} error(s); Play will generate on demand" chunk_color = "#ef4444" status_label = "READY β€” ON-DEMAND AUDIO" status_color = "#f59e0b" else: chunk_text = f"Warming first {target_chunks} chunks: {warmed} / {target_chunks} cached β€” tap Play anytime" chunk_color = "#f59e0b" status_label = "WARMING AUDIO" status_color = "#f59e0b" chunk_html = f"""
{chunk_text}
""" status_html = f"""
{status_label}
""" story_html = render_story_text(paras, 0) return ( player_html, gr.Audio(visible=True), status_html, chunk_html, gr.Button(visible=True), # play_btn gr.Button(visible=False), # pause_btn gr.Button(visible=False), # ask_btn gr.HTML(visible=False), # story_finished_panel paras, # paragraphs_state tts_chunks, # tts_chunks_state gr.HTML(value=story_html, visible=True), # story_text_display gr.Slider(minimum=0, maximum=total, step=1, value=0, visible=True), # timeline_slider ) book_selector.change( handle_book_select, inputs=[book_selector, books_state, voice_profile_state], outputs=[player_title_info, player_audio_control, player_status_bar, chunk_status, play_btn, pause_btn, ask_btn, story_finished_panel, paragraphs_state, tts_chunks_state, story_text_display, timeline_slider] ) # 3. Extract vocal weights trigger def animate_cloning_pipeline(v_name, v_gender, recorder_data, progress=gr.Progress()): if not v_name.strip(): return ( """
Please provide a comforting nickname for your cloning candidate.
""", gr.HTML(visible=False), gr.Audio(visible=False), gr.Button(visible=False), None, ) if recorder_data is None: return ( """
Microphone recording sample missing. Speak some script lines!
""", gr.HTML(visible=False), gr.Audio(visible=False), gr.Button(visible=False), None, ) from voice_clone import create_voice_profile, synthesize_cloned_preview import soundfile as sf progress(0.1, desc="Extracting speaker embedding from recording...") try: profile_id = create_voice_profile(recorder_data, voice_name=v_name.strip()) except Exception as e: import logging, traceback logging.getLogger(__name__).exception("Voice cloning failed") return ( f"""
Voice cloning failed: {html.escape(str(e))}
""", gr.HTML(visible=False), gr.Audio(visible=False), gr.Button(visible=False), None, ) progress(0.7, desc="Generating voice preview...") try: preview_wav, preview_sr = synthesize_cloned_preview(profile_id) preview_path = SAMPLE_SOUNDS_DIR / f"clone_preview_{profile_id}.wav" sf.write(str(preview_path), preview_wav, preview_sr) except Exception as e: import logging logging.getLogger(__name__).exception("Preview synthesis failed") preview_path = None progress(1.0, desc="Voice cloned successfully!") safe_name = html.escape(v_name) avatar = "πŸ‘©" if v_gender == "Female" else "πŸ‘¨" cloned_card_html = f"""
{avatar}

{safe_name} Cloned successfully

Synthesized today using Qwen3-TTS voice cloning

Voice profile ready. Select a story and tap Play to hear narration in this voice.
""" return ( gr.HTML(visible=False), gr.HTML(value=cloned_card_html, visible=True), gr.Audio(value=str(preview_path) if preview_path else None, visible=preview_path is not None), gr.Button(visible=True), profile_id, ) extract_btn.click( animate_cloning_pipeline, inputs=[new_voice_name, new_voice_gender, mic_recorder], outputs=[cloning_progress_msg, voice_cloning_success_panel, voice_sample_preview_widget, add_to_library_btn, voice_profile_state] ) # 4. Add cloned voice to inventory def save_new_cloned_voice(v_name, v_gender, list_voices, current_profile_id): avatar = "πŸ‘©" if v_gender == "Female" else "πŸ‘¨" new_voice_item = { "id": f"v-{time.time()}", "name": v_name, "avatar_url": avatar, "description": f"Custom cloned {v_gender.lower()} companion voice. Highly optimized.", "gender": v_gender, "created_date": "2026-06-06", "status": "ready", "profile_id": current_profile_id, } updated = [new_voice_item] + list_voices return ( updated, render_cloned_voices_html(updated), gr.HTML(visible=True, value="""
πŸŽ™οΈ

Vocal profile saved successfully!

"""), gr.HTML(visible=False), gr.Audio(visible=False), gr.Button(visible=False), gr.Textbox(value="") ) add_to_library_btn.click( save_new_cloned_voice, inputs=[new_voice_name, new_voice_gender, voices_state, voice_profile_state], outputs=[voices_state, cloned_voices_list_grid, cloning_progress_msg, voice_cloning_success_panel, voice_sample_preview_widget, add_to_library_btn, new_voice_name] ) # Voice card selection β€” update voice_profile_state when a card is clicked def handle_voice_select(selected_id, voices): """Map selected voice ID to profile_id for TTS.""" if not selected_id: return gr.update() # Check if it matches a saved profile ID directly from voice_clone import has_profile if has_profile(selected_id): return selected_id # Check if it matches a voice in the list with a profile_id for v in voices: if v.get('profile_id') == selected_id or v.get('id') == selected_id: pid = v.get('profile_id') if pid and has_profile(pid): return pid # No valid profile β€” use stock voice (None) return None voice_selector.change( handle_voice_select, inputs=[voice_selector, voices_state], outputs=[voice_profile_state] ) # 5. Play button β€” streams TTS audio chunk by chunk def stream_tts(tts_chunks, paras, profile_id): import logging as _log _log.getLogger(__name__).info("stream_tts called with profile_id=%s", profile_id) voice_mode = "πŸŽ™οΈ Cloned Voice" if profile_id else "πŸ”Š Stock Voice" _status_playing = f"""
PLAYING β€” {voice_mode}
""" _status_done = """
DONE
""" n_paras = len(paras) if paras else 1 if not tts_chunks: yield (gr.update(), _status_done, "
No content loaded
", gr.Button(visible=True), gr.Button(visible=False), gr.Button(visible=False), gr.Slider(value=0), render_story_text(paras, 0) if paras else "", gr.HTML(visible=False), gr.Button(visible=False)) return n = len(tts_chunks) yield (gr.update(), _status_playing, f"
β–Ά Paragraph 1 / {n_paras}
", gr.Button(visible=False), gr.Button(visible=True), gr.Button(visible=False), gr.Slider(value=0), render_story_text(paras, 0) if paras else "", gr.HTML(visible=False), gr.Button(visible=False)) for sample_rate, wav, i, total, err in generate_audio_stream(tts_chunks, voice_profile_id=profile_id): para_idx = min(int(i * n_paras / total), n_paras - 1) if total > 0 else 0 if err: yield (gr.update(), f"
Something went wrong. Try again.
", "", gr.Button(visible=True), gr.Button(visible=False), gr.Button(visible=False), gr.Slider(value=para_idx), gr.update(), gr.HTML(visible=False), gr.Button(visible=False)) return para_idx = min(int(i * n_paras / total), n_paras - 1) if total > 0 else 0 is_last = (i + 1 >= total) para_status = f"
β–Ά Paragraph {para_idx + 1} / {n_paras}
" yield ((sample_rate, wav.astype(np.float32) if hasattr(wav, 'astype') else wav), _status_playing, para_status, gr.Button(visible=False), gr.Button(visible=True), gr.Button(visible=True), gr.Slider(value=para_idx), render_story_text(paras, para_idx), gr.HTML(visible=is_last), gr.Button(visible=True)) yield (gr.update(), _status_done, f"
βœ… Story complete
", gr.Button(visible=True), gr.Button(visible=False), gr.Button(visible=False), gr.Slider(value=max(n_paras - 1, 0)), render_story_text(paras, max(n_paras - 1, 0)), gr.HTML(visible=True), gr.Button(visible=True)) play_event = play_btn.click( stream_tts, inputs=[tts_chunks_state, paragraphs_state, voice_profile_state], outputs=[player_audio_control, player_status_bar, chunk_status, play_btn, pause_btn, ask_btn, timeline_slider, story_text_display, story_finished_panel, resume_btn] ) # 6. Pause button β€” PAUSED state def enter_paused_state(slider_val, paras): total = len(paras) if paras else 1 chunk_html = f"""
Paragraph {int(slider_val) + 1} / {total} β€” paused
""" status_html = """
PAUSED
""" return ( status_html, chunk_html, gr.Button(visible=True), gr.Button(visible=False), ) pause_btn.click( enter_paused_state, inputs=[timeline_slider, paragraphs_state], outputs=[player_status_bar, chunk_status, play_btn, pause_btn], cancels=[play_event] ) # 7. Timeline slider β†’ highlight current paragraph def update_story_highlight(slider_val, paras): idx = int(slider_val) total = len(paras) chunk_html = f"""
Paragraph {idx + 1} / {total}
""" return render_story_text(paras, idx), chunk_html timeline_slider.change(update_story_highlight, inputs=[timeline_slider, paragraphs_state], outputs=[story_text_display, chunk_status]) # 8. Ask button β€” PAUSED-ASKING state def enter_asking_state(): cancel_pregeneration() status_html = """
PAUSED β€” ASKING
""" return ( status_html, gr.Button(visible=False), # play_btn gr.Button(visible=False), # pause_btn gr.Button(visible=False), # ask_btn gr.Button(visible=True), # resume_btn gr.HTML(value=""), # answer_display β€” clear previous answer gr.Audio(visible=False), # answer_audio ) ask_btn.click( enter_asking_state, inputs=[], outputs=[player_status_bar, play_btn, pause_btn, ask_btn, resume_btn, answer_display, answer_audio], cancels=[play_event], ) # 9. Submit question β€” LFM2.5-Audio end-to-end (single model: audio-in β†’ audio+text-out) _GENERATING_HTML = """
🎧
GENERATING ANSWER...
""" def _build_answer_outputs(result): if not result["ok"]: error_html = """
Please type a question or record one with the microphone.
""" return gr.HTML(value=error_html, visible=True), gr.Audio(visible=False) safe_answer = html.escape(str(result["answer_text"])) safe_question = html.escape(str(result["display_question"])) answer_html = f"""
Answer (LFM2.5-Audio)
“{safe_answer}”
Q: {safe_question}
""" answer_audio_path_out = result["audio_path"] if answer_audio_path_out: return gr.HTML(value=answer_html, visible=True), gr.Audio(value=str(answer_audio_path_out), visible=True, autoplay=True) return gr.HTML(value=answer_html, visible=True), gr.Audio(visible=False) def handle_question_submit(question_txt, question_audio_path, paragraphs, slider_val, profile_id): q_txt = (question_txt or "").strip() has_audio = question_audio_path is not None and question_audio_path != "" yield gr.HTML(value=_GENERATING_HTML, visible=True), gr.Audio(visible=False) result = build_qa_response( question_text=q_txt, question_audio_path=question_audio_path if has_audio else None, paragraphs=paragraphs, output_dir=SAMPLE_SOUNDS_DIR, answer_fn=answer_question_audio, max_new_tokens=512, ) yield _build_answer_outputs(result) submit_question_btn.click( handle_question_submit, inputs=[question_text, question_audio, paragraphs_state, timeline_slider, voice_profile_state], outputs=[answer_display, answer_audio] ) # Auto-submit immediately when recording stops β€” show waiting UI then run inference def auto_submit_on_stop(audio_path, question_txt, paragraphs, slider_val, profile_id): if not audio_path: return if not claim_audio_submission(audio_path): return yield gr.HTML(value=_GENERATING_HTML, visible=True), gr.Audio(visible=False) result = build_qa_response( question_text=(question_txt or "").strip(), question_audio_path=audio_path, paragraphs=paragraphs, output_dir=SAMPLE_SOUNDS_DIR, answer_fn=answer_question_audio, max_new_tokens=512, ) yield _build_answer_outputs(result) question_audio.stop_recording( auto_submit_on_stop, inputs=[question_audio, question_text, paragraphs_state, timeline_slider, voice_profile_state], outputs=[answer_display, answer_audio], ) # 10. Resume button β€” dismiss Q&A panel, show Play to continue def enter_resume_state(): status_html = """
READY β€” Tap Play to continue
""" return ( status_html, gr.Button(visible=True), # play_btn gr.Button(visible=False), # pause_btn gr.Button(visible=True), # ask_btn gr.Button(visible=False), # resume_btn gr.Textbox(value=""), # question_text β€” clear previous question None, # question_audio β€” clear previous recording ) resume_btn.click( enter_resume_state, inputs=[], outputs=[player_status_bar, play_btn, pause_btn, ask_btn, resume_btn, question_text, question_audio] ) # 11. Pause playback when switching to other tabs def pause_on_tab_switch(): status_html = """
PAUSED
""" return status_html, gr.Button(visible=True), gr.Button(visible=False) explore_tab.select( pause_on_tab_switch, inputs=[], outputs=[player_status_bar, play_btn, pause_btn], cancels=[play_event] ) clone_tab.select( pause_on_tab_switch, inputs=[], outputs=[player_status_bar, play_btn, pause_btn], cancels=[play_event] ) if __name__ == "__main__": demo.queue(max_size=8).launch(allowed_paths=gradio_allowed_paths())