diff options
Diffstat (limited to 'app/ui/hub.py')
| -rw-r--r-- | app/ui/hub.py | 637 |
1 files changed, 637 insertions, 0 deletions
diff --git a/app/ui/hub.py b/app/ui/hub.py new file mode 100644 index 0000000..6a94096 --- /dev/null +++ b/app/ui/hub.py @@ -0,0 +1,637 @@ +#!/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 app/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 -> app/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 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']}") + 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 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 _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 |