aboutsummaryrefslogtreecommitdiff
path: root/app/backends/qwen.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-08-26 17:46:48 -0400
committerhistoria <historiavg@proton.me>2026-08-26 17:46:48 -0400
commit6ccb6d443d2fb871b43d96ea61a95bc3e6a92355 (patch)
treed692ddccd6ec212cf4ebabb25a85e83af479d0fe /app/backends/qwen.py
parentc147087c9d4707bffaeee58d390653637a21cce8 (diff)
downloadtts-audiobook-generator-6ccb6d443d2fb871b43d96ea61a95bc3e6a92355.tar.gz
feat: combined install/configure tui screens into one menu, removed extraneous wizard screens
Diffstat (limited to 'app/backends/qwen.py')
-rw-r--r--app/backends/qwen.py207
1 files changed, 49 insertions, 158 deletions
diff --git a/app/backends/qwen.py b/app/backends/qwen.py
index c0aafef..5f45580 100644
--- a/app/backends/qwen.py
+++ b/app/backends/qwen.py
@@ -3,14 +3,15 @@
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 ``app/converter/config.py``, and print the launch commands. It is
-driven by ``audiobook.py``'s hub but can also be run directly with flags.
+cloning) models on separate ports. This module sets it up end-to-end:
+pip-install the package into the managed venv. There are no questions to
+ask — the ports live in ``app/converter/config.py`` (edit them in the
+hub's Settings screen) and the speaker is chosen per run on the
+Generate-audiobooks screen. It is driven by ``audiobook.py``'s hub but
+can also be run directly:
Usage:
- python app/backends/qwen.py [--port-custom PORT] [--port-clone PORT]
- [--speaker NAME] [--skip-install]
+ python app/backends/qwen.py [--skip-install]
"""
import argparse
@@ -60,193 +61,83 @@ def _config_port(url: str, fallback: int) -> int:
return fallback
-def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]:
- """Linear TUI wizard collecting every qwen-setup decision.
+def _wizard(stdscr, args: argparse.Namespace) -> dict:
+ """Collect the setup settings without asking anything.
- Driven by ``tui.Wizard`` as a stack of screen closures: each screen
- shows one widget and returns the next screen, ``Wizard.BACK`` (Esc/q —
- pop to the previous screen), or the settings dict. Steps whose value is
- already provided by a flag (``--port-custom``, ``--port-clone``,
- ``--speaker``, ``--skip-install``) are folded into the ``_after_*``
- guards and never become screens, so Esc always lands on the previous
- real screen. Esc on the first screen aborts the wizard.
+ The qwen backend has no per-install choices: install happens when the
+ package is missing (and not skipped by flag), and every other value —
+ ports, speaker — lives in app/converter/config.py, managed from the
+ hub's Settings and Generate-audiobooks screens.
"""
- _GO_BACK = object()
- s: dict = {}
-
- def _after_install():
- if args.port_custom is None:
- return screen_custom_port
- s["custom_port"] = args.port_custom
- return _after_custom_port()
-
- def screen_custom_port():
- 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)"],
- back_value=_GO_BACK)
- if port_text is _GO_BACK:
- return tui.Wizard.BACK
- s["custom_port"] = int(port_text)
- return _after_custom_port()
-
- def _after_custom_port():
- if args.port_clone is None:
- return screen_clone_port
- s["clone_port"] = args.port_clone
- return _after_clone_port()
-
- def screen_clone_port():
- 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)"],
- back_value=_GO_BACK)
- if port_text is _GO_BACK:
- return tui.Wizard.BACK
- s["clone_port"] = int(port_text)
- return _after_clone_port()
-
- def _after_clone_port():
- if args.speaker is None:
- return screen_speaker
- s["speaker"] = args.speaker
- return _finalize()
-
- def screen_speaker():
- 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"],
- back_value=_GO_BACK)
- if speaker is _GO_BACK:
- return tui.Wizard.BACK
- s["speaker"] = speaker
- return _finalize()
-
- def _finalize() -> dict:
- return {
- "do_install": s.get("do_install", False),
- "custom_port": s["custom_port"],
- "clone_port": s["clone_port"],
- "speaker": s["speaker"],
- }
-
- # pip install happens without asking: when the package is missing (and
- # not skipped by flag), the wizard just does it and moves to the next
- # screen.
- s["do_install"] = (not _is_installed()) and not args.skip_install
- return tui.Wizard().run(_after_install())
+ 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, split into named steps so
- the view can show per-step state and progress. The pip install streams
- through EMIT and aborts on CANCEL; print()-based steps are captured by
- the view's stdout redirect.
+ 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 settings["do_install"]:
- 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))
-
- def sync(emit, cancel):
- 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 "
- "app/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 "
- "app/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 "
- "app/converter/config.py by hand")
- return 0
- steps.append(taskview.TaskStep("Sync config & ports", sync))
-
+ 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: install, sync config, advise."""
+ """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 wizard on an existing curses screen (the hub's).
+ """Run the setup on an existing curses screen (the hub's).
- See backends.setup.screen_flow for the shared flow. Returns 0 on
- completion, 1 when the user aborted.
+ There are no questions: settings are computed up front and the install
+ runs inside the TUI task view on this same screen (skipped entirely
+ when nothing needs installing). Returns 0 always — the flow cannot be
+ aborted, so Esc/Ctrl-C never short-circuits it.
"""
- return setup.screen_flow(stdscr, wizard=_wizard,
- steps_of=_execute_steps,
- title="Setting up qwen-tts",
- parser_factory=build_parser)
+ args = build_parser().parse_args([])
+ settings = _wizard(stdscr, args)
+ if not settings["do_install"]:
+ tui.flash(stdscr, "qwen-tts is already installed.", "ok")
+ 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 wizard end-to-end."""
+ """Run the qwen setup end-to-end."""
if args is None:
args = build_parser().parse_args([])
- return setup.tui_flow(_wizard, _execute, args=args,
- aborted_message="[INFO] Aborted")
+ 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,
- "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})")
+ 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