diff options
Diffstat (limited to 'app/ui/hub.py')
| -rw-r--r-- | app/ui/hub.py | 608 |
1 files changed, 327 insertions, 281 deletions
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: |