diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/backends/__init__.py | 24 | ||||
| -rwxr-xr-x | app/backends/audiocpp.py | 925 | ||||
| -rwxr-xr-x | app/backends/faster.py | 221 | ||||
| -rw-r--r-- | app/backends/qwen.py | 128 | ||||
| -rw-r--r-- | app/tests/test_backends.py | 27 | ||||
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 88 | ||||
| -rw-r--r-- | app/tests/test_backends_faster.py | 25 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 350 | ||||
| -rw-r--r-- | app/tests/test_tui.py | 71 | ||||
| -rw-r--r-- | app/ui/hub.py | 608 | ||||
| -rw-r--r-- | app/ui/tui.py | 42 |
11 files changed, 1605 insertions, 904 deletions
diff --git a/app/backends/__init__.py b/app/backends/__init__.py index 3dff306..ef713ba 100644 --- a/app/backends/__init__.py +++ b/app/backends/__init__.py @@ -16,9 +16,12 @@ 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``, ``run_tui() -> int`` and -``uninstall() -> int``, then append a ``BackendInfo`` in ``_build_registry`` -below. ``audiobook.py`` and the hub pick it up automatically. +``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 @@ -131,11 +134,16 @@ def format_launch_hint(servers: List[ServerSpec]) -> str: @dataclass class BackendInfo: - """One registry entry: identity, detector, setup wizard, uninstaller.""" + """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. + """ key: str label: str detect: Callable[[], BackendStatus] - setup_tui: Callable[[], int] + setup_screen: Callable[[object], int] uninstall: Callable[[], int] = lambda: 0 @@ -153,21 +161,21 @@ def _build_registry() -> None: key="audiocpp", label="audio.cpp", detect=audiocpp.detect, - setup_tui=audiocpp.run_tui, + setup_screen=audiocpp.setup_screen, uninstall=audiocpp.uninstall, )) REGISTRY.append(BackendInfo( key="qwen", label="qwen-tts", detect=qwen.detect, - setup_tui=qwen.run_tui, + setup_screen=qwen.setup_screen, uninstall=qwen.uninstall, )) REGISTRY.append(BackendInfo( key="faster", label="faster-qwen3-tts", detect=faster.detect, - setup_tui=faster.run_tui, + setup_screen=faster.setup_screen, uninstall=faster.uninstall, )) for info in REGISTRY: diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py index f74f726..feeb22a 100755 --- a/app/backends/audiocpp.py +++ b/app/backends/audiocpp.py @@ -100,14 +100,16 @@ _GO_BACK = object() class _GoBack(Exception): - """Raised inside the TUI wizard to fall back to the previous screen group. - - Every wizard widget is passed ``back_value=_GO_BACK`` so Esc returns the - sentinel instead of aborting; pickers and confirmations that call into - callbacks (task/id pickers, the transcription plan, the download prompt) - convert that sentinel into this exception so the enclosing step can catch - it and step back. Only the first screen (the checkout browser) lets Esc - abort the whole wizard. + """Internal signal: Esc was pressed inside one of a screen's sub-prompts. + + The wizard drives a stack of screens via ``tui.Wizard``. Helpers that ask + several questions through callbacks (the task/id pickers inside + ``_build_entries``, the transcription plan, the download prompt) cannot + themselves return the wizard's ``BACK`` sentinel, so they convert the + ``_GO_BACK`` value passed to each widget into this exception. The screen + that invoked the helper catches it and returns ``tui.Wizard.BACK``, which + pops back to the previous screen. Esc on the first screen aborts the + whole wizard. """ # Package names that mark a voice-design model (hosted with task "vdes"). @@ -897,13 +899,20 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser ) -> Optional[dict]: """Run every TUI screen; return the collected settings, or None to abort. - The wizard is a step state machine; each screen group is one step, and - Esc anywhere but the first step falls back to the previous group (the - widget returns the _GO_BACK sentinel, or a callback raises _GoBack). On - the first screen (the audio.cpp checkout browser) Esc aborts the whole - wizard as before. + The wizard is driven by ``tui.Wizard`` as a stack of screen closures: + each screen shows one interactive widget and returns the next screen + (a closure), ``Wizard.BACK`` (Esc/q pressed — pop to the previous + screen), or the final settings dict. Only screens that actually render + are pushed, so Esc always lands on the previous real screen. A step + whose value is already provided by a flag (``--host``, ``--port``, + ``--families``, ...) or does not apply (e.g. the port-sync prompt when + the port did not change) is folded into the ``_after_*`` guards and + never becomes a screen. Esc on the first screen aborts the whole + wizard. """ + s: dict = {} + def ask_confirm(question: str, default: bool) -> bool: result = tui.confirm(stdscr, question, default=default, cancel_value=_GO_BACK) @@ -911,411 +920,481 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser raise _GoBack() return result - step = 0 - while True: - if step == 0: - # Checkout browser. The browser asks for the checkout root and - # finds model_specs/ inside it (picking the model_specs directory - # itself works too — its parent is used). A highlighted - # subdirectory named "audio.cpp" that already contains - # model_specs/ is auto-accepted on Enter/Right, skipping the - # "[ Use this directory ]" step. Esc on the browser is the first - # step, so it aborts the wizard. - auto_accept = True - browser_start: Path = Path.cwd() - - def do_browse(): - return tui.browse_directory( - stdscr, "Select your audio.cpp directory", - validate=lambda p: None if _resolve_audiocpp_root(p) - else "No model_specs/ directory here", - info=_audiocpp_root_status, - preview=_audiocpp_root_preview, - help_lines=["The root folder of your audio.cpp " - "checkout;", - "it is the one that contains " - "model_specs/"], - start=browser_start, - auto_select=_checkout_auto_select if auto_accept - else None) - - while True: - audiocpp_dir = args.audiocpp_dir - if audiocpp_dir is None: - audiocpp_dir = find_local_checkout() - if audiocpp_dir is None: - # No checkout found anywhere: offer to clone one into - # ./app/audio.cpp or browse for an existing checkout. - # Esc on this first menu aborts the wizard. - choice = tui.menu( - stdscr, "No audio.cpp checkout found", - [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} " - f"(from {AUDIOCPP_GIT_URL})", "clone"), - ("Browse for an existing checkout", "browse")], - help_lines=[ - "audio.cpp hosts the TTS model families " - "this generator uses.", - "Clone it into the project's app " - "directory, or point at an existing " - "checkout."]) - if choice == "clone": - target = APP_DIR / AUDIOCPP_DIR_NAME - with tui.suspend(stdscr): - rc = common.git_clone(AUDIOCPP_GIT_URL, - target) - if rc != 0: - raise _TuiError( - f"git clone failed (exit {rc}). Clone " - f"audio.cpp manually: git clone " - f"{AUDIOCPP_GIT_URL} {target}") - audiocpp_dir = target - else: - audiocpp_dir = do_browse() - audiocpp_dir = Path(audiocpp_dir).resolve() - if not audiocpp_dir.is_dir(): - raise _TuiError(f"audio.cpp checkout not found: " - f"{audiocpp_dir}") - root = _resolve_audiocpp_root(audiocpp_dir) - if root is None: - raise _TuiError( - f"{audiocpp_dir} has no model_specs/ directory; " - "select the root of your audio.cpp checkout") - audiocpp_dir = root - try: - catalog = load_model_catalog(audiocpp_dir) - except NotADirectoryError as exc: - raise _TuiError(str(exc)) - if not catalog: - raise _TuiError(f"No TTS model families found in " - f"{audiocpp_dir}/model_specs; check the " - "checkout is up to date") - catalog_by_family = {entry["family"]: entry - for entry in catalog} - - output_path = args.output if args.output is not None \ - else audiocpp_dir / "server.json" - break - - # Modify flow: an existing server.json seeds the wizard's - # screens instead of being overwritten from scratch (an explicit - # --force still starts fresh). - existing_config = load_server_config(output_path) \ - if not args.force else None - if existing_config is not None: - existing_selected, existing_tasks = \ - server_config_selections(existing_config, catalog) - else: - existing_selected, existing_tasks = {}, {} - existing_host = existing_config.get("host") \ - if existing_config else None - existing_port = existing_config.get("port") \ - if existing_config else None - existing_backend = existing_config.get("backend") \ - if existing_config else None - existing_lazy = existing_config.get("lazy_load") \ - if existing_config else None - existing_voice_dir = existing_config.get("voice_dir") \ - if existing_config else None - detected_backend = detect_backend(audiocpp_dir) - step = 1 - continue + def resolve_checkout(audiocpp_dir: Path) -> None: + """Validate AUDIOCPP_DIR and populate the wizard state ``s``.""" + audiocpp_dir = Path(audiocpp_dir).resolve() + if not audiocpp_dir.is_dir(): + raise _TuiError(f"audio.cpp checkout not found: " + f"{audiocpp_dir}") + root = _resolve_audiocpp_root(audiocpp_dir) + if root is None: + raise _TuiError( + f"{audiocpp_dir} has no model_specs/ directory; " + "select the root of your audio.cpp checkout") + audiocpp_dir = root + try: + catalog = load_model_catalog(audiocpp_dir) + except NotADirectoryError as exc: + raise _TuiError(str(exc)) + if not catalog: + raise _TuiError(f"No TTS model families found in " + f"{audiocpp_dir}/model_specs; check the " + "checkout is up to date") + catalog_by_family = {entry["family"]: entry for entry in catalog} + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + # Modify flow: an existing server.json seeds the wizard's screens + # instead of being overwritten from scratch (an explicit --force + # still starts fresh). + existing_config = load_server_config(output_path) \ + if not args.force else None + if existing_config is not None: + existing_selected, existing_tasks = \ + server_config_selections(existing_config, catalog) + else: + existing_selected, existing_tasks = {}, {} + s.update({ + "audiocpp_dir": audiocpp_dir, + "catalog": catalog, + "catalog_by_family": catalog_by_family, + "output_path": output_path, + "existing_config": existing_config, + "existing_selected": existing_selected, + "existing_tasks": existing_tasks, + "existing_host": existing_config.get("host") + if existing_config else None, + "existing_port": existing_config.get("port") + if existing_config else None, + "existing_backend": existing_config.get("backend") + if existing_config else None, + "existing_lazy": existing_config.get("lazy_load") + if existing_config else None, + "existing_voice_dir": existing_config.get("voice_dir") + if existing_config else None, + "detected_backend": detect_backend(audiocpp_dir), + }) - if step == 1: - # Families and packages (flag or tree). Esc returns to the - # checkout browser (step 0). - chosen: Dict[str, List[dict]] = {} - if args.families is not None: - requested = [f.strip() for f in args.families.split(",") - if f.strip()] - unknown = [f for f in requested if f not in catalog_by_family] - if unknown: - raise _TuiError( - f"Unknown family in --families: {', '.join(unknown)}. " - f"Available: {', '.join(catalog_by_family)}") - family_keys: List[str] = [] - for family in requested: - if family not in family_keys: - family_keys.append(family) - chosen[family] = [opt for opt in package_dir_options( - catalog_by_family[family]) if opt["recommended"]] - else: - tree_families = _build_tree_families(catalog) - # Modify flow: pre-check the models an existing server.json - # hosts, so the tree opens as a "modify" list rather than a - # fresh one. - checked_set = set() - for family, dirs in existing_selected.items(): - if family not in catalog_by_family: - continue - family_index = catalog.index(catalog_by_family[family]) - valid_dirs = {opt["target_directory"] - for opt in package_dir_options( - catalog_by_family[family])} - for target in dirs: - if target in valid_dirs: - checked_set.add((family_index, target)) - picked = tui.checkbox_tree( - stdscr, "Select TTS model families to host", - tree_families, expand_all=args.all_packages, - back_value=_GO_BACK, checked=checked_set) - if picked is _GO_BACK: - step = 0 - continue - family_keys = [] - for family_index, option_key in picked: - family = catalog[family_index]["family"] - if family not in chosen: - chosen[family] = [] - family_keys.append(family) - chosen[family].append(option_key) - for family in list(chosen): - keyed = {opt["target_directory"]: opt - for opt in package_dir_options( - catalog_by_family[family])} - chosen[family] = [keyed[key] for key in chosen[family]] - step = 2 - continue + def _families_from_flag() -> None: + requested = [f.strip() for f in args.families.split(",") if f.strip()] + unknown = [f for f in requested if f not in s["catalog_by_family"]] + if unknown: + raise _TuiError( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(s['catalog_by_family'])}") + chosen: Dict[str, List[dict]] = {} + family_keys: List[str] = [] + for family in requested: + if family not in family_keys: + family_keys.append(family) + chosen[family] = [opt for opt in package_dir_options( + s["catalog_by_family"][family]) if opt["recommended"]] + s["chosen"] = chosen + s["family_keys"] = family_keys + + def _compute_entries() -> None: + # Design task menus and duplicate-id renames. Esc on any of them + # raises _GoBack, which the caller turns into Wizard.BACK (the + # design/duplicate-id prompts are grouped: Esc returns to the + # families tree). + def task_picker(install_id: str) -> str: + result = tui.menu( + stdscr, + f"How should the '{install_id}' package be hosted?", + [ + ("design (vdes) - describe the voice with " + "--instructions", TASK_VDES), + ("tts - normal synthesis", TASK_TTS), + ], default_index=0, back_value=_GO_BACK) + if result is _GO_BACK: + raise _GoBack() + return result + + def id_picker(display_name: str, install_id: str, + default: str) -> str: + result = tui.line_edit( + stdscr, + f"Server model id for {display_name} package " + f"'{install_id}'", default, back_value=_GO_BACK) + if result is _GO_BACK: + raise _GoBack() + return result + + model_entries, entry_ids, install_guidance, \ + design_entry_ids, include_clone = _build_entries( + s["family_keys"], s["chosen"], s["catalog_by_family"], + task_picker, id_picker, known_tasks=s["existing_tasks"]) + s.update({ + "model_entries": model_entries, + "entry_ids": entry_ids, + "install_guidance": install_guidance, + "design_entry_ids": design_entry_ids, + "include_clone": include_clone, + }) - if step == 2: - # Design task menus and duplicate-id renames. Esc anywhere here - # falls back to the families tree (step 1). - def task_picker(install_id: str) -> str: - result = tui.menu( - stdscr, - f"How should the '{install_id}' package be hosted?", - [ - ("design (vdes) - describe the voice with " - "--instructions", TASK_VDES), - ("tts - normal synthesis", TASK_TTS), - ], default_index=0, back_value=_GO_BACK) - if result is _GO_BACK: - raise _GoBack() - return result - - def id_picker(display_name: str, install_id: str, - default: str) -> str: - result = tui.line_edit( - stdscr, - f"Server model id for {display_name} package " - f"'{install_id}'", default, back_value=_GO_BACK) - if result is _GO_BACK: - raise _GoBack() - return result - - try: - model_entries, entry_ids, install_guidance, \ - design_entry_ids, include_clone = _build_entries( - family_keys, chosen, catalog_by_family, - task_picker, id_picker, known_tasks=existing_tasks) - except _GoBack: - step = 1 + def _finalize() -> dict: + return { + "audiocpp_dir": s["audiocpp_dir"], + "catalog": s["catalog"], + "catalog_by_family": s["catalog_by_family"], + "output_path": s["output_path"], + "family_keys": s["family_keys"], + "chosen": s["chosen"], + "model_entries": s["model_entries"], + "entry_ids": s["entry_ids"], + "install_guidance": s["install_guidance"], + "design_entry_ids": s["design_entry_ids"], + "include_clone": s["include_clone"], + "host": s["host"], + "port": s["port"], + "backend": s["backend"], + "build": s["build"], + "lazy_load": s["lazy_load"], + "sync_port": s["sync_port"], + "sync_model_ids": s["sync_model_ids"], + "wav_dir": s["wav_dir"], + "plan": s["plan"], + "download": s["download"], + "delete_unused": s["delete_unused"], + "unused_entries": s["unused_entries"], + } + + def screen_no_checkout(): + """First screen when no checkout exists: clone or browse. + + No ``back_value``: Esc aborts the whole wizard (nothing before it). + """ + choice = tui.menu( + stdscr, "No audio.cpp checkout found", + [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} " + f"(from {AUDIOCPP_GIT_URL})", "clone"), + ("Browse for an existing checkout", "browse")], + help_lines=[ + "audio.cpp hosts the TTS model families " + "this generator uses.", + "Clone it into the project's app " + "directory, or point at an existing " + "checkout."]) + if choice == "clone": + target = APP_DIR / AUDIOCPP_DIR_NAME + with tui.suspend(stdscr): + rc = common.git_clone(AUDIOCPP_GIT_URL, target) + if rc != 0: + raise _TuiError( + f"git clone failed (exit {rc}). Clone " + f"audio.cpp manually: git clone " + f"{AUDIOCPP_GIT_URL} {target}") + resolve_checkout(target) + else: + return screen_browse_checkout + return _after_families() + + def screen_browse_checkout(): + """Browse for an existing checkout (Esc returns to the clone menu).""" + audiocpp_dir = tui.browse_directory( + stdscr, "Select your audio.cpp directory", + validate=lambda p: None if _resolve_audiocpp_root(p) + else "No model_specs/ directory here", + info=_audiocpp_root_status, + preview=_audiocpp_root_preview, + help_lines=["The root folder of your audio.cpp checkout;", + "it is the one that contains model_specs/"], + start=Path.cwd(), + auto_select=_checkout_auto_select, + back_value=_GO_BACK) + if audiocpp_dir is _GO_BACK: + return tui.Wizard.BACK + resolve_checkout(audiocpp_dir) + return _after_families() + + def screen_families(): + """Pick TTS model families and packages (the modify tree).""" + tree_families = _build_tree_families(s["catalog"]) + # Modify flow: pre-check the models an existing server.json hosts, + # so the tree opens as a "modify" list rather than a fresh one. + checked_set = set() + for family, dirs in s["existing_selected"].items(): + if family not in s["catalog_by_family"]: continue - step = 3 - continue - - if step == 3: - # Server settings (host, port, port-sync, backend, lazy). Esc on - # any of them falls back to the previous group (step 2). - if args.host: - host = args.host - else: - host = tui.line_edit( - stdscr, "Bind host", - existing_host if isinstance(existing_host, str) - else DEFAULT_HOST, - help_lines=["The IP address audiocpp will be hosted on", - "127.0.0.1 (this machine) is probably " - "correct"], back_value=_GO_BACK) - if host is _GO_BACK: - step = 2 - continue - if args.port is not None: - port = args.port - else: - port_text = tui.line_edit( - stdscr, "Port", - str(existing_port) if isinstance(existing_port, int) - else str(config_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 audiocpp will be hosted on"], - back_value=_GO_BACK) - if port_text is _GO_BACK: - step = 2 - continue - port = int(port_text) - sync_port: Optional[bool] = None - if port != config_port(): - sync_port = tui.confirm( - stdscr, f"Update AUDIOCPP_API_URL in app/converter/config.py " - f"to port {port} so audiobook.py talks to this server", - default=True, cancel_value=_GO_BACK) - if sync_port is _GO_BACK: - step = 2 - continue - if args.build_backend: - backend = args.build_backend - build = detected_backend is None - elif args.backend: - backend = args.backend - build = False - elif detected_backend is not None: - # Already built: use the detected backend, no menu, no build. - backend = detected_backend - build = False - elif existing_backend in BACKENDS: - # Modify flow: keep the backend an existing server.json - # records (already configured, no rebuild needed). - backend = existing_backend - build = False - else: - backend_options, backend_default = _backend_options(None) - backend = tui.menu( - stdscr, "Which inference backend was audiocpp_server " - "built for?", backend_options, - default_index=backend_default, back_value=_GO_BACK) - if backend is _GO_BACK: - step = 2 - continue - # Not built for any backend yet: offer to build it now. The - # build itself runs in the console tail after the wizard. - build = tui.confirm( - stdscr, f"audiocpp_server is not built for {backend}. " - f"Build it now (runs scripts/build_*)?", - default=True, cancel_value=_GO_BACK) - if build is _GO_BACK: - step = 2 - continue - default_lazy = len(model_entries) > 1 - if isinstance(existing_lazy, bool): - default_lazy = existing_lazy - if args.lazy_load: - lazy_load = True - else: - lazy_load = tui.confirm( - stdscr, "Load models lazily (on first use instead of at " - "startup)", default=default_lazy, cancel_value=_GO_BACK) - if lazy_load is _GO_BACK: - step = 2 - continue - step = 4 - continue - - if step == 4: - # Wav directory (flag, browsed when cloning, else skipped). Esc - # falls back to the server settings (step 3). - if args.input_dir is not None: - wav_dir = args.input_dir - elif include_clone: - wav_start = detect_wav_dir(audiocpp_dir, TTS_ROOT) - # Modify flow: an existing voice_dir seeds the browser so the - # user can accept it on Enter instead of re-navigating. - if isinstance(existing_voice_dir, str) and existing_voice_dir: - wav_start = Path(existing_voice_dir) - wav_dir = tui.browse_directory( - stdscr, "Select the directory with your .wav voices", - info=_wav_dir_info, preview=_wav_dir_preview, - start=wav_start if wav_start is not None else VOICES_DIR, - back_value=_GO_BACK) - if wav_dir is _GO_BACK: - step = 3 - continue - else: - wav_dir = None - step = 5 - continue - - if step == 5: - # Transcription plan (questions only; transcription runs after). - # Esc falls back to the wav browser (step 4). - plan: Optional[dict] = None - if include_clone and wav_dir is not None: - wav_files = find_wav_files(wav_dir) - if wav_files: - prompt_path = wav_dir / PROMPT_TEXT_FILENAME - existing = read_prompt_text(prompt_path) if ( - prompt_path.exists() and not args.force) else {} - try: - plan = _decide_transcription( - wav_files, existing, prompt_path.exists(), - args.force, ask_confirm) - except _GoBack: - step = 4 - continue - step = 6 - continue - - if step == 6: - # Single-model id sync decision. Esc falls back to the - # transcription plan (step 5). - sync_model_ids: Optional[bool] = None - if len(entry_ids) == 1 and not ( - config.AUDIOCPP_MODEL_ID == entry_ids[0] - and config.AUDIOCPP_CLONE_MODEL_ID == entry_ids[0]): - sync_model_ids = tui.confirm( - stdscr, "Update AUDIOCPP_MODEL_ID and " - "AUDIOCPP_CLONE_MODEL_ID in app/converter/config.py to " - f"'{entry_ids[0]}' so audiobook.py uses this model", - default=True, cancel_value=_GO_BACK) - if sync_model_ids is _GO_BACK: - step = 5 - continue - step = 7 - continue - - if step == 7: - # Delete unused models: already-downloaded models that the new - # selection no longer hosts. Only offered in the modify flow (an - # existing config was loaded), since a fresh --force run is an - # explicit overwrite. Esc falls back to the model-id sync (6). - new_paths = {entry["path"] for entry in model_entries} - unused_entries = unused_installed_entries(output_path, new_paths) \ - if existing_config is not None else [] - delete_unused = False - if unused_entries: - delete_unused = tui.confirm( - stdscr, "Delete unused models?", default=False, - cancel_value=_GO_BACK) - if delete_unused is _GO_BACK: - step = 6 - continue - step = 8 - continue + family_index = s["catalog"].index(s["catalog_by_family"][family]) + valid_dirs = {opt["target_directory"] + for opt in package_dir_options( + s["catalog_by_family"][family])} + for target in dirs: + if target in valid_dirs: + checked_set.add((family_index, target)) + picked = tui.checkbox_tree( + stdscr, "Select TTS model families to host", + tree_families, expand_all=args.all_packages, + back_value=_GO_BACK, checked=checked_set) + if picked is _GO_BACK: + return tui.Wizard.BACK + chosen: Dict[str, List[dict]] = {} + family_keys: List[str] = [] + for family_index, option_key in picked: + family = s["catalog"][family_index]["family"] + if family not in chosen: + chosen[family] = [] + family_keys.append(family) + chosen[family].append(option_key) + for family in list(chosen): + keyed = {opt["target_directory"]: opt + for opt in package_dir_options( + s["catalog_by_family"][family])} + chosen[family] = [keyed[key] for key in chosen[family]] + s["chosen"] = chosen + s["family_keys"] = family_keys + return screen_host + + def _after_families(): + if args.families is not None: + _families_from_flag() + return screen_host + return screen_families + + def screen_host(): + """Build the model entries, then ask the bind host. + + The task/id pickers (when any) run here too and are grouped with + this screen: Esc on one of them (or on the host field) returns to + the families tree. + """ + try: + _compute_entries() + except _GoBack: + return tui.Wizard.BACK + if args.host is not None: + s["host"] = args.host + return _after_host() + host = tui.line_edit( + stdscr, "Bind host", + s["existing_host"] if isinstance(s["existing_host"], str) + else DEFAULT_HOST, + help_lines=["The IP address audiocpp will be hosted on", + "127.0.0.1 (this machine) is probably " + "correct"], back_value=_GO_BACK) + if host is _GO_BACK: + return tui.Wizard.BACK + s["host"] = host + return _after_host() + + def _after_host(): + if args.port is None: + return screen_port + s["port"] = args.port + return _after_port() + + def screen_port(): + port_text = tui.line_edit( + stdscr, "Port", + str(s["existing_port"]) if isinstance(s["existing_port"], int) + else str(config_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 audiocpp will be hosted on"], + back_value=_GO_BACK) + if port_text is _GO_BACK: + return tui.Wizard.BACK + s["port"] = int(port_text) + return _after_port() + + def _after_port(): + s["sync_port"] = None + if s["port"] != config_port(): + return screen_sync_port + return _after_sync() + + def screen_sync_port(): + sync_port = tui.confirm( + stdscr, "Update AUDIOCPP_API_URL in app/converter/config.py " + f"to port {s['port']} so audiobook.py talks to this server", + default=True, cancel_value=_GO_BACK) + if sync_port is _GO_BACK: + return tui.Wizard.BACK + s["sync_port"] = sync_port + return _after_sync() + + def _after_sync(): + if args.build_backend: + s["backend"] = args.build_backend + s["build"] = s["detected_backend"] is None + return _after_backend() + if args.backend: + s["backend"] = args.backend + s["build"] = False + return _after_backend() + if s["detected_backend"] is not None: + # Already built: use the detected backend, no menu, no build. + s["backend"] = s["detected_backend"] + s["build"] = False + return _after_backend() + if s["existing_backend"] in BACKENDS: + # Modify flow: keep the backend an existing server.json records + # (already configured, no rebuild needed). + s["backend"] = s["existing_backend"] + s["build"] = False + return _after_backend() + return screen_backend + + def screen_backend(): + backend_options, backend_default = _backend_options(None) + backend = tui.menu( + stdscr, "Which inference backend was audiocpp_server " + "built for?", backend_options, + default_index=backend_default, back_value=_GO_BACK) + if backend is _GO_BACK: + return tui.Wizard.BACK + s["backend"] = backend + return screen_build + + def screen_build(): + # Not built for any backend yet: offer to build it now. The build + # itself runs in the console tail after the wizard. + build = tui.confirm( + stdscr, f"audiocpp_server is not built for {s['backend']}. " + f"Build it now (runs scripts/build_*)?", + default=True, cancel_value=_GO_BACK) + if build is _GO_BACK: + return tui.Wizard.BACK + s["build"] = build + return _after_backend() + + def _after_backend(): + if args.lazy_load: + s["lazy_load"] = True + return _after_lazy() + return screen_lazy + + def screen_lazy(): + default_lazy = len(s["model_entries"]) > 1 + if isinstance(s["existing_lazy"], bool): + default_lazy = s["existing_lazy"] + lazy_load = tui.confirm( + stdscr, "Load models lazily (on first use instead of at " + "startup)", default=default_lazy, cancel_value=_GO_BACK) + if lazy_load is _GO_BACK: + return tui.Wizard.BACK + s["lazy_load"] = lazy_load + return _after_lazy() + + def _after_lazy(): + if args.input_dir is not None: + s["wav_dir"] = args.input_dir + return _after_wav() + if s["include_clone"]: + return screen_wav + s["wav_dir"] = None + return _after_wav() + + def screen_wav(): + wav_start = detect_wav_dir(s["audiocpp_dir"], TTS_ROOT) + # Modify flow: an existing voice_dir seeds the browser so the user + # can accept it on Enter instead of re-navigating. + if isinstance(s["existing_voice_dir"], str) and s["existing_voice_dir"]: + wav_start = Path(s["existing_voice_dir"]) + wav_dir = tui.browse_directory( + stdscr, "Select the directory with your .wav voices", + info=_wav_dir_info, preview=_wav_dir_preview, + start=wav_start if wav_start is not None else VOICES_DIR, + back_value=_GO_BACK) + if wav_dir is _GO_BACK: + return tui.Wizard.BACK + s["wav_dir"] = wav_dir + return _after_wav() + + def _after_wav(): + s["plan"] = None + if s["include_clone"] and s["wav_dir"] is not None: + wav_files = find_wav_files(s["wav_dir"]) + if wav_files: + prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME + if prompt_path.exists() and not args.force: + return screen_transcription + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + s["plan"] = _decide_transcription( + wav_files, existing, prompt_path.exists(), + args.force, ask_confirm) + return _after_transcription() + + def screen_transcription(): + # Transcription plan (questions only; transcription runs after). + wav_files = find_wav_files(s["wav_dir"]) + prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + try: + s["plan"] = _decide_transcription( + wav_files, existing, prompt_path.exists(), + args.force, ask_confirm) + except _GoBack: + return tui.Wizard.BACK + return _after_transcription() + + def _after_transcription(): + s["sync_model_ids"] = None + if len(s["entry_ids"]) == 1 and not ( + config.AUDIOCPP_MODEL_ID == s["entry_ids"][0] + and config.AUDIOCPP_CLONE_MODEL_ID == s["entry_ids"][0]): + return screen_model_sync + return _after_model_sync() + + def screen_model_sync(): + sync_model_ids = tui.confirm( + stdscr, "Update AUDIOCPP_MODEL_ID and " + "AUDIOCPP_CLONE_MODEL_ID in app/converter/config.py to " + f"'{s['entry_ids'][0]}' so audiobook.py uses this model", + default=True, cancel_value=_GO_BACK) + if sync_model_ids is _GO_BACK: + return tui.Wizard.BACK + s["sync_model_ids"] = sync_model_ids + return _after_model_sync() + + def _after_model_sync(): + new_paths = {entry["path"] for entry in s["model_entries"]} + s["unused_entries"] = unused_installed_entries( + s["output_path"], new_paths) \ + if s["existing_config"] is not None else [] + s["delete_unused"] = False + if s["unused_entries"]: + return screen_delete_unused + return _after_delete() + + def screen_delete_unused(): + delete_unused = tui.confirm( + stdscr, "Delete unused models?", default=False, + cancel_value=_GO_BACK) + if delete_unused is _GO_BACK: + return tui.Wizard.BACK + s["delete_unused"] = delete_unused + return _after_delete() + + def _after_delete(): + manager = s["audiocpp_dir"] / "tools" / "model_manager_v2.py" + if manager.is_file(): + return screen_download + s["download"] = False + return _finalize() + + def screen_download(): + # Automatic model download (or print the install commands). + try: + s["download"] = _decide_download(s["audiocpp_dir"], ask_confirm) + except _GoBack: + return tui.Wizard.BACK + return _finalize() - if step == 8: - # Automatic model download (or print the install commands). Esc - # falls back to the delete-unused step (7). - try: - download = _decide_download(audiocpp_dir, ask_confirm) - except _GoBack: - step = 7 - continue - return { - "audiocpp_dir": audiocpp_dir, - "catalog": catalog, - "catalog_by_family": catalog_by_family, - "output_path": output_path, - "family_keys": family_keys, - "chosen": chosen, - "model_entries": model_entries, - "entry_ids": entry_ids, - "install_guidance": install_guidance, - "design_entry_ids": design_entry_ids, - "include_clone": include_clone, - "host": host, - "port": port, - "backend": backend, - "build": build, - "lazy_load": lazy_load, - "sync_port": sync_port, - "sync_model_ids": sync_model_ids, - "wav_dir": wav_dir, - "plan": plan, - "download": download, - "delete_unused": delete_unused, - "unused_entries": unused_entries, - } + # First screen: resolve the checkout directly when it already exists + # (the modify flow), so the wizard starts on a real screen. + audiocpp_dir = args.audiocpp_dir + if audiocpp_dir is None: + audiocpp_dir = find_local_checkout() + if audiocpp_dir is None: + first = screen_no_checkout + else: + resolve_checkout(audiocpp_dir) + first = _after_families() + return tui.Wizard().run(first) def load_server_config(server_json: Path) -> Optional[dict]: @@ -1826,6 +1905,24 @@ def _execute(settings: dict, args: argparse.Namespace) -> int: return 0 +def setup_screen(stdscr) -> int: + """Run the setup wizard on an existing curses screen (the hub's). + + The hub drives this as one screen of its own ``tui.Wizard`` stack, so + Esc on the wizard's first screen simply returns here and the hub pops + back to the menu that launched it. The console tail (build/transcribe/ + write) runs under ``tui.suspend`` so the hub's curses session stays + intact. Returns 0 on completion, 1 when the user aborted. + """ + parser = build_parser() + args = parser.parse_args([]) + settings = _wizard(stdscr, args, parser) + if settings is None: + return 1 + with tui.suspend(stdscr): + return _execute(settings, args) + + def run_tui(args: Optional[argparse.Namespace] = None, parser: Optional[argparse.ArgumentParser] = None) -> int: """Run the audio.cpp setup wizard end-to-end. diff --git a/app/backends/faster.py b/app/backends/faster.py index 7e1be74..e209ff2 100755 --- a/app/backends/faster.py +++ b/app/backends/faster.py @@ -185,42 +185,37 @@ def _write_voices_json(output_path: Path, wav_dir: Path, language: str, def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: - """Linear TUI wizard collecting every faster-setup decision.""" + """Linear TUI wizard collecting every faster-setup decision. + + 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 (``--wavs``, ``--language``, + ``--whisper-model``, ``--port``, ``--skip-install``, ``--skip-clone``) + or that do not apply (the transcription plan when there is nothing to + decide) 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. + """ _GO_BACK = object() + s: dict = {} - def confirm(question: str, default: bool = True) -> Optional[bool]: + 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("faster-qwen3-tts is not installed. " - "pip install it now?", default=True) - if choice is None: - return None - do_install = choice - - # Step 1: clone (if not cloned and not skipped). - do_clone = False - if not _is_cloned() and not args.skip_clone: - choice = confirm(f"faster-qwen3-tts repo not cloned. Clone it into " - f"./app/{FASTER_DIR_NAME}?", default=True) - if choice is None: - return None - do_clone = choice - - # Step 2: voices.json — an existing one seeds the defaults (modify flow) - # instead of an overwrite prompt. - existing_voices = {} + # An existing voices.json seeds the defaults (modify flow) instead of an + # overwrite prompt; its voices also seed the wav-directory browser. default_output = args.output if default_output is None and _is_cloned(): default_output = _checkout() / "voices.json" + existing_voices = {} if default_output is not None and default_output.exists() \ and not args.force: existing_voices = load_voices(default_output) - + s["default_output"] = default_output + s["existing_voices"] = existing_voices wav_start = VOICES_DIR if existing_voices: ref_dirs = {Path(voice["ref_audio"]).parent @@ -228,17 +223,58 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: if isinstance(voice, dict) and voice.get("ref_audio")} if len(ref_dirs) == 1: wav_start = next(iter(ref_dirs)) - - wav_dir = args.input_dir - if wav_dir is None: + s["wav_start"] = wav_start + + def screen_install(): + choice = tui.confirm(stdscr, "faster-qwen3-tts is not installed. " + "pip install it now?", default=True, + cancel_value=_GO_BACK) + if choice is _GO_BACK: + return tui.Wizard.BACK + s["do_install"] = choice + return _after_install() + + def _after_install(): + if not _is_cloned() and not args.skip_clone: + return screen_clone + s["do_clone"] = False + return _after_clone() + + def screen_clone(): + choice = tui.confirm( + stdscr, f"faster-qwen3-tts repo not cloned. Clone it into " + f"./app/{FASTER_DIR_NAME}?", default=True, + cancel_value=_GO_BACK) + if choice is _GO_BACK: + return tui.Wizard.BACK + s["do_clone"] = choice + return _after_clone() + + def _after_clone(): + if args.input_dir is None: + return screen_wav + s["wav_dir"] = args.input_dir + return _after_wav() + + def screen_wav(): wav_dir = tui.browse_directory( stdscr, "Select the directory with your .wav voices", info=common.wav_dir_info, preview=common.wav_dir_preview, - start=wav_start) - language = args.language - if language is None: + start=s["wav_start"], back_value=_GO_BACK) + if wav_dir is _GO_BACK: + return tui.Wizard.BACK + s["wav_dir"] = wav_dir + return _after_wav() + + def _after_wav(): + if args.language is None: + return screen_language + s["language"] = args.language + return _after_language() + + def screen_language(): default_language = config.LANGUAGE - for voice in existing_voices.values(): + for voice in s["existing_voices"].values(): if isinstance(voice, dict) and voice.get("language"): default_language = voice["language"] break @@ -247,51 +283,89 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: validate=lambda s: None if _try_language(s) else "Unknown language (e.g. English, en)", help_lines=["Language for every voice, as passed to the TTS " - "model (names or short codes accepted)"]) - language = lang_text - whisper_model = args.whisper_model - if whisper_model is None: + "model (names or short codes accepted)"], + back_value=_GO_BACK) + if lang_text is _GO_BACK: + return tui.Wizard.BACK + s["language"] = lang_text + return _after_language() + + def _after_language(): + if args.whisper_model is None: + return screen_whisper + s["whisper_model"] = args.whisper_model + return _after_whisper() + + def screen_whisper(): whisper_model = tui.menu( stdscr, "Whisper model for transcription", [(m, m) for m in WHISPER_MODELS], - default_index=WHISPER_MODELS.index("base")) - output_path = args.output - if output_path is None: + default_index=WHISPER_MODELS.index("base"), + back_value=_GO_BACK) + if whisper_model is _GO_BACK: + return tui.Wizard.BACK + s["whisper_model"] = whisper_model + return _after_whisper() + + def _after_whisper(): # Default into the cloned checkout; fall back to the wav directory # when the checkout is not present (so a flag-only run still works). - output_path = (_checkout() / "voices.json") if _is_cloned() \ - else (wav_dir / "voices.json") - - # Transcription plan: re-transcribe only new voices (or all of them) — - # the "re-transcribe anyway?" offer appears even when nothing is new. - plan: Optional[dict] = {"mode": "all", "missing": [], "existing": {}} - wav_files = find_wav_files(wav_dir) - if wav_files and existing_voices and not args.force: - plan = _decide_faster_transcription(wav_files, existing_voices, - confirm) + s["output_path"] = args.output + if s["output_path"] is None: + s["output_path"] = (_checkout() / "voices.json") if _is_cloned() \ + else (s["wav_dir"] / "voices.json") + wav_files = find_wav_files(s["wav_dir"]) + if wav_files and s["existing_voices"] and not args.force: + return screen_transcription + s["plan"] = {"mode": "all", "missing": [], "existing": {}} + return _after_transcription() + + def screen_transcription(): + # Re-transcribe only new voices (or all of them) — the + # "re-transcribe anyway?" offer appears even when nothing is new. + wav_files = find_wav_files(s["wav_dir"]) + plan = _decide_faster_transcription( + wav_files, s["existing_voices"], _confirm) if plan is None: - return None + return tui.Wizard.BACK + s["plan"] = plan + return _after_transcription() + + def _after_transcription(): + if args.port is None: + return screen_port + s["port"] = args.port + return _finalize() - # Step 3: port + default voice. - port = args.port - if port is None: + def screen_port(): port_text = tui.line_edit( stdscr, "Server port", str(_config_port()), validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) - else "Enter a port number between 1 and 65535") - port = int(port_text) + else "Enter a port number between 1 and 65535", + back_value=_GO_BACK) + if port_text is _GO_BACK: + return tui.Wizard.BACK + s["port"] = int(port_text) + return _finalize() + + def _finalize() -> dict: + return { + "do_install": s.get("do_install", False), + "do_clone": s.get("do_clone", False), + "wav_dir": s["wav_dir"], + "language": s["language"], + "whisper_model": s["whisper_model"], + "output_path": s["output_path"], + "port": s["port"], + "force": args.force, + "plan": s["plan"], + } - return { - "do_install": do_install, - "do_clone": do_clone, - "wav_dir": wav_dir, - "language": language, - "whisper_model": whisper_model, - "output_path": output_path, - "port": port, - "force": args.force, - "plan": plan, - } + if not _is_installed() and not args.skip_install: + first = screen_install + else: + first = _after_install() + return tui.Wizard().run(first) def _try_language(value: str) -> bool: @@ -359,6 +433,23 @@ def _print_launch_hint(voices_path: Path, port: int) -> None: print(f" then run it with --voices {voices_path} --port {port}") +def setup_screen(stdscr) -> int: + """Run the setup wizard on an existing curses screen (the hub's). + + The hub drives this as one screen of its own ``tui.Wizard`` stack, so + Esc on the wizard's first screen simply returns here and the hub pops + back to the menu that launched it. The console tail (install/clone/ + transcribe/write) runs under ``tui.suspend`` so the hub's curses + session stays intact. Returns 0 on completion, 1 when the user aborted. + """ + args = build_parser().parse_args([]) + settings = _wizard(stdscr, args) + if settings is None: + return 1 + with tui.suspend(stdscr): + return _execute(settings) + + def run_tui(args: Optional[argparse.Namespace] = None) -> int: """Run the faster setup wizard end-to-end.""" import curses diff --git a/app/backends/qwen.py b/app/backends/qwen.py index b170eb8..3416ebe 100644 --- a/app/backends/qwen.py +++ b/app/backends/qwen.py @@ -58,59 +58,98 @@ def _config_port(url: str, fallback: int) -> int: def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: - """Linear TUI wizard collecting every qwen-setup decision.""" + """Linear TUI wizard collecting every qwen-setup decision. + + 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. + """ _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: + s: dict = {} + + def screen_install(): + choice = tui.confirm(stdscr, "qwen-tts is not installed. " + "pip install it now?", default=True, + cancel_value=_GO_BACK) + if choice is _GO_BACK: + return tui.Wizard.BACK + s["do_install"] = choice + return _after_install() + + 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)"]) - custom_port = int(port_text) - clone_port = args.port_clone - if clone_port is None: + 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)"]) - clone_port = int(port_text) - - # Step 2: built-in speaker. - speaker = args.speaker - if speaker is None: + 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"]) + 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"], + } - return { - "do_install": do_install, - "custom_port": custom_port, - "clone_port": clone_port, - "speaker": speaker, - } + if not _is_installed() and not args.skip_install: + first = screen_install + else: + first = _after_install() + return tui.Wizard().run(first) def _execute(settings: dict) -> int: @@ -160,6 +199,23 @@ def _print_launch_hint(custom_port: int, clone_port: int) -> None: print("Then run: python audiobook.py --backend qwen") +def setup_screen(stdscr) -> int: + """Run the setup wizard on an existing curses screen (the hub's). + + The hub drives this as one screen of its own ``tui.Wizard`` stack, so + Esc on the wizard's first screen simply returns here and the hub pops + back to the menu that launched it. The console tail (pip install / + config sync) runs under ``tui.suspend`` so the hub's curses session + stays intact. Returns 0 on completion, 1 when the user aborted. + """ + args = build_parser().parse_args([]) + settings = _wizard(stdscr, args) + if settings is None: + return 1 + with tui.suspend(stdscr): + return _execute(settings) + + def run_tui(args: Optional[argparse.Namespace] = None) -> int: """Run the qwen setup wizard end-to-end.""" import curses diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 0e260be..9ecedd1 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -34,7 +34,7 @@ class RegistryTests(unittest.TestCase): def test_every_entry_has_detect_setup_and_uninstall(self): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) - self.assertTrue(callable(info.setup_tui), info.key) + self.assertTrue(callable(info.setup_screen), info.key) self.assertTrue(callable(info.uninstall), info.key) def test_get_returns_entry_by_key(self): @@ -296,3 +296,28 @@ class RemoteSuppressionTests(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class QwenSetupScreenTests(unittest.TestCase): + """qwen.setup_screen: the wizard run on the hub's screen.""" + + def test_abort_returns_one_without_executing(self): + from backends import qwen + with patch.object(qwen, "_wizard", return_value=None) as mk_wizard, \ + patch.object(qwen, "_execute") as mk_execute: + rc = qwen.setup_screen(None) + self.assertEqual(rc, 1) + mk_wizard.assert_called_once() + mk_execute.assert_not_called() + + def test_success_executes_the_tail_under_suspend(self): + import contextlib + from backends import qwen + settings = {"custom_port": 7860} + with patch.object(qwen, "_wizard", return_value=settings), \ + patch.object(qwen, "_execute", return_value=0) as mk_execute, \ + patch.object(qwen.tui, "suspend", contextlib.nullcontext): + rc = qwen.setup_screen(None) + self.assertEqual(rc, 0) + mk_execute.assert_called_once() + self.assertIs(mk_execute.call_args[0][0], settings) diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 3d042db..dd19cd5 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1,5 +1,6 @@ """Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" +import contextlib import io import json import sys @@ -11,6 +12,7 @@ from unittest.mock import MagicMock, patch from converter import config from backends import audiocpp as make_server +from ui import tui FAKE_CONFIG = ( 'LANGUAGE = "English"\n' @@ -1542,6 +1544,68 @@ class HandInstallGuidanceTests(unittest.TestCase): self.assertIn("download", message.lower()) +class WizardNavigationTests(unittest.TestCase): + """Esc in the audio.cpp wizard goes back one screen (via tui.Wizard).""" + + def _args(self): + return make_server.build_parser().parse_args([]) + + def _checkout(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + return _make_checkout(Path(tmp.name)) + + def test_esc_on_first_screen_aborts(self): + # Configure audio.cpp (modify flow): the families tree is the first + # screen, so Esc on it must abort the wizard — not re-show itself. + checkout = self._checkout() + with patch.object(make_server, "find_local_checkout", + return_value=checkout), \ + patch.object(tui, "checkbox_tree", + return_value=make_server._GO_BACK): + settings = make_server._wizard(None, self._args(), + make_server.build_parser()) + self.assertIsNone(settings) + + def test_bind_host_esc_returns_to_families_tree(self): + # Esc on "Bind host" must fall back to the model-family tree, then + # re-selecting proceeds through the rest of the wizard. + checkout = self._checkout() + catalog = make_server.load_model_catalog(checkout) + supertonic = next(i for i, entry in enumerate(catalog) + if entry["family"] == "supertonic") + tree_calls = [] + hosts = iter([make_server._GO_BACK, "127.0.0.1"]) + + def fake_tree(*args, **kwargs): + tree_calls.append(1) + return [(supertonic, "Supertonic-GGUF")] + + def fake_line_edit(stdscr, title, default, **kwargs): + if title == "Bind host": + return next(hosts) + if title == "Port": + return "8080" + return default + + with patch.object(make_server, "find_local_checkout", + return_value=checkout), \ + patch.object(tui, "checkbox_tree", + side_effect=fake_tree), \ + patch.object(tui, "line_edit", + side_effect=fake_line_edit), \ + patch.object(tui, "menu", return_value="cuda"), \ + patch.object(tui, "confirm", return_value=True): + settings = make_server._wizard(None, self._args(), + make_server.build_parser()) + self.assertIsNotNone(settings) + # The tree was re-shown after the host screen's Esc. + self.assertEqual(len(tree_calls), 2) + self.assertEqual(settings["host"], "127.0.0.1") + self.assertEqual([m["id"] for m in settings["model_entries"]], + ["supertonic"]) + + class UninstallTests(unittest.TestCase): """uninstall: stop the server and remove the checkout.""" @@ -1568,3 +1632,27 @@ class UninstallTests(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class SetupScreenTests(unittest.TestCase): + """setup_screen: the wizard run on the hub's screen, console tail via + suspend.""" + + def test_abort_returns_one_without_executing(self): + with patch.object(make_server, "_wizard", return_value=None) as mk_wizard, \ + patch.object(make_server, "_execute") as mk_execute: + rc = make_server.setup_screen(None) + self.assertEqual(rc, 1) + mk_wizard.assert_called_once() + mk_execute.assert_not_called() + + def test_success_executes_the_tail_under_suspend(self): + settings = {"audiocpp_dir": Path("/x")} + with patch.object(make_server, "_wizard", return_value=settings), \ + patch.object(make_server, "_execute", + return_value=0) as mk_execute, \ + patch.object(tui, "suspend", contextlib.nullcontext): + rc = make_server.setup_screen(None) + self.assertEqual(rc, 0) + mk_execute.assert_called_once() + self.assertIs(mk_execute.call_args[0][0], settings) diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py index 21baea7..0da461a 100644 --- a/app/tests/test_backends_faster.py +++ b/app/tests/test_backends_faster.py @@ -4,6 +4,7 @@ import json import sys import tempfile import unittest +import contextlib from pathlib import Path from unittest.mock import patch @@ -247,3 +248,27 @@ class MainTests(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class SetupScreenTests(unittest.TestCase): + """setup_screen: the wizard run on the hub's screen, console tail via + suspend.""" + + def test_abort_returns_one_without_executing(self): + with patch.object(make_voices, "_wizard", return_value=None) as mk_wizard, \ + patch.object(make_voices, "_execute") as mk_execute: + rc = make_voices.setup_screen(None) + self.assertEqual(rc, 1) + mk_wizard.assert_called_once() + mk_execute.assert_not_called() + + def test_success_executes_the_tail_under_suspend(self): + settings = {"wav_dir": Path("/x")} + with patch.object(make_voices, "_wizard", return_value=settings), \ + patch.object(make_voices, "_execute", + return_value=0) as mk_execute, \ + patch.object(make_voices.tui, "suspend", contextlib.nullcontext): + rc = make_voices.setup_screen(None) + self.assertEqual(rc, 0) + mk_execute.assert_called_once() + self.assertIs(mk_execute.call_args[0][0], settings) diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index a8e3ac1..d6340da 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -113,7 +113,7 @@ class HubHelperTests(unittest.TestCase): class HubMenuTests(unittest.TestCase): - """Drive _hub_menu with a fake screen (no terminal).""" + """Drive the hub's screen stack with a fake screen (no terminal).""" def setUp(self): tui._THEME.clear() @@ -133,7 +133,7 @@ class HubMenuTests(unittest.TestCase): # Settings, Quit]. Quit is the 3rd option (Down twice) then Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) with patch.object(hub, "detect_all", return_value=[]): - result = hub._hub_menu(screen) + result = hub._Hub(screen).run() self.assertIsNone(result) def test_menu_has_only_configure_settings_and_quit_without_backends(self): @@ -148,7 +148,7 @@ class HubMenuTests(unittest.TestCase): screen = FakeScreen() with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]): - hub._hub_menu(screen) + hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual(labels, ["Configure backends", "Settings", "Quit"]) @@ -165,7 +165,7 @@ class HubMenuTests(unittest.TestCase): st.installed = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[st]): - hub._hub_menu(screen) + hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual( labels, @@ -190,7 +190,7 @@ class HubMenuTests(unittest.TestCase): with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[dead, external]): - hub._hub_menu(screen) + hub._Hub(screen).run() # Unusable backend: dim name. Running-but-not-installed stays # bright and is tagged remote (found at its remote URL). self.assertEqual( @@ -212,7 +212,7 @@ class HubMenuTests(unittest.TestCase): st.running = True with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[st]): - hub._hub_menu(screen) + hub._Hub(screen).run() labels = [label for label, _ in captured["options"]] self.assertEqual( labels, @@ -230,7 +230,7 @@ class HubMenuTests(unittest.TestCase): with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.shutil, "which", return_value=None): - hub._hub_menu(screen) + hub._Hub(screen).run() self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) @@ -246,7 +246,7 @@ class HubMenuTests(unittest.TestCase): with patch.object(hub.tui, "menu", fake_menu), \ patch.object(hub, "detect_all", return_value=[]), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): - hub._hub_menu(screen) + hub._Hub(screen).run() self.assertIsNone(captured["notice_lines"]) def test_convert_with_no_available_backend_flashes(self): @@ -273,7 +273,7 @@ class HubMenuTests(unittest.TestCase): FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) - result = hub._hub_menu(screen) + result = hub._Hub(screen).run() self.assertIsNone(result) self.assertEqual(len(flashed), 1) self.assertIn("No backend is ready", flashed[0]) @@ -291,7 +291,7 @@ class SubmenuStatusTableTests(unittest.TestCase): captured["title"] = title captured["options"] = options captured.update(kwargs) - return hub._GO_BACK # Esc: back out immediately + return tui.Wizard.BACK # Esc: back out immediately return fake_menu @@ -300,7 +300,7 @@ class SubmenuStatusTableTests(unittest.TestCase): captured["title"] = title captured["fields"] = fields captured.update(kwargs) - return hub._GO_BACK # Cancel: back out immediately + return tui.Wizard.BACK # Cancel: back out immediately return fake_form @@ -316,16 +316,17 @@ class SubmenuStatusTableTests(unittest.TestCase): configured=False, running=True, remote=True), ] with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): - result = hub._configure_backends_menu(None, statuses) - self.assertIsNone(result) - # Install (faster uninstalled), Configure (qwen installed), then + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) + # Configure (qwen installed), Install (faster uninstalled), then # Uninstall (qwen installed); no audio.cpp means no model actions. self.assertEqual([label for label, _ in captured["options"]], - ["Install Backend", "Configure qwen-tts", + ["Configure qwen-tts", "Install Backend", "Uninstall Backend"]) # ...the shared status table carries the states instead. self.assertEqual(captured["table_title"], "Backend status") @@ -341,11 +342,12 @@ class SubmenuStatusTableTests(unittest.TestCase): statuses = [BackendStatus("qwen", "qwen-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._configure_backends_menu(None, statuses) - self.assertIsNone(result) + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) # Nothing installed: only the install entry is offered. self.assertEqual([label for label, _ in captured["options"]], ["Install Backend"]) @@ -364,13 +366,14 @@ class SubmenuStatusTableTests(unittest.TestCase): (checkout / "models" / "present").mkdir(parents=True) (checkout / "models" / "present" / "m.gguf").write_bytes(b"x") with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.audiocpp_backend, "find_local_checkout", return_value=checkout), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._configure_backends_menu(None, statuses) - self.assertIsNone(result) + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) labels = [label for label, _ in captured["options"]] # A model is missing (download), plus the installed backend's # configure + uninstall entries. Deleting unused models now lives @@ -386,9 +389,10 @@ class SubmenuStatusTableTests(unittest.TestCase): configured=True) with patch.object(hub.tui, "form", self._capture_form(captured)), \ + patch.object(hub, "detect_all", return_value=[st]), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._convert_menu(None, [st]) - self.assertIsNone(result) + result = hub._Hub(None).screen_convert() + self.assertIs(result, tui.Wizard.BACK) self.assertEqual(captured["title"], "Convert books") # One form, no picker menu: the first field is the Backend picker, # and only convertible backends are offered in it. @@ -406,7 +410,7 @@ class SubmenuStatusTableTests(unittest.TestCase): def fake_menu(*args, **kwargs): menus.append((args, kwargs)) - return hub._GO_BACK + return tui.Wizard.BACK def fake_flash(stdscr, text, kind="warn"): flashed.append(text) @@ -414,9 +418,10 @@ class SubmenuStatusTableTests(unittest.TestCase): st = BackendStatus("qwen", "qwen-tts", installed=True, configured=False) with patch.object(hub.tui, "menu", fake_menu), \ - patch.object(hub.tui, "flash", fake_flash): - result = hub._convert_menu(None, [st]) - self.assertIsNone(result) + patch.object(hub.tui, "flash", fake_flash), \ + patch.object(hub, "detect_all", return_value=[st]): + result = hub._Hub(None).screen_convert() + self.assertIs(result, tui.Wizard.BACK) self.assertEqual(menus, []) self.assertIn("No backend is ready", flashed[0]) @@ -426,11 +431,12 @@ class SubmenuStatusTableTests(unittest.TestCase): statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._configure_backends_menu(None, statuses) - self.assertIsNone(result) + result = hub._Hub(None).screen_configure() + self.assertIs(result, tui.Wizard.BACK) self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], [("qwen-tts", "installed", "warn", @@ -446,9 +452,11 @@ class SubmenuStatusTableTests(unittest.TestCase): configured=False) with patch.object(hub.tui, "menu", self._capture_menu(captured)), \ + patch.object(hub, "detect_all", + return_value=[installed, remote, gone]), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._server_menu(None, [installed, remote, gone]) - self.assertIsNone(result) + result = hub._Hub(None).screen_server() + self.assertIs(result, tui.Wizard.BACK) # Only the installed backend is offered; a running external server # (remote) can't be stopped from here and must not appear. self.assertEqual([label for label, _ in captured["options"]], @@ -465,9 +473,10 @@ class SubmenuStatusTableTests(unittest.TestCase): remote = BackendStatus("qwen", "qwen-tts", installed=False, configured=False, running=True) - with patch.object(hub.tui, "flash", fake_flash): - result = hub._server_menu(None, [remote]) - self.assertIsNone(result) + with patch.object(hub.tui, "flash", fake_flash), \ + patch.object(hub, "detect_all", return_value=[remote]): + result = hub._Hub(None).screen_server() + self.assertIs(result, tui.Wizard.BACK) self.assertEqual(len(flashed), 1) self.assertIn("No backend is installed", flashed[0]) @@ -477,18 +486,19 @@ class SubmenuStatusTableTests(unittest.TestCase): statuses = [BackendStatus("qwen", "qwen-tts", installed=True, configured=True)] with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value=None): - hub._configure_backends_menu(None, statuses) + hub._Hub(None).screen_configure() self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) class ConvertFlowTests(unittest.TestCase): - """_convert_menu: one form whose first field is the Backend picker, - followed by that backend's options (local config or live remote - queries).""" + """_convert_form / screen_convert: one form whose first field is the + Backend picker, followed by that backend's options (local config or live + remote queries).""" def setUp(self): self.tui = _ScriptedTUI() @@ -515,6 +525,27 @@ class ConvertFlowTests(unittest.TestCase): """A backend status that is ready to convert with.""" return BackendStatus(key, label, installed=True, configured=True) + def _convert(self, stdscr, statuses): + """Run the convert flow with STATUSES, returning the command tuple. + + ``_run_conversion`` is stubbed so the accepted command is captured + instead of launching the run view; None is returned when the flow + aborts before reaching a conversion (nothing ready, a flash). + """ + captured = {} + + def fake_run_conversion(self_, backend, kwargs): + captured["backend"] = backend + captured["kwargs"] = kwargs + + with patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub._Hub, "_run_conversion", + fake_run_conversion): + hub._Hub(None).screen_convert() + if "backend" not in captured: + return None + return ("convert", captured["backend"], captured["kwargs"]) + # ------------------------------------------------------------------ # audio.cpp: remote server (no local checkout / server.json) # ------------------------------------------------------------------ @@ -550,7 +581,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="", speed="1.5") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], hub.BACKEND_AUDIOCPP) @@ -588,7 +619,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="audiocpp-remote", model_id="qwen", audiocpp_voice="(built-in speaker)", instructions="") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) # The sentinel maps to "no voice" (built-in speaker). self.assertIsNone(cmd[2]["voice"]) @@ -608,7 +639,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="audiocpp-remote", model_id="legacy", audiocpp_voice="(built-in speaker)", instructions="") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) @@ -620,7 +651,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="audiocpp-remote", model_id="design", audiocpp_voice=None, instructions="A warm British narrator") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd[2]["voice"]) self.assertEqual(cmd[2]["instructions"], "A warm British narrator") @@ -639,7 +670,7 @@ class ConvertFlowTests(unittest.TestCase): with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") - hub._convert_menu(None, + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) voice_field = self._field("audiocpp_voice") self.assertIsNotNone(voice_field["validate"]("")) @@ -647,14 +678,14 @@ class ConvertFlowTests(unittest.TestCase): def test_audiocpp_remote_unreachable_models_flash_and_abort(self): self._patch_remote(None) # endpoint did not answer valid JSON - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd) self.assertIn("Could not list models", self.tui.flashes[0]) def test_audiocpp_remote_empty_models_flash_and_abort(self): self._patch_remote([]) - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNone(cmd) self.assertIn("hosts no model entries", self.tui.flashes[0]) @@ -668,7 +699,7 @@ class ConvertFlowTests(unittest.TestCase): with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="", instructions="") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp")]) self.assertIsNotNone(cmd) self.assertIsNone(cmd[2]["voice"]) @@ -702,7 +733,7 @@ class ConvertFlowTests(unittest.TestCase): patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp", model_id="qwen", audiocpp_voice="Narrator", instructions="") - cmd = hub._convert_menu(None, + cmd = self._convert(None, [self._ready("audiocpp", "audio.cpp")]) self.assertEqual(queried, []) self.assertIsNotNone(cmd) @@ -728,7 +759,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="audiocpp", model_id="qwen", audiocpp_voice="(built-in speaker)", instructions="") - cmd = hub._convert_menu(None, [ + cmd = self._convert(None, [ self._ready("audiocpp", "audio.cpp"), self._remote("audiocpp", "audio.cpp")]) self.assertEqual(cmd[0], "convert") @@ -748,7 +779,7 @@ class ConvertFlowTests(unittest.TestCase): with patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="audiocpp-remote", model_id="higgs", audiocpp_voice="narrator", instructions="") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._remote("audiocpp", "audio.cpp", url="http://10.0.0.5:8080")]) self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:8080") @@ -777,7 +808,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="qwen", mode="custom", speaker="Serena", clone="") with patch.object(hub.common, "update_config_value") as mk_update: - cmd = hub._convert_menu(None, + cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) speaker_in_memory = hub.config.SPEAKER self.assertEqual(cmd[0], "convert") @@ -809,7 +840,7 @@ class ConvertFlowTests(unittest.TestCase): self._answer_form(backend="qwen", mode="clone", speaker="Vivian", clone="/tmp/ref.wav") with patch.object(hub.common, "update_config_value") as mk_update: - cmd = hub._convert_menu(None, + cmd = self._convert(None, [self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") # Clone mode does not touch the global speaker. @@ -824,7 +855,7 @@ class ConvertFlowTests(unittest.TestCase): with patch.object(hub.faster_backend, "_checkout", return_value=Path(td)): self._answer_form(backend="faster", faster_voice="obama") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._ready("faster", "faster-qwen3-tts")]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], "faster") @@ -839,7 +870,7 @@ class ConvertFlowTests(unittest.TestCase): with patch.object(hub.faster_backend, "_checkout", return_value=checkout): self._answer_form(backend="faster", faster_voice="obama") - cmd = hub._convert_menu( + cmd = self._convert( None, [self._ready("faster", "faster-qwen3-tts")]) self.assertEqual(cmd[2]["voice"], "obama") voice_field = self._field("faster_voice") @@ -851,7 +882,7 @@ class ConvertFlowTests(unittest.TestCase): st = self._remote("faster", "faster-qwen3-tts", url="http://10.0.0.5:8000") self._answer_form(backend="faster-remote", faster_voice="obama") - cmd = hub._convert_menu(None, [st]) + cmd = self._convert(None, [st]) self.assertEqual(cmd[0], "convert") self.assertEqual(cmd[1], "faster") self.assertEqual(cmd[2]["voice"], "obama") @@ -869,7 +900,7 @@ class ConvertFlowTests(unittest.TestCase): patch.object(hub.config, "SPEAKER", "Vivian"): self._answer_form(backend="qwen-remote", mode="clone", speaker="Vivian", clone="/tmp/ref.wav") - cmd = hub._convert_menu(None, [st]) + cmd = self._convert(None, [st]) self.assertEqual(cmd[1], hub.BACKEND_QWEN) self.assertEqual(cmd[2]["clone"], "/tmp/ref.wav") self.assertEqual(cmd[2]["api_url"], "http://10.0.0.5:7861") @@ -895,7 +926,7 @@ class ConvertFlowTests(unittest.TestCase): patch.object(hub.config, "AUDIOCPP_INSTRUCTIONS", ""): self._answer_form(backend="qwen", mode="custom", speaker="Vivian", clone="") - cmd = hub._convert_menu(None, [ + cmd = self._convert(None, [ self._ready("audiocpp", "audio.cpp"), self._ready("qwen", "qwen-tts")]) self.assertEqual(cmd[0], "convert") @@ -1059,30 +1090,55 @@ class PreflightTests(unittest.TestCase): self.assertFalse(hub._preflight(stdscr, self._cmd())) mk_flash.assert_called_once() + def test_confirm_esc_raises_back_to_form(self): + # Esc on an overwrite confirm backs out to the form (one screen), + # not "No" — which could dump the user on the main menu. + stdscr = object() + with patch.object(hub.AudiobookConverter, "preflight_overwrites", + return_value=(["book.txt"], [("book.txt", "x")])) \ + as mk_pre: + hub._preflight(stdscr, self._cmd()) + confirm = mk_pre.call_args.kwargs["confirm"] + with patch.object(hub.tui, "confirm", return_value=hub._CANCEL): + with self.assertRaises(hub._BackToForm): + confirm("overwrite?", True) + class DispatchConversionTests(unittest.TestCase): - """_dispatch_conversion: builds the config and runs the run view.""" + """_Hub._run_conversion: builds the config and runs the run view.""" + + def test_runs_run_view_on_the_hub_screen(self): + timeouts = [] - def test_runs_run_view_inside_curses(self): - from tests.test_tui import FakeScreen + class Screen: + def timeout(self, ms): + timeouts.append(ms) class FakeView: def __init__(self, scr, config): self.config = config + self.scr = scr def run(self): pass - made = [] + screen = Screen() with patch.object(hub, "_prepare_run_config", return_value=hub.runview.RunConfig( backend="qwen", backend_label="qwen-tts", kwargs={}, book_files=[], planned=[])) as mk_cfg, \ - patch("curses.wrapper", - side_effect=lambda cb: cb(FakeScreen())) as mk_wrapper, \ patch.object(hub.runview, "RunView", FakeView): - hub._dispatch_conversion("qwen", {}) + hub._Hub(screen)._run_conversion("qwen", {}) + mk_cfg.assert_called_once() + # The run view leaves a timed getch behind; it is reset so the hub + # menus block for keys again. + self.assertEqual(timeouts, [-1]) + + def test_no_run_config_skips_the_view(self): + with patch.object(hub, "_prepare_run_config", return_value=None) as mk_cfg, \ + patch.object(hub.runview, "RunView") as mk_view: + hub._Hub(None)._run_conversion("qwen", {}) mk_cfg.assert_called_once() - mk_wrapper.assert_called_once() + mk_view.assert_not_called() class AddAutostartTests(unittest.TestCase): @@ -1276,7 +1332,7 @@ class SettingsTests(unittest.TestCase): with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply), \ patch.object(hub.tui, "flash", fake_flash): - hub._settings_menu(None) + hub._Hub(None).screen_settings() self.assertEqual([f["key"] for f in captured["fields"]], ["audio_format", "audio_bitrate", "language", "chunk_size", "unload_models", "audiocpp_port", @@ -1324,7 +1380,7 @@ class SettingsTests(unittest.TestCase): with patch.object(hub.tui, "form", fake_form), \ patch.object(hub, "_apply_settings", fake_apply): - hub._settings_menu(None) + hub._Hub(None).screen_settings() self.assertEqual(applied, []) def test_settings_menu_writes_config_end_to_end(self): @@ -1372,7 +1428,7 @@ class SettingsTests(unittest.TestCase): FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10, 21, ord("3"), ord("0"), ord("0"), 10, 9, 10, 10]) - hub._settings_menu(screen) + hub._Hub(screen).screen_settings() text = path.read_text(encoding="utf-8") self.assertIn('AUDIO_FORMAT = "m4b"', text) self.assertIn("CHUNK_SIZE = 300", text) @@ -1456,37 +1512,39 @@ class AudiocppServerConfigTests(unittest.TestCase): class ConfigureBackendsDispatchTests(unittest.TestCase): - """run() and the configure-backends submenus dispatch their commands.""" + """The configure-backends screens dispatch their backend actions.""" - def test_run_dispatches_install_to_setup_tui(self): + def test_setup_screen_runs_the_backend_wizard_and_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) + with patch.object(info, "setup_screen") as mk_setup: + result = hub._Hub(None).screen_setup(info)() + mk_setup.assert_called_once_with(None) + self.assertIs(result, tui.Wizard.BACK) - def fake_wrapper(cb): - fake_wrapper.calls += 1 - return ("install", "qwen") if fake_wrapper.calls == 1 else None - fake_wrapper.calls = 0 - - import curses - with patch.object(curses, "wrapper", fake_wrapper), \ - patch.object(hub, "get", return_value=info), \ - patch.object(info, "setup_tui") as mk_setup: - hub.run() - mk_setup.assert_called_once_with() - - def test_run_dispatches_uninstall(self): + def test_setup_screen_flashes_on_crash_and_goes_back(self): info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) + flashes = [] - def fake_wrapper(cb): - fake_wrapper.calls += 1 - return ("uninstall", "qwen") if fake_wrapper.calls == 1 else None - fake_wrapper.calls = 0 + def boom(scr): + raise RuntimeError("kaboom") - import curses - with patch.object(curses, "wrapper", fake_wrapper), \ - patch.object(hub, "get", return_value=info), \ - patch.object(info, "uninstall") as mk_uninstall: - hub.run() + with patch.object(info, "setup_screen", boom), \ + patch.object(hub.tui, "flash", + lambda scr, text, kind="warn": + flashes.append(text)): + result = hub._Hub(None).screen_setup(info)() + self.assertIs(result, tui.Wizard.BACK) + self.assertEqual(flashes, ["kaboom"]) + + def test_screen_uninstall_runs_uninstall_and_goes_back(self): + import contextlib + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) + with patch.object(hub._Hub, "_pick_backend", return_value=info), \ + patch.object(info, "uninstall") as mk_uninstall, \ + patch.object(hub.tui, "suspend", contextlib.nullcontext): + result = hub._Hub(None).screen_uninstall() mk_uninstall.assert_called_once_with() + self.assertIs(result, tui.Wizard.BACK) def _capture_flashes(self): flashes = [] @@ -1570,12 +1628,12 @@ class ConfigureBackendsDispatchTests(unittest.TestCase): self.assertEqual(len(flashes), 1) self.assertEqual(flashes[0][1], "err") - def test_pick_backend_menu_install_lists_uninstalled_only(self): + def test_pick_backend_install_lists_uninstalled_only(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): captured["options"] = options - return hub._GO_BACK + return tui.Wizard.BACK infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), BackendInfo("faster", "faster-qwen3-tts", lambda: None, @@ -1585,14 +1643,14 @@ class ConfigureBackendsDispatchTests(unittest.TestCase): BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "menu", fake_menu): - result = hub._pick_backend_menu(None, statuses, "Install Backend", - installed_only=False) + result = hub._Hub(None)._pick_backend(installed_only=False) self.assertIsNone(result) self.assertEqual([label for label, _ in captured["options"]], ["faster-qwen3-tts"]) - def test_pick_backend_menu_uninstall_lists_installed_only(self): + def test_pick_backend_uninstall_lists_installed_only(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -1607,12 +1665,106 @@ class ConfigureBackendsDispatchTests(unittest.TestCase): BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False)] with patch.object(hub, "REGISTRY", infos), \ - patch.object(hub.tui, "menu", fake_menu): - result = hub._pick_backend_menu(None, statuses, "Uninstall Backend", - installed_only=True) - self.assertEqual(result, ("uninstall", "qwen")) + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "get", return_value=infos[0]) as mk_get: + result = hub._Hub(None)._pick_backend(installed_only=True) + self.assertEqual(result, infos[0]) + mk_get.assert_called_once_with("qwen") self.assertEqual([label for label, _ in captured["options"]], ["qwen-tts"]) + +class HubNavigationTests(unittest.TestCase): + """Esc (and q) steps back exactly one screen across the whole hub.""" + + def setUp(self): + tui._THEME.clear() + self.addCleanup(tui._THEME.clear) + + def _info(self, key="audiocpp", label="audio.cpp"): + return BackendInfo(key, label, lambda: None, lambda: 0) + + def _status(self, key="audiocpp", label="audio.cpp"): + return BackendStatus(key, label, installed=True, configured=True) + + def _drive(self, script, statuses, registry): + """Run the hub, feeding SCRIPT (one value per menu) to tui.menu. + + Records the title of every menu shown, in order. ``tui.Wizard.BACK`` + in the script simulates Esc on that menu. + """ + titles = [] + + def menu(stdscr, title, options, **kwargs): + titles.append(title) + return script.pop(0) + + def get(key): + return next((i for i in registry if i.key == key), None) + + with patch.object(hub, "REGISTRY", registry), \ + patch.object(hub, "detect_all", return_value=statuses), \ + patch.object(hub, "get", side_effect=get), \ + patch.object(hub.tui, "menu", menu), \ + patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=None): + hub._Hub(None).run() + return titles + + def test_esc_on_wizard_first_screen_returns_to_configure(self): + # The reported bug: Esc on the audio.cpp "Select TTS model + # families" tree (the wizard's first screen) must land back on + # "Configure backends", not the main menu. + info = self._info() + with patch.object(info, "setup_screen", return_value=1): + titles = self._drive( + ["configure_backends", ("configure", "audiocpp"), + tui.Wizard.BACK, tui.Wizard.BACK], + [self._status()], [info]) + self.assertEqual( + titles, + ["tts-audiobook-generator", "Configure backends", + "Configure backends", "tts-audiobook-generator"]) + + def test_esc_on_install_picker_returns_to_configure(self): + registry = [self._info("audiocpp", "audio.cpp"), + self._info("qwen", "qwen-tts")] + statuses = [self._status("audiocpp", "audio.cpp"), + BackendStatus("qwen", "qwen-tts", installed=False, + configured=False)] + titles = self._drive( + ["configure_backends", "install", tui.Wizard.BACK, + tui.Wizard.BACK, tui.Wizard.BACK], + statuses, registry) + self.assertEqual( + titles, + ["tts-audiobook-generator", "Configure backends", + "Install Backend", "Configure backends", + "tts-audiobook-generator"]) + + def test_esc_on_server_action_returns_one_screen_at_a_time(self): + specs = [ServerSpec("qwen-custom", "http://127.0.0.1:7860", []), + ServerSpec("qwen-clone", "http://127.0.0.1:7861", [])] + status = BackendStatus("qwen", "qwen-tts", installed=True, + configured=True, servers=specs) + registry = [self._info("qwen", "qwen-tts")] + with patch.object(hub.common, "server_running", return_value=False): + titles = self._drive( + ["server", "qwen", "qwen-clone", tui.Wizard.BACK, + tui.Wizard.BACK, tui.Wizard.BACK, tui.Wizard.BACK], + [status], registry) + self.assertEqual( + titles, + ["tts-audiobook-generator", "Start / Stop a server", + "qwen-tts server", "qwen-clone (stopped)", + "qwen-tts server", "Start / Stop a server", + "tts-audiobook-generator"]) + + def test_esc_on_main_menu_quits(self): + titles = self._drive([tui.Wizard.BACK], [], []) + self.assertEqual(titles, ["tts-audiobook-generator"]) + + if __name__ == "__main__": - unittest.main() + unittest.main()
\ No newline at end of file diff --git a/app/tests/test_tui.py b/app/tests/test_tui.py index 49960a1..b206de7 100644 --- a/app/tests/test_tui.py +++ b/app/tests/test_tui.py @@ -947,5 +947,76 @@ class FlashTests(TuiTestCase): self.assert_inside_border(screen) +class WizardTests(unittest.TestCase): + """The tui.Wizard screen-stack driver: Esc steps back one screen.""" + + def test_advances_to_the_final_value(self): + called = [] + + def one(): + called.append("one") + return two + + def two(): + called.append("two") + return {"done": True} + + wizard = tui.Wizard() + self.assertEqual(wizard.run(one), {"done": True}) + self.assertEqual(called, ["one", "two"]) + + def test_back_on_first_screen_aborts(self): + wizard = tui.Wizard() + + def first(): + return tui.Wizard.BACK + + self.assertIsNone(wizard.run(first)) + + def test_back_pops_to_the_previous_screen(self): + calls = [] + + def first(): + calls.append("first") + # Esc on the re-shown first screen finishes the wizard. + return second if calls.count("first") == 1 else {"done": True} + + def second(): + calls.append("second") + return tui.Wizard.BACK + + wizard = tui.Wizard() + self.assertEqual(wizard.run(first), {"done": True}) + self.assertEqual(calls, ["first", "second", "first"]) + + def test_back_goes_one_screen_at_a_time(self): + order = [] + + def a(): + order.append("a") + return b + + def b(): + order.append("b") + return c if order.count("b") == 1 else {"done": True} + + def c(): + order.append("c") + return tui.Wizard.BACK + + wizard = tui.Wizard() + self.assertEqual(wizard.run(a), {"done": True}) + # Back from c lands on b (one screen), not a. + self.assertEqual(order, ["a", "b", "c", "b"]) + + def test_screen_returning_none_aborts(self): + wizard = tui.Wizard() + + def first(): + return None + + self.assertIsNone(wizard.run(first)) + + if __name__ == "__main__": unittest.main() diff --git a/app/ui/hub.py b/app/ui/hub.py index 2ac1551..d71fa4c 100644 --- a/app/ui/hub.py +++ b/app/ui/hub.py @@ -1,19 +1,24 @@ #!/usr/bin/env python3 -"""The TUI main menu for the audiobook generator (run via ``audiobook.py``). +"""The TUI hub 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, or install/configure/remove a backend via the "Configure backends" menu. -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 ('q' mirrors Esc on every screen). -Esc inside a sub-menu falls back to the main menu. +The entire hub runs in one curses session, driven by a single ``tui.Wizard`` +stack of screens (the ``_Hub`` class below). Every menu/action is a screen +that returns the next screen, ``Wizard.BACK`` (Esc/q) to pop one screen, or +None to quit. Backend setup wizards and the conversion run view run as +opaque leaf screens on this same session (console tails under +``tui.suspend``); a leaf screen finishes by returning ``Wizard.BACK``, so +the stack lands back on the menu that launched it. Esc therefore steps back +exactly one screen everywhere — on the main menu (an empty stack) it quits. +'q' mirrors Esc on every screen that has no typed text. """ import contextlib +import functools import io import json import re @@ -53,41 +58,51 @@ from converter.tts import ( ) from ui import runview, tui -_GO_BACK = object() +_CANCEL = object() # sentinel: a convert preflight confirm backed out + + +class _BackToForm(Exception): + """Raised when Esc backs out of a preflight confirm (re-show the form).""" def run() -> int: - """Run the hub menu loop until the user quits. Returns exit code.""" + """Run the hub as one curses session; return the 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 in ("install", "configure"): - info = get(command[1]) - if info is not None: - info.setup_tui() - elif kind == "uninstall": - info = get(command[1]) - if info is not None: - info.uninstall() - elif kind == "convert": - _dispatch_conversion(command[1], command[2]) - elif kind == "server": - _run_server_action(command[1], command[2]) - - -def _hub_menu(stdscr) -> Optional[tuple]: - """Show the main menu; return a command tuple, or None to quit.""" - while True: + try: + curses.wrapper(_app) + except tui.WizardCancelled: + return 0 + except KeyboardInterrupt: + return 130 + return 0 + + +def _app(stdscr) -> None: + """Drive the whole hub as one ``tui.Wizard`` stack of screens.""" + _Hub(stdscr).run() + + +class _Hub: + """The hub as a single ``tui.Wizard`` stack of screens. + + Every menu and action is a zero-argument bound method driven by + ``tui.Wizard``: a screen returns the next screen (advance), + ``Wizard.BACK`` (Esc/q — pop exactly one screen), or None (quit, only + reached from the main menu). Backend setup wizards and the conversion + run view run as opaque leaf screens on this same session; a leaf screen + finishes by returning ``Wizard.BACK``, so the stack naturally lands back + on the menu that launched it. + """ + + def __init__(self, stdscr): + self.stdscr = stdscr + + def run(self) -> None: + tui.Wizard().run(self.screen_main) + + # -- top level ------------------------------------------------------ + + def screen_main(self): statuses = detect_all() options = [("Configure backends", "configure_backends")] # Converting works against an external (remote) server too, but @@ -100,114 +115,273 @@ def _hub_menu(stdscr) -> Optional[tuple]: options.append(("Settings", "settings")) options.append(("Quit", "quit")) choice = tui.menu( - stdscr, "tts-audiobook-generator", options, + self.stdscr, "tts-audiobook-generator", options, + back_value=tui.Wizard.BACK, table_title="Backend status", table_rows=_status_rows(statuses), notice_lines=_notice_lines()) - if choice is None or choice == "quit": + if choice is tui.Wizard.BACK or choice == "quit": return None if choice == "convert": - cmd = _convert_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "configure_backends": - cmd = _configure_backends_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "server": - cmd = _server_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "settings": - _settings_menu(stdscr) - - -def _configure_backends_menu(stdscr, statuses) -> Optional[tuple]: - """One flat menu of backend setup/configure/cleanup actions. - - Replaces the old "Set up a backend" + "Configure a backend" pair with a - single screen whose options are populated from the detected statuses: - install (any uninstalled backend), configure each installed backend, - download/delete audio.cpp models (when a server.json references models - on/off disk), and uninstall. Each option returns a command tuple that - ``run`` dispatches after the curses session ends. - """ - by_key = {st.key: st for st in statuses} - installed = [info for info in REGISTRY - if by_key.get(info.key) is not None - and by_key[info.key].installed] - options: list = [] - if any(info.key not in by_key or not by_key[info.key].installed - for info in REGISTRY): - options.append(("Install Backend", "install")) - for info in installed: - options.append((f"Configure {info.label}", ("configure", info.key))) - - audiocpp_status = by_key.get("audiocpp") - missing = [] - if audiocpp_status is not None and audiocpp_status.installed: - checkout = audiocpp_backend.find_local_checkout() - server_json = checkout / "server.json" if checkout else None - if server_json is not None and server_json.exists(): - missing = audiocpp_backend.missing_model_entries(server_json) - if missing: - options.append(("Download Missing Models (audio.cpp)", - "download_models")) - - if installed: - options.append(("Uninstall Backend", "uninstall")) - - choice = tui.menu( - stdscr, "Configure backends", options, - back_value=_GO_BACK, - help_lines=["Install, configure, or remove a TTS backend."], - table_title="Backend status", - table_rows=_status_rows(statuses), - notice_lines=_notice_lines()) - if choice is _GO_BACK or choice is None: - return None - if choice == "install": - return _pick_backend_menu(stdscr, statuses, "Install Backend", - installed_only=False) - if choice == "uninstall": - return _pick_backend_menu(stdscr, statuses, "Uninstall Backend", - installed_only=True) - if choice == "download_models": - _download_models_action(stdscr) - return None - kind, key = choice - return (kind, key) + return self.screen_convert + if choice == "configure_backends": + return self.screen_configure + if choice == "server": + return self.screen_server + return self.screen_settings + + # -- configure / install / uninstall -------------------------------- + + def screen_configure(self): + """One flat menu of backend setup/configure/cleanup actions. + + Options are populated from the detected statuses: install (any + uninstalled backend), configure each installed backend, + download/delete audio.cpp models (when a server.json references + models on/off disk), and uninstall. Selecting one pushes the next + screen; Esc pops back to the main menu. + """ + while True: + statuses = detect_all() + by_key = {st.key: st for st in statuses} + installed = [info for info in REGISTRY + if by_key.get(info.key) is not None + and by_key[info.key].installed] + options = [(f"Configure {info.label}", ("configure", info.key)) + for info in installed] + if any(info.key not in by_key or not by_key[info.key].installed + for info in REGISTRY): + options.append(("Install Backend", "install")) + + audiocpp_status = by_key.get("audiocpp") + missing = [] + if audiocpp_status is not None and audiocpp_status.installed: + checkout = audiocpp_backend.find_local_checkout() + server_json = checkout / "server.json" if checkout else None + if server_json is not None and server_json.exists(): + missing = audiocpp_backend.missing_model_entries( + server_json) + if missing: + options.append(("Download Missing Models (audio.cpp)", + "download_models")) + + if installed: + options.append(("Uninstall Backend", "uninstall")) + + choice = tui.menu( + self.stdscr, "Configure backends", options, + back_value=tui.Wizard.BACK, + help_lines=["Install, configure, or remove a TTS backend."], + table_title="Backend status", + table_rows=_status_rows(statuses), + notice_lines=_notice_lines()) + if choice is tui.Wizard.BACK: + return tui.Wizard.BACK + if choice == "install": + return self.screen_install + if choice == "uninstall": + return self.screen_uninstall + if choice == "download_models": + _download_models_action(self.stdscr) + continue # an inline action: re-show this same menu + _kind, key = choice + info = get(key) + if info is None: + continue + return self.screen_setup(info) + + def screen_setup(self, info): + """Run one backend's setup wizard as a leaf screen of the stack. + + The wizard drives its own internal ``tui.Wizard`` on this screen; + Esc on its first screen (or Ctrl-C) returns here and the hub pops + back to the menu that launched it. A crash flashes and does the same. + """ + def screen(): + try: + info.setup_screen(self.stdscr) + except tui.WizardCancelled: + pass + except Exception as exc: # noqa: BLE001 - keep the hub alive + tui.flash(self.stdscr, str(exc), "err") + return tui.Wizard.BACK + return screen + + def screen_install(self): + info = self._pick_backend(installed_only=False) + if info is None: + return tui.Wizard.BACK + return self.screen_setup(info) + + def screen_uninstall(self): + info = self._pick_backend(installed_only=True) + if info is None: + return tui.Wizard.BACK + with tui.suspend(self.stdscr): + info.uninstall() + return tui.Wizard.BACK + + def _pick_backend(self, installed_only: bool): + """Pick a backend for the Install/Uninstall actions. + + With INSTALLED_ONLY False every backend is listed (the install + list); with it True only the currently-installed ones are (the + uninstall list). Returns a registry entry, or None to go back. + """ + statuses = detect_all() + by_key = {st.key: st for st in statuses} + if installed_only: + candidates = [info for info in REGISTRY + if by_key.get(info.key) is not None + and by_key[info.key].installed] + else: + candidates = [info for info in REGISTRY + if by_key.get(info.key) is None + or not by_key[info.key].installed] + if not candidates: + tui.flash(self.stdscr, "No backends to list here.") + return None + options = [(info.label, info.key) for info in candidates] + title = "Uninstall Backend" if installed_only else "Install Backend" + key = tui.menu(self.stdscr, title, options, + back_value=tui.Wizard.BACK, + table_title="Backend status", + table_rows=_status_rows(statuses), + notice_lines=_notice_lines()) + if key is tui.Wizard.BACK: + return None + return get(key) + + # -- convert -------------------------------------------------------- + + def screen_convert(self): + """Collect run settings, preflight, then run the conversion view. + + Esc on the form (or Cancel) pops back to the main menu; Esc on a + preflight "overwrite?" confirm returns to the form (one screen). + """ + prepared = _convert_form(self.stdscr) + if prepared is None: + return tui.Wizard.BACK + fields, builders, statuses = prepared + while True: + result = tui.form(self.stdscr, "Convert books", fields, + buttons=("Generate!", "Cancel"), + start_on_buttons=True, + back_value=tui.Wizard.BACK) + if result is tui.Wizard.BACK or result is None: + return tui.Wizard.BACK + _, mapper = builders[result["backend"]] + cmd = mapper(result) + if cmd is None: + return tui.Wizard.BACK + _add_autostart(cmd, statuses) + try: + ok = _preflight(self.stdscr, cmd) + except _BackToForm: + continue + if not ok: + return tui.Wizard.BACK + self._run_conversion(cmd[1], cmd[2]) + return tui.Wizard.BACK + + def _run_conversion(self, backend: str, kwargs: dict) -> None: + """Run a conversion in the full-screen run view on this session. + + A crash inside the view cancels the worker and flashes an error + instead of taking the whole hub down; the timed getch the run view + leaves behind is reset so the hub menus still block for keys. + """ + run_config = _prepare_run_config(backend, kwargs) + if run_config is None: + return + view = runview.RunView(self.stdscr, run_config) + try: + view.run() + except tui.WizardCancelled: + pass + except KeyboardInterrupt: + pass + except Exception as exc: # noqa: BLE001 - keep the hub alive + view._cancel.set() + view._worker.join(timeout=30) + tui.flash(self.stdscr, f"The run view failed: {exc}", "err") + finally: + try: + self.stdscr.timeout(-1) + except Exception: + pass + # -- settings ------------------------------------------------------- -def _pick_backend_menu(stdscr, statuses, title: str, - installed_only: bool) -> Optional[tuple]: - """Pick a backend for the Install/Uninstall actions. + def screen_settings(self): + result = tui.form(self.stdscr, "Settings", _settings_fields(), + back_value=tui.Wizard.BACK) + if result is tui.Wizard.BACK or result is None: + return tui.Wizard.BACK + try: + _apply_settings(result) + except ValueError as exc: + tui.flash(self.stdscr, str(exc), "err") + return tui.Wizard.BACK + tui.flash(self.stdscr, "Settings saved.", "ok") + return tui.Wizard.BACK - With INSTALLED_ONLY False every backend is listed (the install list); - with it True only the currently-installed ones are (the uninstall list). - Returns ``(action, key)`` where action is "install" or "uninstall". - """ - by_key = {st.key: st for st in statuses} - if installed_only: - candidates = [info for info in REGISTRY - if by_key.get(info.key) is not None - and by_key[info.key].installed] - else: - candidates = [info for info in REGISTRY - if by_key.get(info.key) is None - or not by_key[info.key].installed] - if not candidates: - tui.flash(stdscr, "No backends to list here.") - return None - options = [(info.label, info.key) for info in candidates] - key = tui.menu(stdscr, title, options, - back_value=_GO_BACK, - table_title="Backend status", - table_rows=_status_rows(statuses), - notice_lines=_notice_lines()) - if key is _GO_BACK or key is None: - return None - action = "uninstall" if installed_only else "install" - return (action, key) + # -- servers -------------------------------------------------------- + + def screen_server(self): + statuses = detect_all() + candidates = [st for st in statuses if st.installed] + if not candidates: + tui.flash(self.stdscr, "No backend is installed yet — use " + "'Configure backends' first.") + return tui.Wizard.BACK + options = [(st.label, st.key) for st in candidates] + key = tui.menu(self.stdscr, "Start / Stop a server", options, + back_value=tui.Wizard.BACK, + table_title="Backend status", + table_rows=_status_rows(statuses), + notice_lines=_notice_lines()) + if key is tui.Wizard.BACK: + return tui.Wizard.BACK + status = next((s for s in statuses if s.key == key), None) + if status is None: + return tui.Wizard.BACK + specs = status.servers + if not specs: + tui.flash(self.stdscr, f"{status.label} has no server " + "configured. Run 'Configure backends' first.") + return tui.Wizard.BACK + if len(specs) == 1: + return functools.partial(self._server_action, specs[0]) + return functools.partial(self._server_spec, status, specs) + + def _server_spec(self, status, specs): + options = [(f"{s.name} ({'running' if common.server_running(s.url) else 'stopped'})", + s.name) for s in specs] + name = tui.menu(self.stdscr, f"{status.label} server", options, + back_value=tui.Wizard.BACK) + if name is tui.Wizard.BACK: + return tui.Wizard.BACK + spec = next((s for s in specs if s.name == name), None) + if spec is None: + return tui.Wizard.BACK + return functools.partial(self._server_action, spec) + + def _server_action(self, spec): + running = common.server_running(spec.url) + action = tui.menu( + self.stdscr, + f"{spec.name} ({'running' if running else 'stopped'})", + [("Start", "start"), ("Stop", "stop")], + back_value=tui.Wizard.BACK) + if action is tui.Wizard.BACK: + return tui.Wizard.BACK + with tui.suspend(self.stdscr): + if action == "start": + servers.start(spec) + else: + servers.stop(spec.name) + return tui.Wizard.BACK def _download_models_action(stdscr) -> None: @@ -298,8 +472,8 @@ def _notice_lines() -> Optional[list]: return None -def _convert_menu(stdscr, statuses) -> Optional[tuple]: - """Collect run settings on one form: pick a backend, then its options. +def _convert_form(stdscr) -> Optional[tuple]: + """Build the Convert-books form (fields + builders), or None to go back. The first field is the Backend picker; the remaining fields are that backend's options (audio.cpp: model/voice/instructions; qwen: @@ -312,7 +486,11 @@ def _convert_menu(stdscr, statuses) -> Optional[tuple]: prepared up front so the Backend field lists only backends whose options could be gathered — an entry whose data is unavailable (e.g. an unreachable remote audio.cpp server) is dropped here. + + Returns ``(fields, builders, statuses)``; None (after a flash) when + there is nothing to convert with. """ + statuses = detect_all() entries = [] for st in statuses: if st.ready: @@ -367,18 +545,7 @@ def _convert_menu(stdscr, statuses) -> Optional[tuple]: field["visible"] = _gate_backend(field, key) fields += backend_fields fields += _common_fields() - - result = _show_convert_form(stdscr, "Convert books", fields) - if result is None: - return None - _, mapper = builders[result["backend"]] - cmd = mapper(result) - if cmd is None: - return None - _add_autostart(cmd, statuses) - if not _preflight(stdscr, cmd): - return None - return cmd + return fields, builders, statuses def _preflight(stdscr, cmd: tuple) -> bool: @@ -396,8 +563,11 @@ def _preflight(stdscr, cmd: tuple) -> bool: kwargs.get("clone")) def confirm(message: str, default: bool) -> bool: - return tui.confirm(stdscr, message, default=default, - cancel_value=False) + answer = tui.confirm(stdscr, message, default=default, + cancel_value=_CANCEL) + if answer is _CANCEL: + raise _BackToForm() + return answer with contextlib.redirect_stdout(io.StringIO()): book_files, planned = AudiobookConverter.preflight_overwrites( @@ -486,17 +656,6 @@ def _common_kwargs(values: dict) -> dict: } -def _show_convert_form(stdscr, title: str, fields: list) -> Optional[dict]: - """Show the single conversion form (Generate!/Cancel, focus on - Generate!) and return its values, or None/_GO_BACK to go back.""" - result = tui.form(stdscr, title, fields, - buttons=("Generate!", "Cancel"), - start_on_buttons=True, back_value=_GO_BACK) - if result is None or result is _GO_BACK: - return None - return result - - def _audiocpp_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]: """audio.cpp-specific fields and a result mapper for the Convert form. @@ -797,9 +956,9 @@ def _faster_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]: # Settings menu (global output options -> app/converter/config.py) # --------------------------------------------------------------------------- -def _settings_menu(stdscr) -> None: - """Edit the global output settings; Save writes them back to config.py.""" - fields = [ +def _settings_fields() -> list: + """The global output-settings field list (Save writes to config.py).""" + return [ {"key": "audio_format", "label": "Audio format", "kind": "choice", "value": config.AUDIO_FORMAT, "choices": list(AUDIO_FORMATS)}, {"key": "audio_bitrate", "label": "Audio bitrate", "kind": "text", @@ -850,15 +1009,6 @@ def _settings_menu(stdscr) -> None: "value": config.CLONE_REMOTE_URL, "validate": _validate_remote_url}, ] - result = tui.form(stdscr, "Settings", fields, back_value=_GO_BACK) - if result is None or result is _GO_BACK: - return - try: - _apply_settings(result) - except ValueError as exc: - tui.flash(stdscr, str(exc), "err") - return - tui.flash(stdscr, "Settings saved.", "ok") def _validate_bitrate(value: str) -> Optional[str]: @@ -1019,46 +1169,6 @@ def _write_config(updates: dict) -> None: path.write_text(text, encoding="utf-8") -def _dispatch_conversion(backend: str, kwargs: dict) -> None: - """Run a conversion in the full-screen run view (its own curses session). - - ``_prepare_run_config`` turns the accepted form (plus the autostart - decision the convert menu recorded) into everything the run view needs; - the view then boots the server when required, runs the conversion with - progress events, and asks the cancel/stop-server questions itself. A - crash inside the view cancels the worker and returns to the menu - instead of taking the whole hub down. - """ - import curses - run_config = _prepare_run_config(backend, kwargs) - if run_config is None: - return - holder: dict = {} - - def main(stdscr) -> None: - view = runview.RunView(stdscr, run_config) - holder["view"] = view - view.run() - - try: - curses.wrapper(main) - except tui.WizardCancelled: - pass - except KeyboardInterrupt: - pass - except Exception as exc: # noqa: BLE001 - keep the hub alive - view = holder.get("view") - if view is not None: - view._cancel.set() - view._worker.join(timeout=30) - print(f"[ERROR] The run view failed: {exc}") - finally: - try: - curses.curs_set(1) # restore the text cursor hidden by the TUI - except Exception: - pass - - def _prepare_run_config(backend: str, kwargs: dict ) -> Optional[runview.RunConfig]: """Build the run view's config from the accepted conversion kwargs. @@ -1167,70 +1277,6 @@ def _find_spec(name: str) -> Optional[ServerSpec]: return None -def _run_server_action(spec_name: str, action: str) -> None: - """Run a Start/Stop action in the plain console (after the TUI returns).""" - if action == "start": - spec = _find_spec(spec_name) - if spec is None: - print(f"[ERROR] no server named '{spec_name}'") - return - servers.start(spec) - elif action == "stop": - servers.stop(spec_name) - - -def _server_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend, then one of its servers and a Start/Stop action.""" - # Only backends installed on this machine: a merely-running external - # server cannot be stopped from here (stop() refuses without our pid - # file), so listing it would dead-end. - candidates = [st for st in statuses if st.installed] - if not candidates: - tui.flash(stdscr, "No backend is installed yet — use " - "'Configure backends' first.") - return None - options = [(st.label, st.key) for st in candidates] - key = tui.menu(stdscr, "Start / Stop a server", options, - back_value=_GO_BACK, - table_title="Backend status", - table_rows=_status_rows(statuses), - notice_lines=_notice_lines()) - if key is _GO_BACK or key is None: - return None - status = next((s for s in statuses if s.key == key), None) - if status is None: - return None - return _server_actions(stdscr, status) - - -def _server_actions(stdscr, status) -> Optional[tuple]: - """Pick a server spec (qwen has two) and a Start or Stop action.""" - specs = status.servers - if not specs: - tui.flash(stdscr, f"{status.label} has no server configured. " - "Run 'Configure backends' first.") - return None - if len(specs) == 1: - spec = specs[0] - else: - options = [(f"{s.name} ({'running' if common.server_running(s.url) else 'stopped'})", - s.name) for s in specs] - name = tui.menu(stdscr, f"{status.label} server", options, - back_value=_GO_BACK) - if name is _GO_BACK or name is None: - return None - spec = next((s for s in specs if s.name == name), None) - if spec is None: - return None - running = common.server_running(spec.url) - action = tui.menu( - stdscr, f"{spec.name} ({'running' if running else 'stopped'})", - [("Start", "start"), ("Stop", "stop")], back_value=_GO_BACK) - if action is _GO_BACK or action is None: - return None - return ("server", spec.name, action) - - def _list_voices(voice_dir: str) -> list: """Return sorted .wav stems in VOICE_DIR (best-effort).""" try: diff --git a/app/ui/tui.py b/app/ui/tui.py index 95130cf..0a119e9 100644 --- a/app/ui/tui.py +++ b/app/ui/tui.py @@ -46,6 +46,48 @@ class WizardCancelled(Exception): """Raised when the user presses Esc to abort the wizard.""" +class Wizard: + """Drive a stack of screen closures with "Esc goes back one screen". + + Each screen is a zero-argument callable that shows exactly one + interactive screen and returns a navigation result: + + Wizard.BACK the user pressed Esc/q; go back one screen + a callable advance to that screen (it is the next screen) + None abort the whole wizard + any value finish the wizard and return that value (the settings) + + ``run(first_screen)`` returns the final value, or None when the user + pressed Esc on the first screen (or a screen returned None). Only + screens that actually render are pushed onto the stack, so Esc always + lands on the previous real screen; a step whose value is already known + (a flag, or a condition that does not apply) is folded into the screen + that precedes it and never appears on the stack, so it cannot be backed + into. + """ + + BACK = object() + + def __init__(self): + self._stack = [] + + def run(self, first_screen) -> Optional[object]: + screen = first_screen + while True: + nxt = screen() + if nxt is Wizard.BACK: + if not self._stack: + return None + screen = self._stack.pop() + continue + if nxt is None: + return None + if not callable(nxt): + return nxt + self._stack.append(screen) + screen = nxt + + @contextlib.contextmanager def suspend(scr): """Temporarily leave curses to run plain-console code. |