aboutsummaryrefslogtreecommitdiff
path: root/app/ui/taskview.py
diff options
context:
space:
mode:
Diffstat (limited to 'app/ui/taskview.py')
-rw-r--r--app/ui/taskview.py171
1 files changed, 36 insertions, 135 deletions
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:
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 15:40:28 +0000