path: root/hub.py

#!/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())