diff options
Diffstat (limited to 'ui/hub.py')
| -rw-r--r-- | ui/hub.py | 637 |
1 files changed, 0 insertions, 637 deletions
diff --git a/ui/hub.py b/ui/hub.py deleted file mode 100644 index b2a2b08..0000000 --- a/ui/hub.py +++ /dev/null @@ -1,637 +0,0 @@ -#!/usr/bin/env python3 -"""The TUI main menu 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, set up a new backend, or configure an existing one. -Each backend's setup wizard runs in its own curses session, so the hub -collects a "command" inside its own wrapper, returns to the plain terminal, -and then dispatches — no nested curses sessions. - -Esc on the main menu quits the hub. Esc inside a sub-menu falls back to the -main menu. -""" - -import json -import re -from pathlib import Path -from typing import 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 converter import config -from converter.converter import AUDIO_FORMATS -from converter.tts import ( - AUDIOCPP_FAMILY_QWEN3_TTS, - BACKEND_AUDIOCPP, - BACKEND_FASTER, - BACKEND_QWEN, - normalize_language, -) -from ui import tui - -_GO_BACK = object() - - -def run() -> int: - """Run the hub menu loop until the user quits. Returns exit code.""" - import curses - while True: - try: - command = curses.wrapper(_hub_menu) - except tui.WizardCancelled: - return 0 - except KeyboardInterrupt: - return 130 - if command is None: - return 0 - kind = command[0] - if kind == "quit": - return 0 - if kind == "setup": - info = get(command[1]) - if info is not None: - info.setup_tui() - elif kind == "configure": - info = get(command[1]) - if info is not None and command[2] < len(info.configure_actions): - info.configure_actions[command[2]].run() - elif kind == "convert": - _run_conversion(command[1], command[2]) - elif kind == "server": - _run_server_action(command[1], command[2]) - - -def _hub_menu(stdscr) -> Optional[tuple]: - """Show the main menu; return a command tuple, or None to quit.""" - while True: - statuses = detect_all() - options = [("Set up a backend...", "setup")] - if any(st.installed or st.running for st in statuses): - options.insert(0, ("Convert books...", "convert")) - options.append(("Configure a backend...", "configure")) - options.append(("Server...", "server")) - options.append(("Settings...", "settings")) - options.append(("Quit", "quit")) - rows = [(st.label, *_status_mark(st)) for st in statuses] - choice = tui.menu( - stdscr, "tts-audiobook-generator", options, - table_title="Backend status", table_rows=rows) - if choice is None or choice == "quit": - return None - if choice == "convert": - cmd = _convert_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "setup": - cmd = _setup_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "configure": - cmd = _configure_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "server": - cmd = _server_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "settings": - _settings_menu(stdscr) - - -def _setup_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend to set up. Returns ("setup", key) or None to go back.""" - by_key = {st.key: st for st in statuses} - options = [(f"{info.label} ({_status_mark(by_key.get(info.key))[0]})", - info.key) for info in REGISTRY] - choice = tui.menu(stdscr, "Set up a backend", options, - back_value=_GO_BACK, - help_lines=["Clone/build/install a backend so you can " - "convert with it."]) - if choice is _GO_BACK or choice is None: - return None - return ("setup", choice) - - -def _configure_menu(stdscr, statuses) -> Optional[tuple]: - """Pick an installed backend and one of its configure actions.""" - 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] - if not installed: - tui.flash(stdscr, "No backend is installed yet — use 'Set up a " - "backend' first.") - return None - options = [(info.label, info.key) for info in installed] - key = tui.menu(stdscr, "Configure a backend", options, back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - info = get(key) - actions = info.configure_actions - choice = tui.menu( - stdscr, f"Configure {info.label}", - [(action.label, index) for index, action in enumerate(actions)], - back_value=_GO_BACK) - if choice is _GO_BACK or choice is None: - return None - return ("configure", key, choice) - - -def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]: - """Map a backend's state to (status_text, status_kind, name_kind). - - 'running' (green/ok) takes priority — an external server is already up; - 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). - 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: - return ("running", "ok", "body") - if status is not None and status.installed: - return ("installed", "warn", "body") - return ("unavailable", "err", "dim") - - -def _convert_menu(stdscr, statuses) -> Optional[tuple]: - """Pick an available backend and collect per-backend run settings.""" - available = [st for st in statuses if st.ready or st.running] - options = [(st.label, st.key) for st in available] - if not available: - choice = tui.menu( - stdscr, "No backend is available", - [("Set up a backend...", "__setup__")], - help_lines=["Set up a backend (clone/build/configure) before " - "converting."]) - if choice == "__setup__": - return _setup_menu(stdscr, statuses) - return None - options.append(("Set up a backend...", "__setup__")) - key = tui.menu(stdscr, "Convert books with...", options, - back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - if key == "__setup__": - return _setup_menu(stdscr, statuses) - if key == BACKEND_AUDIOCPP: - cmd = _convert_audiocpp(stdscr, statuses) - elif key == BACKEND_QWEN: - cmd = _convert_qwen(stdscr) - elif key == BACKEND_FASTER: - cmd = _convert_faster(stdscr) - else: - return None - if cmd is None: - return None - _add_autostart(stdscr, cmd, statuses) - return cmd - - -def _convert_audiocpp(stdscr, statuses) -> Optional[tuple]: - """Collect audio.cpp run settings by reading ./audio.cpp/server.json.""" - checkout = audiocpp_backend.find_local_checkout() - server_json = checkout / "server.json" if checkout else None - if not server_json or not server_json.exists(): - tui.flash(stdscr, "No server.json found in the audio.cpp checkout. " - "Run 'Set up a backend' 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 - model_options = [(f"{m.get('id')} ({m.get('family')}, {m.get('task', 'tts')})", - m.get("id")) for m in models] - model_id = tui.menu(stdscr, "Select the audio.cpp model to use", - model_options, back_value=_GO_BACK) - if model_id is _GO_BACK or model_id is None: - return None - entry = next((m for m in models if m.get("id") == model_id), {}) - family = entry.get("family") - task = entry.get("task", "tts") - - # Voice: optional for qwen3_tts (built-in speaker), required otherwise. - voice = None - voice_dir = data.get("voice_dir") - voices = _list_voices(voice_dir) if voice_dir else [] - if task == "vdes": - # Voice design: no voice, instructions required. - pass - elif family == AUDIOCPP_FAMILY_QWEN3_TTS: - # Speaker mode available; voice optional. - if voices: - opts = [("(built-in speaker)", None)] + [(v, v) for v in voices] - voice = tui.menu(stdscr, "Voice", opts, back_value=_GO_BACK) - if voice is _GO_BACK: - return None - else: - voice = None - else: - if not voices: - tui.flash(stdscr, f"This model needs a --voice but voice_dir " - f"{voice_dir} has no .wav voices. Reconfigure " - "audio.cpp or add voices.") - return None - voice = tui.menu(stdscr, "Select the voice to clone", [(v, v) for v in voices], - back_value=_GO_BACK) - if voice is _GO_BACK or voice is None: - return None - - # Instructions: required for vdes, optional otherwise. - instructions = None - if task == "vdes": - instructions = tui.line_edit( - stdscr, "Voice design instructions (required for this model)", - config.AUDIOCPP_INSTRUCTIONS, - validate=lambda s: None if s.strip() - else "Describe the voice, e.g. 'A warm female narrator'", - back_value=_GO_BACK) - if instructions is _GO_BACK: - return None - else: - instructions = tui.line_edit( - stdscr, "Style instructions (optional, blank for none)", - config.AUDIOCPP_INSTRUCTIONS, back_value=_GO_BACK) - if instructions is _GO_BACK: - return None - if not instructions.strip(): - instructions = None - - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_AUDIOCPP, { - "model_id": model_id, "voice": voice, "instructions": instructions, - **common_kw, - }) - - -def _convert_qwen(stdscr) -> Optional[tuple]: - """Collect qwen run settings: built-in speaker or clone a .wav.""" - mode = tui.menu( - stdscr, "qwen-tts mode", - [("Custom voice (built-in speaker)", "custom"), - ("Voice clone from a .wav file", "clone")], - back_value=_GO_BACK, - help_lines=[f"Speaker: {config.SPEAKER} (change it via Configure " - "qwen-tts)"]) - if mode is _GO_BACK or mode is None: - return None - clone = None - if mode == "clone": - clone = tui.line_edit( - stdscr, "Path to a reference .wav (10-15s is ideal)", - "", - 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", - back_value=_GO_BACK) - if clone is _GO_BACK: - return None - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_QWEN, {"clone": clone, **common_kw}) - - -def _convert_faster(stdscr) -> Optional[tuple]: - """Collect faster run settings: pick a voice from voices.json.""" - checkout = faster_backend._checkout() - voices_json = checkout / "voices.json" - if not voices_json.exists(): - tui.flash(stdscr, f"No voices.json at {voices_json}. Run 'Set up a " - "backend' for faster first.") - return None - 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 - default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \ - next(iter(voices)) - voice = tui.menu( - stdscr, "Select the voice to clone", - [(k, k) for k in voices], - default_index=list(voices).index(default), back_value=_GO_BACK) - if voice is _GO_BACK or voice is None: - return None - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_FASTER, {"voice": voice, **common_kw}) - - -def _common_options(stdscr) -> Optional[dict]: - """Collect output format, speed, single-file, chunk, debug.""" - fmt_options = [(f, f) for f in AUDIO_FORMATS] - fmt_default = AUDIO_FORMATS.index(config.AUDIO_FORMAT) \ - if config.AUDIO_FORMAT in AUDIO_FORMATS else 0 - output_format = tui.menu(stdscr, "Output format", fmt_options, - default_index=fmt_default, back_value=_GO_BACK) - if output_format is _GO_BACK or output_format is None: - return None - speed_text = tui.line_edit( - stdscr, "Playback speed (1.0 = normal)", "1.0", - validate=lambda s: None if (_is_float(s) and float(s) > 0) - else "Enter a positive number, e.g. 1.0", - back_value=_GO_BACK) - if speed_text is _GO_BACK: - return None - single_file = tui.confirm(stdscr, "Combine all chapters into one file?", - default=False, cancel_value=_GO_BACK) - if single_file is _GO_BACK: - return None - chunk = tui.confirm(stdscr, "Force client-side chunking (--chunk)?", - default=False, cancel_value=_GO_BACK) - if chunk is _GO_BACK: - return None - debug = tui.confirm(stdscr, "Debug mode (dump per-chunk audio/text)?", - default=False, cancel_value=_GO_BACK) - if debug is _GO_BACK: - return None - return { - "output_format": output_format, - "speed": float(speed_text), - "single_file": single_file, - "chunk": chunk, - "debug": debug, - } - - -# --------------------------------------------------------------------------- -# Settings menu (global output options -> converter/config.py) -# --------------------------------------------------------------------------- - -def _settings_menu(stdscr) -> None: - """Edit the global output settings; Save writes them back to config.py.""" - fields = [ - {"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}, - ] - result = tui.form(stdscr, "Settings", fields, back_value=_GO_BACK) - if result is None or result is _GO_BACK: - return - try: - _apply_settings(result) - except ValueError as exc: - tui.flash(stdscr, str(exc), "err") - return - tui.flash(stdscr, "Settings saved.", "ok") - - -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 _apply_settings(values: dict) -> None: - """Write VALUES to 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']}") - updates = { - "AUDIO_FORMAT": values["audio_format"], - "AUDIO_BITRATE": bitrate, - "LANGUAGE": normalize_language(values["language"]), - "CHUNK_SIZE": chunk_size, - } - _write_config(updates) - for name, value in updates.items(): - setattr(config, name, value) - - -def _write_config(updates: dict) -> None: - """Rewrite the ``NAME = value`` lines for UPDATES in 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 _run_conversion(backend: str, kwargs: dict) -> None: - """Run a conversion in the plain console (after the TUI returns). - - When the convert menu recorded an ``autostart`` server (the user opted to - have the hub start it), spawn it now and abort the conversion if it does - not come up. After the conversion, offer to stop a server we started. - """ - autostart = kwargs.pop("autostart", None) - status = next((s for s in detect_all() if s.key == backend), None) - if status is not None and not status.ready and not status.running: - print(f"[WARNING] {status.label} is not fully set up.") - if autostart: - spec = _find_spec(autostart) - if spec is None: - print(f"[WARNING] no server named '{autostart}'; continuing") - elif not servers.start(spec): - print("[ERROR] could not start the server; aborting conversion.") - if status is not None and status.launch_hint: - print("Start it manually and run the conversion again:") - print(f" {status.launch_hint}") - return - elif status is not None and not status.running and status.launch_hint: - print("[INFO] Make sure the server is running. Start it with:") - print(f" {status.launch_hint}") - try: - audiobook.convert(backend=backend, **kwargs) - finally: - if autostart: - _maybe_stop_server(autostart) - - -def _maybe_stop_server(name: str) -> None: - """Ask (in the plain console) whether to stop a server we auto-started.""" - try: - ans = input(f"\n[?] Stop the '{name}' server now? [y/N] ").strip().lower() - except EOFError: - return - if ans in ("y", "yes"): - servers.stop(name) - - -def _add_autostart(stdscr, cmd: tuple, statuses) -> None: - """Offer to auto-start the conversion's target server when it isn't running. - - Records the chosen server spec name as ``kwargs['autostart']`` for - ``_run_conversion`` to act on. Mode-aware for qwen (custom vs clone). - """ - _, key, kwargs = cmd - 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 - choice = tui.confirm(stdscr, f"The {status.label} server is not running. " - "Start it automatically?", default=True, - cancel_value=False) - if choice is True: - 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 _run_server_action(spec_name: str, action: str) -> None: - """Run a Start/Stop action in the plain console (after the TUI returns).""" - if action == "start": - spec = _find_spec(spec_name) - if spec is None: - print(f"[ERROR] no server named '{spec_name}'") - return - servers.start(spec) - elif action == "stop": - servers.stop(spec_name) - - -def _server_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend, then one of its servers and a Start/Stop action.""" - candidates = [st for st in statuses if st.servers or st.running] - if not candidates: - tui.flash(stdscr, "No backend with a server is available. " - "Set one up first.") - return None - options = [(st.label, st.key) for st in candidates] - key = tui.menu(stdscr, "Start / Stop a server", options, - back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - status = next((s for s in statuses if s.key == key), None) - if status is None: - return None - return _server_actions(stdscr, status) - - -def _server_actions(stdscr, status) -> Optional[tuple]: - """Pick a server spec (qwen has two) and a Start or Stop action.""" - specs = status.servers - if not specs: - tui.flash(stdscr, f"{status.label} has no server configured. " - "Run 'Set up a backend' first.") - return None - if len(specs) == 1: - spec = specs[0] - else: - options = [(f"{s.name} ({'running' if common.server_running(s.url) else 'stopped'})", - s.name) for s in specs] - name = tui.menu(stdscr, f"{status.label} server", options, - back_value=_GO_BACK) - if name is _GO_BACK or name is None: - return None - spec = next((s for s in specs if s.name == name), None) - if spec is None: - return None - running = common.server_running(spec.url) - action = tui.menu( - stdscr, f"{spec.name} ({'running' if running else 'stopped'})", - [("Start", "start"), ("Stop", "stop")], back_value=_GO_BACK) - if action is _GO_BACK or action is None: - return None - return ("server", spec.name, action) - - -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 |