#!/usr/bin/env python3 """The TUI hub for the audiobook generator (run via ``audiobook.py``). The hub is the single entry point for the whole workflow: it detects which backends are already set up and offers to convert the input directory with one of them, or install/configure/remove a backend via the "Configure Backends" menu. The entire hub runs in one curses session, driven by a single ``tui.Wizard`` stack of screens (the ``_Hub`` class below). Every menu/action is a screen that returns the next screen, ``Wizard.BACK`` (Esc/q) to pop one screen, or None to quit. Backend setup wizards and the conversion run view run as opaque leaf screens on this same session; every long step — setup tails, model downloads, server start/stop, and uninstall — runs inside the ``ui.taskview`` task view, so the user is never dropped to the console. A leaf screen finishes by returning ``Wizard.BACK``, so the stack lands back on the menu that launched it. Esc therefore steps back exactly one screen everywhere — on the main menu (an empty stack) it quits. 'q' mirrors Esc on every screen that has no typed text. """ import contextlib import functools import io import json import shutil import sys import urllib.parse from datetime import datetime from pathlib import Path from typing import Callable, Optional, Tuple import logging_kit from backends import ( REGISTRY, BackendStatus, ServerSpec, common, detect_all, invalidate_detect_cache, get, servers, ) from backends import audiocpp as audiocpp_backend from backends import faster as faster_backend from backends import probe as backend_probe from backends import qwen as qwen_backend from converter import config from converter import converter as converter_mod from converter.converter import ( AUDIO_FORMATS, AudiobookConverter, LOGS_FOLDER, voice_mode_for, ) from converter.clients import ( AUDIOCPP_VOICE_CLONE, AUDIOCPP_VOICE_DESIGN, AUDIOCPP_VOICE_SPEAKER, BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, LANGUAGE_CHOICES, QWEN3_TTS_SPEAKERS, audiocpp_entry_voice_capability, normalize_language, ) from ui import runview, taskview, tui _CANCEL = object() # sentinel: a convert preflight confirm backed out class _BackToForm(Exception): """Raised when Esc backs out of a preflight confirm (re-show the form).""" def run() -> int: """Run the hub as one curses session; return the exit code.""" import curses try: curses.wrapper(_app) except tui.WizardCancelled: return 0 except KeyboardInterrupt: return 130 finally: # The curses session is over and the terminal is restored: surface # anything setup steps queued for the console (e.g. a failed audio.cpp # build's copy-pastable command and build log path). for notice in common.drain_post_tui_notices(): print(notice) return 0 def _app(stdscr) -> None: """Drive the whole hub as one ``tui.Wizard`` stack of screens.""" _Hub(stdscr).run() class _Hub: """The hub as a single ``tui.Wizard`` stack of screens. Every menu and action is a zero-argument bound method driven by ``tui.Wizard``: a screen returns the next screen (advance), ``Wizard.BACK`` (Esc/q — pop exactly one screen), or None (quit, only reached from the main menu). Backend setup wizards and the conversion run view run as opaque leaf screens on this same session; a leaf screen finishes by returning ``Wizard.BACK``, so the stack naturally lands back on the menu that launched it. """ def __init__(self, stdscr): self.stdscr = stdscr def run(self) -> None: tui.Wizard().run(self.screen_main) # -- top level ------------------------------------------------------ def screen_main(self): statuses = detect_all() options = [("Configure Backends", "configure_backends")] # Converting works against an external (remote) server too, but # configuring one needs it on this machine. if any(st.installed or st.running for st in statuses): options.insert(0, ("Generate Audiobooks", "convert")) options.append(("Settings", "settings")) options.append(("Help", "help")) options.append(("Quit", "quit")) choice = tui.menu( self.stdscr, "tts-audiobook-generator", options, back_value=tui.Wizard.BACK, table_rows=_status_rows(statuses), notice_lines=_notice_lines()) if choice is tui.Wizard.BACK or choice == "quit": return None if choice == "convert": return self.screen_convert if choice == "configure_backends": return self.screen_configure if choice == "help": return self.screen_help return self.screen_settings # -- configure / install / uninstall -------------------------------- def screen_configure(self): """One flat menu of backend setup/configure/cleanup actions. The audio.cpp "next step" — build its server (when a checkout has no binary) or download its missing models (only once built, so build > configure > download — Build and Download never appear together) — heads the menu with a yellow ``[recommended]`` tag, separated from the rest by a blank line. The remaining actions are populated from the detected statuses: configure each configurable backend (qwen offers its per-model weight (un)installer there), start/stop the installed backends' local servers, install (backends with nothing on disk), update (every installed backend refreshed to the latest upstream version in one task-view run), and uninstall. Selecting one pushes the next screen; Esc pops back to the main menu. The Build action downloads any missing models alongside the build (a split view), so it heals a configured-but-unbuilt backend in one step; Download Missing Models stays as the fallback for when a download fails or is interrupted. """ while True: statuses = detect_all() by_key = {st.key: st for st in statuses} installed = [info for info in REGISTRY if by_key.get(info.key) is not None and by_key[info.key].installed] audiocpp_status = by_key.get("audiocpp") missing = [] needs_build = False if audiocpp_status is not None: checkout = audiocpp_backend.find_local_checkout() if checkout is not None: built = audiocpp_backend.find_audiocpp_server_bin( checkout) is not None if not built: needs_build = True server_json = checkout / "server.json" # Models can only be downloaded once the server binary # exists (build > configure > download), so Build and # Download never appear together. if built and audiocpp_status.configured \ and server_json.exists(): missing = audiocpp_backend.missing_model_entries( server_json) options = [] if needs_build: options.append(("Build audio.cpp Server", "build_audiocpp", ("[recommended]", "warn"))) elif missing: options.append(("Download Missing Models (audio.cpp)", "download_models", ("[recommended]", "warn"))) if needs_build or missing: options.append(tui.MENU_SEPARATOR) options += [(f"Configure {info.label}", ("configure", info.key)) for info in installed if _configurable(info)] if any(st.installed for st in statuses): options.append(("Start/Stop Backend Servers", "server")) if any(_installable(info, by_key) for info in REGISTRY): options.append(("Install Backend", "install")) if any(_updatable(info, by_key) for info in REGISTRY): options.append(("Update Backends", "update")) if any(_uninstallable(info, by_key) for info in REGISTRY): options.append(("Uninstall Backend", "uninstall")) choice = tui.menu( self.stdscr, "Configure Backends", options, back_value=tui.Wizard.BACK, help_lines=["Install, update, configure, or remove a TTS " "backend."], table_rows=_status_rows(statuses), notice_lines=_notice_lines()) if choice is tui.Wizard.BACK: return tui.Wizard.BACK if choice == "install": return self.screen_install if choice == "server": return self.screen_server if choice == "update": _update_backends_action(self.stdscr) invalidate_detect_cache() continue # an inline action: re-show this same menu if choice == "uninstall": return self.screen_uninstall if choice == "download_models": _download_models_action(self.stdscr) invalidate_detect_cache() continue # an inline action: re-show this same menu if choice == "build_audiocpp": audiocpp_backend.build_screen(self.stdscr) invalidate_detect_cache() continue # an inline action: re-show this same menu _kind, key = choice info = get(key) if info is None: continue return self.screen_setup(info) def screen_setup(self, info): """Run one backend's setup/configure wizard as a leaf screen. The wizard drives its own internal ``tui.Wizard`` on this screen; Esc on its first screen (or Ctrl-C) returns here and the hub pops back to the menu that launched it. A crash flashes and does the same. """ def screen(): # A dedicated configure screen (qwen's per-model manager) takes # precedence over the plain setup wizard here; Install Backend # keeps running the setup wizard either way. if info.configure_screen is not None: self._run_configure(info) else: self._run_setup(info) invalidate_detect_cache() return tui.Wizard.BACK return screen def _run_configure(self, info) -> None: """Run one backend's dedicated configure screen on this session.""" try: info.configure_screen(self.stdscr) except tui.WizardCancelled: pass except Exception as exc: # noqa: BLE001 - keep the hub alive tui.flash(self.stdscr, str(exc), "err") def _run_setup(self, info) -> None: """Run one backend's setup wizard on this session (no stack frame).""" try: info.setup_screen(self.stdscr) except tui.WizardCancelled: pass except Exception as exc: # noqa: BLE001 - keep the hub alive tui.flash(self.stdscr, str(exc), "err") def screen_install(self): """Pick a backend to install and run its setup inline. The setup is not pushed as a stack frame: when it finishes this screen returns BACK, popping straight past the picker to the Configure menu, whose status table re-detects the new install. Esc on the picker still pops back one screen normally. """ info = self._pick_backend(installed_only=False) if info is None: return tui.Wizard.BACK self._run_setup(info) return tui.Wizard.BACK def screen_uninstall(self): """Pick a backend, confirm, then uninstall it inside the task view. Esc on the picker or the confirm (or answering No) backs out untouched. A confirmed uninstall runs as one task-view step on this session — no console drop: the backend's ``uninstall`` stops its servers, pip-uninstalls, and deletes its files, honoring cancel between phases only. A flash summarizes the result and the stack lands back on the Configure menu, whose status table re-detects the removal. """ info = self._pick_backend(installed_only=True) if info is None: return tui.Wizard.BACK answer = tui.confirm( self.stdscr, f"Uninstall {info.label}?", body=[f"This permanently removes {info.label} from this " "machine: managed servers are stopped and every installed " "file — including downloaded models — is deleted."], default=False, cancel_value=tui.Wizard.BACK) if answer is not True: return tui.Wizard.BACK title = f"Uninstall {info.label}" step = taskview.TaskStep( title, lambda emit, cancel: info.uninstall(emit=emit, cancel=cancel)) rc = taskview.run_steps(self.stdscr, title, [step], wait_on_finish=False) invalidate_detect_cache() # The uninstallers warn-and-continue (a failed pip step still # returns 0), so rc == 0 means "finished"; anything else covers a # failure or an Esc-cancelled run between phases. if rc == 0: tui.flash(self.stdscr, f"{info.label} uninstalled.", "ok") else: tui.flash(self.stdscr, f"Could not fully uninstall {info.label}.", "err") return tui.Wizard.BACK def _pick_backend(self, installed_only: bool): """Pick a backend for the Install/Uninstall actions. With INSTALLED_ONLY False every backend with nothing on disk yet is listed (the install list — audio.cpp only without a checkout, since a downloaded-but-unbuilt checkout is past install); with it True the ones with something on disk to remove are (the uninstall list — including a downloaded-but-unbuilt audio.cpp checkout, which ``uninstall`` deletes whole). Returns a registry entry, or None to go back. """ statuses = detect_all() by_key = {st.key: st for st in statuses} if installed_only: candidates = [info for info in REGISTRY if _uninstallable(info, by_key)] else: candidates = [info for info in REGISTRY if _installable(info, by_key)] if not candidates: tui.flash(self.stdscr, "No backends to list here.") return None options = [(info.label, info.key) for info in candidates] title = "Uninstall Backend" if installed_only else "Install Backend" key = tui.menu(self.stdscr, title, options, back_value=tui.Wizard.BACK, table_rows=_status_rows(statuses), notice_lines=_notice_lines()) if key is tui.Wizard.BACK: return None return get(key) # -- convert -------------------------------------------------------- def screen_convert(self): """Collect run settings, preflight, then run the conversion view. Esc on the form (or Cancel) pops back to the main menu; Esc on a preflight "overwrite?" confirm returns to the form (one screen). """ prepared = _convert_form(self.stdscr) if prepared is None: return tui.Wizard.BACK fields, builders, statuses = prepared while True: result = tui.form(self.stdscr, "Generate Audiobooks", fields, buttons=("Generate!", "Cancel"), start_on_buttons=True, back_value=tui.Wizard.BACK) if result is tui.Wizard.BACK or result is None: return tui.Wizard.BACK _, mapper = builders[result["backend"]] cmd = mapper(result) if cmd is None: return tui.Wizard.BACK # Nothing is persisted anymore, but server state may have # changed since the form opened — the autostart plan must # read fresh statuses. invalidate_detect_cache() statuses = detect_all(refresh=True) autostart_error = _add_autostart(cmd, statuses) if autostart_error: tui.flash(self.stdscr, autostart_error, "err") continue try: ok = _preflight(self.stdscr, cmd) except _BackToForm: continue if not ok: return tui.Wizard.BACK if self._run_conversion(cmd[1], cmd[2]): # "Stop server and exit" was on: returning # None ends the wizard stack (the whole TUI). return None return tui.Wizard.BACK def _run_conversion(self, backend: str, kwargs: dict) -> bool: """Run a conversion in the full-screen run view on this session. Returns True when the run view's stop-and-exit toggle was on — the caller then quits the whole TUI (results are printed after curses closes) instead of landing back on the menu. A crash inside the view cancels the worker and flashes an error instead of taking the whole hub down; the timed getch the run view leaves behind is reset so the hub menus still block for keys. """ run_config = _prepare_run_config(backend, kwargs) if run_config is None: return False view = runview.RunView(self.stdscr, run_config) try: return bool(view.run()) except tui.WizardCancelled: pass except KeyboardInterrupt: pass except Exception as exc: # noqa: BLE001 - keep the hub alive view._cancel.set() view._worker.join(timeout=30) tui.flash(self.stdscr, f"The run view failed: {exc}", "err") finally: try: self.stdscr.timeout(-1) except Exception: pass # The run may have autostarted a server or changed on-disk # state; every exit path (stop-and-exit, key press, cancel, # crash) must drop the cached statuses. invalidate_detect_cache() return False # -- settings ------------------------------------------------------- def screen_settings(self): fields = _settings_fields() original = {field["key"]: field["value"] for field in fields} while True: result = tui.form(self.stdscr, "Settings", fields, back_value=tui.Wizard.BACK) if not (result is tui.Wizard.BACK or result is None): # Save pressed: apply as before, no prompt. try: _apply_settings(result) except ValueError as exc: tui.flash(self.stdscr, str(exc), "err") return tui.Wizard.BACK # q/Esc (or the Cancel button) left the form without saving: # with no edits there is nothing to keep, so go straight back; # otherwise ask whether the edits should be preserved. if not _settings_changed(fields, original): return tui.Wizard.BACK answer = tui.confirm_yn_cancel(self.stdscr, "Save settings?") if answer == "cancel": continue # back into the form, edits intact if answer == "yes": values = {field["key"]: field["value"] for field in fields} try: _apply_settings(values) except ValueError as exc: tui.flash(self.stdscr, str(exc), "err") return tui.Wizard.BACK # -- help ------------------------------------------------------------ def screen_help(self): """Show the quick-start Help text in a scrollable dialog. A leaf screen: the viewer closes on Esc/q/Enter (its back_value), so the stack pops back to the menu that opened it. """ tui.text_viewer(self.stdscr, "Help", _help_lines(), back_value=tui.Wizard.BACK) return tui.Wizard.BACK # -- servers -------------------------------------------------------- def screen_server(self): """One flat menu of every installed backend's servers. A status table above the menu shows each server's live state — "running" (green) or "stopped" (red) — and selecting an entry toggles it directly (starts a stopped server, stops a running one) without an extra action menu. The state lives in the table, not on the entries, because the menu's selection bar would cover inline colors. Each server is labelled by its backend's name; qwen hosts one model at a time (its default start runs CustomVoice — a Generate-audiobooks run needing another model restarts it). """ statuses = detect_all() candidates = [st for st in statuses if st.installed] if not candidates: tui.flash(self.stdscr, "No backend is installed yet — use " "'Configure Backends' first.") return tui.Wizard.BACK options = [] rows = [] for st in candidates: specs = st.servers for spec in specs: running = common.server_running(spec.url) label = st.label if len(specs) == 1 \ else f"{st.label} — {spec.name}" options.append((label, spec)) rows.append((label, "running" if running else "stopped", "ok" if running else "err", "body")) if not options: tui.flash(self.stdscr, "No backend server is configured yet — " "use 'Configure Backends' first.") return tui.Wizard.BACK spec = tui.menu(self.stdscr, "Start / Stop A Server", options, back_value=tui.Wizard.BACK, help_lines=["Start/stop local servers manually.", "'Generate Audiobooks' handles this " "automatically."], table_rows=rows, notice_lines=_notice_lines()) if spec is tui.Wizard.BACK: return tui.Wizard.BACK return functools.partial(self._server_toggle, spec) def _server_toggle(self, spec): """Start SPEC's server when stopped, stop it when running. Runs inside the task view (no console drop); the server module's plain-console output is tee'd to a log file under ``app/logs`` so nothing is lost, and on failure a flash points the user at that file. Returns BACK so the stack lands back on the server list, which re-reads each server's live state. """ running = common.server_running(spec.url) action = "stop" if running else "start" step, log_path = _server_action_step(spec, action) taskview.run_steps(self.stdscr, f"{action.capitalize()} " f"{spec.name} server", [step], wait_on_finish=False) # Re-check the server instead of trusting the step's exit code # (cancel and failure both come back non-zero): did the toggle take? invalidate_detect_cache() now_running = common.server_running(spec.url) if action == "start" and not now_running: tui.flash(self.stdscr, f"Could not start the {spec.name} " f"server. See the log: {log_path}", "err") elif action == "stop" and now_running: tui.flash(self.stdscr, f"Could not stop the {spec.name} " f"server. See the log: {log_path}", "err") return tui.Wizard.BACK def _server_action_step(spec, action: str): """Build a task step that starts/stops SPEC's server, logged to a file. ACTION is "start" or "stop". The step runs inside the task view (no console drop): the server module's output is tee'd to a timestamped ``__*.log`` artifact under ``servers.LOG_DIR`` (see ``logging_kit.run_artifact``) and to the view's log tail. Returns ``(TaskStep, log_path)`` so the caller can point the user at the file on failure. """ title = (f"Start {spec.name} server" if action == "start" else f"Stop {spec.name} server") log_path, logf = logging_kit.run_artifact(f"{spec.name}_{action}", log_dir=servers.LOG_DIR) def work(emit, cancel): inner = sys.stdout # the task view's line-writer, when run in TUI with contextlib.redirect_stdout(logging_kit.TeeWriter(logf, inner)): if action == "start": ok = servers.start(spec, cancel=cancel) else: ok = servers.stop(spec.name) return 0 if ok else 1 return taskview.TaskStep(title, work), log_path def _configurable(info) -> bool: """True when INFO has a configure screen worth running from the hub. A backend with a dedicated ``configure_screen`` (qwen manages its three model weight installs there) is always configurable; other backends count via their non-trivial setup wizard. Bare qwen — whose wizard asks no questions: ports live in Settings, the speaker is chosen per run on Generate Audiobooks — would only ever flash "already installed", so it stays excluded until it ships a dedicated screen. """ return info.configure_screen is not None or info.key != "qwen" def _installable(info, by_key: dict) -> bool: """True when INFO has nothing on disk yet — an install-entry candidate. audio.cpp is installable only without a checkout: a downloaded-but-unbuilt checkout is already past the install step (its next action is the hub's Build entry), so listing it under "Install Backend" would duplicate that and suggest re-running setup from scratch. The other backends are installable while not installed. """ if info.key == "audiocpp": return audiocpp_backend.find_local_checkout() is None status = by_key.get(info.key) return status is None or not status.installed def _uninstallable(info, by_key: dict) -> bool: """True when INFO has something on disk that uninstall removes. audio.cpp's ``installed`` flag means *built*, so a downloaded-but-unbuilt checkout would otherwise miss the Uninstall menu — but its checkout (binary, models, server.json) lives on disk and ``uninstall()`` removes it, so it counts too. The other backends' ``installed`` already covers everything their uninstaller touches. """ if info.key == "audiocpp": return audiocpp_backend.find_local_checkout() is not None status = by_key.get(info.key) return status is not None and status.installed def _updatable(info, by_key: dict) -> bool: """True when the "Update Backends" action has something to do for INFO. The same on-disk predicate as _uninstallable — update acts on exactly what uninstall removes (the pip package / the checkout) — plus the backend must implement an update action at all. """ return info.update is not None and _uninstallable(info, by_key) def _download_models_action(stdscr) -> None: """Run the "Download Missing Models (audio.cpp)" action inside the TUI. Computes the missing models; when they map to install commands it runs the downloads in the task view (with real byte progress and cancellation) instead of dropping to the console — a successful run returns silently (the view already shows [OK] and waits for a key), and only a cancelled or failed run flashes. When the checkout/server.json is missing, nothing is missing, or the models do not map to an install command, it flashes an explanatory notice (the latter explaining how to install each model by hand). """ checkout = audiocpp_backend.find_local_checkout() if checkout is None: tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp " "first.", "err") return server_json = checkout / "server.json" if not server_json.exists(): tui.flash(stdscr, "No audio.cpp server.json found — configure " "audio.cpp first.", "err") return missing = audiocpp_backend.missing_model_entries(server_json) if not missing: tui.flash(stdscr, "Every configured audio.cpp model is already " "downloaded.", "ok") return guidance = audiocpp_backend.missing_model_install_guidance( checkout, missing) if not guidance: tui.flash(stdscr, audiocpp_backend.hand_install_guidance( checkout, missing), "err") return def run(emit, cancel): return audiocpp_backend.install_models(checkout, guidance, emit=emit, cancel=cancel) rc = taskview.run_steps(stdscr, "Download models", [taskview.TaskStep("Download missing models", run)]) if rc == 130: tui.flash(stdscr, "Model download cancelled — re-run it any time.", "warn") elif rc: tui.flash(stdscr, "Some model downloads failed. Re-run 'Download " "Missing Models' or install them by hand (see the log).", "err") def _update_backends_action(stdscr) -> None: """Run the "Update Backends" action inside the TUI. One task-view step per installed backend that implements update, in registry order; each update stops its managed server first (best- effort) and then refreshes — pip install -U for the pip backends, git fetch + hard reset for the checkouts, with audio.cpp's binary rebuilt when its checkout moved. A failing backend's step is marked [FAIL] and the remaining backends still update (the run's exit code is the first failure). A successful run returns silently (the view already shows [OK] and waits for a key); only a cancelled or failed run flashes. The status table re-detects when the menu re-shows. """ statuses = detect_all() by_key = {st.key: st for st in statuses} targets = [info for info in REGISTRY if _updatable(info, by_key)] if not targets: tui.flash(stdscr, "No installed backend supports updating.") return def make_work(info): def work(emit, cancel): return info.update(emit=emit, cancel=cancel) return work steps = [taskview.TaskStep(f"Update {info.label}", make_work(info)) for info in targets] rc = taskview.run_steps(stdscr, "Update Backends", steps) if rc == 130: tui.flash(stdscr, "Update cancelled — re-run 'Update Backends' " "any time.", "warn") elif rc: tui.flash(stdscr, "Some updates did not complete (failed or " "cancelled) — see the log above. Re-run 'Update " "Backends' to retry.", "err") def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]: """Map a backend's state to (status_text, status_kind, name_kind). A backend is 'running' (green/ok) when it is usable either locally — a server this tool started (``status.managed``) — or remotely — a server found by probing its remote URL (``status.remote``); the text names which, e.g. "running [local]", "running [remote]", or "running [local, remote]". Otherwise a backend set up only part-way (``status.partial``) shows that label verbatim (amber), e.g. audio.cpp's "downloaded (not built)" or "built (not configured)"; 'installed' (green/ok) when the backend is present on disk — amber (warn) instead when its models are missing — or 'unavailable' (red/err). A backend that is neither installed nor running is unusable, so its name is dimmed (NAME_KIND). A multi-model backend (qwen) also names which models answered in parentheses, e.g. "running [local, remote] (Base, CustomVoice)". CURSES has no true orange, so the theme's yellow 'warn' is used; it renders amber/orange on most terminals. """ if status is not None and status.running: tags = [] if status.managed: tags.append("local") if status.remote: tags.append("remote") text = "running" if tags: text += " [" + ", ".join(tags) + "]" if status.running_models: text += " (" + ", ".join(status.running_models) + ")" return (text, "ok", "body") if status is not None and status.partial: # Part-way set up (audio.cpp: "downloaded (not built)" / # "built (not configured)"): amber text, name dimmed while the # backend is still unusable. name_kind = "dim" if not status.installed else "body" return (status.partial, "warn", name_kind) if status is not None and status.installed: if status.models_missing and not status.running: return ("installed (models missing)", "warn", "body") return ("installed", "ok", "body") return ("unavailable", "err", "dim") def _status_rows(statuses) -> list: """Status-table rows for tui.menu: (label, status, kind, name_kind). One row per detected backend, in detect order — the same table the main menu shows, reused on each flow's first picker screen so the backend states stay visible there. """ return [(st.label, *_status_mark(st)) for st in statuses] def _notice_lines() -> Optional[list]: """Warning lines shown above the status table, or None when all good.""" if shutil.which("ffmpeg") is None: return [("Warning: ffmpeg not installed!", "err")] return None def _help_lines() -> list: """The Help screen's quick-start text (folder paths resolved live). Items are text_viewer rows: "" (a blank line) or a (segments, indent) pair — SEGMENTS are (text, kind) with KIND a theme key (None = body). Numbered steps start at the margin; every other line is indented three spaces — Frame's two-space indent unit plus a leading space in the row's first segment — so it lines up with the step text after the "N. " prefixes. The input/output folders are read from the converter module at call time, so a Settings change this session is reflected without a restart. """ return [ ([("1. ", "title"), ("Put your ebooks (epub, txt, or pdf) here:", None)], 0), ([(" " + str(converter_mod.BOOKS_FOLDER), "input")], 1), "", ([("2. ", "title"), ("Put any .wavs of voices to clone here:", None)], 0), ([(" " + str(common.VOICES_DIR), "input")], 1), "", ([("3. ", "title"), ("If no backend is installed, go to ", None), ("Configure Backends", "accent"), (" > ", None), ("Install Backend", "accent"), (" and install audio.cpp.", None)], 0), "", ([("4. ", "title"), ("Select TTS models to install. If you're unsure, try " "these qwen3-tts models:", None)], 0), "", ([(" Voice cloning:", None), (" ", None), ("qwen3_tts_1_7b_base_q8_0", "ok")], 1), ([(" Built-in-voice:", None), (" ", None), ("qwen3_tts_1_7b_customvoice_q8_0", "ok")], 1), "", ([(" It will take a while to build audio.cpp and download " "the model files.", None)], 1), "", ([("5. ", "title"), ("Go to ", None), ("Generate Audiobooks", "accent"), (". It will automatically start the necessary server, " "generate the books, and stop it.", None)], 0), ([(" There is no need to manually start/stop servers.", None)], 1), "", ([("6. ", "title"), ("Generated audiobooks (m4b, mp3, etc.) will output here:", None)], 0), ([(" " + str(converter_mod.AUDIOBOOKS_FOLDER), "input")], 1), ] def _convert_form(stdscr) -> Optional[tuple]: """Build the Convert-books form (fields + builders), or None to go back. The first field is the Backend picker; the remaining fields are that backend's options (audio.cpp: model/voice/instructions; qwen: model + speaker / clone .wav / design instruction; faster: voice) plus the per-run combine-all-chapters toggle (the shared output settings live in the Settings menu). A backend appears once as a managed entry ("audio.cpp") when it is installed+configured here, and once as a remote entry ("audio.cpp [remote]") when a running server was found at its remote URL. Managed entries read the local server.json / voices.json; remote entries query the remote server live. Each available entry's data is prepared up front so the Backend field lists only backends whose options could be gathered — an entry whose data is unavailable (e.g. an unreachable remote audio.cpp server) is dropped here. Returns ``(fields, builders, statuses)``; None (after a flash) when there is nothing to convert with. """ statuses = detect_all() entries = [] for st in statuses: if st.ready: entries.append((st.key, st.label, st, False)) if st.remote: entries.append((f"{st.key}-remote", f"{st.label} [remote]", st, True)) if not entries: tui.flash(stdscr, "No backend is ready to convert with yet — use " "'Configure Backends' first.") return None builders = {} # A backend can appear twice (managed + "[remote]"), so the remote # entry's fields are keyed under "." (e.g. "audiocpp-remote. # model_id"): the form returns one flat {key: value} dict, and duplicate # keys would make one entry's value silently win over the other's. for key, _label, st, remote in entries: prefix = f"{key}." if remote else "" if remote: if st.key == BACKEND_AUDIOCPP: built = _audiocpp_fields( stdscr, api_url=st.remote_urls.get("audiocpp"), prefix=prefix) elif st.key == BACKEND_QWEN: built = _qwen_fields(remote_modes=st.remote_models, urls=st.remote_urls, prefix=prefix) elif st.key == BACKEND_FASTER: built = _faster_fields( stdscr, api_url=st.remote_urls.get("faster"), prefix=prefix) else: continue else: if st.key == BACKEND_AUDIOCPP: built = _audiocpp_fields(stdscr) elif st.key == BACKEND_QWEN: built = _qwen_fields() elif st.key == BACKEND_FASTER: built = _faster_fields(stdscr) else: continue if built is not None: builders[key] = built if not builders: return None choices = [(label, key) for key, label, _st, _remote in entries if key in builders] fields = [{ "key": "backend", "label": "Backend", "kind": "choice", "value": choices[0][1], "choices": choices, }] for key, _label, _st, _remote in entries: if key not in builders: continue backend_fields, _ = builders[key] for field in backend_fields: field["visible"] = _gate_backend(field, key) fields += backend_fields fields += _common_fields() return fields, builders, statuses def _preflight(stdscr, cmd: tuple) -> bool: """Run the overwrite checks in the TUI; stash the plan on the command. Asks every "output exists — overwrite?" question now (tui.confirm instead of the console input()) so the run view itself is unattended, and records the discovered books / accepted plan in the command's kwargs (``book_files``/``planned``) for ``audiobook.convert``. Returns False when nothing would be converted (a flash explains why), so the user stays in the menu instead of entering an empty run. """ _kind, backend, kwargs = cmd voice_mode = voice_mode_for(backend, kwargs.get("voice"), kwargs.get("clone"), kwargs.get("instructions")) def confirm(message: str, default: bool) -> bool: answer = tui.confirm(stdscr, message, default=default, cancel_value=_CANCEL) if answer is _CANCEL: raise _BackToForm() return answer with contextlib.redirect_stdout(io.StringIO()): book_files, planned = AudiobookConverter.preflight_overwrites( backend=backend, voice=kwargs.get("voice"), voice_mode=voice_mode, voice_clone_ref_audio=kwargs.get("clone"), output_format=kwargs.get("output_format") or config.AUDIO_FORMAT, instructions=kwargs.get("instructions"), confirm=confirm) if not book_files: tui.flash(stdscr, "No books to convert. Add a .txt, .pdf or .epub " "file to the input folder first.") return False if not planned: tui.flash(stdscr, "Nothing to convert — every existing output was " "kept.") return False kwargs["book_files"] = book_files kwargs["planned"] = planned return True def _gate_backend(field: dict, key: str) -> Callable: """A visible() that shows FIELD only when the Backend field is KEY. Composes with any ``visible`` callable the field already carries (audio.cpp's task-driven Voice field, qwen's mode-driven fields), so both the backend gate and the field's own rule must pass. """ base = field.get("visible", True) def visible(fields) -> bool: if _field_value(fields, "backend") != key: return False if callable(base): return base(fields) return bool(base) return visible def _field_value(fields, key: str, default=None): """Current value of the field named KEY, or DEFAULT when absent.""" for field in fields: if field.get("key") == key: return field["value"] return default # Dim hint shown while editing the Settings Language field: which # languages a model accepts varies by backend/model. _LANGUAGE_EDIT_HINT = ["Check model documentation for supported languages."] def _common_fields() -> list: """The per-run Generate-form fields (the global output options — output format, language, speed, debug, stop-server-and-exit — live in the Settings menu and reach the run via _common_kwargs()). The single-file field is hidden for m4b (always a single file with embedded chapter markers), so its "visible" callable reads the configured output format. """ return [ {"key": "single_file", "label": "Combine all chapters", "kind": "bool", "value": False, "visible": lambda fs: config.AUDIO_FORMAT != "m4b"}, ] def _common_kwargs(values: dict) -> dict: """Map the common form fields plus the configured output settings to converter keyword arguments. Output format, language, speed, debug, and stop-server-and-exit are configured once in the Settings menu (config.py) and apply to every run; only per-run choices (combine-all-chapters) come from the form. """ output_format = config.AUDIO_FORMAT return { "language": config.LANGUAGE, "output_format": output_format, "speed": float(config.SPEED), "single_file": bool(values["single_file"]) and output_format != "m4b", "debug": bool(config.DEBUG), "stop_and_exit": bool(config.STOP_SERVER_AND_EXIT), } def _audiocpp_fields(stdscr, api_url: Optional[str] = None, prefix: str = "") -> Optional[tuple]: """audio.cpp-specific fields and a result mapper for the Convert form. Returns ``(fields, mapper)`` where FIELDS are the audio.cpp options (Model / Voice / Instructions) and MAPPER turns a submitted form values dict into the audio.cpp converter kwargs. Returns None when the model list cannot be gathered (a flash explains why), so the caller drops audio.cpp from the Backend choices. PREFIX namespaces the field keys ("" for the managed entry) so two entries of this backend can share one form without overwriting each other. The Voice field tracks the selected entry's capability — built-in speakers on CustomVoice, the server's clone voices on every other entry. A clone-capable entry whose server lists no voices cannot be picked from (empty menu) and refuses Generate! with a hint pointing at the voice-clone .wav directory instead of crashing or producing a run that fails at model-load time. With API_URL None (the managed entry) the model list is fed from the local checkout's server.json — the config of the server this tool manages. With API_URL set (the "[remote]" entry) the models and voices are queried live from that server instead (the same GET /v1/models and GET /v1/audio/voices endpoints the converter resolves at run time). """ if api_url is None: checkout = audiocpp_backend.find_local_checkout() server_json = checkout / "server.json" if checkout else None if not (server_json and server_json.exists()): tui.flash(stdscr, "No audio.cpp server.json found — run " "'Configure Backends' first.") return None try: data = json.loads(server_json.read_text(encoding="utf-8")) except (OSError, ValueError): tui.flash(stdscr, f"Could not read {server_json}.") return None models = data.get("models") or [] if not models: tui.flash(stdscr, "No model entries in server.json. Reconfigure " "audio.cpp first.") return None local = True url = config.AUDIOCPP_API_URL else: # Remote flow: the backend only reaches the convert menu while a # server is running, so query it — the local config says nothing # about an external server. url = api_url local = False data = {} models = audiocpp_backend.fetch_server_models(url) if models is None: tui.flash(stdscr, f"Could not list models from the audio.cpp " f"server at {url}. Is an audiocpp_server answering " "there?") return None if not models: tui.flash(stdscr, f"The audio.cpp server at {url} hosts no " "model entries.") return None # Normalize each entry so the form logic sees a family/task always. # A missing family is left empty (an unknown family resolves to the # clone capability, requiring a --voice) rather than guessing a specific # one — audiocpp_server always reports family for entries it hosts. models = [dict(m) for m in models] for entry in models: entry["family"] = entry.get("family") or "" entry["task"] = entry.get("task") or "tts" if local: # Only offer entries whose model files are actually on disk: a # server.json can reference a package that was never downloaded, # and picking it would fail the whole run at model-load time. missing = audiocpp_backend.missing_model_entries(server_json) if missing: missing_ids = {item["id"] for item in missing} models = [entry for entry in models if entry.get("id") not in missing_ids] if not models: hints = audiocpp_backend.model_install_hints(checkout, missing) message = hints[0] if hints \ else "Download the models first." tui.flash(stdscr, "No model files are downloaded for " f"audio.cpp. {message}") return None local_voices = _list_voices(data.get("voice_dir")) \ if data.get("voice_dir") else [] voice_cache: dict = {} # model id -> voices (local: shared list) # Per-family request-option support comes from this machine's audio.cpp # checkout model_specs, best effort for both entries: the server's HTTP # API does not report it. A "[remote]" entry is classified from the same # local specs when a family matches; with no checkout every family counts # as unknown and the Request options field stays hidden. specs_checkout = checkout if local \ else audiocpp_backend.find_local_checkout() option_families = ( audiocpp_backend.request_options_families(specs_checkout) if specs_checkout is not None else {}) def voices_for(model_id: str) -> list: if local: return local_voices if model_id not in voice_cache: fetched = audiocpp_backend.fetch_server_voices(url, model_id) voice_cache[model_id] = fetched or [] return voice_cache[model_id] def model_entry(fields): model_id = _field_value(fields, prefix + "model_id") return next((m for m in models if m.get("id") == model_id), models[0]) def model_capability(fields) -> str: entry = model_entry(fields) return audiocpp_entry_voice_capability( entry.get("family") or "", entry.get("task") or "tts", entry.get("id") or "") def reset_voice(fields) -> None: """Re-point the Voice field at the newly selected model's voice. A model switch that keeps the same voice list (two clone entries sharing one server's voices) keeps the current pick: only a value the new list cannot offer is re-pointed at its default. """ voice_field = next(f for f in fields if f.get("key") == prefix + "audiocpp_voice") capability = model_capability(fields) if capability == AUDIOCPP_VOICE_DESIGN: voice_field["value"] = None return if capability == AUDIOCPP_VOICE_SPEAKER: voices = QWEN3_TTS_SPEAKERS else: # clone voices = voices_for(_field_value(fields, prefix + "model_id")) if voice_field.get("value") in voices: return # the new list still offers the pick: keep it voice_field["value"] = voices[0] if voices else "" def voice_choices(fields) -> list: capability = model_capability(fields) if capability == AUDIOCPP_VOICE_SPEAKER: # Built-in Qwen3-TTS CustomVoice speakers; no server query needed. return [(s, s) for s in QWEN3_TTS_SPEAKERS] if capability == AUDIOCPP_VOICE_CLONE: return [(v, v) for v in voices_for(_field_value( fields, prefix + "model_id"))] return [] # design: the field is hidden def no_voices_hint(_fs=None) -> str: """Why a clone-capable entry has no selectable voices. The form invokes ``on_empty_choices`` with the field list (see tui.form); validate's echo calls it without one. """ if local: return ("No .wav files available to clone\n" "Go to Configure Backends → Configure audio.cpp and " "set a voice clone directory.\n") return ("No .wav files available to clone — the audio.cpp server " f"at {url} hosts none. Configure its voice-clone .wav " "directory on that machine.") def voice_validate(value): """Refuse Generate! when this entry's clone voice is unavailable. An Instructions text substitutes for the voice: on families that condition synthesis on instructions alone the client designs the voice from it (instruction-voice mode), so an empty Voice is accepted when an instruction is present. """ if model_capability(fields) != AUDIOCPP_VOICE_CLONE: return None has_instruction = bool(str(_field_value( fields, prefix + "instructions") or "").strip()) if not voices_for(_field_value(fields, prefix + "model_id")): return None if has_instruction else no_voices_hint() return None if (value or has_instruction) \ else "This model needs a voice — pick one or switch models" model_ids = [m.get("id") for m in models] default_model = model_ids[0] default_entry = next((m for m in models if m.get("id") == default_model), models[0]) default_capability = audiocpp_entry_voice_capability( default_entry.get("family") or "", default_entry.get("task") or "tts", default_entry.get("id") or "") initial_voice = None if default_capability == AUDIOCPP_VOICE_SPEAKER: initial_voice = QWEN3_TTS_SPEAKERS[0] elif default_capability == AUDIOCPP_VOICE_CLONE: initial = voices_for(default_model) initial_voice = initial[0] if initial else "" # The Model picker reads as a two-column table: pad every id to the # widest one so the (type) column starts on the same position. id_width = max(len(entry.get("id") or "") for entry in models) def _label(entry: dict) -> str: capability = audiocpp_entry_voice_capability( entry.get("family") or "", entry.get("task") or "tts", entry.get("id") or "") return f"{entry.get('id') or '':<{id_width}} ({capability})" def entry_supports_options(fs) -> bool: """True when the selected entry's family defines request options. Resolved strictly from this machine's model_specs: a family the specs prove unable to read options, or cannot classify at all, keeps the field hidden (unknown support is treated as no). """ family = model_entry(fs).get("family") or "" return audiocpp_backend.supports_request_options( option_families, family) is True # Edit-dialog help lines: short, and identical for every capability # (design-model validation already explains its own requirement). INSTRUCTIONS_HELP = [ "TTS style instructions. Supported by some clone models. Example:", '"Speak in a calm, soothing, and happy tone."', ] # Edit-dialog help for the Request options field — at most 2 lines; # unsupported keys are ignored server-side, so nothing else needs # spelling out here. OPTIONS_HELP = [ "KEY=VALUE items, comma/space separated; unsupported keys ignored.", "Examples: emotion=neutral, speed=1.1, temperature=0.8", ] fields = [ {"key": prefix + "model_id", "label": "Model", "kind": "choice", "value": default_model, "choices": [(_label(m), m.get("id")) for m in models], "on_change": reset_voice}, # The label tracks the entry's capability: a built-in speaker on # CustomVoice, otherwise the name of a server-side voice to clone. {"key": prefix + "audiocpp_voice", "label": lambda fs: ("Built-in voice" if model_capability(fs) == AUDIOCPP_VOICE_SPEAKER else "Voice to clone"), "kind": "choice", "value": initial_voice, "choices": lambda fs: voice_choices(fs), "visible": lambda fs: model_capability(fs) != AUDIOCPP_VOICE_DESIGN, "on_empty_choices": no_voices_hint, "validate": voice_validate}, # Style/voice-design instruction. Required for design entries (the # voice comes from it); on every other entry an optional style/ # delivery instruction — or, on families without built-in speakers # that read instructions, the voice itself (instruction-voice mode). {"key": prefix + "instructions", "label": "Instructions", "kind": "text", "value": "", "help": INSTRUCTIONS_HELP, "validate": lambda value: None if (model_capability(fields) != AUDIOCPP_VOICE_DESIGN or str(value).strip()) else "Describe the voice, e.g. 'A warm female narrator'"}, # Free-form per-model controls (--option KEY=VALUE on the CLI). # Shown only for families whose audio.cpp spec declares request # options; unknown-support families keep it hidden. {"key": prefix + "request_options", "label": "Request options", "kind": "text", "value": "", "visible": entry_supports_options, "help": OPTIONS_HELP, "validate": _validate_request_options}, ] def mapper(result) -> Optional[tuple]: model_id = result[prefix + "model_id"] # The picked voice (a built-in speaker name on a CustomVoice entry, # a server-side preset otherwise); the client resolves which it is. voice = result[prefix + "audiocpp_voice"] or None # The instruction is forwarded for every capability: required on # design entries, optional style/delivery control elsewhere. With # no voice it defines the voice on instruction-conditioned families. instructions = ((result.get(prefix + "instructions") or "").strip() or None) try: request_options = common.parse_request_options( result.get(prefix + "request_options") or "") except ValueError: request_options = {} # submit-time validation already caught this kwargs = { "model_id": model_id, "voice": voice, "instructions": instructions, "request_options": request_options, **_common_kwargs(result), } if api_url is not None: kwargs["api_url"] = api_url return ("convert", BACKEND_AUDIOCPP, kwargs) return fields, mapper def _qwen_fields(remote_modes: Optional[list] = None, urls: Optional[dict] = None, prefix: str = "") -> Optional[tuple]: """qwen-specific fields and a result mapper for the Convert form. Returns ``(fields, mapper)`` where FIELDS are the qwen options — which model the demo server hosts (Base (voice cloning) / CustomVoice (built-in voices) / VoiceDesign (design)), plus the per-model controls: Speaker on CustomVoice, a Clone .wav directory browser (default ./voices) + Voice- to-clone .wav picker on Base, Instructions on VoiceDesign — and MAPPER turns a submitted form values dict into the qwen converter kwargs. qwen always has options to offer, so it never signals unavailability. PREFIX namespaces the field keys ("" for the managed entry) so two entries of this backend can share one form without overwriting each other. One demo server hosts one model at a time, so the picked model decides which server must be up. For the managed entry REMOTE_MODES/URLS are None: the picker offers every model and the autostart/restart plan boots exactly the picked model. For a "[remote]" entry REMOTE_MODES names which demos answered remotely ("Base", "CustomVoice" and/or "VoiceDesign") and URLS maps "qwen" to its URL: the picker is limited to the available models and the mapper passes the URL as ``api_url``. """ remote_modes = list(remote_modes or []) urls = dict(urls or {}) mode_keys = (("custom", "CustomVoice"), ("clone", "Base"), ("design", "VoiceDesign")) purposes = {"custom": "built-in voices", "clone": "voice cloning", "design": "design"} # The Model picker reads as a two-column table (like the audio.cpp # picker): pad every model name to the widest one so the (purpose) # column starts on the same position. name_width = max(len(model) for _mode, model in mode_keys) model_choices = [(f"{model:<{name_width}} ({purposes[mode]})", mode) for mode, model in mode_keys] if remote_modes: available = set(remote_modes) model_choices = [(label, value) for (label, value) in model_choices if dict(mode_keys)[value] in available] by_value = {value: label for label, value in model_choices} default_mode = "custom" if "custom" in by_value else model_choices[0][1] speakers = list(qwen_backend.QWEN_SPEAKERS) default_speaker = speakers[0] # Voice cloning references: the directory the .wavs live in — browsed # with the directory widget, defaulting to the project's ./voices (the # folder the Help screen points at) — plus a picker of the .wav files # found there (the same directory + voice picker the audio.cpp form # uses; the demo uploads exactly one reference file). def clone_wav_choices(fs) -> list: """(file name, full path) pairs for the clone directory's .wavs.""" return [(p.name, str(p)) for p in _list_wavs( _field_value(fs, prefix + "clone_dir"))] def reset_clone_wav(fs) -> None: """Re-point the .wav picker at the newly chosen directory.""" wav_field = next(f for f in fields if f.get("key") == prefix + "clone") wav_field["value"] = next( (path for _name, path in clone_wav_choices(fs)), "") def no_wavs_hint(_fs=None) -> str: """Why the .wav picker is empty (validate echoes it on Generate!).""" directory = next((f.get("value") for f in fields if f.get("key") == prefix + "clone_dir"), None) return (f"No .wav files in {directory} — put a reference .wav " "there or pick another directory.") def clone_wav_validate(value) -> Optional[str]: """Refuse Generate! when no reference .wav is available to clone.""" if value: return None return no_wavs_hint() initial_wavs = _list_wavs(common.VOICES_DIR) initial_clone = str(initial_wavs[0]) if initial_wavs else "" fields = [ {"key": prefix + "mode", "label": "Model", "kind": "choice", "value": default_mode, "choices": model_choices}, {"key": prefix + "speaker", "label": "Speaker", "kind": "choice", "value": default_speaker, "choices": speakers, "visible": lambda fs: _field_value(fs, prefix + "mode") == "custom"}, {"key": prefix + "clone_dir", "label": "Clone .wav directory", "kind": "dir", "value": common.VOICES_DIR, "info": common.wav_dir_info, "preview": common.wav_dir_preview, "on_change": reset_clone_wav, "visible": lambda fs: _field_value(fs, prefix + "mode") == "clone"}, {"key": prefix + "clone", "label": "Voice to clone", "kind": "choice", "value": initial_clone, "choices": clone_wav_choices, "on_empty_choices": no_wavs_hint, "validate": clone_wav_validate, "visible": lambda fs: _field_value(fs, prefix + "mode") == "clone"}, {"key": prefix + "qwen_instructions", "label": "Instructions", "kind": "text", "value": "", "help": ["Describe the voice to design, e.g.", '"A warm adult female narrator with a British accent".'], "validate": lambda s: None if s.strip() else "Describe the voice to design", "visible": lambda fs: _field_value(fs, prefix + "mode") == "design"}, ] def mapper(result) -> Optional[tuple]: mode = result[prefix + "mode"] clone = None if mode == "clone": clone = str(result[prefix + "clone"] or "").strip() or None kwargs = {"clone": clone, **_common_kwargs(result)} if mode == "custom": kwargs["voice"] = result[prefix + "speaker"] if mode == "design": kwargs["instructions"] = result[prefix + "qwen_instructions"] if urls: api_url = urls.get("qwen") if api_url: kwargs["api_url"] = api_url return ("convert", BACKEND_QWEN, kwargs) return fields, mapper def _faster_fields(stdscr, api_url: Optional[str] = None, prefix: str = "") -> Optional[tuple]: """faster-specific fields and a result mapper for the Convert form. Returns ``(fields, mapper)`` where FIELDS are the faster options (Voice to clone, as a picker when a local voices.json lists them, else typed free text) and MAPPER turns a submitted form values dict into the faster converter kwargs. Returns None when a local voices.json exists but cannot be read/used (a flash explains why), so the caller drops faster from the Backend choices. PREFIX namespaces the field keys ("" for the managed entry) so two entries of this backend can share one form without overwriting each other. With API_URL None (the managed entry) a local checkout's voices.json drives the picker. With API_URL set (the "[remote]" entry) the running server was configured elsewhere and its voice names are unknown here, so the name is typed instead — safe for any value, since the server falls back to its first configured voice when the name is not defined. """ voices = None if api_url is None: checkout = faster_backend._checkout() voices_json = checkout / "voices.json" if voices_json.exists(): try: voices = json.loads(voices_json.read_text(encoding="utf-8")) except (OSError, ValueError): tui.flash(stdscr, f"Could not read {voices_json}.") return None if not voices: tui.flash(stdscr, "voices.json has no voices. Reconfigure faster.") return None if voices is None: # No local voices.json: prompt for a server-side voice name. fields = [ {"key": prefix + "faster_voice", "label": "Voice to clone", "kind": "text", "value": "", "validate": lambda s: None if s.strip() else "Enter a voice name"}, ] else: default = next(iter(voices)) fields = [ {"key": prefix + "faster_voice", "label": "Voice to clone", "kind": "choice", "value": default, "choices": [(k, k) for k in voices]}, ] def mapper(result) -> Optional[tuple]: voice_value = result[prefix + "faster_voice"] voice = voice_value.strip() \ if isinstance(voice_value, str) else voice_value kwargs = { "voice": voice or None, **_common_kwargs(result), } if api_url is not None: kwargs["api_url"] = api_url return ("convert", BACKEND_FASTER, kwargs) return fields, mapper # --------------------------------------------------------------------------- # Settings menu (global output options -> app/converter/config.py) # --------------------------------------------------------------------------- def _settings_changed(fields: list, original: dict) -> bool: """True when any field's current value differs from its ORIGINAL. Text values compare whitespace-stripped (the form's editor and _apply_settings trim them anyway), so retyping a setting with stray spaces does not count as a change. """ for field in fields: value = field["value"] base = original[field["key"]] if isinstance(value, str) and isinstance(base, str): changed = value.strip() != base.strip() else: changed = value != base if changed: return True return False def _settings_fields() -> list: """The global settings field list (Save writes to config.py).""" return [ {"key": "audio_format", "label": "Audio format", "kind": "choice", "value": config.AUDIO_FORMAT, "choices": list(AUDIO_FORMATS)}, {"key": "audio_bitrate", "label": "Audio bitrate", "kind": "text", "value": config.AUDIO_BITRATE, "validate": _validate_bitrate}, {"key": "language", "label": "Language", "kind": "choice", "value": config.LANGUAGE, "choices": list(LANGUAGE_CHOICES), "help": _LANGUAGE_EDIT_HINT, "validate": _validate_language}, {"key": "chunk_size", "label": "Chunk size (words)", "kind": "text", "value": str(config.CHUNK_SIZE), "validate": _validate_chunk_size}, {"key": "input_dir", "label": "Input Directory", "kind": "dir", "value": converter_mod.resolve_dir(config.INPUT_DIR, "input"), "validate": _validate_dir}, {"key": "output_dir", "label": "Output Directory", "kind": "dir", "value": converter_mod.resolve_dir(config.OUTPUT_DIR, "output"), "validate": _validate_dir}, {"key": "speed", "label": "Speed", "kind": "text", "value": str(config.SPEED), "validate": _validate_speed}, {"key": "debug", "label": "Debug", "kind": "bool", "value": config.DEBUG}, {"key": "stop_and_exit", "label": "Stop server and exit", "kind": "bool", "value": config.STOP_SERVER_AND_EXIT, "note": "Automatically stop the TTS server and exit TUI " "after generating audiobooks"}, {"key": "unload_models", "label": "Unload models", "kind": "bool", "value": config.AUDIOCPP_UNLOAD_MODELS, "note": "audio.cpp: Unload previously-loaded models before converting to prevent VRAM exhaustion."}, {"key": "audiocpp_port", "label": "audio.cpp port", "kind": "text", "value": str(_port_from_url(config.AUDIOCPP_API_URL, 8080)), "validate": _validate_port, "note": "Ports apply to servers this tool starts (local instances)"}, {"key": "faster_port", "label": "faster-qwen3-tts port", "kind": "text", "value": str(_port_from_url(config.FASTER_API_URL, 8000)), "validate": _validate_port}, {"key": "qwen_port", "label": "qwen-tts port", "kind": "text", "value": str(_port_from_url(config.QWEN_API_URL, 7860)), "validate": _validate_port}, {"key": "audiocpp_remote_url", "label": "audio.cpp remote URL", "kind": "text", "value": config.AUDIOCPP_REMOTE_URL, "validate": _validate_remote_url, "note": "Remote (externally-run) servers. Empty disables probing."}, {"key": "faster_remote_url", "label": "faster-qwen3-tts remote URL", "kind": "text", "value": config.FASTER_REMOTE_URL, "validate": _validate_remote_url}, {"key": "qwen_remote_url", "label": "qwen-tts remote URL", "kind": "text", "value": config.QWEN_REMOTE_URL, "validate": _validate_remote_url}, ] def _validate_bitrate(value: str) -> Optional[str]: """Error message for a blank audio bitrate, or None to accept it.""" if value.strip(): return None return "Audio bitrate must not be empty" def _validate_language(value: str) -> Optional[str]: """Error message for an unrecognized LANGUAGE, or None to accept it.""" try: normalize_language(value) return None except ValueError as exc: return str(exc) def _validate_chunk_size(value: str) -> Optional[str]: """Error message for an invalid CHUNK_SIZE, or None to accept it.""" try: number = int(value.strip()) except ValueError: return "Enter a whole number of words, e.g. 250" if number < 1: return "Chunk size must be at least 1" return None def _validate_speed(value: str) -> Optional[str]: """Error message for an invalid SPEED, or None to accept it.""" if _is_float(value.strip()) and float(value) > 0: return None return "Enter a positive number, e.g. 1.0" def _validate_dir(value) -> Optional[str]: """Error message for a blank directory setting, or None to accept it.""" if str(value).strip(): return None return "Directory must not be empty" def _validate_port(value: str) -> Optional[str]: """Error message for an invalid port, or None to accept it.""" try: number = int(value.strip()) except ValueError: return "Enter a port number, e.g. 8080" if not 1 <= number <= 65535: return "Port must be between 1 and 65535" return None def _validate_remote_url(value: str) -> Optional[str]: """Error message for an invalid remote URL, or None to accept it.""" try: common.normalize_remote_url(value) return None except ValueError as exc: return str(exc) def _validate_request_options(value: str) -> Optional[str]: """Error message for malformed KEY=VALUE request options, or None.""" try: common.parse_request_options(value) return None except ValueError as exc: return str(exc) def _port_from_url(url: str, default: int) -> int: """Return the port in URL, or DEFAULT when it has none/unparsable.""" try: return urllib.parse.urlsplit(url).port or default except ValueError: return default def _apply_settings(values: dict) -> None: """Write VALUES to app/converter/config.py and reload them in-memory.""" chunk_size = int(values["chunk_size"].strip()) if chunk_size < 1: raise ValueError("Chunk size must be at least 1") bitrate = values["audio_bitrate"].strip() if not bitrate: raise ValueError("Audio bitrate must not be empty") if values["audio_format"] not in AUDIO_FORMATS: raise ValueError(f"Unsupported audio format: {values['audio_format']}") speed = float(str(values["speed"]).strip()) if speed <= 0: raise ValueError("Speed must be a positive number") input_dir = str(values["input_dir"]).strip() output_dir = str(values["output_dir"]).strip() if not input_dir: raise ValueError("Input Directory must not be empty") if not output_dir: raise ValueError("Output Directory must not be empty") ports = { "qwen_port": _read_port(values, "qwen_port"), "faster_port": _read_port(values, "faster_port"), "audiocpp_port": _read_port(values, "audiocpp_port"), } remote_urls = { "QWEN_REMOTE_URL": common.normalize_remote_url( values.get("qwen_remote_url", "")), "FASTER_REMOTE_URL": common.normalize_remote_url( values.get("faster_remote_url", "")), "AUDIOCPP_REMOTE_URL": common.normalize_remote_url( values.get("audiocpp_remote_url", "")), } updates = { "AUDIO_FORMAT": values["audio_format"], "AUDIO_BITRATE": bitrate, "LANGUAGE": normalize_language(values["language"]), "CHUNK_SIZE": chunk_size, "INPUT_DIR": input_dir, "OUTPUT_DIR": output_dir, "SPEED": speed, "DEBUG": bool(values["debug"]), "STOP_SERVER_AND_EXIT": bool(values["stop_and_exit"]), "AUDIOCPP_UNLOAD_MODELS": bool(values["unload_models"]), "QWEN_API_URL": common.url_with_port( config.QWEN_API_URL, ports["qwen_port"]), "FASTER_API_URL": common.url_with_port( config.FASTER_API_URL, ports["faster_port"]), "AUDIOCPP_API_URL": common.url_with_port( config.AUDIOCPP_API_URL, ports["audiocpp_port"]), **remote_urls, } # Sync the audio.cpp server.json first: if it fails, neither the file # nor the in-memory settings are touched, so the save is not reported # as successful while the two are out of sync. _sync_audiocpp_server_port(ports["audiocpp_port"]) # update_config_value rewrites app/converter/config.py AND mirrors # each value onto the imported config module. for name, value in updates.items(): if not common.update_config_value(name, value): raise ValueError(f"Could not save {name} to " f"{common.CONFIG_PATH}") # The input/output folders changed: re-derive the converter module's # folder globals so this session's preflight/runs (and the Help text) # see the new directories without a restart. converter_mod.BOOKS_FOLDER = converter_mod.resolve_dir( input_dir, "input") converter_mod.AUDIOBOOKS_FOLDER = converter_mod.resolve_dir( output_dir, "output") # Ports/URLs may have changed: the cached backend statuses are stale. invalidate_detect_cache() def _read_port(values: dict, key: str) -> int: """Parse a port field value, raising ValueError on a bad number.""" try: number = int(values[key].strip()) except (KeyError, ValueError) as exc: raise ValueError(f"Enter a valid port for {key}") from exc if not 1 <= number <= 65535: raise ValueError("Port must be between 1 and 65535") return number def _sync_audiocpp_server_port(port: int) -> None: """Rewrite the audio.cpp server.json 'port' to PORT when it exists. A missing checkout/server.json is a no-op (the config URL still changes; the file is regenerated on reconfigure). An existing server.json that cannot be updated raises, so the save is not reported as successful while the two are out of sync. """ checkout = audiocpp_backend.find_local_checkout() if checkout is None: return server_json = checkout / "server.json" if not server_json.exists(): return if not audiocpp_backend.update_server_config_port(port): raise ValueError( f"Could not update {server_json}; the audio.cpp port was " "left as-is") def _prepare_run_config(backend: str, kwargs: dict ) -> Optional[runview.RunConfig]: """Build the run view's config from the accepted conversion kwargs. A remote conversion (``api_url``) targets an externally-run server, so no autostart is attempted and the managed instance's setup state is irrelevant. Otherwise, when the convert menu recorded an ``autostart`` server (the server was not running), the run view boots it first; a managed server whose port is already occupied by a server this tool did not start is left alone but flagged with a notice. Returns None when the backend disappeared between the menu and the dispatch. """ label = backend info = get(backend) if info is not None: label = info.label log_path = str(LOGS_FOLDER / f"audiobook_{datetime.now():%Y%m%d}.log") # The run view points failures at this file, so make sure it exists # from the moment a run starts — even when the run dies before the # converter's setup_logging creates it. LOGS_FOLDER.mkdir(parents=True, exist_ok=True) Path(log_path).touch() autostart = kwargs.pop("autostart", None) # A running managed qwen server hosting another model than the run's # selection: stop it and boot the new model before converting. restart_name = kwargs.pop("restart_server", None) # The run-view behavior toggle (not a converter kwarg): stop the server # and quit the TUI once the generation ends. stop_and_exit = bool(kwargs.pop("stop_and_exit", False)) # book_files/planned travel on the dedicated RunConfig fields; keeping # them in kwargs too would collide with convert()'s named parameters. book_files = kwargs.pop("book_files", None) or [] planned = kwargs.pop("planned", None) or [] api_url = kwargs.get("api_url") if api_url: identity = _remote_identity(backend, kwargs) return runview.RunConfig( backend=backend, backend_label=f"{label} [remote]", kwargs=kwargs, book_files=book_files, planned=planned, server_url=api_url, server_identity=identity, log_path=log_path, stop_and_exit=stop_and_exit) status = next((s for s in detect_all(refresh=True) if s.key == backend), None) notice = "" spec: Optional[ServerSpec] = None if autostart: spec = _find_spec(autostart) elif status is not None: spec = _select_spec(status, kwargs) if spec is not None and common.server_running(spec.url) \ and not servers.alive(spec.name): notice = (f"a server this tool did not start is running at " f"{spec.url} — the conversion will talk to it") if autostart and spec is None: # The recorded server vanished (backend reconfigured meanwhile): # converting without it is still meaningful, so continue. notice = (f"no server named '{autostart}' — starting it was skipped") if restart_name and spec is None: spec = _find_spec(restart_name) if spec is None: notice = (f"no server named '{restart_name}' — the model " "switch restart was skipped") if spec is not None and backend == BACKEND_QWEN: # One demo server hosts one model: aim the spec at the model this # run selected (same URL/port, matching probe identity), so an # autostart or model-switch restart boots exactly what the run # needs instead of the Start/Stop menu's default model. spec = qwen_backend.build_spec(_qwen_wanted_model(kwargs)) return runview.RunConfig( backend=backend, backend_label=label, kwargs=kwargs, book_files=book_files, planned=planned, server_name=spec.name if spec is not None else None, server_url=spec.url if spec is not None else None, server_identity=spec.identity if spec is not None else None, autostart_spec=spec if (autostart or restart_name) else None, restart_first=bool(restart_name) and spec is not None, log_path=log_path, notice=notice, stop_and_exit=stop_and_exit) def _qwen_wanted_model(kwargs: dict) -> str: """The qwen model a conversion with these kwargs needs hosted. One demo server hosts one model; the selected voice mode picks it (mirrors ``voice_mode_for``): instructions design the voice (VoiceDesign), a reference .wav clones (Base), otherwise built-in speakers (CustomVoice). """ if (kwargs.get("instructions") or "").strip(): return "VoiceDesign" return "Base" if kwargs.get("clone") else "CustomVoice" def _remote_identity(backend: str, kwargs: dict) -> Optional[str]: """The probe identity of the remote server a conversion targets.""" if backend == BACKEND_AUDIOCPP: return backend_probe.IDENTITY_AUDIOCPP if backend == BACKEND_QWEN: wanted = _qwen_wanted_model(kwargs) return {"CustomVoice": backend_probe.IDENTITY_QWEN_CUSTOM, "Base": backend_probe.IDENTITY_QWEN_CLONE, "VoiceDesign": backend_probe.IDENTITY_QWEN_DESIGN}[wanted] if backend == BACKEND_FASTER: return backend_probe.IDENTITY_FASTER return None def _add_autostart(cmd: tuple, statuses) -> Optional[str]: """Auto-start the conversion's target server when it isn't running. Records the chosen server spec name as ``kwargs['autostart']`` for ``_prepare_run_config`` to act on. The user already accepted the run on the Generate! screen, so no start-server prompt is asked here — the server is simply started. Remote conversions (an ``api_url`` in the kwargs) never autostart: the server is external to this tool. The single-port qwen backend additionally checks the model the running server hosts against the one this run selected: a managed server hosting another model is recorded in ``kwargs['restart_server']`` (stopped and rebooted with the new model before converting), while a foreign server with the wrong model refuses the run — an explanatory message is returned for the caller to flash. Returns None when no message is owed. """ _, key, kwargs = cmd if kwargs.get("api_url"): return None status = next((s for s in statuses if s.key == key), None) if status is None or not status.servers: return None spec = _select_spec(status, kwargs) if spec is None: return None if not common.server_running(spec.url): kwargs["autostart"] = spec.name return None if status.key != BACKEND_QWEN or len(status.servers) != 1: return None wanted = _qwen_wanted_model(kwargs) running = qwen_backend.model_for_identity( backend_probe.identify_server(spec.url)) if running == wanted: return None if servers.alive(spec.name): # Ours: the run view stops it and boots the newly-selected model. kwargs["restart_server"] = spec.name return None return (f"a server this tool did not start is running at {spec.url} " f"hosting {running or 'an unknown'} — this run needs " f"{wanted}. Stop that server first, or convert with it by " f"picking {running} as the Model.") def _select_spec(status, kwargs) -> Optional[ServerSpec]: """The server spec this conversion needs (qwen has exactly one).""" return status.servers[0] if status.servers else None def _find_spec(name: str) -> Optional[ServerSpec]: """Look up a server spec by name across every backend's detect().""" for st in detect_all(refresh=True): for spec in st.servers: if spec.name == name: return spec return None def _list_wavs(directory) -> list: """Return the .wav file Paths directly inside DIRECTORY (best-effort).""" try: path = Path(directory) if not path.is_dir(): return [] return sorted((p for p in path.iterdir() if p.is_file() and p.suffix.lower() == ".wav"), key=lambda p: p.name.lower()) except OSError: return [] def _list_voices(voice_dir: str) -> list: """Return sorted .wav stems in VOICE_DIR (best-effort).""" return [p.stem for p in _list_wavs(voice_dir)] def _is_float(value: str) -> bool: try: float(value) return True except ValueError: return False