path: root/app/backends/__init__.py

"""Registry of the TTS backends the audiobook generator can talk to.

Each backend (audio.cpp, qwen, faster) lives in its own module and owns
its setup wizard, its status detection, its uninstaller, and the launch
command it prints once configured. This package aggregates them into a single
registry so ``audiobook.py``'s TUI hub and future tools can iterate backends
without hardcoding their names: ``backends.detect_all()`` reports which are set
up (and whether their server is currently running), and the registry
drives the hub's "Configure backends" menu.

The registry is built lazily on the first call to ``get``/``detect_all``/
``detect`` (not at package import time), because the backend modules pull
in ``converter.clients`` and its third-party dependencies, which are only
available inside the managed venv that ``audiobook.py`` bootstraps before
importing them. ``backends.envs`` is imported during that bootstrap, so
importing this package must stay cheap and dependency-free.

Adding a backend: create ``backends/<name>.py`` exposing
``detect() -> BackendStatus``, ``setup_screen(stdscr) -> int`` (the setup
wizard run on the hub's own screen) and ``uninstall() -> int``, then append
a ``BackendInfo`` in ``_build_registry`` below. ``audiobook.py`` and the hub
pick it up automatically. A backend's standalone CLI keeps its own
``run_tui()`` entry (its own curses session), which is not part of the
registry.
"""

import shlex
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Dict, List, Optional

# How long detect_all() results stay fresh (see detect_all). Long enough to
# cover a burst of menu renders, short enough that state changed by an
# outside actor (a remote server appearing) surfaces promptly.
DETECT_TTL_SECONDS = 2.0
_detect_cache = None


@dataclass
class ServerSpec:
    """One launchable server process for a backend.

    Most backends expose exactly one server. ARGV is the exact command line
    the hub spawns (using
    the managed venv's absolute binaries, so no shell activation is needed);
    URL is the endpoint ``common.server_running`` probes to decide readiness.

    CWD is the working directory the server is spawned in. It matters for
    servers that discover resources relative to their process working
    directory (audio.cpp resolves ``model_specs/<family>.json`` by walking up
    from its cwd), so the hub starts them from their checkout root. None
    inherits the hub's cwd (today's behavior).

    IDENTITY is the ``backends.probe.IDENTITY_*`` constant the server is
    expected to answer as once it is truly ready. When set, ``servers.start``
    waits for the server to answer HTTP with that identity — not merely to
    accept TCP connections — so "listening but still starting" servers are
    caught. None keeps the plain TCP-connect readiness check.
    """
    name: str
    url: str
    argv: List[str]
    cwd: Optional[Path] = None
    identity: Optional[str] = None


@dataclass
class BackendStatus:
    """How far a backend is set up, plus the command to start it.

    INSTALLED means the backend itself is present (a cloned + built
    checkout, or a pip package). CONFIGURED means the supporting files are
    in place (a server.json / voices.json and an app/converter/config.py that
    points at the right port). DETAILS are short status lines for the hub.
    LAUNCH_HINT is the human-readable command(s) the user runs to start the
    server, derived from SERVERS by ``format_launch_hint``. SERVERS is the
    machine-usable list of server processes the hub can start/stop (empty
    when the backend is not yet configured).

    MANAGED says a running server was started by this tool: ``servers``
    contains a spec whose pid file still names a live process (see
    ``servers.manages``); when a server is running but not MANAGED the hub
    tags it "[remote]".

    REMOTE says a server answering at the backend's configured remote URL
    (``*_REMOTE_URL`` in ``app/converter/config.py``) was identified as this
    backend by ``backends.probe.identify_server`` — a server this tool did
    not start (it is suppressed when the remote URL equals the local URL and
    this tool's own pid is still alive). REMOTE_URLS maps each server spec
    name ("audiocpp", "faster", "qwen") to the remote
    URL that answered, so the hub's convert menu can target it. RUNNING is
    true when the backend is usable either locally (MANAGED) or remotely
    (REMOTE), and drives both the status table ("running [local]",
    "running [remote]", "running [local, remote]") and the hub menu gating.
    RUNNING_MODELS names which of a multi-model backend's models answered
    (qwen: "Base", "CustomVoice" and/or "VoiceDesign", local and remote
    combined), shown in parentheses in the hub's status table.

    MODELS_MISSING says the server config references model files that are not
    on disk (e.g. an audio.cpp server.json entry whose ``path`` was never
    downloaded); DETAILS then names them. The backend still counts as ready
    (the hub surfaces the warning), but a conversion would fail until the
    models are installed.

    PARTIAL is an optional, more specific label for a backend that is set up
    only part-way (neither running nor fully installed): audio.cpp reports
    "downloaded (not built)" when its checkout exists but ``audiocpp_server``
    was never built, and "built (not configured)" when the binary exists but
    no ``server.json`` does. The hub shows it verbatim (amber) instead of the
    generic "unavailable"/"installed" text, and dims the name while the
    backend is still unusable.
    """
    key: str
    label: str
    installed: bool
    configured: bool
    running: bool = False
    details: List[str] = field(default_factory=list)
    launch_hint: str = ""
    servers: List[ServerSpec] = field(default_factory=list)
    managed: bool = False
    running_models: List[str] = field(default_factory=list)
    remote: bool = False
    remote_urls: Dict[str, str] = field(default_factory=dict)
    remote_models: List[str] = field(default_factory=list)
    models_missing: bool = False
    partial: str = ""

    @property
    def ready(self) -> bool:
        """True when the backend is installed and configured for use."""
        return self.installed and self.configured


def format_launch_hint(servers: List[ServerSpec]) -> str:
    """Join a backend's server argvs into a copy-pasteable launch hint.

    A server with a CWD is prefixed with ``cd <cwd> &&`` so the hint works
    pasted into a shell (the server relies on that working directory).
    """
    parts = []
    for server in servers:
        command = shlex.join(server.argv)
        if server.cwd is not None:
            command = f"cd {shlex.quote(str(server.cwd))} && {command}"
        parts.append(command)
    return " ; ".join(parts)


@dataclass
class BackendInfo:
    """One registry entry: identity, detector, setup wizard, uninstaller.

    SETUP_SCREEN runs the setup wizard on an already-open curses screen
    (the hub's), returning 0 on completion and non-zero when aborted; the
    hub calls it as one screen of its own ``tui.Wizard`` stack. UNINSTALL
    removes the backend (stops its servers, pip-uninstalls, deletes its
    files); the hub runs it inside the task view, calling it with optional
    ``emit``/``cancel`` keywords (cancel honored between phases only).
    """
    key: str
    label: str
    detect: Callable[[], BackendStatus]
    setup_screen: Callable[[object], int]
    uninstall: Callable[..., int] = lambda *args, **kwargs: 0


REGISTRY: List[BackendInfo] = []
_BY_KEY: dict = {}


def _build_registry() -> None:
    """Import the backend modules and wire up REGISTRY (once)."""
    if REGISTRY:
        return
    from . import audiocpp, faster, qwen

    REGISTRY.append(BackendInfo(
        key="audiocpp",
        label="audio.cpp",
        detect=audiocpp.detect,
        setup_screen=audiocpp.setup_screen,
        uninstall=audiocpp.uninstall,
    ))
    REGISTRY.append(BackendInfo(
        key="qwen",
        label="qwen-tts",
        detect=qwen.detect,
        setup_screen=qwen.setup_screen,
        uninstall=qwen.uninstall,
    ))
    REGISTRY.append(BackendInfo(
        key="faster",
        label="faster-qwen3-tts",
        detect=faster.detect,
        setup_screen=faster.setup_screen,
        uninstall=faster.uninstall,
    ))
    for info in REGISTRY:
        _BY_KEY[info.key] = info


def get(key: str) -> Optional[BackendInfo]:
    """Return the registry entry for KEY, or None."""
    _build_registry()
    return _BY_KEY.get(key)


def detect_all(*, refresh: bool = False) -> List[BackendStatus]:
    """Detect every registered backend's status, in registry order.

    Detection is not free — each backend probes the filesystem and, for
    remote servers, the network — so results are cached for a short
    window (DETECT_TTL_SECONDS). Menu renders that happen in quick
    succession (popping back and forth between hub screens) reuse the
    cached statuses; anything past the TTL re-probes. REFRESH forces an
    immediate re-detection: callers use it right after an action that can
    change status (setup, uninstall, server start/stop) so the next render
    never shows stale state.
    """
    _build_registry()
    global _detect_cache
    now = time.monotonic()
    if not refresh and _detect_cache is not None:
        at, statuses = _detect_cache
        if now - at < DETECT_TTL_SECONDS:
            return list(statuses)
    statuses = [info.detect() for info in REGISTRY]
    _detect_cache = (now, statuses)
    return list(statuses)


def invalidate_detect_cache() -> None:
    """Drop the cached statuses so the next detect_all() re-probes.

    Called by the hub after any action that can change a backend's on-disk
    or running state (setup wizards, uninstallers, server toggles,
    conversion runs with autostart, settings writes).
    """
    global _detect_cache
    _detect_cache = None


def detect(key: str) -> Optional[BackendStatus]:
    """Detect a single backend by key."""
    info = get(key)
    return info.detect() if info is not None else None