diff options
Diffstat (limited to 'hub.py')
| -rw-r--r-- | hub.py | 377 |
1 files changed, 0 insertions, 377 deletions
@@ -1,377 +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 modify/reconfigure 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 sys -from pathlib import Path -from typing import Optional - -import tui -import audiobook -from backends import REGISTRY, detect_all, get -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 - -_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 == "modify": - info = get(command[1]) - if info is not None and command[2] < len(info.modify_actions): - info.modify_actions[command[2]].run() - elif kind == "convert": - _run_conversion(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() - summary = ["Backend status:"] - for st in statuses: - mark = "ready" if st.ready else ( - "installed" if st.installed else "not set up") - summary.append(f" {st.label}: {mark}") - choice = tui.menu( - stdscr, "tts-audiobook-generator", - [("Convert books...", "convert"), - ("Set up a backend...", "setup"), - ("Modify a backend...", "modify"), - ("Quit", "quit")], - help_lines=summary) - 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 == "modify": - cmd = _modify_menu(stdscr, statuses) - if cmd is not None: - return cmd - - -def _setup_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend to set up. Returns ("setup", key) or None to go back.""" - options = [(f"{info.label} ({_status_mark(info.key, statuses)})", - 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 _modify_menu(stdscr, statuses) -> Optional[tuple]: - """Pick an installed backend and one of its modify actions.""" - installed = [info for info in REGISTRY - if _status_mark(info.key, statuses) != "not set up"] - if not installed: - tui.flash(stdscr, "No backend is set up yet — use 'Set up a backend' first.") - return None - options = [(info.label, info.key) for info in installed] - key = tui.menu(stdscr, "Modify a backend", options, back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - info = get(key) - actions = info.modify_actions - choice = tui.menu( - stdscr, f"Modify {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 ("modify", key, choice) - - -def _status_mark(key: str, statuses) -> str: - for st in statuses: - if st.key == key: - return "ready" if st.ready else ( - "installed" if st.installed else "not set up") - return "not set up" - - -def _convert_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a ready backend and collect per-backend run settings.""" - ready = [st for st in statuses if st.ready] - options = [(f"{st.label}", st.key) for st in ready] - if not ready: - choice = tui.menu( - stdscr, "No backend is ready", - [("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: - return _convert_audiocpp(stdscr, statuses) - if key == BACKEND_QWEN: - return _convert_qwen(stdscr) - if key == BACKEND_FASTER: - return _convert_faster(stdscr) - return None - - -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, "Qwen3-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 Modify Qwen)"]) - 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, - } - - -def _run_conversion(backend: str, kwargs: dict) -> None: - """Run a conversion in the plain console (after the TUI returns).""" - status = next((s for s in detect_all() if s.key == backend), None) - if status is not None and not status.ready: - print(f"[WARNING] {status.label} is not fully set up.") - if status is not None and status.launch_hint: - print("[INFO] Make sure the server is running. Start it with:") - print(f" {status.launch_hint}") - audiobook.convert(backend=backend, **kwargs) - - -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 - - -if __name__ == "__main__": - sys.exit(run()) |