aboutsummaryrefslogtreecommitdiff
path: root/app/backends
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-24 18:57:09 -0400
committerhistoria <historiavg@proton.me>2026-08-24 18:57:09 -0400
commit0522e73b68291af62e43c387ab8f9b8ffa2cab47 (patch)
tree1800168c1ed5de7540624265945faba8251a7cab /app/backends
parentd950fc8e64ee508334e608f6045d687d73a464be (diff)
downloadtts-audiobook-generator-0522e73b68291af62e43c387ab8f9b8ffa2cab47.tar.gz
feat: configure [backend] loads existing backend config rather than just overwriting it
Diffstat (limited to 'app/backends')
-rw-r--r--app/backends/__init__.py32
-rwxr-xr-xapp/backends/audiocpp.py465
-rw-r--r--app/backends/common.py10
-rw-r--r--app/backends/envs.py14
-rwxr-xr-xapp/backends/faster.py161
-rw-r--r--app/backends/qwen.py21
6 files changed, 546 insertions, 157 deletions
diff --git a/app/backends/__init__.py b/app/backends/__init__.py
index 488c36e..3dff306 100644
--- a/app/backends/__init__.py
+++ b/app/backends/__init__.py
@@ -1,12 +1,12 @@
"""Registry of the TTS backends the audiobook generator can talk to.
Each backend (audio.cpp, qwen, faster) lives in its own module and owns
-its setup wizard, its status detection, and the launch command it prints
-once configured. This package aggregates them into a single registry so
-``audiobook.py``'s TUI hub and future tools can iterate backends without
-hardcoding their names: ``backends.detect_all()`` reports which are set
+its setup wizard, its status detection, its uninstaller, and the launch
+command it prints once configured. This package aggregates them into a single
+registry so ``audiobook.py``'s TUI hub and future tools can iterate backends
+without hardcoding their names: ``backends.detect_all()`` reports which are set
up (and whether their server is currently running), and the registry
-drives the hub's setup/configure menus.
+drives the hub's "Configure backends" menu.
The registry is built lazily on the first call to ``get``/``detect_all``/
``detect`` (not at package import time), because the backend modules pull
@@ -17,9 +17,8 @@ importing this package must stay cheap and dependency-free.
Adding a backend: create ``backends/<name>.py`` exposing
``detect() -> BackendStatus``, ``run_tui() -> int`` and
-``configure_actions: list[ConfigureAction]``, then append a ``BackendInfo`` in
-``_build_registry`` below. ``audiobook.py`` and the hub pick it up
-automatically.
+``uninstall() -> int``, then append a ``BackendInfo`` in ``_build_registry``
+below. ``audiobook.py`` and the hub pick it up automatically.
"""
import shlex
@@ -131,20 +130,13 @@ def format_launch_hint(servers: List[ServerSpec]) -> str:
@dataclass
-class ConfigureAction:
- """A per-backend "configure" menu entry (e.g. "New server.json")."""
- label: str
- run: Callable[[], int]
-
-
-@dataclass
class BackendInfo:
- """One registry entry: identity, detector, setup wizard, configure menu."""
+ """One registry entry: identity, detector, setup wizard, uninstaller."""
key: str
label: str
detect: Callable[[], BackendStatus]
setup_tui: Callable[[], int]
- configure_actions: List[ConfigureAction] = field(default_factory=list)
+ uninstall: Callable[[], int] = lambda: 0
REGISTRY: List[BackendInfo] = []
@@ -162,21 +154,21 @@ def _build_registry() -> None:
label="audio.cpp",
detect=audiocpp.detect,
setup_tui=audiocpp.run_tui,
- configure_actions=audiocpp.configure_actions,
+ uninstall=audiocpp.uninstall,
))
REGISTRY.append(BackendInfo(
key="qwen",
label="qwen-tts",
detect=qwen.detect,
setup_tui=qwen.run_tui,
- configure_actions=qwen.configure_actions,
+ uninstall=qwen.uninstall,
))
REGISTRY.append(BackendInfo(
key="faster",
label="faster-qwen3-tts",
detect=faster.detect,
setup_tui=faster.run_tui,
- configure_actions=faster.configure_actions,
+ uninstall=faster.uninstall,
))
for info in REGISTRY:
_BY_KEY[info.key] = info
diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py
index a502c8f..f74f726 100755
--- a/app/backends/audiocpp.py
+++ b/app/backends/audiocpp.py
@@ -26,12 +26,20 @@ Usage:
With no flags and a terminal, the TUI wizard runs. Without a terminal
(or with all flags supplied), it runs non-interactively from the flags;
any missing required value is a hard error with a remediation hint.
+
+When the target ``server.json`` already exists, the TUI wizard runs as a
+"modify": it loads the existing models, host, port, backend, lazy-load
+and voice directory and pre-fills the screens with them (the model tree
+opens with the installed models already checked) instead of prompting to
+overwrite, and offers to delete already-downloaded models that are no
+longer selected.
"""
import argparse
import json
import os
import re
+import shutil
import subprocess
import sys
import urllib.parse
@@ -44,7 +52,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from backends import (
BackendStatus,
- ConfigureAction,
ServerSpec,
common,
format_launch_hint,
@@ -721,14 +728,18 @@ def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None
def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]],
catalog_by_family: Dict[str, dict],
task_picker: Callable[[str], str],
- id_picker: Callable[[str, str, str], str]
+ id_picker: Callable[[str, str, str], str],
+ known_tasks: Optional[Dict[Tuple[str, str], str]] = None
) -> Tuple[List[dict], List[str], List[Tuple[str, str]],
List[str], bool]:
"""Build server.json model entries from the selected families/packages.
TASK_PICKER is called for each design package to choose vdes/tts;
- ID_PICKER resolves a duplicate server entry id. Returns (model_entries,
- entry_ids, install_guidance, design_entry_ids, include_clone).
+ ID_PICKER resolves a duplicate server entry id. KNOWN_TASKS maps
+ ``(family, target_directory)`` to a previously-stored task ("tts" or
+ "vdes") so a modify run preserves how a design package was hosted
+ instead of re-asking. Returns (model_entries, entry_ids,
+ install_guidance, design_entry_ids, include_clone).
"""
model_entries: List[dict] = []
entry_ids: List[str] = []
@@ -739,7 +750,13 @@ def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]],
entry = catalog_by_family[family]
include_clone = include_clone or entry["clone_capable"]
for opt in chosen[family]:
- task = task_picker(opt["install_id"]) if opt["design"] else TASK_TTS
+ if opt["design"]:
+ task = known_tasks.get((family, opt["target_directory"])) \
+ if known_tasks else None
+ if task is None:
+ task = task_picker(opt["install_id"])
+ else:
+ task = TASK_TTS
base_id = (f"{entry['preferred_id']}-design"
if task == TASK_VDES else entry["preferred_id"])
model_id = base_id
@@ -861,8 +878,6 @@ def _build_tree_families(catalog: List[dict]) -> List[dict]:
if "design" in entry["tasks"]:
capabilities.append("design")
name = entry["display_name"]
- if name != entry["family"]:
- name = f"{name} ({entry['family']})"
options = []
for opt in package_dir_options(entry):
options.append({
@@ -899,21 +914,15 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
step = 0
while True:
if step == 0:
- # Checkout browser + the output path/overwrite confirmation. The
- # browser asks for the checkout root and finds model_specs/ inside
- # it (picking the model_specs directory itself works too — its
- # parent is used). A highlighted subdirectory named "audio.cpp"
- # that already contains model_specs/ is auto-accepted on
- # Enter/Right, skipping the "[ Use this directory ]" step.
- # Pressing Esc on an overwrite confirmation returns here instead
- # of aborting: the browser then restarts inside the previously
- # accepted checkout with auto-accept disabled, so a wrong guess
- # can be corrected. An explicit --audiocpp-dir flag has no
- # browser to return to, so Esc still aborts there. Esc on the
- # browser itself is the first step, so it aborts the wizard.
+ # Checkout browser. The browser asks for the checkout root and
+ # finds model_specs/ inside it (picking the model_specs directory
+ # itself works too — its parent is used). A highlighted
+ # subdirectory named "audio.cpp" that already contains
+ # model_specs/ is auto-accepted on Enter/Right, skipping the
+ # "[ Use this directory ]" step. Esc on the browser is the first
+ # step, so it aborts the wizard.
auto_accept = True
browser_start: Path = Path.cwd()
- force_browse = False
def do_browse():
return tui.browse_directory(
@@ -932,44 +941,36 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
while True:
audiocpp_dir = args.audiocpp_dir
- if audiocpp_dir is None and not force_browse:
+ if audiocpp_dir is None:
audiocpp_dir = find_local_checkout()
- if force_browse:
- audiocpp_dir = None
if audiocpp_dir is None:
- if force_browse:
- # Esc on an overwrite confirmation came back here: go
- # straight back into the browser inside the previously
- # accepted checkout (auto-accept disabled).
- audiocpp_dir = do_browse()
+ # No checkout found anywhere: offer to clone one into
+ # ./app/audio.cpp or browse for an existing checkout.
+ # Esc on this first menu aborts the wizard.
+ choice = tui.menu(
+ stdscr, "No audio.cpp checkout found",
+ [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} "
+ f"(from {AUDIOCPP_GIT_URL})", "clone"),
+ ("Browse for an existing checkout", "browse")],
+ help_lines=[
+ "audio.cpp hosts the TTS model families "
+ "this generator uses.",
+ "Clone it into the project's app "
+ "directory, or point at an existing "
+ "checkout."])
+ if choice == "clone":
+ target = APP_DIR / AUDIOCPP_DIR_NAME
+ with tui.suspend(stdscr):
+ rc = common.git_clone(AUDIOCPP_GIT_URL,
+ target)
+ if rc != 0:
+ raise _TuiError(
+ f"git clone failed (exit {rc}). Clone "
+ f"audio.cpp manually: git clone "
+ f"{AUDIOCPP_GIT_URL} {target}")
+ audiocpp_dir = target
else:
- # No checkout found anywhere: offer to clone one into
- # ./app/audio.cpp or browse for an existing checkout.
- # Esc on this first menu aborts the wizard.
- choice = tui.menu(
- stdscr, "No audio.cpp checkout found",
- [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} "
- f"(from {AUDIOCPP_GIT_URL})", "clone"),
- ("Browse for an existing checkout", "browse")],
- help_lines=[
- "audio.cpp hosts the TTS model families "
- "this generator uses.",
- "Clone it into the project's app "
- "directory, or point at an existing "
- "checkout."])
- if choice == "clone":
- target = APP_DIR / AUDIOCPP_DIR_NAME
- with tui.suspend(stdscr):
- rc = common.git_clone(AUDIOCPP_GIT_URL,
- target)
- if rc != 0:
- raise _TuiError(
- f"git clone failed (exit {rc}). Clone "
- f"audio.cpp manually: git clone "
- f"{AUDIOCPP_GIT_URL} {target}")
- audiocpp_dir = target
- else:
- audiocpp_dir = do_browse()
+ audiocpp_dir = do_browse()
audiocpp_dir = Path(audiocpp_dir).resolve()
if not audiocpp_dir.is_dir():
raise _TuiError(f"audio.cpp checkout not found: "
@@ -993,37 +994,28 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
output_path = args.output if args.output is not None \
else audiocpp_dir / "server.json"
- esc_back = args.audiocpp_dir is None
- went_back = False
- if not args.force and output_path.exists():
- decision = tui.confirm(
- stdscr, f"{output_path} already exists. Overwrite?",
- default=True,
- cancel_value=_GO_BACK if esc_back else None)
- if decision is _GO_BACK:
- went_back = True
- elif decision is False:
- if args.output is None:
- output_path = Path.cwd() / "server.json"
- if output_path.exists():
- decision = tui.confirm(
- stdscr,
- f"{output_path} already exists. "
- "Overwrite?",
- default=True,
- cancel_value=_GO_BACK if esc_back else None)
- if decision is _GO_BACK:
- went_back = True
- elif decision is False:
- return None
- else:
- return None
- if went_back:
- auto_accept = False
- browser_start = audiocpp_dir
- force_browse = True
- continue
break
+
+ # Modify flow: an existing server.json seeds the wizard's
+ # screens instead of being overwritten from scratch (an explicit
+ # --force still starts fresh).
+ existing_config = load_server_config(output_path) \
+ if not args.force else None
+ if existing_config is not None:
+ existing_selected, existing_tasks = \
+ server_config_selections(existing_config, catalog)
+ else:
+ existing_selected, existing_tasks = {}, {}
+ existing_host = existing_config.get("host") \
+ if existing_config else None
+ existing_port = existing_config.get("port") \
+ if existing_config else None
+ existing_backend = existing_config.get("backend") \
+ if existing_config else None
+ existing_lazy = existing_config.get("lazy_load") \
+ if existing_config else None
+ existing_voice_dir = existing_config.get("voice_dir") \
+ if existing_config else None
detected_backend = detect_backend(audiocpp_dir)
step = 1
continue
@@ -1048,10 +1040,24 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
catalog_by_family[family]) if opt["recommended"]]
else:
tree_families = _build_tree_families(catalog)
+ # Modify flow: pre-check the models an existing server.json
+ # hosts, so the tree opens as a "modify" list rather than a
+ # fresh one.
+ checked_set = set()
+ for family, dirs in existing_selected.items():
+ if family not in catalog_by_family:
+ continue
+ family_index = catalog.index(catalog_by_family[family])
+ valid_dirs = {opt["target_directory"]
+ for opt in package_dir_options(
+ catalog_by_family[family])}
+ for target in dirs:
+ if target in valid_dirs:
+ checked_set.add((family_index, target))
picked = tui.checkbox_tree(
stdscr, "Select TTS model families to host",
tree_families, expand_all=args.all_packages,
- back_value=_GO_BACK)
+ back_value=_GO_BACK, checked=checked_set)
if picked is _GO_BACK:
step = 0
continue
@@ -1100,7 +1106,7 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
model_entries, entry_ids, install_guidance, \
design_entry_ids, include_clone = _build_entries(
family_keys, chosen, catalog_by_family,
- task_picker, id_picker)
+ task_picker, id_picker, known_tasks=existing_tasks)
except _GoBack:
step = 1
continue
@@ -1114,7 +1120,9 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
host = args.host
else:
host = tui.line_edit(
- stdscr, "Bind host", DEFAULT_HOST,
+ stdscr, "Bind host",
+ existing_host if isinstance(existing_host, str)
+ else DEFAULT_HOST,
help_lines=["The IP address audiocpp will be hosted on",
"127.0.0.1 (this machine) is probably "
"correct"], back_value=_GO_BACK)
@@ -1125,7 +1133,9 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
port = args.port
else:
port_text = tui.line_edit(
- stdscr, "Port", str(config_port()),
+ stdscr, "Port",
+ str(existing_port) if isinstance(existing_port, int)
+ else str(config_port()),
validate=lambda s: None if (s.isdigit()
and 1 <= int(s) <= 65535)
else "Enter a port number between 1 and 65535",
@@ -1154,6 +1164,11 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
# Already built: use the detected backend, no menu, no build.
backend = detected_backend
build = False
+ elif existing_backend in BACKENDS:
+ # Modify flow: keep the backend an existing server.json
+ # records (already configured, no rebuild needed).
+ backend = existing_backend
+ build = False
else:
backend_options, backend_default = _backend_options(None)
backend = tui.menu(
@@ -1173,6 +1188,8 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
step = 2
continue
default_lazy = len(model_entries) > 1
+ if isinstance(existing_lazy, bool):
+ default_lazy = existing_lazy
if args.lazy_load:
lazy_load = True
else:
@@ -1192,6 +1209,10 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
wav_dir = args.input_dir
elif include_clone:
wav_start = detect_wav_dir(audiocpp_dir, TTS_ROOT)
+ # Modify flow: an existing voice_dir seeds the browser so the
+ # user can accept it on Enter instead of re-navigating.
+ if isinstance(existing_voice_dir, str) and existing_voice_dir:
+ wav_start = Path(existing_voice_dir)
wav_dir = tui.browse_directory(
stdscr, "Select the directory with your .wav voices",
info=_wav_dir_info, preview=_wav_dir_preview,
@@ -1240,16 +1261,35 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
if sync_model_ids is _GO_BACK:
step = 5
continue
+ step = 7
+ continue
+
+ if step == 7:
+ # Delete unused models: already-downloaded models that the new
+ # selection no longer hosts. Only offered in the modify flow (an
+ # existing config was loaded), since a fresh --force run is an
+ # explicit overwrite. Esc falls back to the model-id sync (6).
+ new_paths = {entry["path"] for entry in model_entries}
+ unused_entries = unused_installed_entries(output_path, new_paths) \
+ if existing_config is not None else []
+ delete_unused = False
+ if unused_entries:
+ delete_unused = tui.confirm(
+ stdscr, "Delete unused models?", default=False,
+ cancel_value=_GO_BACK)
+ if delete_unused is _GO_BACK:
+ step = 6
+ continue
step = 8
continue
if step == 8:
# Automatic model download (or print the install commands). Esc
- # falls back to the model-id sync (step 6).
+ # falls back to the delete-unused step (7).
try:
download = _decide_download(audiocpp_dir, ask_confirm)
except _GoBack:
- step = 6
+ step = 7
continue
return {
"audiocpp_dir": audiocpp_dir,
@@ -1273,9 +1313,64 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser
"wav_dir": wav_dir,
"plan": plan,
"download": download,
+ "delete_unused": delete_unused,
+ "unused_entries": unused_entries,
}
+def load_server_config(server_json: Path) -> Optional[dict]:
+ """Read server.json into a dict, or None when it cannot be used.
+
+ Returns None for a missing file, unreadable content, or a non-dict
+ document. Used by the wizard's modify flow to pre-fill its screens
+ from an existing config instead of prompting to overwrite it.
+ """
+ if not server_json.exists():
+ return None
+ try:
+ data = json.loads(server_json.read_text(encoding="utf-8"))
+ except (OSError, ValueError):
+ return None
+ if not isinstance(data, dict):
+ return None
+ return data
+
+
+def server_config_selections(server_config: dict,
+ catalog: List[dict]
+ ) -> Tuple[Dict[str, List[str]],
+ Dict[Tuple[str, str], str]]:
+ """Map an existing server.json's models back to catalog selections.
+
+ Returns ``(selected_dirs, tasks)``: ``selected_dirs`` maps a catalog
+ family to the target directories it hosts (``models/<target>`` paths
+ with the ``models/`` prefix stripped, in server.json order), and
+ ``tasks`` maps ``(family, target_directory)`` to the entry's task
+ (``"tts"`` or ``"vdes"``) so the wizard can preserve how design
+ packages were hosted. Entries whose family is not in the CATALOG are
+ ignored — the wizard cannot offer them again.
+ """
+ families = {entry["family"] for entry in catalog}
+ selected_dirs: Dict[str, List[str]] = {}
+ tasks: Dict[Tuple[str, str], str] = {}
+ for entry in server_config.get("models") or []:
+ if not isinstance(entry, dict):
+ continue
+ family = entry.get("family")
+ if not isinstance(family, str) or family not in families:
+ continue
+ path = entry.get("path")
+ if not isinstance(path, str):
+ continue
+ target = path[len("models/"):] if path.startswith("models/") else path
+ if family not in selected_dirs:
+ selected_dirs[family] = []
+ if target not in selected_dirs[family]:
+ selected_dirs[family].append(target)
+ tasks[(family, target)] = str(entry.get("task") or TASK_TTS)
+ return selected_dirs, tasks
+
+
def _model_path_present(path: Path) -> bool:
"""True when a server.json model path holds actual model files.
@@ -1322,6 +1417,70 @@ def missing_model_entries(server_json: Path) -> List[dict]:
return missing
+def _install_id_by_path(audiocpp_dir: Path) -> Dict[str, str]:
+ """Map ``models/<target_directory>`` -> catalog install id.
+
+ The catalog package that installs a model is derived from the
+ ``default_path`` of each TTS family; an entry whose path matches no
+ catalog package has no install id.
+ """
+ by_path: Dict[str, str] = {}
+ try:
+ for entry in load_model_catalog(audiocpp_dir):
+ by_path[entry["default_path"]] = entry["install_id"]
+ except (NotADirectoryError, OSError):
+ pass
+ return by_path
+
+
+def installed_model_entries(server_json: Path) -> List[dict]:
+ """Return the server.json model entries whose files ARE on disk.
+
+ The complement of ``missing_model_entries``: each returned entry carries
+ the entry ``id`` and ``rel`` (the configured path string), resolved
+ exactly like ``missing_model_entries`` (relative against the server.json's
+ directory). Used by the wizard's "Delete unused models?" step to find
+ already-downloaded models that were unselected.
+ """
+ try:
+ data = json.loads(server_json.read_text(encoding="utf-8"))
+ except (OSError, ValueError):
+ return []
+ if not isinstance(data, dict):
+ return []
+ base = server_json.parent
+ installed: List[dict] = []
+ for entry in data.get("models") or []:
+ if not isinstance(entry, dict):
+ continue
+ rel = entry.get("path")
+ if not isinstance(rel, str) or not rel:
+ continue
+ path = Path(rel) if Path(rel).is_absolute() else base / rel
+ if _model_path_present(path):
+ installed.append({"id": str(entry.get("id") or rel), "rel": rel})
+ return installed
+
+
+def missing_model_install_guidance(audiocpp_dir: Path,
+ missing: List[dict]) -> List[Tuple[str, str]]:
+ """Map MISSING model entries to (display name, install id) pairs.
+
+ The install id is derived from each entry's configured path via the
+ catalog (see ``_install_id_by_path``); entries whose path matches no
+ catalog package are skipped (there is no ``model_manager_v2.py install``
+ command for them). Feeds ``_install_models`` for the "Download Missing
+ Models" action.
+ """
+ by_path = _install_id_by_path(audiocpp_dir)
+ guidance: List[Tuple[str, str]] = []
+ for item in missing:
+ install_id = by_path.get(item["rel"])
+ if install_id:
+ guidance.append((item["id"], install_id))
+ return guidance
+
+
def model_install_hints(audiocpp_dir: Path,
missing: List[dict]) -> List[str]:
"""Remediation lines for MISSING model entries (see missing_model_entries).
@@ -1331,12 +1490,7 @@ def model_install_hints(audiocpp_dir: Path,
carries the exact ``model_manager_v2.py install`` command; entries whose
directory matches no catalog package just name the path.
"""
- by_path: Dict[str, str] = {}
- try:
- for entry in load_model_catalog(audiocpp_dir):
- by_path[entry["default_path"]] = entry["install_id"]
- except (NotADirectoryError, OSError):
- pass
+ by_path = _install_id_by_path(audiocpp_dir)
hints: List[str] = []
for item in missing:
install_id = by_path.get(item["rel"])
@@ -1348,6 +1502,104 @@ def model_install_hints(audiocpp_dir: Path,
return hints
+def install_models(audiocpp_dir: Path,
+ guidance: List[Tuple[str, str]]) -> None:
+ """Download the (display name, install id) models via the helper script.
+
+ Runs ``model_manager_v2.py install`` for each de-duped install id in the
+ checkout, streaming to the console; a failing install is reported as a
+ warning and does not abort the rest. Used by the hub's "Download Missing
+ Models" action (see ``missing_model_install_guidance`` for the mapping).
+ """
+ _install_models(audiocpp_dir, guidance, download=True)
+
+
+def hand_install_guidance(audiocpp_dir: Path,
+ missing: List[dict]) -> str:
+ """Explain how to install MISSING model entries by hand.
+
+ Returns a multi-line message listing each missing model's id and the
+ path its files must be placed in (``rel``, resolved against the
+ AUDIOCPP_DIR checkout). Used when the missing models cannot be mapped to
+ a ``model_manager_v2.py install`` command, so the user still knows what
+ to download and where to put it.
+ """
+ lines = [
+ "None of the missing models map to a model_manager_v2.py install "
+ "command.",
+ "Download them by hand and place the files at these paths:",
+ ]
+ for item in missing:
+ lines.append(f" {item['id']} -> {item['rel']}")
+ lines.append(f"(paths are relative to {audiocpp_dir})")
+ return "\n".join(lines)
+
+
+def unused_installed_entries(server_json: Path,
+ new_paths: Set[str]) -> List[dict]:
+ """Return installed server.json entries whose path is not in NEW_PATHS.
+
+ The already-downloaded models (see ``installed_model_entries``) that the
+ new selection does not host any more — the candidates for the wizard's
+ "Delete unused models?" prompt. Entries whose files are not on disk are
+ never listed (there is nothing to delete).
+ """
+ return [entry for entry in installed_model_entries(server_json)
+ if entry["rel"] not in new_paths]
+
+
+def delete_model_files(server_json: Path, entries: List[dict]) -> int:
+ """Remove the on-disk model files for ENTRIES ({id, rel}) from disk.
+
+ Each entry's ``rel`` is resolved exactly like the server resolves it
+ (relative against ``server_json``'s directory; absolute paths honored),
+ then removed as a directory tree or a single file. Missing entries are
+ ignored. Returns the number of paths removed. Used by the wizard's
+ "Delete unused models?" step — the regenerated server.json already only
+ lists the kept models, so no entry cleanup is needed here.
+ """
+ base = server_json.parent
+ removed = 0
+ for item in entries:
+ rel = item.get("rel")
+ if not isinstance(rel, str) or not rel:
+ continue
+ path = Path(rel) if Path(rel).is_absolute() else base / rel
+ try:
+ if not path.exists():
+ continue
+ if path.is_dir():
+ shutil.rmtree(path, ignore_errors=True)
+ else:
+ path.unlink()
+ except OSError as exc:
+ print(f"[WARNING] Could not remove {path}: {exc}")
+ continue
+ print(f"[OK] Removed unused model {path}")
+ removed += 1
+ return removed
+
+
+def uninstall() -> int:
+ """Remove the audio.cpp backend entirely: stop its server, delete the checkout.
+
+ The checkout (``app/audio.cpp``, or wherever ``find_local_checkout``
+ resolves it) holds the built binary, the downloaded models, and the
+ server.json, so removing the directory uninstalls the backend. A running
+ server this tool started is stopped first (best-effort). Returns the exit
+ code.
+ """
+ servers.stop("audiocpp")
+ checkout = find_local_checkout()
+ if checkout is None:
+ print("[INFO] No audio.cpp checkout to remove.")
+ return 0
+ print(f"[INFO] Removing audio.cpp checkout {checkout}...")
+ shutil.rmtree(checkout, ignore_errors=True)
+ print("[OK] audio.cpp removed.")
+ return 0
+
+
def find_local_checkout() -> Optional[Path]:
"""Best-effort location of an audio.cpp checkout with model_specs.
@@ -1555,6 +1807,15 @@ def _execute(settings: dict, args: argparse.Namespace) -> int:
settings["host"], settings["port"], settings["backend"],
settings["lazy_load"], transcripts, write_prompt)
+ # Delete-unused cleanup (modify flow): remove the already-downloaded
+ # models the new selection dropped. The regenerated server.json already
+ # only lists the kept entries.
+ if settings.get("delete_unused"):
+ removed = delete_model_files(settings["output_path"],
+ settings["unused_entries"])
+ print(f"[OK] Deleted {removed} unused model "
+ f"{'entry' if removed == 1 else 'entries'} from disk.")
+
if len(settings["entry_ids"]) == 1:
_offer_config_model_id_sync(settings["entry_ids"][0],
settings["sync_model_ids"])
@@ -1806,7 +2067,9 @@ def build_parser() -> argparse.ArgumentParser:
"(default: base)")
parser.add_argument("--force", action="store_true",
help="Overwrite the output file (and prompt_text) "
- "without prompting")
+ "without prompting; in the TUI, start the "
+ "wizard fresh instead of loading the existing "
+ "server.json")
parser.add_argument("--download", action="store_true",
help="Run model_manager_v2.py install for each hosted "
"model automatically (default: print the commands "
@@ -1893,12 +2156,6 @@ def _detect_remote(managed: bool = False) -> Tuple[bool, dict]:
return False, {}
-configure_actions: List[ConfigureAction] = [
- ConfigureAction("Reconfigure audio.cpp (models, voices, server.json)",
- run_tui),
-]
-
-
def main() -> int:
parser = build_parser()
args = parser.parse_args()
diff --git a/app/backends/common.py b/app/backends/common.py
index d5e1b6b..08c8863 100644
--- a/app/backends/common.py
+++ b/app/backends/common.py
@@ -299,3 +299,13 @@ def pip_install(packages: List[str]) -> int:
"""
from backends import envs
return envs.pip_install(packages)
+
+
+def pip_uninstall(packages: List[str]) -> int:
+ """pip uninstall PACKAGES from the managed venv. Returns exit code.
+
+ Delegates to ``backends.envs.pip_uninstall`` (local import to avoid a
+ circular import). Used by the backends' ``uninstall`` action.
+ """
+ from backends import envs
+ return envs.pip_uninstall(packages)
diff --git a/app/backends/envs.py b/app/backends/envs.py
index 6e5b6cc..5a51a33 100644
--- a/app/backends/envs.py
+++ b/app/backends/envs.py
@@ -107,6 +107,20 @@ def pip_install(packages: List[str]) -> int:
[str(env_python()), "-m", "pip", "install", *packages])
+def pip_uninstall(packages: List[str]) -> int:
+ """pip uninstall PACKAGES from the venv. Returns pip's exit code.
+
+ Used by the backends' ``uninstall`` action to remove pip-installed TTS
+ packages from the managed environment. A missing env is a no-op (there
+ is nothing to uninstall from), reported as success.
+ """
+ if not env_exists():
+ return 0
+ print(f"[INFO] pip uninstall {' '.join(packages)} from {ENV_DIR}...")
+ return common.run_console_subprocess(
+ [str(env_python()), "-m", "pip", "uninstall", "-y", *packages])
+
+
def module_available(module: str) -> bool:
"""True when MODULE imports inside the venv (e.g. qwen_tts, faster_qwen3_tts).
diff --git a/app/backends/faster.py b/app/backends/faster.py
index e1249ca..7e1be74 100755
--- a/app/backends/faster.py
+++ b/app/backends/faster.py
@@ -14,10 +14,17 @@ 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]
+
+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.
"""
import argparse
import json
+import shutil
import sys
from pathlib import Path
from typing import List, Optional
@@ -26,7 +33,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from backends import (
BackendStatus,
- ConfigureAction,
ServerSpec,
common,
envs,
@@ -95,9 +101,68 @@ def build_voices(wav_files: list, language: str, whisper_model: str) -> dict:
return voices
+def load_voices(path: Path) -> dict:
+ """Read voices.json into a name -> voice-entry dict, or {} when unusable.
+
+ Returns {} for a missing file, unreadable content, or a non-dict
+ document. Used by the wizard's modify flow to seed its defaults from an
+ existing voices.json instead of prompting to overwrite it.
+ """
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, ValueError):
+ return {}
+ if not isinstance(data, dict):
+ return {}
+ 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.
+
+ 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.
+ """
+ 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}
+
+
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."""
+ whisper_model: str, plan: Optional[dict]) -> Optional[dict]:
+ """Transcribe the wav dir and write voices.json; return the voices dict.
+
+ PLAN (built by ``_decide_faster_transcription`` in the wizard, or an
+ "all" plan for a fresh/flag run) decides whether every voice is
+ re-transcribed ("all"), only the new ones ("missing" — merged into the
+ existing entries), or nothing changes ("keep" — the existing file is
+ left untouched and returned as-is). None (cancelled) writes nothing.
+ """
+ if plan is None:
+ return None
+ if plan["mode"] == "keep":
+ return dict(plan["existing"])
wav_files = find_wav_files(wav_dir)
if not wav_files:
print(f"[ERROR] No .wav files found in {wav_dir}")
@@ -106,7 +171,11 @@ def _write_voices_json(output_path: Path, wav_dir: Path, language: str,
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)
+ if plan["mode"] == "missing":
+ voices = dict(plan["existing"])
+ voices.update(build_voices(plan["missing"], language, whisper_model))
+ else:
+ 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")
@@ -142,17 +211,39 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
return None
do_clone = choice
- # Step 2: voices.json — wav dir, language, whisper model, output path.
+ # Step 2: voices.json — an existing one seeds the defaults (modify flow)
+ # instead of an overwrite prompt.
+ existing_voices = {}
+ default_output = args.output
+ if default_output is None and _is_cloned():
+ default_output = _checkout() / "voices.json"
+ if default_output is not None and default_output.exists() \
+ and not args.force:
+ existing_voices = load_voices(default_output)
+
+ wav_start = VOICES_DIR
+ if existing_voices:
+ ref_dirs = {Path(voice["ref_audio"]).parent
+ for voice in existing_voices.values()
+ if isinstance(voice, dict) and voice.get("ref_audio")}
+ if len(ref_dirs) == 1:
+ wav_start = next(iter(ref_dirs))
+
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)
+ start=wav_start)
language = args.language
if language is None:
+ default_language = config.LANGUAGE
+ for voice in existing_voices.values():
+ if isinstance(voice, dict) and voice.get("language"):
+ default_language = voice["language"]
+ break
lang_text = tui.line_edit(
- stdscr, "Language", config.LANGUAGE,
+ 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 "
@@ -170,12 +261,16 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
# 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"
+
+ # Transcription plan: re-transcribe only new voices (or all of them) —
+ # the "re-transcribe anyway?" offer appears even when nothing is new.
+ plan: Optional[dict] = {"mode": "all", "missing": [], "existing": {}}
+ wav_files = find_wav_files(wav_dir)
+ if wav_files and existing_voices and not args.force:
+ plan = _decide_faster_transcription(wav_files, existing_voices,
+ confirm)
+ if plan is None:
+ return None
# Step 3: port + default voice.
port = args.port
@@ -195,6 +290,7 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
"output_path": output_path,
"port": port,
"force": args.force,
+ "plan": plan,
}
@@ -226,7 +322,7 @@ def _execute(settings: dict) -> int:
voices = _write_voices_json(settings["output_path"], settings["wav_dir"],
settings["language"], settings["whisper_model"],
- settings["force"])
+ settings["plan"])
if voices is None:
return 1
@@ -308,6 +404,7 @@ def _collect_from_flags(args: argparse.Namespace,
"output_path": output_path,
"port": args.port if args.port is not None else _config_port(),
"force": args.force,
+ "plan": {"mode": "all", "missing": [], "existing": {}},
}
@@ -332,7 +429,8 @@ def build_parser() -> argparse.ArgumentParser:
"(default: base)")
parser.add_argument("--force", action="store_true",
help="Overwrite an existing voices.json without "
- "prompting")
+ "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)")
@@ -394,22 +492,27 @@ def _detect_remote(managed: bool = False):
return False, {}
-def _run_voices_only_tui() -> int:
- """Rebuild voices.json via the TUI (the "configure" action).
+def uninstall() -> int:
+ """Remove the faster-qwen3-tts backend entirely.
- Runs the same wizard but skips the pip/clone prerequisites so it goes
- straight to picking the .wav directory and writing voices.json.
+ Uninstalls the pip package (``faster-qwen3-tts``) from the managed venv
+ and deletes the cloned checkout (``app/faster-qwen3-tts``, which holds
+ examples/openai_server.py and voices.json). A running server this tool
+ started is stopped first (best-effort). Returns the exit code.
"""
- 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),
-]
+ servers.stop("faster")
+ rc = common.pip_uninstall(["faster-qwen3-tts"])
+ if rc != 0:
+ print("[WARNING] pip uninstall failed (exit "
+ f"{rc}); remove faster-qwen3-tts from the managed venv manually")
+ else:
+ print("[OK] faster-qwen3-tts removed.")
+ checkout = _checkout()
+ if checkout.is_dir():
+ print(f"[INFO] Removing checkout {checkout}...")
+ shutil.rmtree(checkout, ignore_errors=True)
+ print("[OK] checkout removed.")
+ return 0
def main() -> int:
diff --git a/app/backends/qwen.py b/app/backends/qwen.py
index 7f821fa..b170eb8 100644
--- a/app/backends/qwen.py
+++ b/app/backends/qwen.py
@@ -22,7 +22,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from backends import (
BackendStatus,
- ConfigureAction,
ServerSpec,
common,
envs,
@@ -282,9 +281,23 @@ def _detect_remote(managed: bool = False):
return remote_models, remote_urls
-configure_actions: List[ConfigureAction] = [
- ConfigureAction("Reconfigure qwen-tts (ports/speaker)", run_tui),
-]
+def uninstall() -> int:
+ """Remove the qwen-tts backend entirely: stop its servers, pip uninstall.
+
+ qwen-tts is a pip package (``qwen_tts`` + the ``qwen-tts-demo`` script)
+ installed into the managed venv, so uninstalling it removes the backend.
+ Any server this tool started is stopped first (best-effort). Returns the
+ exit code.
+ """
+ servers.stop("qwen-custom")
+ servers.stop("qwen-clone")
+ rc = common.pip_uninstall([QWEN_PIP_PKG])
+ if rc != 0:
+ print(f"[WARNING] pip uninstall failed (exit {rc}); remove "
+ f"{QWEN_PIP_PKG} from the managed venv manually")
+ else:
+ print(f"[OK] {QWEN_PIP_PKG} removed.")
+ return 0
def main() -> int: