From c147087c9d4707bffaeee58d390653637a21cce8 Mon Sep 17 00:00:00 2001 From: historia Date: Wed, 26 Aug 2026 03:02:23 -0400 Subject: refactor: put shared ui screen code into ui.viewkit --- app/ui/taskview.py | 171 +++++++++++------------------------------------------ 1 file changed, 36 insertions(+), 135 deletions(-) (limited to 'app/ui/taskview.py') diff --git a/app/ui/taskview.py b/app/ui/taskview.py index 6708304..0d3b445 100644 --- a/app/ui/taskview.py +++ b/app/ui/taskview.py @@ -47,6 +47,10 @@ from queue import Empty, Queue from typing import Callable, List, Optional, Tuple from ui import tui +from ui.viewkit import (TERMINAL_PHASES as _TERMINAL, + DRAW_TIMEOUT_MS as _DRAW_TIMEOUT_MS, + ScreenView, _box, _fit, _format_elapsed, _sep, + _text) # Redraw cadence for the timed getch (milliseconds). _DRAW_TIMEOUT_MS = 250 @@ -192,22 +196,17 @@ def run_lanes(scr, title: str, lanes: List[TaskLane]) -> int: return LanesView(scr, title, lanes).run() -class TaskView: +class TaskView(ScreenView): """Draws and drives one list of setup steps; see the module docstring.""" def __init__(self, scr, title: str, steps: List[TaskStep], clock: Callable[[], float] = time.time, wait_on_finish: bool = True): - import curses - self.curses = curses - self.scr = scr + super().__init__(scr, clock=clock) self.title = title self.steps = steps self.wait_on_finish = wait_on_finish - self.theme = tui._ensure_theme(curses) - self._clock = clock # -- state ----------------------------------------------------- - self.phase = "running" # running | done | error | cancelled self.current: Optional[int] = None # index of the running step self.results: List[Optional[int]] = [None] * len(steps) self.cancelled_step: Optional[int] = None @@ -312,46 +311,21 @@ class TaskView: if len(self.log_tail) > _LOG_TAIL: del self.log_tail[: len(self.log_tail) - _LOG_TAIL] - def _finish(self, phase: str) -> None: - self.phase = phase - if self.finished_at is None: - self.finished_at = self._now() + # ScreenView hooks ------------------------------------------------ - def _now(self) -> float: - return self._clock() + def _early_exit(self): + """A no-wait run returns as soon as the steps are over.""" + if self.phase in _TERMINAL and not self.wait_on_finish: + return self._result_rc() + return None - # ------------------------------------------------------------------ - # Main loop - # ------------------------------------------------------------------ + def _terminal_result(self) -> int: + return self._result_rc() - def run(self) -> int: - scr = self.scr - try: - scr.timeout(_DRAW_TIMEOUT_MS) - except Exception: - pass - self._worker.start() - first_failure = 0 - try: - while True: - self._drain() - self.render() - if self.phase in _TERMINAL and not self.wait_on_finish: - return self._result_rc() - key = self._get_key() - if key is None: - continue - if self.phase in _TERMINAL: - return self._result_rc() - if key in (27, ord("q"), 3) and not self.cancelling: - if self._prompt_cancel(): - self._drain() - return self._result_rc() - finally: - 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 _after_cancel(self) -> int: + # Fold the worker's final events in so the result reflects them. + self._drain() + return self._result_rc() def _result_rc(self) -> int: """The exit code for the whole run (cancelled counts as failure).""" @@ -359,23 +333,6 @@ class TaskView: return 1 return next((rc for rc in self.results if rc), 0) - def _get_key(self) -> Optional[int]: - try: - key = self.scr.getch() - except KeyboardInterrupt: - return 3 - if key == -1: - return None - return key - - def _drain(self) -> None: - while True: - try: - event = self._queue.get_nowait() - except Empty: - return - self.handle_event(event) - def _prompt_cancel(self) -> bool: """Esc/q: confirm cancel, then wait for the worker to wind down.""" self._blocking() @@ -391,18 +348,6 @@ class TaskView: self._worker.join(timeout=60) return True - def _blocking(self) -> None: - try: - self.scr.timeout(-1) - except Exception: - pass - - def _nonblocking(self) -> None: - try: - self.scr.timeout(_DRAW_TIMEOUT_MS) - except Exception: - pass - # ------------------------------------------------------------------ # Drawing # ------------------------------------------------------------------ @@ -548,55 +493,6 @@ def _find_line_end(text: str) -> int: return min(newline, carriage) -def _text(scr, theme, y, x, text, attr) -> None: - try: - scr.addstr(y, x, text, attr) - except Exception: - pass - - -def _box(scr, curses, theme, height, width) -> None: - 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: - 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: - if width < 1: - return "" - if len(text) <= width: - return text - return text[: max(0, width - 1)] + "~" - - -def _format_elapsed(seconds: float) -> str: - 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 _progress_label(progress: Tuple[float, float], kind: str) -> str: done, total = progress if kind == "bytes": @@ -696,7 +592,24 @@ class _LaneState: self.finished = False -class LanesView: + +class _GetchModes: + """The blocking/non-blocking getch switching shared by all views.""" + + def _blocking(self) -> None: + try: + self.scr.timeout(-1) + except Exception: + pass + + def _nonblocking(self) -> None: + try: + self.scr.timeout(_DRAW_TIMEOUT_MS) + except Exception: + pass + + +class LanesView(_GetchModes): """A full-screen task view that runs two step lists in parallel. The two-lane counterpart of ``TaskView``: each lane gets its own worker @@ -895,18 +808,6 @@ class LanesView: lane.worker.join(timeout=60) return True - def _blocking(self) -> None: - try: - self.scr.timeout(-1) - except Exception: - pass - - def _nonblocking(self) -> None: - try: - self.scr.timeout(_DRAW_TIMEOUT_MS) - except Exception: - pass - # -- drawing ----------------------------------------------------- def render(self) -> None: -- cgit v1.2.3