diff options
Diffstat (limited to 'app/backends/faster.py')
| -rwxr-xr-x | app/backends/faster.py | 347 |
1 files changed, 158 insertions, 189 deletions
diff --git a/app/backends/faster.py b/app/backends/faster.py index 6e2735c..57e6752 100755 --- a/app/backends/faster.py +++ b/app/backends/faster.py @@ -13,13 +13,12 @@ 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] + [--voice NAME] [--skip-install] [--skip-clone] When the target ``voices.json`` already exists, the TUI wizard runs as a -"modify": it loads the existing voices and pre-fills the language and -wav directory from them instead of prompting to overwrite, asks whether -to only transcribe new voices or re-transcribe everything, and writes -back to the same file. +"modify": it loads the existing voices, pre-fills the language, wav +directory and transcription choice from them instead of prompting to +overwrite, and writes back to the same file. """ import argparse @@ -118,36 +117,25 @@ def load_voices(path: Path) -> dict: return data -def _decide_faster_transcription(wav_files: list, existing_voices: dict, - confirm) -> Optional[dict]: - """Decide which voices to transcribe when a voices.json already exists. +def _decide_faster_transcription(wav_files: list, existing_voices: dict + ) -> tuple: + """Shape the transcription question for the setup form. - CONFIRM asks the yes/no question (returning True/False, or None when the - user backs out). With new .wavs present it offers to transcribe only - those (default Yes); otherwise — and always, per the modify design — it - offers to re-transcribe everything (default No), so a stale transcript - can be refreshed even when every voice is already known. Returns a plan - dict: ``{"mode": "missing"|"all"|"keep", "missing": [...], "existing": - {...}}``, or None when CONFIRM cancelled. + Returns ``(choices, default_mode)``: CHOICES is a list of + ``(label, mode)`` pairs where MODE is ``"missing"`` (only the new + voices), ``"all"`` (re-transcribe everything) or ``"keep"`` + (reuse voices.json untouched). With new .wavs present transcribing + only those is offered first (and is the default); otherwise — and + always, per the modify design — re-transcribing everything stays + available, but keeping the existing file is the default. """ existing = dict(existing_voices) new_wavs = [wav for wav in wav_files if wav.stem not in existing] if new_wavs: - choice = confirm("Existing voices.json found. Only transcribe the " - "new voices?", True) - if choice is None: - return None - if choice: - return {"mode": "missing", "missing": new_wavs, - "existing": existing} - return {"mode": "all", "missing": [], "existing": existing} - choice = confirm("All voices already in voices.json. Re-transcribe " - "anyway?", False) - if choice is None: - return None - if choice: - return {"mode": "all", "missing": [], "existing": existing} - return {"mode": "keep", "missing": [], "existing": existing} + return ([("Only transcribe new voices", "missing"), + ("Re-transcribe all", "all")], "missing") + return ([("Keep the existing voices.json", "keep"), + ("Re-transcribe all", "all")], "keep") def _write_voices_json(output_path: Path, wav_dir: Path, language: str, @@ -185,29 +173,35 @@ def _write_voices_json(output_path: Path, wav_dir: Path, language: str, return voices +def _plan_for(mode: str, wav_dir: Path, existing_voices: dict) -> dict: + """Build the transcription PLAN for the chosen form MODE. + + The plan dict is what ``_write_voices_json`` consumes: "missing" + carries the new .wavs (computed here from the final directory choice) + plus the existing entries; "all"/"keep" only name the mode. + """ + if mode == "missing": + missing = [wav for wav in find_wav_files(wav_dir) + if wav.stem not in existing_voices] + return {"mode": mode, "missing": missing, + "existing": dict(existing_voices)} + return {"mode": mode, "missing": [], "existing": {}} + + def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: - """Linear TUI wizard collecting every faster-setup decision. - - Driven by ``tui.Wizard`` as a stack of screen closures: each screen - shows one widget and returns the next screen, ``Wizard.BACK`` (Esc/q — - pop to the previous screen), or the settings dict. Steps whose value is - already provided by a flag (``--wavs``, ``--language``, - ``--whisper-model``, ``--port``, ``--skip-install``, ``--skip-clone``) - or that do not apply (the transcription plan when there is nothing to - decide) are folded into the ``_after_*`` guards and never become - screens, so Esc always lands on the previous real screen. Esc on the - first screen aborts the wizard. + """Single-form setup: every faster-setup decision on one screen. + + The form mirrors the Generate-audiobooks screen: a Voices-directory + picker, Language, Whisper model, and — on a modify run with an + existing voices.json — which voices to transcribe. There is no port + question: the server port lives in app/converter/config.py (edit it + in the hub's Settings screen). Install and clone happen without + asking; Esc or Cancel aborts the whole setup. """ _GO_BACK = object() - s: dict = {} - - 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 - # An existing voices.json seeds the defaults (modify flow) instead of an - # overwrite prompt; its voices also seed the wav-directory browser. + # An existing voices.json seeds the defaults (modify flow) instead of + # an overwrite prompt; its voices also seed the directory field. default_output = args.output if default_output is None and _is_cloned(): default_output = _checkout() / "voices.json" @@ -215,8 +209,6 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: if default_output is not None and default_output.exists() \ and not args.force: existing_voices = load_voices(default_output) - s["default_output"] = default_output - s["existing_voices"] = existing_voices wav_start = VOICES_DIR if existing_voices: ref_dirs = {Path(voice["ref_audio"]).parent @@ -224,131 +216,117 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: if isinstance(voice, dict) and voice.get("ref_audio")} if len(ref_dirs) == 1: wav_start = next(iter(ref_dirs)) - s["wav_start"] = wav_start # Install and clone happen without asking: when the package or repo is - # missing (and not skipped by flag), the wizard just does it and moves - # to the next screen. - s["do_install"] = (not _is_installed()) and not args.skip_install - s["do_clone"] = (not _is_cloned()) and not args.skip_clone - - def _after_clone(): - if args.input_dir is None: - return screen_wav - s["wav_dir"] = args.input_dir - return _after_wav() - - def screen_wav(): - wav_dir = tui.browse_directory( - stdscr, "Select the directory with your .wav voices", - info=common.wav_dir_info, preview=common.wav_dir_preview, - start=s["wav_start"], back_value=_GO_BACK) - if wav_dir is _GO_BACK: - return tui.Wizard.BACK - s["wav_dir"] = wav_dir - return _after_wav() - - def _after_wav(): - if args.language is None: - return screen_language - s["language"] = args.language - return _after_language() - - def screen_language(): - default_language = config.LANGUAGE - for voice in s["existing_voices"].values(): - if isinstance(voice, dict) and voice.get("language"): - default_language = voice["language"] - break - lang_text = tui.line_edit( - stdscr, "Language", default_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)"], - back_value=_GO_BACK) - if lang_text is _GO_BACK: - return tui.Wizard.BACK - s["language"] = lang_text - return _after_language() - - def _after_language(): - if args.whisper_model is None: - return screen_whisper - s["whisper_model"] = args.whisper_model - return _after_whisper() - - def screen_whisper(): - whisper_model = tui.menu( - stdscr, "Whisper model for transcription", - [(m, m) for m in WHISPER_MODELS], - default_index=WHISPER_MODELS.index("base"), - back_value=_GO_BACK) - if whisper_model is _GO_BACK: - return tui.Wizard.BACK - s["whisper_model"] = whisper_model - return _after_whisper() - - def _after_whisper(): - # Default into the cloned checkout — also when the clone is still - # pending in this run's steps (do_clone): detect() and the server - # launch only read voices.json from there, so a fresh install must - # not leave the file in the wav directory. The wav-directory - # fallback keeps flag-only runs working without any checkout. - s["output_path"] = args.output - if s["output_path"] is None: - if _is_cloned() or s.get("do_clone"): - s["output_path"] = _checkout() / "voices.json" - else: - s["output_path"] = s["wav_dir"] / "voices.json" - wav_files = find_wav_files(s["wav_dir"]) - if wav_files and s["existing_voices"] and not args.force: - return screen_transcription - s["plan"] = {"mode": "all", "missing": [], "existing": {}} - return _after_transcription() - - def screen_transcription(): - # Re-transcribe only new voices (or all of them) — the - # "re-transcribe anyway?" offer appears even when nothing is new. - wav_files = find_wav_files(s["wav_dir"]) - plan = _decide_faster_transcription( - wav_files, s["existing_voices"], _confirm) - if plan is None: - return tui.Wizard.BACK - s["plan"] = plan - return _after_transcription() - - def _after_transcription(): - if args.port is None: - return screen_port - s["port"] = args.port - return _finalize() - - def screen_port(): - 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", - back_value=_GO_BACK) - if port_text is _GO_BACK: - return tui.Wizard.BACK - s["port"] = int(port_text) - return _finalize() - - def _finalize() -> dict: - return { - "do_install": s.get("do_install", False), - "do_clone": s.get("do_clone", False), - "wav_dir": s["wav_dir"], - "language": s["language"], - "whisper_model": s["whisper_model"], - "output_path": s["output_path"], - "port": s["port"], - "force": args.force, - "plan": s["plan"], - } + # missing (and not skipped by flag), the tail just does it afterwards. + do_install = (not _is_installed()) and not args.skip_install + do_clone = (not _is_cloned()) and not args.skip_clone - return tui.Wizard().run(_after_clone()) + default_language = config.LANGUAGE + for voice in existing_voices.values(): + if isinstance(voice, dict) and voice.get("language"): + default_language = voice["language"] + break + + fields: List[dict] = [ + {"key": "wav_dir", "label": "Voices directory", "kind": "dir", + "value": Path(args.input_dir) if args.input_dir is not None + else wav_start}, + {"key": "language", "label": "Language", "kind": "text", + "value": default_language, + "validate": lambda s: None if _try_language(s) + else "Unknown language (e.g. English, en)"}, + {"key": "whisper_model", "label": "Whisper model", "kind": "choice", + "value": args.whisper_model or "base", + "choices": list(WHISPER_MODELS)}, + ] + modifying = bool(existing_voices) and not args.force + if modifying: + # Modify flow: offer keep/new-only/all when the picked directory + # holds .wavs. Recomputed live so switching directories updates it. + + def current_dir(fields_list): + value = next(f["value"] for f in fields_list + if f.get("key") == "wav_dir") + return Path(value) if value else wav_start + + choices_cache: dict = {} + + def transcription_field() -> dict: + wav_files = find_wav_files(current_dir(fields)) + if choices_cache.get("dir") != wav_files: + choices, default = _decide_faster_transcription( + wav_files, existing_voices) + choices_cache.clear() + choices_cache.update({"dir": wav_files, + "choices": choices, + "default": default}) + return choices_cache + + def transcription_choices(_fields_list): + return list(transcription_field()["choices"]) + + def reset_transcription(fields_list) -> None: + field = next(f for f in fields_list + if f.get("key") == "transcription") + modes = [mode for _label, mode + in transcription_field()["choices"]] + if field["value"] not in modes: + field["value"] = transcription_field()["default"] + + fields.append({ + "key": "transcription", "label": "Transcription", + "kind": "choice", + "value": transcription_field()["default"], + "choices": transcription_choices, + "visible": lambda fs: bool(find_wav_files(current_dir(fs))), + "note": "An existing voices.json was found.", + }) + # Changing the directory refreshes the transcription offer; + # tui.form calls the field's `on_change` with the field list. + fields[0]["on_change"] = reset_transcription + + result = tui.form( + stdscr, "Set up faster-qwen3-tts", fields, + buttons=("Continue!", "Cancel"), + start_on_buttons=False, back_value=_GO_BACK) + if result is _GO_BACK: + return None + + wav_dir = Path(result["wav_dir"]) + language = normalize_language(result["language"]) + whisper_model = result["whisper_model"] + if not modifying: + plan = {"mode": "all", "missing": [], "existing": {}} + elif find_wav_files(wav_dir): + plan = _plan_for(result["transcription"], wav_dir, existing_voices) + else: + # Directory without .wavs on a modify run: keep the existing file. + plan = {"mode": "keep", "missing": [], + "existing": dict(existing_voices)} + + # Default into the cloned checkout — also when the clone is still + # pending in this run's steps (do_clone): detect() and the server + # launch only read voices.json from there, so a fresh install must + # not leave the file in the wav directory. The wav-directory + # fallback keeps runs working without any checkout. + output_path = default_output + if output_path is None: + if _is_cloned() or do_clone: + output_path = _checkout() / "voices.json" + else: + output_path = wav_dir / "voices.json" + + return { + "do_install": do_install, + "do_clone": do_clone, + "wav_dir": wav_dir, + "language": language, + "whisper_model": whisper_model, + "output_path": output_path, + "force": args.force, + "plan": plan, + } def _try_language(value: str) -> bool: @@ -403,15 +381,9 @@ def _execute_steps(settings: dict) -> List[taskview.TaskStep]: 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") + # Sync app/converter/config.py default voice. (The server port is + # not touched here: it lives in FASTER_API_URL, edited in the + # Settings screen.) default_voice = next(iter(voices)) if default_voice != config.FASTER_VOICE: if common.update_config_value("FASTER_VOICE", default_voice): @@ -420,7 +392,7 @@ def _execute_steps(settings: dict) -> List[taskview.TaskStep]: print("[WARNING] Could not update FASTER_VOICE; edit " "app/converter/config.py by hand") - _print_launch_hint(settings["output_path"], port) + _print_launch_hint(settings["output_path"]) return 0 steps.append(taskview.TaskStep( "Write voices.json & sync config", write)) @@ -433,7 +405,7 @@ def _execute(settings: dict) -> int: return taskview.run_steps_inline(_execute_steps(settings)) -def _print_launch_hint(voices_path: Path, port: int) -> None: +def _print_launch_hint(voices_path: Path) -> None: """Remediation only (troubleshooting): what's missing when not cloned. The hub starts and stops the server itself, so a working install gets @@ -442,7 +414,8 @@ def _print_launch_hint(voices_path: Path, port: int) -> None: if _is_cloned(): return print("[INFO] Clone faster-qwen3-tts to get examples/openai_server.py,") - print(f" then run it with --voices {voices_path} --port {port}") + print(f" then run it with --voices {voices_path} " + f"--port {_config_port()}") def setup_screen(stdscr) -> int: @@ -496,7 +469,6 @@ def _collect_from_flags(args: argparse.Namespace, "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, "plan": {"mode": "all", "missing": [], "existing": {}}, } @@ -525,9 +497,6 @@ def build_parser() -> argparse.ArgumentParser: help="Overwrite an existing voices.json without " "prompting; in the TUI, re-transcribe every " "voice instead of reusing the existing file") - 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", |