aboutsummaryrefslogtreecommitdiff
path: root/app/backends/faster.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-26 17:46:48 -0400
committerhistoria <historiavg@proton.me>2026-08-26 17:46:48 -0400
commit6ccb6d443d2fb871b43d96ea61a95bc3e6a92355 (patch)
treed692ddccd6ec212cf4ebabb25a85e83af479d0fe /app/backends/faster.py
parentc147087c9d4707bffaeee58d390653637a21cce8 (diff)
downloadtts-audiobook-generator-6ccb6d443d2fb871b43d96ea61a95bc3e6a92355.tar.gz
feat: combined install/configure tui screens into one menu, removed extraneous wizard screens
Diffstat (limited to 'app/backends/faster.py')
-rwxr-xr-xapp/backends/faster.py347
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",