ui-base.py

#!/usr/bin/env python3
"""Simple Gradio UI for qwen-tts via base.sh (no voice cloning)."""

import os
import signal
import subprocess
import tempfile
from pathlib import Path

from dotenv import load_dotenv
import gradio as gr

# ── load .env if present ─────────────────────────────────────────────
load_dotenv(Path(__file__).parent / ".env")

SCRIPT = str(Path(__file__).parent / "base.sh")
OUTPUT_WAV = "/tmp/out.wav"


def synthesize(prompt: str, lang: str, seed: int, greedy: bool, temp: float,
               top_k: int, top_p: float, rep_pen: float, sub_temp: float,
               fmt: str, max_new: int, stream: bool, no_fa: bool,
               clamp_fp16: bool, output_path: str):
    """Run base.sh with user-provided inputs."""

    if not (Path(__file__).parent / ".env").exists():
        raise gr.Error("Missing .env — copy .env.example to .env and configure it (see README.md).")

    prompt_path = tempfile.NamedTemporaryFile(
        mode="w", suffix=".txt", delete=False
    ).name
    Path(prompt_path).write_text(prompt)

    # Build env overrides for advanced flags
    env = os.environ.copy()
    if seed != -1:
        env["QWENTTS_SEED"] = str(seed)
    if greedy:
        env["QWENTTS_GREEDY"] = "1"
    env["QWENTTS_TEMP"] = str(temp)
    env["QWENTTS_TOP_K"] = str(top_k)
    env["QWENTTS_TOP_P"] = str(top_p)
    env["QWENTTS_REP_PEN"] = str(rep_pen)
    env["QWENTTS_SUB_TEMP"] = str(sub_temp)
    env["QWENTTS_FORMAT"] = fmt
    env["QWENTTS_MAX_NEW"] = str(max_new)
    if stream:
        env["QWENTTS_STREAM"] = "1"
    if no_fa:
        env["QWENTTS_NO_FA"] = "1"
    if clamp_fp16:
        env["QWENTTS_CLAMP_FP16"] = "1"
    if output_path:
        env["QWENTTS_OUTPUT"] = output_path

    try:
        subprocess.run(
            ["bash", SCRIPT, prompt_path, lang],
            check=True, env=env,
        )
        return env.get("QWENTTS_OUTPUT", OUTPUT_WAV)
    except subprocess.CalledProcessError as e:
        raise gr.Error(f"base.sh failed (exit {e.returncode}). Check terminal for details.")
    finally:
        Path(prompt_path).unlink(missing_ok=True)


_LANGS = ["English", "French", "Spanish", "German", "Italian", "Portuguese", "Chinese", "Japanese", "Korean", "Russian"]
_DEFAULT_LANG = os.environ.get("LANG", "French") if os.environ.get("LANG", "") in _LANGS else "French"

with gr.Blocks(title="Qwen TTS - Base") as demo:
    gr.Markdown("# Qwen TTS - Base")

    prompt = gr.Textbox(
        label="Prompt Text (text to synthesize)",
        lines=6,
        placeholder="Enter the text you want spoken…",
    )
    lang = gr.Dropdown(
        label="Language",
        choices=_LANGS,
        value=_DEFAULT_LANG,
    )

    with gr.Accordion("Advanced Options", open=False):
        seed = gr.Number(label="Seed (-1 = random)", value=-1, precision=0)
        greedy = gr.Checkbox(label="Greedy decoding", value=False)
        temp = gr.Slider(label="Temperature", minimum=0.0, maximum=2.0, step=0.01, value=0.7)
        top_k = gr.Number(label="Top-k", value=40, precision=0)
        top_p = gr.Slider(label="Top-p", minimum=0.0, maximum=1.0, step=0.01, value=0.9)
        rep_pen = gr.Slider(label="Repetition Penalty", minimum=1.0, maximum=2.0, step=0.01, value=1.1)
        sub_temp = gr.Slider(label="Sub-quantizer Temp", minimum=0.0, maximum=1.0, step=0.01, value=0.1)
        fmt = gr.Dropdown(label="Output Format", choices=["wav16", "wav24", "wav32"], value="wav16")
        max_new = gr.Number(label="Max New Tokens", value=4096, precision=0)
        stream = gr.Checkbox(label="Stream by line", value=False)
        no_fa = gr.Checkbox(label="Disable Flash Attention", value=False)
        clamp_fp16 = gr.Checkbox(label="Clamp FP16", value=False)
        output_path = gr.Textbox(label="Output Path (optional)", placeholder="/tmp/out.wav")

    btn = gr.Button("Synthesize", variant="primary")
    stop_btn = gr.Button("Stop", variant="stop")
    output = gr.Audio(label="Output Audio", type="filepath", buttons=["download"])

    task = btn.click(synthesize, inputs=[prompt, lang, seed, greedy, temp, top_k, top_p,
                                         rep_pen, sub_temp, fmt, max_new, stream, no_fa,
                                         clamp_fp16, output_path], outputs=output)
    stop_btn.click(None, None, None, cancels=[task])

demo.launch(server_name="127.0.0.1", server_port=7861, share=False)