diff options
Diffstat (limited to 'app/backends/faster.py')
| -rwxr-xr-x | app/backends/faster.py | 416 |
1 files changed, 416 insertions, 0 deletions
diff --git a/app/backends/faster.py b/app/backends/faster.py new file mode 100755 index 0000000..0d34a0f --- /dev/null +++ b/app/backends/faster.py @@ -0,0 +1,416 @@ +#!/usr/bin/env python3 +"""Set up the faster-qwen3-tts backend for the audiobook generator. + +faster-qwen3-tts is an OpenAI-compatible Qwen3-TTS server with CUDA-graph +inference (NVIDIA GPU required). It always uses voice cloning, with the +reference voice configured on the server through a ``voices.json``. This +module sets the whole backend up end-to-end as a TUI: pip-install the +package, clone the repo (for ``examples/openai_server.py``), build a +``voices.json`` from a directory of .wav references (transcribed with +Whisper), sync ``app/converter/config.py``, and print the launch command. It is +driven by ``audiobook.py``'s hub but can also be run directly with flags. + +Usage: + python app/backends/faster.py [--wavs WAV_DIR] [--output PATH] + [--language LANG] [--whisper-model NAME] [--force] + [--port PORT] [--voice NAME] [--skip-install] [--skip-clone] +""" + +import argparse +import json +import sys +from pathlib import Path +from typing import List, Optional + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from backends import ( + BackendStatus, + ConfigureAction, + ServerSpec, + common, + envs, + format_launch_hint, +) +from backends.common import ( + APP_DIR, + VOICES_DIR, + find_wav_files, + normalize_dir_arg, +) +from converter import config +from converter.tts import ( + normalize_language, + transcribe_reference_audio, + whisper_backend_available, +) +from ui import tui + +FASTER_DIR_NAME = "faster-qwen3-tts" +FASTER_GIT_URL = "https://github.com/andimarafioti/faster-qwen3-tts" +FASTER_PIP_PKG = "faster-qwen3-tts[demo]" +WHISPER_MODELS = ("tiny", "base", "small", "medium", "large-v3") + + +def _checkout() -> Path: + return APP_DIR / FASTER_DIR_NAME + + +def _is_installed() -> bool: + return envs.module_available("faster_qwen3_tts") + + +def _is_cloned() -> bool: + return (_checkout() / "examples" / "openai_server.py").is_file() + + +def _config_port() -> int: + import urllib.parse + try: + return urllib.parse.urlsplit(config.FASTER_API_URL).port or 8000 + except ValueError: + return 8000 + + +def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: + """Transcribe each wav file and build the voices mapping.""" + voices = {} + for wav_file in wav_files: + name = wav_file.stem + print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") + text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) + if text: + print(f"[OK] {name}: {text}") + else: + print(f"[WARNING] No transcript for '{name}'; the faster backend " + "strongly recommends an accurate transcript — consider " + "editing voices.json by hand before starting the server") + voices[name] = { + "ref_audio": str(wav_file.resolve()), + "ref_text": text or "", + "language": language, + } + return voices + + +def _write_voices_json(output_path: Path, wav_dir: Path, language: str, + whisper_model: str, force: bool) -> Optional[dict]: + """Transcribe the wav dir and write voices.json; return the voices dict.""" + wav_files = find_wav_files(wav_dir) + if not wav_files: + print(f"[ERROR] No .wav files found in {wav_dir}") + return None + if whisper_backend_available() is None: + print("[WARNING] Neither faster_whisper nor whisper was found, so " + "transcripts will be empty — install one or edit voices.json " + "by hand.") + voices = build_voices(wav_files, language, whisper_model) + with output_path.open("w", encoding="utf-8") as handle: + json.dump(voices, handle, indent=4, ensure_ascii=False) + handle.write("\n") + print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " + f"{', '.join(voices)}") + return voices + + +def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: + """Linear TUI wizard collecting every faster-setup decision.""" + _GO_BACK = object() + + def confirm(question: str, default: bool = True) -> Optional[bool]: + res = tui.confirm(stdscr, question, default=default, + cancel_value=_GO_BACK) + return None if res is _GO_BACK else res + + # Step 0: pip install (if not installed and not skipped). + do_install = False + if not _is_installed() and not args.skip_install: + choice = confirm("faster-qwen3-tts is not installed. " + "pip install it now?", default=True) + if choice is None: + return None + do_install = choice + + # Step 1: clone (if not cloned and not skipped). + do_clone = False + if not _is_cloned() and not args.skip_clone: + choice = confirm(f"faster-qwen3-tts repo not cloned. Clone it into " + f"./app/{FASTER_DIR_NAME}?", default=True) + if choice is None: + return None + do_clone = choice + + # Step 2: voices.json — wav dir, language, whisper model, output path. + wav_dir = args.input_dir + if wav_dir is None: + wav_dir = tui.browse_directory( + stdscr, "Select the directory with your .wav voices", + info=common.wav_dir_info, preview=common.wav_dir_preview, + start=VOICES_DIR) + language = args.language + if language is None: + lang_text = tui.line_edit( + stdscr, "Language", config.LANGUAGE, + validate=lambda s: None if _try_language(s) + else "Unknown language (e.g. English, en)", + help_lines=["Language for every voice, as passed to the TTS " + "model (names or short codes accepted)"]) + language = lang_text + whisper_model = args.whisper_model + if whisper_model is None: + whisper_model = tui.menu( + stdscr, "Whisper model for transcription", + [(m, m) for m in WHISPER_MODELS], + default_index=WHISPER_MODELS.index("base")) + output_path = args.output + if output_path is None: + # Default into the cloned checkout; fall back to the wav directory + # when the checkout is not present (so a flag-only run still works). + output_path = (_checkout() / "voices.json") if _is_cloned() \ + else (wav_dir / "voices.json") + if output_path.exists() and not args.force: + choice = confirm(f"{output_path} already exists. Overwrite?", + default=True) + if choice is None or choice is False: + # Fall back to a path in the current directory. + output_path = Path.cwd() / "voices.json" + + # Step 3: port + default voice. + port = args.port + if port is None: + port_text = tui.line_edit( + stdscr, "Server port", str(_config_port()), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535") + port = int(port_text) + + return { + "do_install": do_install, + "do_clone": do_clone, + "wav_dir": wav_dir, + "language": language, + "whisper_model": whisper_model, + "output_path": output_path, + "port": port, + "force": args.force, + } + + +def _try_language(value: str) -> bool: + try: + normalize_language(value) + return True + except ValueError: + return False + + +def _execute(settings: dict) -> int: + """Console tail: install, clone, write voices.json, sync, advise.""" + if settings["do_install"]: + rc = common.pip_install([FASTER_PIP_PKG]) + if rc != 0: + print(f"[WARNING] pip install failed (exit {rc}); install " + f"{FASTER_PIP_PKG} manually") + else: + print("[OK] faster-qwen3-tts installed") + + if settings["do_clone"]: + rc = common.git_clone(FASTER_GIT_URL, _checkout()) + if rc != 0: + print(f"[WARNING] git clone failed (exit {rc}); clone manually: " + f"git clone {FASTER_GIT_URL} {_checkout()}") + else: + print(f"[OK] cloned into {_checkout()}") + + voices = _write_voices_json(settings["output_path"], settings["wav_dir"], + settings["language"], settings["whisper_model"], + settings["force"]) + if voices is None: + return 1 + + # Sync app/converter/config.py port + default voice. + port = settings["port"] + new_url = common.url_with_port(config.FASTER_API_URL, port) + if new_url != config.FASTER_API_URL: + if common.update_config_value("FASTER_API_URL", new_url): + print(f"[OK] Updated FASTER_API_URL to {new_url}") + else: + print("[WARNING] Could not update FASTER_API_URL; edit " + "app/converter/config.py by hand") + default_voice = next(iter(voices)) + if default_voice != config.FASTER_VOICE: + if common.update_config_value("FASTER_VOICE", default_voice): + print(f"[OK] Updated FASTER_VOICE to {default_voice}") + else: + print("[WARNING] Could not update FASTER_VOICE; edit " + "app/converter/config.py by hand") + + _print_launch_hint(settings["output_path"], port) + return 0 + + +def _print_launch_hint(voices_path: Path, port: int) -> None: + print() + if _is_cloned(): + py = envs.env_python() + print("Start the server with (or use the hub's 'Server' menu):") + print(f" {py} {_checkout()}/examples/openai_server.py " + f"--voices {voices_path} --port {port}") + else: + print("[INFO] Clone faster-qwen3-tts to get examples/openai_server.py,") + print(f" then run it with --voices {voices_path} --port {port}") + + +def run_tui(args: Optional[argparse.Namespace] = None) -> int: + """Run the faster setup wizard end-to-end.""" + import curses + if args is None: + args = build_parser().parse_args([]) + try: + settings = curses.wrapper(_wizard, args) + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + try: + curses.curs_set(1) + except curses.error: + pass + if settings is None: + print("[INFO] Aborted") + return 1 + return _execute(settings) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> Optional[dict]: + """Build the settings dict from flags for a non-interactive run.""" + wav_dir = args.input_dir if args.input_dir is not None else VOICES_DIR + if not wav_dir.is_dir(): + parser.error(f"WAV directory not found: {wav_dir}") + try: + language = normalize_language(args.language or config.LANGUAGE) + except ValueError as exc: + parser.error(str(exc)) + output_path = args.output if args.output is not None \ + else ((_checkout() / "voices.json") if _is_cloned() + else (wav_dir / "voices.json")) + if output_path.exists() and not args.force: + print("[INFO] Aborted; existing voices.json kept") + return None + return { + "do_install": (not _is_installed()) and not args.skip_install, + "do_clone": (not _is_cloned()) and not args.skip_clone, + "wav_dir": wav_dir, + "language": language, + "whisper_model": args.whisper_model or "base", + "output_path": output_path, + "port": args.port if args.port is not None else _config_port(), + "force": args.force, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Set up the faster-qwen3-tts backend: pip install, clone, " + "build voices.json, and sync app/converter/config.py.") + parser.add_argument("input_dir", type=normalize_dir_arg, nargs="?", + default=None, metavar="WAV_DIR", + help="Directory with .wav reference files " + f"(default: {VOICES_DIR}; browsed for in the TUI)") + parser.add_argument("--output", type=Path, default=None, + help="Output path for voices.json (default: " + "./app/faster-qwen3-tts/voices.json, or " + "WAV_DIR/voices.json when not cloned)") + parser.add_argument("--language", type=str, default=None, + help="Language for all voices (default: English; " + "names and short codes accepted)") + parser.add_argument("--whisper-model", type=str, default=None, + choices=WHISPER_MODELS, + help="Whisper model size for transcription " + "(default: base)") + parser.add_argument("--force", action="store_true", + help="Overwrite an existing voices.json without " + "prompting") + parser.add_argument("--port", type=int, default=None, + help="Server port to record in app/converter/config.py " + "(default: the port in FASTER_API_URL)") + parser.add_argument("--skip-install", action="store_true", + help="Do not pip install faster-qwen3-tts[demo]") + parser.add_argument("--skip-clone", action="store_true", + help="Do not clone the faster-qwen3-tts repo") + return parser + + +def detect() -> BackendStatus: + """Detect how far faster-qwen3-tts is set up, plus the launch command.""" + installed = _is_installed() + cloned = _is_cloned() + voices_json = _checkout() / "voices.json" + configured = installed and cloned and voices_json.exists() + running = common.server_running(config.FASTER_API_URL) + details: List[str] = [] + details.append("pip: installed" if installed else + "not installed — run setup to pip install") + details.append(f"checkout: {_checkout()}" if cloned else + f"not cloned — run setup to clone ./app/{FASTER_DIR_NAME}") + details.append(f"voices: {voices_json}" if voices_json.exists() else + "no voices.json — run setup to create one") + launch = "" + servers: List[ServerSpec] = [] + if cloned and voices_json.exists(): + argv = [str(envs.env_python()), + str(_checkout() / "examples" / "openai_server.py"), + "--voices", str(voices_json), "--port", str(_config_port())] + servers = [ServerSpec("faster", config.FASTER_API_URL, argv)] + launch = format_launch_hint(servers) + return BackendStatus("faster", "faster-qwen3-tts", + installed=installed and cloned, + configured=configured, running=running, + details=details, launch_hint=launch, + servers=servers) + + +def _run_voices_only_tui() -> int: + """Rebuild voices.json via the TUI (the "configure" action). + + Runs the same wizard but skips the pip/clone prerequisites so it goes + straight to picking the .wav directory and writing voices.json. + """ + args = build_parser().parse_args([]) + args.skip_install = True + args.skip_clone = True + return run_tui(args) + + +configure_actions: List[ConfigureAction] = [ + ConfigureAction("Rebuild voices.json", _run_voices_only_tui), + ConfigureAction("Reconfigure faster-qwen3-tts", run_tui), +] + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + if _interactive(): + return run_tui(args) + + settings = _collect_from_flags(args, parser) + if settings is None: + return 1 + return _execute(settings) + + +def _interactive() -> bool: + try: + import curses # noqa: F401 + except ImportError: + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +if __name__ == "__main__": + sys.exit(main()) |