"""Ncurses UI adapter for the installer runner. Provides two concrete classes: NcursesUI — minimal skeleton (original interface, unchanged) CursesInstallerUI — full two-panel TUI with number-selection navigation Layout ------ ┌──────────────────────────────────────────────────────────────────────────┐ │ [ k3d ] [ k3s ] [ k8s ] mode strip (line 0) │ ├──────────────────┬───────────────────────────────────────────────────────┤ │ [1] Welcome │ Screen title / content / log scroll │ │ [2] Dependencies│ │ │ ... │ Tab / Shift-Tab — cycle focusable fields │ │ [18] Post Install│ ↑ / ↓ — scroll log │ │ │ q / Esc — return focus to left panel │ ├──────────────────┴───────────────────────────────────────────────────────┤ │ [q] quit [m] cycle mode [↑↓] navigate [Enter] select [Tab] fields │ └──────────────────────────────────────────────────────────────────────────┘ Number shortcuts: type a digit (or two digits within 1 s) to jump to that nav item; press Enter to confirm. Press 'm' at any time to cycle PROLE_MODE k3d → k3s → k8s → k3d. """ from __future__ import annotations import curses import os import time from collections import deque from typing import Deque from ..runner import InstallerRunner from ..state import InstallerState from ..milestone import Milestone # --------------------------------------------------------------------------- # Minimal skeleton (original public interface — unchanged) # --------------------------------------------------------------------------- class NcursesUI: """Minimal ncurses adapter that listens to runner events.""" def __init__(self, runner: InstallerRunner): self.runner = runner self._bind_runner() def _bind_runner(self) -> None: self.runner.on("on_enter_milestone", self.on_enter_milestone) self.runner.on("on_validation_error", self.on_validation_error) self.runner.on("on_progress", self.on_progress) self.runner.on("on_complete", self.on_complete) # Event handlers (override in concrete UI) def on_enter_milestone(self, milestone: Milestone, state: InstallerState) -> None: pass def on_validation_error( self, milestone: Milestone | None, state: InstallerState, errors: list[str] ) -> None: pass def on_progress( self, milestone: Milestone, state: InstallerState, message: str, percent: float | None, ) -> None: pass def on_complete(self, milestone: Milestone | None, state: InstallerState) -> None: pass def start(self, start_id: str | None = None) -> bool: """Run the installer without a GUI main loop.""" return self.runner.run(start_id=start_id) # --------------------------------------------------------------------------- # Full two-panel TUI # --------------------------------------------------------------------------- _NAV_ITEMS = [ ("Welcome", "welcome"), ("Dependencies", "deps_summary"), ("Network", "network_scan"), ("System Environment", "env_setup"), ("Cluster Environment", "init_cluster"), ("Cluster Nodes", "cluster_nodes"), ("Common Services", "common_services"), ("Database Options", "database_options"), ("Docker Build", "init_db_build"), ("Database Creation", "init_password"), ("Initialization Scripts", "init_scripts"), ("Knoe Authority", "kerberos_config"), ("Knoe Users", "knoe_users"), ("ArgoCD", "argocd_config"), ("GitOps", "gitops_config"), ("Supabase", "supabase_config"), ("Deployment", "init_cnpg_deploy"), ("Post Install", "create_installer"), ] _MODES = ["k3d", "k3s", "k8s"] _MODE_PAIRS = list(zip(_MODES, _MODES)) # (label, value) def _left_panel_width() -> int: longest = max(len(label) for label, _ in _NAV_ITEMS) return max(24, longest + 8) def _current_mode() -> str: return os.environ.get("PROLE_MODE", "k3s") def _set_mode(mode: str) -> None: os.environ["PROLE_MODE"] = mode def _cycle_mode() -> str: current = _current_mode() idx = _MODES.index(current) if current in _MODES else 0 new_mode = _MODES[(idx + 1) % len(_MODES)] _set_mode(new_mode) return new_mode class CursesInstallerUI(NcursesUI): """Full two-panel ncurses TUI with number-selection navigation.""" # colour pair indices _CP_NORMAL = 1 _CP_HIGHLIGHT = 2 _CP_HEADER = 3 _CP_MODE_ACTIVE = 4 _CP_MODE_INACTIVE = 5 _CP_STATUS = 6 _CP_LOG = 7 def __init__(self, runner: InstallerRunner): super().__init__(runner) self._log_lines: Deque[str] = deque(maxlen=500) self._scroll_offset: int = 0 self._selected_index: int = 0 # highlighted in left panel self._active_index: int = 0 # confirmed (right panel loaded) self._right_focus: bool = False # True when Tab/arrow are for right panel self._digit_buf: str = "" self._digit_time: float = 0.0 self._running: bool = False # ------------------------------------------------------------------ # Runner event handlers # ------------------------------------------------------------------ def on_enter_milestone(self, milestone: Milestone, state: InstallerState) -> None: self._log_lines.clear() self._scroll_offset = 0 self._log(f"── {milestone.title} ──") def on_progress( self, milestone: Milestone, state: InstallerState, message: str, percent: float | None, ) -> None: prefix = f"[{percent:5.1f}%] " if percent is not None else " " self._log(f"{prefix}{message}") def on_validation_error( self, milestone: Milestone | None, state: InstallerState, errors: list[str] ) -> None: for err in errors: self._log(f"[ERROR] {err}") def on_complete(self, milestone: Milestone | None, state: InstallerState) -> None: self._log("── Complete ──") # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _log(self, msg: str) -> None: for line in msg.splitlines() or [""]: self._log_lines.append(line) # ------------------------------------------------------------------ # curses entry point # ------------------------------------------------------------------ def start(self, start_id: str | None = None) -> bool: # type: ignore[override] return curses.wrapper(self._main, start_id) def _main(self, stdscr: "curses.window", start_id: str | None) -> bool: # type: ignore[name-defined] self._init_colors() curses.curs_set(0) stdscr.nodelay(True) stdscr.keypad(True) self._running = True result: bool = False while self._running: rows, cols = stdscr.getmaxyx() lw = min(_left_panel_width(), cols // 3) rw = max(1, cols - lw - 1) stdscr.erase() self._draw_mode_strip(stdscr, cols) self._draw_left(stdscr, lw, rows) self._draw_divider(stdscr, lw, rows) self._draw_right(stdscr, lw + 1, rw, rows) self._draw_status_bar(stdscr, rows, cols) stdscr.refresh() key = stdscr.getch() if key == -1: time.sleep(0.05) continue result = self._handle_key(key, start_id) if not self._running: break return result # ------------------------------------------------------------------ # Colour initialisation # ------------------------------------------------------------------ def _init_colors(self) -> None: curses.start_color() curses.use_default_colors() curses.init_pair(self._CP_NORMAL, curses.COLOR_WHITE, -1) curses.init_pair(self._CP_HIGHLIGHT, curses.COLOR_BLACK, curses.COLOR_WHITE) curses.init_pair(self._CP_HEADER, curses.COLOR_CYAN, -1) curses.init_pair(self._CP_MODE_ACTIVE, curses.COLOR_BLACK, curses.COLOR_GREEN) curses.init_pair(self._CP_MODE_INACTIVE, curses.COLOR_WHITE, curses.COLOR_BLACK) curses.init_pair(self._CP_STATUS, curses.COLOR_BLACK, curses.COLOR_YELLOW) curses.init_pair(self._CP_LOG, curses.COLOR_GREEN, -1) # ------------------------------------------------------------------ # Drawing # ------------------------------------------------------------------ def _draw_mode_strip(self, stdscr: "curses.window", cols: int) -> None: active = _current_mode() x = 1 try: stdscr.addstr(0, 0, " " * (cols - 1), curses.color_pair(self._CP_NORMAL)) except curses.error: pass for mode in _MODES: label = f" {mode} " if active == mode: attr = curses.color_pair(self._CP_MODE_ACTIVE) | curses.A_BOLD else: attr = curses.color_pair(self._CP_MODE_INACTIVE) try: stdscr.addstr(0, x, label, attr) except curses.error: pass x += len(label) + 1 def _draw_left(self, stdscr: "curses.window", lw: int, rows: int) -> None: # header header = " INSTALLER"[:lw] try: stdscr.addstr(1, 0, header.ljust(lw), curses.color_pair(self._CP_HEADER) | curses.A_BOLD) except curses.error: pass nav_rows = rows - 3 # row 0 = mode strip, row 1 = header, last = status for i, (label, _) in enumerate(_NAV_ITEMS): row = 2 + i if row >= rows - 1: break num = i + 1 text = f" [{num:>2}] {label}"[:lw].ljust(lw) if i == self._selected_index: attr = curses.color_pair(self._CP_HIGHLIGHT) | curses.A_BOLD elif i == self._active_index: attr = curses.color_pair(self._CP_NORMAL) | curses.A_UNDERLINE else: attr = curses.color_pair(self._CP_NORMAL) try: stdscr.addstr(row, 0, text, attr) except curses.error: pass def _draw_divider(self, stdscr: "curses.window", lw: int, rows: int) -> None: for row in range(1, rows - 1): try: stdscr.addch(row, lw, curses.ACS_VLINE) except curses.error: pass def _draw_right(self, stdscr: "curses.window", x0: int, rw: int, rows: int) -> None: label, page_id = _NAV_ITEMS[self._active_index] title = f" {label} ({page_id})"[:rw] try: stdscr.addstr(1, x0, title.ljust(rw), curses.color_pair(self._CP_HEADER) | curses.A_BOLD) except curses.error: pass log_rows = rows - 3 visible = list(self._log_lines) total = len(visible) start = max(0, total - log_rows - self._scroll_offset) end = max(0, total - self._scroll_offset) for i, line in enumerate(visible[start:end]): row = 2 + i if row >= rows - 1: break try: stdscr.addstr(row, x0, line[:rw].ljust(rw), curses.color_pair(self._CP_LOG)) except curses.error: pass def _draw_status_bar(self, stdscr: "curses.window", rows: int, cols: int) -> None: hint = "[q] quit [m] cycle mode [↑↓] navigate [Enter] select [Tab] right panel" try: stdscr.addstr(rows - 1, 0, hint[:cols - 1].ljust(cols - 1), curses.color_pair(self._CP_STATUS)) except curses.error: pass # ------------------------------------------------------------------ # Key handling # ------------------------------------------------------------------ def _handle_key(self, key: int, start_id: str | None) -> bool: # Quit if key in (ord("q"), ord("Q"), 27): # 27 = Esc self._running = False return False # Cycle mode if key == ord("m"): _cycle_mode() return False # Digit accumulation for number jump if ord("0") <= key <= ord("9"): now = time.monotonic() if self._digit_buf and (now - self._digit_time) > 1.0: self._digit_buf = "" self._digit_buf += chr(key) self._digit_time = now # If we have 2 digits, resolve immediately if len(self._digit_buf) == 2: self._resolve_digit_buf() return False # Enter confirms digit buf or selected item if key in (curses.KEY_ENTER, 10, 13): if self._digit_buf: self._resolve_digit_buf() else: self._active_index = self._selected_index self._right_focus = True self._scroll_offset = 0 return False # Arrow navigation if key == curses.KEY_UP: if self._right_focus: self._scroll_offset = min(self._scroll_offset + 1, max(0, len(self._log_lines) - 1)) else: self._selected_index = max(0, self._selected_index - 1) return False if key == curses.KEY_DOWN: if self._right_focus: self._scroll_offset = max(0, self._scroll_offset - 1) else: self._selected_index = min(len(_NAV_ITEMS) - 1, self._selected_index + 1) return False # Tab — switch panel focus if key == ord("\t"): self._right_focus = True self._active_index = self._selected_index return False # Shift-Tab (key code 353 in most terminals) or backtick — back to left if key in (curses.KEY_BTAB, 353): self._right_focus = False return False return False def _resolve_digit_buf(self) -> None: try: n = int(self._digit_buf) if 1 <= n <= len(_NAV_ITEMS): self._selected_index = n - 1 self._active_index = n - 1 self._right_focus = True self._scroll_offset = 0 except ValueError: pass finally: self._digit_buf = ""