path: root/app/backends/qwen.py

#!/usr/bin/env python3
"""Set up the Qwen3-TTS demo backend for the audiobook generator.

qwen-tts is a pip package providing the ``qwen-tts-demo`` server, which hosts
ONE Qwen3-TTS model per process — CustomVoice (built-in speakers), Base
(voice cloning) or VoiceDesign (described voice). This module sets it up
end-to-end: pip-install the package into the managed venv. There are no
questions to ask — the port and which model to run live in
``app/converter/config.py`` (the model is chosen per run on the hub's
Generate-audiobooks screen), and only one server runs at a time. It is driven
by ``audiobook.py``'s hub but can also be run directly:

Usage:
    python app/backends/qwen.py [--skip-install]
"""

import argparse
import sys
from pathlib import Path
from typing import List, Optional

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from backends import (
    BackendStatus,
    ServerSpec,
    common,
    envs,
    format_launch_hint,
    probe,
    servers,
    setup,
)
from converter import config
from converter.clients import QWEN3_TTS_SPEAKERS
from ui import taskview

QWEN_PIP_PKG = "qwen-tts"
DEFAULT_PORT = 7860

# The models a single demo server can host, by config.QWEN_MODEL name.
# A running server identifies itself via its probe identity (backends.probe),
# so "which model is up" is always read off the server, never assumed.
MODEL_REPOS = {
    "CustomVoice": "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice",
    "Base": "Qwen/Qwen3-TTS-12Hz-1.7B-Base",
    "VoiceDesign": "Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign",
}
# Probe identity -> the model name reported in statuses/menus.
IDENTITY_TO_MODEL = {
    probe.IDENTITY_QWEN_CUSTOM: "CustomVoice",
    probe.IDENTITY_QWEN_CLONE: "Base",
    probe.IDENTITY_QWEN_DESIGN: "VoiceDesign",
}
DEFAULT_MODEL = "CustomVoice"

# Built-in CustomVoice speakers (see app/converter/config.py SPEAKER). The
# canonical list lives in converter.clients.speakers (shared with the
# audio.cpp backend's Convert-form Speaker picker).
QWEN_SPEAKERS = QWEN3_TTS_SPEAKERS


def _is_installed() -> bool:
    if envs.env_script("qwen-tts-demo").is_file():
        return True
    return envs.module_available("qwen_tts")


def _config_port(url: str, fallback: int) -> int:
    import urllib.parse
    try:
        return urllib.parse.urlsplit(url).port or fallback
    except ValueError:
        return fallback


def current_model() -> str:
    """The configured model to host (a MODEL_REPOS key; DEFAULT_MODEL on typos)."""
    return config.QWEN_MODEL if config.QWEN_MODEL in MODEL_REPOS else DEFAULT_MODEL


def model_for_identity(identity: Optional[str]) -> Optional[str]:
    """The model name a qwen demo answers as (None when not a known identity)."""
    return IDENTITY_TO_MODEL.get(identity)


def desired_identity(model: str) -> str:
    """The probe identity the model's demo answers as (used while booting)."""
    return {
        "CustomVoice": probe.IDENTITY_QWEN_CUSTOM,
        "Base": probe.IDENTITY_QWEN_CLONE,
        "VoiceDesign": probe.IDENTITY_QWEN_DESIGN,
    }[model]


def _wizard(stdscr, args: argparse.Namespace) -> dict:
    """Collect the setup settings without asking anything.

    The qwen backend has no per-install choices: install happens when the
    package is missing (and not skipped by flag), and every other value —
    port, speaker, which model runs — lives in app/converter/config.py /
    the hub's Settings and Generate-audiobooks screens.
    """
    return {
        "do_install": (not _is_installed()) and not args.skip_install,
    }


def _execute_steps(settings: dict) -> List[taskview.TaskStep]:
    """Build the ordered setup steps for the in-TUI task view.

    The same work ``_execute`` runs on the console. The pip install streams
    through EMIT and aborts on CANCEL; the list is empty (no-op) when
    there is nothing to install.
    """
    steps: List[taskview.TaskStep] = []
    if not settings["do_install"]:
        return steps

    def install(emit, cancel):
        rc = common.pip_install([QWEN_PIP_PKG], emit=emit, cancel=cancel)
        if rc != 0:
            print(f"[WARNING] pip install failed (exit {rc}); install "
                  f"{QWEN_PIP_PKG} manually")
        else:
            print(f"[OK] {QWEN_PIP_PKG} installed")
        return rc

    steps.append(taskview.TaskStep(f"Install {QWEN_PIP_PKG}", install))
    return steps


def _execute(settings: dict) -> int:
    """Console tail: pip install (no-op when already installed)."""
    return taskview.run_steps_inline(_execute_steps(settings))


def setup_screen(stdscr) -> int:
    """Run the setup on an existing curses screen (the hub's).

    There are no questions: settings are computed up front and the install
    runs inside the TUI task view on this same screen — skipped entirely
    (a silent no-op) when nothing needs installing. Returns 0 always —
    the flow cannot be aborted, so Esc/Ctrl-C never short-circuits it.
    """
    args = build_parser().parse_args([])
    settings = _wizard(stdscr, args)
    if not settings["do_install"]:
        return 0
    return taskview.run_steps(stdscr, "Setting up qwen-tts",
                              _execute_steps(settings))


def run_tui(args: Optional[argparse.Namespace] = None) -> int:
    """Run the qwen setup end-to-end."""
    if args is None:
        args = build_parser().parse_args([])
    return _execute(_wizard(None, args))


def _collect_from_flags(args: argparse.Namespace,
                        parser: argparse.ArgumentParser) -> dict:
    return {
        "do_install": (not _is_installed()) and not args.skip_install,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Set up the Qwen3-TTS demo backend: pip install "
                    "qwen-tts into the managed venv.")
    parser.add_argument("--skip-install", action="store_true",
                        help="Do not pip install qwen-tts")
    return parser


def detect() -> BackendStatus:
    """Detect whether qwen-tts is installed, plus the launch command.

    One managed spec exists, hosting ``config.QWEN_MODEL`` on the single
    configured port. Which model currently answers there is read via the
    probe (local pid alive => check our own URL; otherwise the remote URL)
    so the status names the *running* model even when it differs from the
    configured one.
    """
    installed = _is_installed()
    model = current_model()
    url = config.QWEN_API_URL
    details: List[str] = []
    details.append("pip: installed" if installed else
                   "not installed — run setup to pip install qwen-tts")
    details.append(f"port: {_config_port(url, DEFAULT_PORT)}")
    details.append(f"model: {model}")
    details.append(f"speaker: {config.SPEAKER}")
    demo = str(envs.env_script("qwen-tts-demo"))
    specs = [
        ServerSpec("qwen", url,
                   [demo, MODEL_REPOS[model], "--ip", "127.0.0.1",
                    "--port", str(_config_port(url, DEFAULT_PORT))],
                   identity=desired_identity(model)),
    ]
    managed = servers.manages(specs)
    # A locally-managed server names its running model via the probe of the
    # managed URL; a remotely-run demo names it via the remote-URL probe.
    local_models: List[str] = []
    if managed and servers.alive(specs[0].name):
        found = model_for_identity(probe.identify_server(url))
        if found is not None:
            local_models.append(found)
    remote_models, remote_urls = _detect_remote(managed)
    running_models = list(dict.fromkeys(local_models + remote_models))
    return BackendStatus("qwen", "qwen-tts",
                         installed=installed, configured=installed,
                         running=managed or bool(remote_urls),
                         details=details,
                         launch_hint=format_launch_hint(specs),
                         servers=specs,
                         managed=managed,
                         remote=bool(remote_urls),
                         remote_urls=remote_urls,
                         remote_models=remote_models,
                         running_models=running_models)


def _detect_remote(managed: bool = False):
    """Detect an externally-run qwen demo server at the remote URL.

    Returns ``([model, ...], {spec_name: url})``. The remote URL must answer
    as one of the three demos (see probe.identify_server); when it equals
    the local URL and this tool started that server, it is ignored (already
    reported "[local]").
    """
    remote_models = []
    remote_urls = {}
    url = (config.QWEN_REMOTE_URL or "").strip()
    if not url:
        return remote_models, remote_urls
    if managed and probe.same_endpoint(url, config.QWEN_API_URL):
        return remote_models, remote_urls
    model = model_for_identity(probe.identify_server(url))
    if model is not None:
        remote_urls["qwen"] = url
        remote_models.append(model)
    return remote_models, remote_urls


def uninstall(*, emit=None, cancel=None) -> int:
    """Remove the qwen-tts backend entirely: stop its server, pip uninstall.

    qwen-tts is a pip package (``qwen_tts`` + the ``qwen-tts-demo`` script)
    installed into the managed venv, so uninstalling it removes the backend.
    Any server this tool started is stopped first (best-effort). Model
    weights already fetched into the HuggingFace cache stay on disk.

    With EMIT given (the in-TUI task view) pip runs piped, streaming into
    EMIT, so its output never touches the terminal behind curses. CANCEL is
    a ``threading.Event`` honored between phases only (after the server has
    been stopped, before pip starts) — a started phase always completes,
    so pip is never killed mid-run. Returns the exit code (130 when
    cancelled before pip ran).
    """
    if servers.pid_for("qwen") is not None:
        # Only stop when a pid file exists: without one this tool never
        # started the server, so the "not started by this tool" notice
        # would be uninstall-time noise.
        servers.stop("qwen")
    if common.cancel_requested(cancel):
        return 130
    rc = common.pip_uninstall([QWEN_PIP_PKG], emit=emit)
    if rc != 0:
        print(f"[WARNING] pip uninstall failed (exit {rc}); remove "
              f"{QWEN_PIP_PKG} from the managed venv manually")
    else:
        print(f"[OK] {QWEN_PIP_PKG} removed.")
    return rc


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()

    if setup.interactive():
        return run_tui(args)

    settings = _collect_from_flags(args, parser)
    return _execute(settings)


if __name__ == "__main__":
    sys.exit(main())