#!/usr/bin/env python3 """The full-screen run view: server boot + conversion on one screen. Replaces the old plain-console drop after "Generate!": instead of dumping the user into scrolling log output, this widget keeps them in the TUI and shows the two processes that matter — the TTS server (top) and the conversion (bottom, with a chunk progress bar and elapsed time). The screen is fed by two threads the widget spawns: * the worker runs the same code the console path runs — ``backends.servers.start`` (when the conversion needs to boot a managed server; its progress events stream in as they happen) followed by ``audiobook.convert`` with a ``progress`` callback — so behavior is identical to the CLI, only the presentation differs; * a monitor polls the server URL while the conversion runs and reports when it stops answering. Esc and 'q' do the same thing everywhere: a confirmation to cancel processing, then (when this run started the server) a confirmation to shut it down. On the finished screen the behavior follows the convert form's "Stop server and exit" toggle: ON stops the server automatically, quits the whole TUI, and prints the results summary to the real terminal after curses closes; OFF waits for a key press and returns to the hub menu with the server still running. Errors (the server exits while booting, the server stops mid-conversion, a chunk fails and the book aborts) put the corresponding state into error and wait for a key press, so the failure is never scrolled away. """ import contextlib import io import threading import time from dataclasses import dataclass, field from datetime import datetime from queue import Empty, Queue from typing import Callable, List, Optional from backends import common, servers from ui import tui # Terminal states: the run is over and the screen waits for a key. _TERMINAL = ("done", "error", "cancelled") # Server panel states -> (text, theme kind) with the elapsed clock added # while booting. _SERVER_STATES = { "starting": ("starting", "warn"), "ready": ("ready", "ok"), "processing": ("processing", "ok"), "stopping": ("stopping", "warn"), "down": ("not responding", "err"), "error": ("error", "err"), "stopped": ("stopped", "info"), } # Redraw cadence / poll cadence (milliseconds / seconds). _DRAW_TIMEOUT_MS = 250 _MONITOR_INTERVAL = 2.0 @dataclass class RunConfig: """Everything the run view needs to execute one conversion. BACKEND/BACKEND_LABEL identify the chosen backend (label for display); KWARGS are the converter keyword arguments the hub collected (voice, clone, output format, api_url, ...); BOOK_FILES/PLANNED carry the pre-flight overwrite result so the questions are not asked again. SERVER_NAME/SERVER_URL/SERVER_IDENTITY describe the TTS server the conversion talks to (the name is the backends.ServerSpec name; the URL is what the monitor polls). AUTOSTART_SPEC, when not None, is the ServerSpec the worker boots first (the hub only sets it when the server is not already running). LOG_PATH names the converter's log file for the error screen's "details" hint. NOTICE is an optional warning line shown under the progress panel (e.g. a foreign server holding the managed port). STOP_AND_EXIT ("Stop server and exit after generating") skips the finished screen entirely: the server is stopped automatically, the TUI quits, and the results are printed to the real terminal after curses closes. """ backend: str backend_label: str kwargs: dict book_files: list planned: list server_name: Optional[str] = None server_url: Optional[str] = None server_identity: Optional[str] = None autostart_spec: object = None log_path: str = "" notice: str = "" stop_and_exit: bool = False class RunView: """Draws and drives one conversion run; see the module docstring.""" def __init__(self, scr, config: RunConfig, clock: Callable[[], float] = time.time): import curses self.curses = curses self.scr = scr self.config = config self.theme = tui._ensure_theme(curses) self._clock = clock # -- state ----------------------------------------------------- self.phase = "boot" # boot | convert | done | error | cancelled self.server = "starting" self.server_message = "" self.log_tail: List[str] = [] self.book: Optional[tuple] = None # (index, total, name) self.chapter: Optional[tuple] = None # (index, total) self.chunk_done = 0 self.chunk_total = 0 self.book_results: List[tuple] = [] # (name, ok) self.error_message = "" self.cancelled = False self.cancelling = False self.started_server = False self.finished_at: Optional[float] = None self.boot_started: Optional[float] = None self.convert_started: Optional[float] = None self.stop_started: Optional[float] = None self.server_log_path = "" # -- threads --------------------------------------------------- self._queue: Queue = Queue() self._cancel = threading.Event() self._monitor_stop = threading.Event() self._worker = threading.Thread(target=self._worker_main, daemon=True) # ------------------------------------------------------------------ # Event handling (pure state transitions; no drawing) # ------------------------------------------------------------------ def handle_event(self, event: dict) -> None: """Fold one worker/monitor event into the view state.""" kind = event.get("kind") if kind == "starting": self.phase = "boot" self.server = "starting" self.boot_started = self._now() self.started_server = True self.server_log_path = event.get("log_path") or "" elif kind == "running": self.server = "ready" self.boot_started = self.boot_started or self._now() elif kind == "ready": self.server = "ready" elif kind in ("exited", "timeout"): self.server = "error" self.server_message = { "exited": f"server exited with code " f"{event.get('returncode')}", "timeout": "server did not become ready in time", }[kind] self.log_tail = list(event.get("log_tail") or []) self._finish("error") elif kind == "cancelled": self.cancelled = True if self.server in ("starting", "ready", "processing"): self.server = "stopped" self._finish("cancelled") elif kind == "server_down": if self.phase == "convert": self.server = "down" elif kind == "server_stopped": self.server = "stopped" elif kind == "book": self.phase = "convert" self.book = (event.get("index"), event.get("total"), event.get("name") or "") self.chapter = None self.chunk_done = 0 self.chunk_total = 0 self.convert_started = self.convert_started or self._now() if self.server == "ready": self.server = "processing" elif kind == "chapter": self.chapter = (event.get("index"), event.get("total")) self.chunk_done = 0 self.chunk_total = 0 elif kind == "chunks": self.chunk_total = event.get("total") or 0 self.chunk_done = 0 elif kind == "chunk_done": self.chunk_done = event.get("chunk") or self.chunk_done self.chunk_total = event.get("total") or self.chunk_total if self.server in ("ready", "processing"): self.server = "processing" elif kind == "chunk_failed": self.error_message = (f"chunk {event.get('chunk')}/" f"{event.get('total')} failed") if self.server in ("ready", "processing"): self.server = "ready" elif kind == "book_done": self.book_results.append((event.get("name") or "?", bool(event.get("ok")), list(event.get("files") or []), "")) elif kind == "book_failed": self.book_results.append((event.get("name") or "?", False, list(event.get("files") or []), event.get("error") or "conversion failed")) self.error_message = self.error_message or \ (event.get("error") or "conversion failed") elif kind == "done": ok = event.get("ok") or 0 total = event.get("total") or 0 if event.get("cancelled"): self.cancelled = True self._finish("cancelled") elif total and ok >= total and not self.error_message: self._finish("done") else: self.error_message = self.error_message or \ f"{total - ok} of {total} book(s) failed" self._finish("error") elif kind == "error": self.error_message = str(event.get("message") or "error") self._finish("error") elif kind == "worker_exit": if self.phase not in _TERMINAL: self.error_message = self.error_message or \ "the conversion ended unexpectedly" self._finish("error") def _finish(self, phase: str) -> None: """Enter a terminal phase, freezing the elapsed clock.""" self.phase = phase if self.finished_at is None: self.finished_at = self._now() def _now(self) -> float: return self._clock() # ------------------------------------------------------------------ # Threads # ------------------------------------------------------------------ def _worker_main(self) -> None: """Boot the server (when asked) and run the conversion.""" import audiobook config = self.config try: with contextlib.redirect_stdout(io.StringIO()): if config.autostart_spec is not None: ok = servers.start(config.autostart_spec, progress=self._queue.put, cancel=self._cancel) if not ok: if self._cancel.is_set() and self.phase != "error": self._queue.put({"kind": "cancelled"}) return if self._cancel.is_set(): self._queue.put({"kind": "cancelled"}) return # book_files/planned travel on the config fields; dropping # any stray duplicates from kwargs keeps convert()'s call # binding unambiguous. kwargs = {key: value for key, value in config.kwargs.items() if key not in ("book_files", "planned")} audiobook.convert(backend=config.backend, progress=self._queue.put, cancel=self._cancel, book_files=config.book_files, planned=config.planned, **kwargs) except Exception as exc: # noqa: BLE001 - reported to the view self._queue.put({"kind": "error", "message": f"{exc}"}) # The view points failures at the dated log; a crash that # happens before the converter configures logging (e.g. bad # arguments) must still leave its trace there. if self.config.log_path: try: with open(self.config.log_path, "a", encoding="utf-8") as logf: logf.write(f"{datetime.now():%Y-%m-%d %H:%M:%S} - " f"ERROR - {exc}\n") except (OSError, ValueError): pass finally: self._queue.put({"kind": "worker_exit"}) def _monitor_main(self) -> None: """Watch the server URL while converting; report when it drops.""" url = self.config.server_url if not url: return # Give a booting server the full start window before judging it. while not self._monitor_stop.wait(_MONITOR_INTERVAL): if self.phase in _TERMINAL: return if self.phase != "convert": continue if not common.server_running(url): self._queue.put({"kind": "server_down"}) return # ------------------------------------------------------------------ # Main loop # ------------------------------------------------------------------ def run(self) -> bool: """Run the view until the user leaves the terminal screen. Returns True only when the run should end with the whole TUI quitting — the "Stop server and exit" path, which stops the server automatically and records the results as a post-TUI notice. Every other exit (a key press on the summary screen, the Esc cancel flow) lands back on the hub menu. """ scr = self.scr try: self.scr.timeout(_DRAW_TIMEOUT_MS) except Exception: pass self._worker.start() monitor = threading.Thread(target=self._monitor_main, daemon=True) monitor.start() try: while True: self._drain() # The stop-and-exit setting never waits for a key: leave as # soon as the run ends (an explicit Esc cancel keeps its own # interactive flow instead). if self.config.stop_and_exit and self.phase in _TERMINAL \ and self.phase != "cancelled": return self._auto_stop_and_exit() self.render() key = self._get_key() if key is None: continue if self.phase in _TERMINAL: return False if key in (27, ord("q"), 3) and not self.cancelling: if self._prompt_cancel(): return finally: self._monitor_stop.set() self._cancel.set() # Leave the screen blocking again: the timed redraw getch must # not make later hub dialogs (e.g. tui.flash) dismiss themselves. self._blocking() def _get_key(self) -> Optional[int]: """One key from the screen (None on the redraw timeout).""" try: key = self.scr.getch() except KeyboardInterrupt: return 3 if key == -1: return None return key def _drain(self) -> None: """Fold every queued event into the state.""" while True: try: event = self._queue.get_nowait() except Empty: return self.handle_event(event) def _prompt_cancel(self) -> bool: """The Esc/q flow: confirm cancel, then confirm stopping the server. Returns True when the run view should return to the menu (the run is over); False when the user changed their mind and the run keeps going. """ self._blocking() answer = tui.confirm(self.scr, "Cancel processing?", default=False, cancel_value=False) if not answer: self._nonblocking() return False self.cancelling = True self._cancel.set() # When this run booted the server, offer to shut it down too (the # boot path kills it itself when cancelled before ready). self._confirm_stop_server() # Wait for the worker to wind down so the hub menu shows the real # backend state (and the summary screen is drawn at least once). self._worker.join(timeout=60) self._drain() self.render() # One more key press acknowledges the final screen. self._blocking() try: self.scr.getch() except KeyboardInterrupt: pass return True def _confirm_stop_server(self) -> None: """Ask whether to stop the server this run started (Esc-cancel path). The stop runs on a background thread while the screen keeps redrawing the server panel — showing "stopping" with an elapsed clock, mirroring the boot screen — so the SIGTERM grace period never freezes the TUI. Returns once the server is gone. """ if not self.started_server or self._server_stopped_confirmed: return self._server_stopped_confirmed = True name = self.config.server_name if not name or not servers.alive(name): return self._blocking() try: answer = tui.confirm(self.scr, f"Stop the '{name}' server now?", default=True, cancel_value=False) finally: self._nonblocking() if not answer: return self._stop_server_now() def _stop_server_now(self) -> None: """Stop the managed server while the screen keeps repainting. Shared by the Esc-cancel flow and the stop-and-exit path: the stop runs on a background thread and the view drains/render at redraw cadence until it reports done. """ name = self.config.server_name self.server = "stopping" self.stop_started = self._now() done = threading.Event() def _stop() -> None: try: with contextlib.redirect_stdout(io.StringIO()): servers.stop(name) finally: self._queue.put({"kind": "server_stopped"}) done.set() threading.Thread(target=_stop, daemon=True).start() while not done.wait(_DRAW_TIMEOUT_MS / 1000.0): self._drain() self.render() self._drain() self.render() def _auto_stop_and_exit(self) -> bool: """The "Stop server and exit" path. No prompts and no key waits: stop the managed server this run started (if any), record the results summary as a post-TUI notice (printed to the real terminal once curses closes), and report "quit" to the hub. An unmanaged/external server is left alone. """ if self.started_server: name = self.config.server_name if name and servers.alive(name): self._stop_server_now() common.record_post_tui_notice(self._summary_text()) return True def _summary_text(self) -> str: """The results summary printed after the TUI exits. Output directory, one line per book with its generated file names and OK/FAIL status (plus the failure detail), a success count, and the total elapsed time. """ from converter.converter import AUDIOBOOKS_FOLDER lines = ["Audiobook generation finished", f"Output directory: {AUDIOBOOKS_FOLDER}"] ok_count = 0 for name, ok, files, error in self.book_results: ok_count += 1 if ok else 0 lines.append(f"{'[OK]' if ok else '[FAIL]'} {name}" + (f": {', '.join(files)}" if files else "")) if not ok and error: lines.append(f" {error}") total = len(self.book_results) if total: lines.append(f"{ok_count} of {total} book(s) generated " f"successfully") else: lines.append("No books were converted") started = self.convert_started or self.boot_started finished = self.finished_at or self._now() elapsed = finished - (started if started is not None else finished) lines.append(f"Elapsed time: {_format_elapsed(elapsed)}") return "\n".join(lines) def _blocking(self) -> None: """Make getch block (used while a confirm dialog owns the screen).""" try: self.scr.timeout(-1) except Exception: pass def _nonblocking(self) -> None: """Restore the redraw-cadence getch timeout.""" try: self.scr.timeout(_DRAW_TIMEOUT_MS) except Exception: pass _server_stopped_confirmed = False # ------------------------------------------------------------------ # Drawing # ------------------------------------------------------------------ def render(self) -> None: """Repaint the whole screen from the current state.""" curses, theme = self.curses, self.theme scr = self.scr scr.erase() height, width = scr.getmaxyx() if height < 14 or width < 46: _text(scr, theme, height // 2, 2, "Terminal too small", curses.A_BOLD) scr.refresh() return _box(scr, curses, theme, height, width) _text(scr, theme, 0, 2, " Converting audiobooks ", theme["title"]) inner_x = 3 label_w = 9 # "Server", "Status", "Chunk", "Elapsed" value_x = inner_x + label_w + 1 value_w = width - value_x - 3 # -- server panel ------------------------------------------------ y = 2 url = self.config.server_url or "not managed" _text(scr, theme, y, inner_x, "Server".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _fit(f"{self.config.backend_label} @ {url}", value_w), theme["body"]) y += 1 state_text, state_kind = _SERVER_STATES.get( self.server, (self.server, "info")) if self.server == "starting" and self.boot_started is not None: state_text += f" ({int(self._now() - self.boot_started)}s)" elif self.server == "stopping" and self.stop_started is not None: state_text += f" ({int(self._now() - self.stop_started)}s)" if self.config.autostart_spec is None and self.server == "ready": state_text += " (external)" _text(scr, theme, y, inner_x, "Status".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _fit(state_text, value_w), theme.get(state_kind, theme["body"])) y += 2 # -- separator --------------------------------------------------- _sep(scr, curses, theme, y, width) y += 2 if self.phase in _TERMINAL: y = self._draw_summary(scr, theme, y, inner_x, label_w, value_x, value_w, width) else: y = self._draw_progress(scr, theme, y, inner_x, label_w, value_x, value_w, width) # -- footer ------------------------------------------------------ if self.cancelling and self.phase not in _TERMINAL: footer = "cancelling..." elif self.server == "stopping": name = self.config.server_name or "server" footer = f"stopping the {name} server..." elif self.phase in _TERMINAL: footer = "press any key to return to the menu" else: footer = "Esc or q: cancel" _text(scr, theme, height - 2, 2, _fit(footer, width - 4), theme["dim"]) scr.refresh() def _draw_progress(self, scr, theme, y, inner_x, label_w, value_x, value_w, width) -> int: """The live panel: book, chapter, chunk bar, elapsed, message.""" # Book line if self.book is not None: index, total, name = self.book book_text = f"{index}/{total} {name}" else: book_text = "waiting..." if self.phase == "convert" else "-" _text(scr, theme, y, inner_x, "Book".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _fit(book_text, value_w), theme["body"]) y += 1 # Chapter line (only while a multi-chapter book is converting) if self.chapter is not None: _text(scr, theme, y, inner_x, "Chapter".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _fit(f"{self.chapter[0]}/{self.chapter[1]}", value_w), theme["body"]) y += 1 # Chunk bar bar_label = "Chunk".ljust(label_w) _text(scr, theme, y, inner_x, bar_label, theme["dim"]) bar_x = value_x bar_room = max(10, value_w - 12) filled = 0 if self.chunk_total: filled = round(bar_room * self.chunk_done / self.chunk_total) filled = max(0, min(bar_room, filled)) try: scr.addstr(y, bar_x, " " * filled, theme["bar"]) except Exception: pass _text(scr, theme, y, bar_x + bar_room + 1, f"{self.chunk_done}/{self.chunk_total or '?'}", theme["accent"]) y += 1 # Elapsed started = self.convert_started or self.boot_started or self._now() _text(scr, theme, y, inner_x, "Elapsed".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _format_elapsed(self._now() - started), theme["body"]) y += 2 # Message line (last error / current activity) if self.error_message: _text(scr, theme, y, inner_x, _fit(self.error_message, width - inner_x - 3), theme["err"]) y += 1 elif self.server == "down": _text(scr, theme, y, inner_x, _fit("the server stopped responding; the conversion " "will fail", width - inner_x - 3), theme["err"]) y += 1 elif self.config.notice: _text(scr, theme, y, inner_x, _fit(self.config.notice, width - inner_x - 3), theme["warn"]) y += 1 elif self.server_log_path and self.phase == "boot": _text(scr, theme, y, inner_x, _fit(f"loading the model can take a while — log: " f"{self.server_log_path}", width - inner_x - 3), theme["dim"]) y += 1 return y def _draw_summary(self, scr, theme, y, inner_x, label_w, value_x, value_w, width) -> int: """The terminal panel: result, per-book lines, error detail.""" if self.phase == "done": result, kind = "completed", "ok" elif self.phase == "cancelled": result, kind = "cancelled", "warn" else: result, kind = "failed", "err" _text(scr, theme, y, inner_x, "Result".ljust(label_w), theme["dim"]) _text(scr, theme, y, value_x, _fit(result, value_w), theme.get(kind, theme["body"])) y += 1 for name, ok, _files, _error in self.book_results[:5]: mark = "[OK] " if ok else "[FAIL]" _text(scr, theme, y, value_x, _fit(f"{mark} {name}", value_w), theme["ok"] if ok else theme["err"]) y += 1 if len(self.book_results) > 5: _text(scr, theme, y, value_x, _fit(f"... and {len(self.book_results) - 5} more", value_w), theme["dim"]) y += 1 if self.phase == "error": detail = self.error_message or self.server_message if detail: for line in _wrap(detail, width - inner_x - 3)[:2]: _text(scr, theme, y, inner_x, line, theme["err"]) y += 1 if self.log_tail: for line in self.log_tail[:3]: _text(scr, theme, y, inner_x, _fit(line.strip() or " ", width - inner_x - 3), theme["dim"]) y += 1 if self.config.log_path: _text(scr, theme, y, inner_x, _fit(f"details: {self.config.log_path}", width - inner_x - 3), theme["dim"]) y += 1 elif self.phase == "cancelled": _text(scr, theme, y, inner_x, "no audiobook was produced for the cancelled book", theme["dim"]) y += 1 return y # --------------------------------------------------------------------------- # Small drawing/formatting helpers (module-level for testability) # --------------------------------------------------------------------------- def _text(scr, theme, y, x, text, attr) -> None: """addstr wrapper that ignores out-of-bounds errors.""" try: scr.addstr(y, x, text, attr) except Exception: pass def _box(scr, curses, theme, height, width) -> None: """Draw the full-screen frame.""" border = theme["border"] try: scr.addch(0, 0, curses.ACS_ULCORNER, border) scr.addch(0, width - 1, curses.ACS_URCORNER, border) scr.addch(height - 1, 0, curses.ACS_LLCORNER, border) scr.addch(height - 1, width - 1, curses.ACS_LRCORNER, border) scr.hline(0, 1, curses.ACS_HLINE, width - 2, border) scr.hline(height - 1, 1, curses.ACS_HLINE, width - 2, border) for y in range(1, height - 1): scr.addch(y, 0, curses.ACS_VLINE, border) scr.addch(y, width - 1, curses.ACS_VLINE, border) except Exception: pass def _sep(scr, curses, theme, y, width) -> None: """A horizontal separator line inside the frame.""" try: scr.addch(y, 0, curses.ACS_LTEE, theme["border"]) scr.addch(y, width - 1, curses.ACS_RTEE, theme["border"]) scr.hline(y, 1, curses.ACS_HLINE, width - 2, theme["dim"]) except Exception: pass def _fit(text: str, width: int) -> str: """Truncate TEXT to WIDTH columns, appending '~' when cut.""" if width < 1: return "" if len(text) <= width: return text return text[: max(0, width - 1)] + "~" def _wrap(text: str, width: int) -> List[str]: """Greedy word wrap (no textwrap dependency on curses chars).""" lines: List[str] = [] current = "" for word in text.split(): candidate = f"{current} {word}".strip() if len(candidate) <= max(10, width): current = candidate else: if current: lines.append(current) current = word if current: lines.append(current) return lines def _format_elapsed(seconds: float) -> str: """Format a duration as H:MM:SS / M:SS.""" seconds = max(0, int(seconds)) hours, remainder = divmod(seconds, 3600) minutes, secs = divmod(remainder, 60) if hours: return f"{hours}:{minutes:02d}:{secs:02d}" return f"{minutes}:{secs:02d}" def run(scr, config: RunConfig) -> bool: """Enter the run view (called inside curses.wrapper by the hub). Returns True when the stop-and-exit toggle fired — see ``RunView.run``. """ view = RunView(scr, config) return view.run()