aboutsummaryrefslogtreecommitdiff
path: root/backends/qwen.py
diff options
context:
space:
mode:
Diffstat (limited to 'backends/qwen.py')
-rw-r--r--backends/qwen.py272
1 files changed, 0 insertions, 272 deletions
diff --git a/backends/qwen.py b/backends/qwen.py
deleted file mode 100644
index 52f7a3f..0000000
--- a/backends/qwen.py
+++ /dev/null
@@ -1,272 +0,0 @@
-#!/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 the Qwen3-TTS CustomVoice (built-in speakers) and Base (voice
-cloning) models on separate ports. This module sets it up end-to-end as a
-TUI: pip-install the package, configure the two ports and the built-in
-speaker in ``converter/config.py``, and print the launch commands. It is
-driven by ``audiobook.py``'s hub but can also be run directly with flags.
-
-Usage:
- python -m backends.qwen [--port-custom PORT] [--port-clone PORT]
- [--speaker NAME] [--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,
- ConfigureAction,
- ServerSpec,
- common,
- envs,
- format_launch_hint,
-)
-from converter import config
-from ui import tui
-
-QWEN_PIP_PKG = "qwen-tts"
-QWEN_CUSTOMVOICE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"
-QWEN_BASE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-Base"
-DEFAULT_CUSTOM_PORT = 7860
-DEFAULT_CLONE_PORT = 7861
-
-# Built-in CustomVoice speakers (see converter/config.py SPEAKER).
-QWEN_SPEAKERS = ("Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric", "Ryan",
- "Aiden", "Ono_Anna", "Sohee")
-
-
-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 _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
- """Linear TUI wizard collecting every qwen-setup decision."""
- _GO_BACK = object()
-
- def confirm(question: str, default: bool = True) -> Optional[bool]:
- res = tui.confirm(stdscr, question, default=default,
- cancel_value=_GO_BACK)
- return None if res is _GO_BACK else res
-
- # Step 0: pip install (if not installed and not skipped).
- do_install = False
- if not _is_installed() and not args.skip_install:
- choice = confirm("qwen-tts is not installed. pip install it now?",
- default=True)
- if choice is None:
- return None
- do_install = choice
-
- # Step 1: ports.
- custom_port = args.port_custom
- if custom_port is None:
- port_text = tui.line_edit(
- stdscr, "CustomVoice (built-in speaker) port",
- str(_config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT)),
- validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535)
- else "Enter a port number between 1 and 65535",
- help_lines=["The port for qwen-tts-demo CustomVoice (speaker mode)"])
- custom_port = int(port_text)
- clone_port = args.port_clone
- if clone_port is None:
- port_text = tui.line_edit(
- stdscr, "Base (voice clone) port",
- str(_config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT)),
- validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535)
- else "Enter a port number between 1 and 65535",
- help_lines=["The port for qwen-tts-demo Base (voice cloning)"])
- clone_port = int(port_text)
-
- # Step 2: built-in speaker.
- speaker = args.speaker
- if speaker is None:
- speaker = tui.menu(
- stdscr, "Built-in CustomVoice speaker",
- [(s, s) for s in QWEN_SPEAKERS],
- default_index=max(0, QWEN_SPEAKERS.index(config.SPEAKER)
- if config.SPEAKER in QWEN_SPEAKERS else 0),
- help_lines=["Used by audiobook.py --backend qwen without --clone"])
-
- return {
- "do_install": do_install,
- "custom_port": custom_port,
- "clone_port": clone_port,
- "speaker": speaker,
- }
-
-
-def _execute(settings: dict) -> int:
- """Console tail: install, sync config, advise."""
- if settings["do_install"]:
- rc = common.pip_install([QWEN_PIP_PKG])
- 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")
-
- custom_url = common.url_with_port(config.QWEN_API_URL, settings["custom_port"])
- if custom_url != config.QWEN_API_URL:
- if common.update_config_value("QWEN_API_URL", custom_url):
- print(f"[OK] Updated QWEN_API_URL to {custom_url}")
- else:
- print("[WARNING] Could not update QWEN_API_URL; edit "
- "converter/config.py by hand")
- clone_url = common.url_with_port(config.CLONE_API_URL, settings["clone_port"])
- if clone_url != config.CLONE_API_URL:
- if common.update_config_value("CLONE_API_URL", clone_url):
- print(f"[OK] Updated CLONE_API_URL to {clone_url}")
- else:
- print("[WARNING] Could not update CLONE_API_URL; edit "
- "converter/config.py by hand")
- if settings["speaker"] != config.SPEAKER:
- if common.update_config_value("SPEAKER", settings["speaker"]):
- print(f"[OK] Updated SPEAKER to {settings['speaker']}")
- else:
- print("[WARNING] Could not update SPEAKER; edit "
- "converter/config.py by hand")
-
- _print_launch_hint(settings["custom_port"], settings["clone_port"])
- return 0
-
-
-def _print_launch_hint(custom_port: int, clone_port: int) -> None:
- demo = envs.env_script("qwen-tts-demo")
- print()
- print("Start the servers (in separate terminals), or use the hub's")
- print("'Server' menu / let a conversion start one automatically:")
- print(f" {demo} {QWEN_CUSTOMVOICE_MODEL} --ip 127.0.0.1 "
- f"--port {custom_port}")
- print(f" {demo} {QWEN_BASE_MODEL} --ip 127.0.0.1 "
- f"--port {clone_port}")
- print("Then run: python audiobook.py --backend qwen")
-
-
-def run_tui(args: Optional[argparse.Namespace] = None) -> int:
- """Run the qwen setup wizard end-to-end."""
- import curses
- if args is None:
- args = build_parser().parse_args([])
- try:
- settings = curses.wrapper(_wizard, args)
- except tui.WizardCancelled:
- print("\n[INFO] Cancelled; nothing was written")
- return 1
- try:
- curses.curs_set(1)
- except curses.error:
- pass
- if settings is None:
- print("[INFO] Aborted")
- return 1
- return _execute(settings)
-
-
-def _collect_from_flags(args: argparse.Namespace,
- parser: argparse.ArgumentParser) -> dict:
- return {
- "do_install": (not _is_installed()) and not args.skip_install,
- "custom_port": args.port_custom if args.port_custom is not None
- else _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT),
- "clone_port": args.port_clone if args.port_clone is not None
- else _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT),
- "speaker": args.speaker or config.SPEAKER,
- }
-
-
-def build_parser() -> argparse.ArgumentParser:
- parser = argparse.ArgumentParser(
- description="Set up the Qwen3-TTS demo backend: pip install, "
- "configure ports/speaker, and print launch commands.")
- parser.add_argument("--port-custom", type=int, default=None,
- help="CustomVoice (speaker) port (default: "
- f"{DEFAULT_CUSTOM_PORT})")
- parser.add_argument("--port-clone", type=int, default=None,
- help="Base (voice clone) port (default: "
- f"{DEFAULT_CLONE_PORT})")
- parser.add_argument("--speaker", type=str, default=None,
- choices=QWEN_SPEAKERS,
- help="Built-in CustomVoice speaker (default: "
- f"{config.SPEAKER})")
- 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 commands."""
- installed = _is_installed()
- custom_port = _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT)
- clone_port = _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT)
- # Running when either server is up — CustomVoice (speaker mode) or Base
- # (voice clone) each suffice for a conversion on their own.
- running = (common.server_running(config.QWEN_API_URL)
- or common.server_running(config.CLONE_API_URL))
- details: List[str] = []
- details.append("pip: installed" if installed else
- "not installed — run setup to pip install qwen-tts")
- details.append(f"CustomVoice port: {custom_port}")
- details.append(f"Base (clone) port: {clone_port}")
- details.append(f"speaker: {config.SPEAKER}")
- demo = str(envs.env_script("qwen-tts-demo"))
- servers = [
- ServerSpec("qwen-custom", config.QWEN_API_URL,
- [demo, QWEN_CUSTOMVOICE_MODEL, "--ip", "127.0.0.1",
- "--port", str(custom_port)]),
- ServerSpec("qwen-clone", config.CLONE_API_URL,
- [demo, QWEN_BASE_MODEL, "--ip", "127.0.0.1",
- "--port", str(clone_port)]),
- ]
- return BackendStatus("qwen", "qwen-tts",
- installed=installed, configured=installed,
- running=running, details=details,
- launch_hint=format_launch_hint(servers),
- servers=servers)
-
-
-configure_actions: List[ConfigureAction] = [
- ConfigureAction("Reconfigure qwen-tts (ports/speaker)", run_tui),
-]
-
-
-def main() -> int:
- parser = build_parser()
- args = parser.parse_args()
-
- if _interactive():
- return run_tui(args)
-
- settings = _collect_from_flags(args, parser)
- return _execute(settings)
-
-
-def _interactive() -> bool:
- try:
- import curses # noqa: F401
- except ImportError:
- return False
- try:
- return sys.stdin.isatty() and sys.stdout.isatty()
- except (AttributeError, ValueError):
- return False
-
-
-if __name__ == "__main__":
- sys.exit(main())
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 15:22:00 +0000