"""Shared plumbing for full-screen views (the "viewkit"). ``ScreenView`` is the base class behind TaskView and RunView: it owns the event queue and drain loop, the cancellation event, the timed-redraw main loop with its Esc/q cancel flow, and the blocking/non-blocking getch switching that lets confirm dialogs own the screen. Subclasses provide ``handle_event``/``render`` plus small hooks for how a terminal phase exits, and everything else — thread start-up, key handling, cleanup — is identical across views. The drawing helpers at the bottom are the shared primitives both views' render methods build on. """ import threading import time from queue import Empty, Queue from typing import List, Optional from ui import tui # Terminal states: the view's work is over and the screen waits for a key. TERMINAL_PHASES = ("done", "error", "cancelled") # Redraw cadence for the timed getch (milliseconds). DRAW_TIMEOUT_MS = 250 class ScreenView: """Base class for worker-thread-driven full-screen views. Subclasses set ``self._worker`` (a Thread running ``_worker_main``) and implement ``handle_event(event)``, ``render()`` and ``_terminal_result()``. The ``run`` template below drives everything else; its behavior is tuned through the hooks: - ``_start_workers`` start threads (default: just the worker) - ``_early_exit`` pre-render exit check (returns a result or None) - ``_after_cancel`` result once the cancel flow completed - ``_on_stop`` finally-block cleanup (cancel + block getch) ESC/Q/Ctrl-C asks ``_prompt_cancel`` (overridable); confirming sets ``self.cancelling``/``self._cancel`` and winds the worker down. """ def __init__(self, scr, clock=time.time): import curses self.curses = curses self.scr = scr self.theme = tui._ensure_theme(curses) self._clock = clock # -- state ----------------------------------------------------- self.phase = "running" self.finished_at: Optional[float] = None self.cancelled = False self.cancelling = False # -- threads --------------------------------------------------- self._queue: Queue = Queue() self._cancel = threading.Event() self._worker = None def _now(self) -> float: return self._clock() 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() # ------------------------------------------------------------------ # Event plumbing # ------------------------------------------------------------------ def handle_event(self, event: dict) -> None: """Fold one queued event into the view state (no drawing).""" raise NotImplementedError 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) # ------------------------------------------------------------------ # Main loop # ------------------------------------------------------------------ def run(self): """Drive the view until a terminal phase exits the loop.""" try: self.scr.timeout(DRAW_TIMEOUT_MS) except Exception: pass self._start_workers() try: while True: self._drain() early = self._early_exit() if early is not None: return early self.render() key = self._get_key() if key is None: continue if self.phase in TERMINAL_PHASES: return self._terminal_result() if key in (27, ord("q"), 3) and not self.cancelling: if self._prompt_cancel(): return self._after_cancel() finally: self._on_stop() def _start_workers(self) -> None: if self._worker is not None: self._worker.start() def _early_exit(self): """Optional pre-render exit check; a non-None value ends the view.""" return None def _terminal_result(self): """The view's return value when the work reached a terminal phase.""" raise NotImplementedError def _after_cancel(self): """The view's return value after a confirmed cancel flow.""" return self._terminal_result() def _on_stop(self) -> None: 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 _prompt_cancel(self) -> bool: """Esc/q: confirm cancel, then wait for the worker to wind down.""" self._blocking() try: answer = tui.confirm(self.scr, "Cancel this step?", default=False, cancel_value=False) finally: self._nonblocking() if not answer: return False self.cancelling = True self._cancel.set() self._worker.join(timeout=60) return True 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 # ---------------------------------------------------------------------- # Shared drawing primitives # ---------------------------------------------------------------------- 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.""" return tui._fit(text, width) 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 tui._disp_width(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}"