#!/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 (console tails under ``tui.suspend``); 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 re import shutil import urllib.parse from datetime import datetime from pathlib import Path from typing import Callable, Optional, Tuple import audiobook from backends import ( REGISTRY, BackendStatus, ServerSpec, common, detect_all, 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.converter import ( AUDIO_FORMATS, AudiobookConverter, LOGS_FOLDER, voice_mode_for, ) from converter.tts import ( AUDIOCPP_FAMILY_QWEN3_TTS, BACKEND_AUDIOCPP, BACKEND_FASTER, BACKEND_QWEN, normalize_language, ) from ui import runview, 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 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 and starting/stopping its servers need it on # this machine. if any(st.installed or st.running for st in statuses): options.insert(0, ("Convert books", "convert")) if any(st.installed for st in statuses): options.append(("Start/Stop Backend Servers", "server")) options.append(("Settings", "settings")) options.append(("Quit", "quit")) choice = tui.menu( self.stdscr, "tts-audiobook-generator", options, back_value=tui.Wizard.BACK, table_title="Backend status", 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 == "server": return self.screen_server return self.screen_settings # -- configure / install / uninstall -------------------------------- def screen_configure(self): """One flat menu of backend setup/configure/cleanup actions. Options are populated from the detected statuses: install (any uninstalled backend), configure each installed backend, download/delete audio.cpp models (when a server.json references models on/off disk), and uninstall. Selecting one pushes the next screen; Esc pops back to the main menu. """ 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] options = [(f"Configure {info.label}", ("configure", info.key)) for info in installed] if any(info.key not in by_key or not by_key[info.key].installed for info in REGISTRY): options.append(("Install Backend", "install")) audiocpp_status = by_key.get("audiocpp") missing = [] if audiocpp_status is not None and audiocpp_status.installed: checkout = audiocpp_backend.find_local_checkout() server_json = checkout / "server.json" if checkout else None if server_json is not None and server_json.exists(): missing = audiocpp_backend.missing_model_entries( server_json) if missing: options.append(("Download Missing Models (audio.cpp)", "download_models")) if installed: options.append(("Uninstall Backend", "uninstall")) choice = tui.menu( self.stdscr, "Configure backends", options, back_value=tui.Wizard.BACK, help_lines=["Install, configure, or remove a TTS backend."], table_title="Backend status", 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 == "uninstall": return self.screen_uninstall if choice == "download_models": _download_models_action(self.stdscr) 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 wizard as a leaf screen of the stack. 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(): 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") return tui.Wizard.BACK return screen def screen_install(self): info = self._pick_backend(installed_only=False) if info is None: return tui.Wizard.BACK return self.screen_setup(info) def screen_uninstall(self): info = self._pick_backend(installed_only=True) if info is None: return tui.Wizard.BACK with tui.suspend(self.stdscr): info.uninstall() 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 is listed (the install list); with it True only the currently-installed ones are (the uninstall list). 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 by_key.get(info.key) is not None and by_key[info.key].installed] else: candidates = [info for info in REGISTRY if by_key.get(info.key) is None or not by_key[info.key].installed] 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_title="Backend status", 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, "Convert books", 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 _add_autostart(cmd, statuses) try: ok = _preflight(self.stdscr, cmd) except _BackToForm: continue if not ok: return tui.Wizard.BACK self._run_conversion(cmd[1], cmd[2]) return tui.Wizard.BACK def _run_conversion(self, backend: str, kwargs: dict) -> None: """Run a conversion in the full-screen run view on this session. 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 view = runview.RunView(self.stdscr, run_config) try: 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 # -- settings ------------------------------------------------------- def screen_settings(self): result = tui.form(self.stdscr, "Settings", _settings_fields(), back_value=tui.Wizard.BACK) if result is tui.Wizard.BACK or result is None: return tui.Wizard.BACK try: _apply_settings(result) except ValueError as exc: tui.flash(self.stdscr, str(exc), "err") return tui.Wizard.BACK tui.flash(self.stdscr, "Settings saved.", "ok") return tui.Wizard.BACK # -- servers -------------------------------------------------------- def screen_server(self): 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 = [(st.label, st.key) for st in candidates] key = tui.menu(self.stdscr, "Start / Stop a server", options, back_value=tui.Wizard.BACK, table_title="Backend status", table_rows=_status_rows(statuses), notice_lines=_notice_lines()) if key is tui.Wizard.BACK: return tui.Wizard.BACK status = next((s for s in statuses if s.key == key), None) if status is None: return tui.Wizard.BACK specs = status.servers if not specs: tui.flash(self.stdscr, f"{status.label} has no server " "configured. Run 'Configure backends' first.") return tui.Wizard.BACK if len(specs) == 1: return functools.partial(self._server_action, specs[0]) return functools.partial(self._server_spec, status, specs) def _server_spec(self, status, specs): options = [(f"{s.name} ({'running' if common.server_running(s.url) else 'stopped'})", s.name) for s in specs] name = tui.menu(self.stdscr, f"{status.label} server", options, back_value=tui.Wizard.BACK) if name is tui.Wizard.BACK: return tui.Wizard.BACK spec = next((s for s in specs if s.name == name), None) if spec is None: return tui.Wizard.BACK return functools.partial(self._server_action, spec) def _server_action(self, spec): running = common.server_running(spec.url) action = tui.menu( self.stdscr, f"{spec.name} ({'running' if running else 'stopped'})", [("Start", "start"), ("Stop", "stop")], back_value=tui.Wizard.BACK) if action is tui.Wizard.BACK: return tui.Wizard.BACK with tui.suspend(self.stdscr): if action == "start": servers.start(spec) else: servers.stop(spec.name) return tui.Wizard.BACK 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 suspends curses to stream the downloads, then flashes a result — instead of silently returning to the main menu. 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 with tui.suspend(stdscr): audiocpp_backend.install_models(checkout, guidance) tui.flash(stdscr, "Model download finished. See the output above for " "any warnings.", "ok") 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 'installed' (orange/warn) when the backend is present on disk, 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.installed: if status.models_missing and not status.running: return ("installed (models missing)", "warn", "body") return ("installed", "warn", "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 _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: speaker or clone .wav; faster: voice), plus the shared output settings. 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 = {} for key, _label, st, remote in entries: if remote: if st.key == BACKEND_AUDIOCPP: built = _audiocpp_fields( stdscr, api_url=st.remote_urls.get("audiocpp")) elif st.key == BACKEND_QWEN: built = _qwen_fields(remote_modes=st.remote_models, urls=st.remote_urls) elif st.key == BACKEND_FASTER: built = _faster_fields( stdscr, api_url=st.remote_urls.get("faster")) 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] default = config.BACKEND if config.BACKEND in builders \ else choices[0][1] fields = [{ "key": "backend", "label": "Backend", "kind": "choice", "value": default, "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")) 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 # Sentinel value the audio.cpp Voice field uses for "no --voice" (the # built-in CustomVoice speaker); mapped to None when the form returns. _AUDIOCPP_BUILTIN_SPEAKER = "(built-in speaker)" 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 def _common_fields() -> list: """Field dicts for output format, speed, single-file, and debug. The single-file field is hidden for m4b (always a single file with embedded chapter markers), so its "visible" callable reads the live output-format value from the field list. """ fmt_default = config.AUDIO_FORMAT \ if config.AUDIO_FORMAT in AUDIO_FORMATS else AUDIO_FORMATS[0] return [ {"key": "output_format", "label": "Output format", "kind": "choice", "value": fmt_default, "choices": list(AUDIO_FORMATS)}, {"key": "speed", "label": "Speed", "kind": "text", "value": "1.0", "validate": lambda s: None if (_is_float(s) and float(s) > 0) else "Enter a positive number, e.g. 1.0"}, {"key": "single_file", "label": "Combine all chapters", "kind": "bool", "value": False, "visible": lambda fs: _field_value(fs, "output_format") != "m4b"}, {"key": "debug", "label": "Debug", "kind": "bool", "value": False}, ] def _common_kwargs(values: dict) -> dict: """Map the common form fields to converter keyword arguments.""" output_format = values["output_format"] return { "output_format": output_format, "speed": float(values["speed"]), "single_file": bool(values["single_file"]) and output_format != "m4b", "debug": bool(values["debug"]), } def _audiocpp_fields(stdscr, api_url: Optional[str] = None) -> 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. 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. 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 not local and not entry["family"]: # Servers predating the family field omit it; mirror the # converter's default: unknown family means qwen3_tts. entry["family"] = AUDIOCPP_FAMILY_QWEN3_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) 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, "model_id") return next((m for m in models if m.get("id") == model_id), models[0]) def model_task(fields) -> str: return model_entry(fields).get("task", "tts") def model_family(fields) -> str: return model_entry(fields).get("family") or "" def reset_voice(fields) -> None: """Re-point the Voice field at the newly selected model's voice.""" voice_field = next(f for f in fields if f.get("key") == "audiocpp_voice") if model_task(fields) == "vdes": voice_field["value"] = None elif model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS: voice_field["value"] = _AUDIOCPP_BUILTIN_SPEAKER else: voices = voices_for(_field_value(fields, "model_id")) voice_field["value"] = voices[0] if voices else "" def voice_choices(fields) -> list: if model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS: return [(_AUDIOCPP_BUILTIN_SPEAKER, _AUDIOCPP_BUILTIN_SPEAKER)] \ + [(v, v) for v in voices_for(_field_value(fields, "model_id"))] return [(v, v) for v in voices_for(_field_value(fields, "model_id"))] model_ids = [m.get("id") for m in models] default_model = config.AUDIOCPP_MODEL_ID \ if config.AUDIOCPP_MODEL_ID in model_ids else model_ids[0] default_entry = next((m for m in models if m.get("id") == default_model), models[0]) initial_voice = _AUDIOCPP_BUILTIN_SPEAKER if default_entry.get("task") == "vdes": initial_voice = None elif default_entry.get("family") != AUDIOCPP_FAMILY_QWEN3_TTS: initial = voices_for(default_model) initial_voice = initial[0] if initial else "" fields = [ {"key": "model_id", "label": "Model", "kind": "choice", "value": default_model, "choices": [(f"{m.get('id')} ({m.get('family') or '?'}, " f"{m.get('task') or 'tts'})", m.get("id")) for m in models], "on_change": reset_voice}, {"key": "audiocpp_voice", "label": "Voice", "kind": "choice", "value": initial_voice, "choices": lambda fs: voice_choices(fs), "visible": lambda fs: model_task(fs) != "vdes", "validate": lambda value: None if (model_family(fields) == AUDIOCPP_FAMILY_QWEN3_TTS or value) else "This model needs a voice — pick one or switch models"}, {"key": "instructions", "label": "Instructions", "kind": "text", "value": config.AUDIOCPP_INSTRUCTIONS, "validate": lambda value: None if (model_task(fields) != "vdes" or str(value).strip()) else "Describe the voice, e.g. 'A warm female narrator'"}, ] def mapper(result) -> Optional[tuple]: model_id = result["model_id"] voice = result["audiocpp_voice"] if voice == _AUDIOCPP_BUILTIN_SPEAKER or not voice: voice = None entry = next((m for m in models if m.get("id") == model_id), {}) if entry.get("task") == "vdes": voice = None instructions = (result["instructions"] or "").strip() or None kwargs = { "model_id": model_id, "voice": voice, "instructions": instructions, **_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) -> Optional[tuple]: """qwen-specific fields and a result mapper for the Convert form. Returns ``(fields, mapper)`` where FIELDS are the qwen options (Voice mode / Speaker / Clone .wav path) and MAPPER turns a submitted form values dict into the qwen converter kwargs. qwen always has options to offer, so it never signals unavailability. For the managed entry REMOTE_MODES/URLS are None and the mode picker offers both modes, targeting the configured local URLs. For a "[remote]" entry REMOTE_MODES names which demos answered remotely ("CustomVoice" and/or "Base") and URLS maps "qwen-custom"/"qwen-clone" to their URLs: the mode picker is limited to the available demos, and the mapper passes the matching remote URL as ``api_url``. """ remote_modes = list(remote_modes or []) urls = dict(urls or {}) mode_choices = [] if urls.get("qwen-custom") or not remote_modes: mode_choices.append(("Built-in speaker", "custom")) if urls.get("qwen-clone") or not remote_modes: mode_choices.append(("Clone from a .wav file", "clone")) default_mode = mode_choices[0][1] if mode_choices else "custom" speakers = list(qwen_backend.QWEN_SPEAKERS) default_speaker = config.SPEAKER if config.SPEAKER in speakers \ else speakers[0] fields = [ {"key": "mode", "label": "Voice mode", "kind": "choice", "value": default_mode, "choices": mode_choices}, {"key": "speaker", "label": "Speaker", "kind": "choice", "value": default_speaker, "choices": speakers, "visible": lambda fs: _field_value(fs, "mode") == "custom"}, {"key": "clone", "label": "Clone .wav path", "kind": "text", "value": "", "validate": lambda s: None if (s and Path(s).is_file() and s.lower().endswith(".wav")) else "Enter the path to an existing .wav file", "visible": lambda fs: _field_value(fs, "mode") == "clone"}, ] def mapper(result) -> Optional[tuple]: clone = result["clone"].strip() if result["mode"] == "clone" else None speaker = result["speaker"] if result["mode"] == "custom" and speaker != config.SPEAKER: # Persist the speaker choice for this and future runs (mirrors # the qwen setup wizard), so the converter picks it up at # request time. common.update_config_value("SPEAKER", speaker) config.SPEAKER = speaker kwargs = {"clone": clone, **_common_kwargs(result)} if urls: api_url = urls.get("qwen-clone") if result["mode"] == "clone" \ else urls.get("qwen-custom") 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) -> Optional[tuple]: """faster-specific fields and a result mapper for the Convert form. Returns ``(fields, mapper)`` where FIELDS are the faster options (Voice, 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. 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": "faster_voice", "label": "Voice", "kind": "text", "value": config.FASTER_VOICE, "validate": lambda s: None if s.strip() else "Enter a voice name"}, ] else: default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \ next(iter(voices)) fields = [ {"key": "faster_voice", "label": "Voice", "kind": "choice", "value": default, "choices": [(k, k) for k in voices]}, ] def mapper(result) -> Optional[tuple]: voice = result["faster_voice"].strip() \ if isinstance(result["faster_voice"], str) \ else result["faster_voice"] 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_fields() -> list: """The global output-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": "text", "value": config.LANGUAGE, "validate": _validate_language}, {"key": "chunk_size", "label": "Chunk size (words)", "kind": "text", "value": str(config.CHUNK_SIZE), "validate": _validate_chunk_size}, {"key": "unload_models", "label": "Unload models", "kind": "bool", "value": config.AUDIOCPP_UNLOAD_MODELS, "note": "audio.cpp only. Ask the server to unload resident models " "before converting."}, {"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_custom_port", "label": "qwen-tts CustomVoice port", "kind": "text", "value": str(_port_from_url(config.QWEN_API_URL, 7860)), "validate": _validate_port}, {"key": "qwen_clone_port", "label": "qwen-tts Base port", "kind": "text", "value": str(_port_from_url(config.CLONE_API_URL, 7861)), "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. The hub probes each URL and " "offers a \"[remote]\" backend entry when one answers. " "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_custom_remote_url", "label": "qwen-tts CustomVoice remote URL", "kind": "text", "value": config.QWEN_REMOTE_URL, "validate": _validate_remote_url}, {"key": "qwen_clone_remote_url", "label": "qwen-tts Base remote URL", "kind": "text", "value": config.CLONE_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_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 _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']}") ports = { "qwen_custom_port": _read_port(values, "qwen_custom_port"), "qwen_clone_port": _read_port(values, "qwen_clone_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_custom_remote_url", "")), "CLONE_REMOTE_URL": common.normalize_remote_url( values.get("qwen_clone_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, "AUDIOCPP_UNLOAD_MODELS": bool(values["unload_models"]), "QWEN_API_URL": common.url_with_port( config.QWEN_API_URL, ports["qwen_custom_port"]), "CLONE_API_URL": common.url_with_port( config.CLONE_API_URL, ports["qwen_clone_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, } _write_config(updates) for name, value in updates.items(): setattr(config, name, value) _sync_audiocpp_server_port(ports["audiocpp_port"]) 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): raise ValueError(f"Enter a valid port for {key}") 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 _write_config(updates: dict) -> None: """Rewrite the ``NAME = value`` lines for UPDATES in app/converter/config.py. Only the value of each named assignment changes: the indentation, the quotes (double, matching the file's style) and any trailing comment on the line are preserved. Every other line is left untouched. """ path = Path(config.__file__).resolve() text = path.read_text(encoding="utf-8") for name, value in updates.items(): rendered = str(value) if isinstance(value, int) else f'"{value}"' pattern = re.compile( rf"^(\s*{re.escape(name)}\s*=\s*)(\S*)(\s*(#.*))?$", re.MULTILINE) text, count = pattern.subn( lambda m, rendered=rendered: f"{m.group(1)}{rendered}{m.group(3) or ''}", text) if count != 1: raise ValueError(f"Could not find {name} in {path}") path.write_text(text, encoding="utf-8") 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") autostart = kwargs.pop("autostart", None) 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=kwargs.get("book_files") or [], planned=kwargs.get("planned") or [], server_url=api_url, server_identity=identity, log_path=log_path) status = next((s for s in detect_all() 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") return runview.RunConfig( backend=backend, backend_label=label, kwargs=kwargs, book_files=kwargs.get("book_files") or [], planned=kwargs.get("planned") or [], 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 else None, log_path=log_path, notice=notice) 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: return backend_probe.IDENTITY_QWEN_CLONE if kwargs.get("clone") \ else backend_probe.IDENTITY_QWEN_CUSTOM if backend == BACKEND_FASTER: return backend_probe.IDENTITY_FASTER return None def _add_autostart(cmd: tuple, statuses) -> None: """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. Mode-aware for qwen (custom vs clone). Remote conversions (an ``api_url`` in the kwargs) never autostart: the server is external to this tool. """ _, key, kwargs = cmd if kwargs.get("api_url"): return status = next((s for s in statuses if s.key == key), None) if status is None or not status.servers: return spec = _select_spec(status, kwargs) if spec is None: return if common.server_running(spec.url): return kwargs["autostart"] = spec.name def _select_spec(status, kwargs) -> Optional[ServerSpec]: """The server spec this conversion needs (mode-aware for qwen).""" if status.key == BACKEND_QWEN: wanted = "qwen-clone" if kwargs.get("clone") else "qwen-custom" return next((s for s in status.servers if s.name == wanted), None) 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(): for spec in st.servers: if spec.name == name: return spec return None def _list_voices(voice_dir: str) -> list: """Return sorted .wav stems in VOICE_DIR (best-effort).""" try: path = Path(voice_dir) if not path.is_dir(): return [] return sorted( (p.stem for p in path.iterdir() if p.is_file() and p.suffix.lower() == ".wav"), key=str.lower, ) except OSError: return [] def _is_float(value: str) -> bool: try: float(value) return True except ValueError: return False