""" Ncurses-based Knoe Installer. Provides a terminal UI alternative to the Tk GUI in install.py. Maintains the same screen order and operations. """ from __future__ import annotations import curses import getpass import os import re import subprocess import sys import threading import time from pathlib import Path from . import config as inst_config from .config import _write_k3s_kubeconfig, _merge_kubeconfig from .core.env import ( _resolve_k3s_connection, _verify_k3s_services_status, _find_kubeconfig_file, resolve_knoe_home, ) from .core.ops import openbao as openbao_ops from .core.ops import opentofu as opentofu_ops from .core.ops import registry as registry_ops from .core.stream_exec import run_streaming_cmd from .ncurses_ui import CursesWindow, NavFooter, InputField class KnoeNcursesInstaller: """Ncurses-based installer matching install.py screen flow.""" def __init__(self, stdscr, controller): self.stdscr = stdscr self.controller = controller self.project_root = controller.project_root # Screen state self.page_index = 0 self.pages = [] # List of (page_id, render_func) tuples self.running = True self.status_message = "" # Data storage (same as GUI version) self.dependencies = inst_config.load_dependencies() self.verify_mode = False self.deploy_environment = None self.selected_disk_type = None self.selected_removable_disk = None self.selected_local_path = None # Environment/config variables self.cluster_env = "k3d-knoe-dev-cluster" self.cluster_environments = [ "knoe-dev-cluster", "knoe-service-cluster", "knoe-prod-cluster", ] self.selected_env_index = 0 self.selected_dep_index = 0 self.kubectx_list = self._get_kubectx_list() self.selected_kubectx_index = 0 self.k3s_services_status = { "registry": "Unknown", "openbao": "Unknown", "opentofu": "Unknown", } # Defer creation until windows are initialized self.prod_artifacts_path = None # Connection details for remote k3s — resolved from env vars, # knoe.cfg *and* ansible topology so all UIs agree. self.k3s_server, self.k3s_token = _resolve_k3s_connection( project_root=self.project_root, ) try: self.namespace_owner = getpass.getuser() except Exception: self.namespace_owner = "knoe" self.db_username = self.namespace_owner self.db_namespace = os.environ.get("NAMESPACE", "default") self.db_password = "" self.db_password_confirm = "" # Kerberos config self.kerberos_enabled = False self.kerberos_realm = "" self.kerberos_user = "" self.kerberos_password = "" self.kerberos_kdc = "" # Validation state self.validation_running = False self.validation_results = {} # Navigation focus self.focus = "sidebar" # Can be "sidebar" or "main" # Setup curses curses.curs_set(0) # Hide cursor curses.use_default_colors() self._setup_colors() # Get screen dimensions self.height, self.width = stdscr.getmaxyx() # Create windows self._create_windows() # Initialize fields that depend on windows self.prod_artifacts_path = InputField(self.main_content_win, 15, 4, 50) self.prod_artifacts_path.set_value(str(self.project_root / "data" / "staging")) # Register all screens in order self._register_screens() # Setup footer self.footer = NavFooter(self.footer_win, 3, self.width, 0) self._update_footer() def _setup_colors(self): """Initialize color pairs.""" if curses.has_colors(): curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_WHITE) curses.init_pair(2, curses.COLOR_GREEN, -1) curses.init_pair(3, curses.COLOR_RED, -1) curses.init_pair(4, curses.COLOR_YELLOW, -1) curses.init_pair(5, curses.COLOR_CYAN, -1) def _create_windows(self): """Create ncurses windows for layout.""" # Header (3 lines) self.header_win = curses.newwin(3, self.width, 0, 0) # Main content area (full height minus header and footer) content_height = self.height - 6 self.content_win = curses.newwin(content_height, self.width, 3, 0) # Footer (3 lines at bottom) self.footer_win = curses.newwin(3, self.width, self.height - 3, 0) # Sidebar (optional, can be rendered in content area) self.sidebar_width = max(25, self.width // 3) self.sidebar_win = curses.newwin(content_height, self.sidebar_width, 3, 0) self.main_content_win = curses.newwin( content_height, self.width - self.sidebar_width - 1, 3, self.sidebar_width + 1, ) def _register_screens(self): """Register all installer screens in order.""" self.pages = [ ("welcome", self._render_welcome), ("deps_summary", self._render_deps_summary), ("network_scan", self._render_network_scan), ("env_setup", self._render_env_setup), ("init_cluster", self._render_init_cluster), ("init_password", self._render_init_password), ("init_db_build", self._render_init_db_build), ("kerberos_config", self._render_kerberos_config), ("init_scripts", self._render_init_scripts), ("init_cnpg_deploy", self._render_init_cnpg_deploy), ("create_installer", self._render_create_installer), ] # Add dynamic dependency screens for dep in self.dependencies: dep_id = f"dep_{dep['id']}" self.pages.append((dep_id, lambda dep=dep: self._render_dependency(dep))) def run(self): """Main event loop.""" while self.running: self._render() self._handle_input() def _render(self): """Render the current screen with flicker-free updates.""" try: # Erase all windows to prevent ghosting from previous frames # Using erase() instead of clear() to avoid flickering self.stdscr.erase() self.header_win.erase() self.sidebar_win.erase() self.main_content_win.erase() self.footer_win.erase() # Render header self._render_header() # Render sidebar self._render_sidebar() # Render current page if 0 <= self.page_index < len(self.pages): _, render_func = self.pages[self.page_index] render_func() # Render footer self.footer.render(focused=(self.focus == "footer")) # Use noutrefresh for all windows to batch updates # stdscr is the background, so refresh it first self.stdscr.noutrefresh() self.header_win.noutrefresh() self.sidebar_win.noutrefresh() self.main_content_win.noutrefresh() self.footer_win.noutrefresh() # Perform a single physical update to the terminal curses.doupdate() except curses.error: pass def _render_header(self): """Render the header.""" try: title = "Knoe.DB Installer" self.header_win.addstr(0, 2, title, curses.A_BOLD) self.header_win.addstr(1, 2, "─" * (self.width - 4)) if self.status_message: self.header_win.addstr(2, 2, self.status_message[: self.width - 4]) except curses.error: pass def _render_sidebar(self): """Render the navigation sidebar.""" try: nav_items = [ ("Welcome", "welcome"), ("Dependencies", "deps_summary"), ("Network Scan", "network_scan"), ("Environment", "env_setup"), ("Cluster Environment", "init_cluster"), ("Password", "init_password"), ("Build", "init_db_build"), ("Scripts", "init_scripts"), ("Kerberos", "kerberos_config"), ("Deploy", "init_cnpg_deploy"), ("Installer", "create_installer"), ] current_page_id = ( self.pages[self.page_index][0] if self.page_index < len(self.pages) else "" ) self.sidebar_win.addstr(0, 1, "Navigation", curses.A_BOLD) self.sidebar_win.addstr(1, 1, "─" * (self.sidebar_width - 2)) for i, (label, page_id) in enumerate(nav_items): y = i + 3 if y >= self.height - 6: break if page_id == current_page_id: if self.focus == "sidebar": attr = curses.A_REVERSE | curses.A_BOLD else: attr = curses.A_BOLD # Selected but not focused else: attr = curses.A_NORMAL display_label = f" {label}"[: self.sidebar_width - 2] self.sidebar_win.addstr( y, 1, display_label.ljust(self.sidebar_width - 3), attr ) # Draw vertical separator for y in range(self.height - 6): self.sidebar_win.addch(y, self.sidebar_width - 1, curses.ACS_VLINE) except curses.error: pass def _handle_input(self): """Handle keyboard input.""" self.stdscr.timeout(100) # 100ms timeout for non-blocking input try: key = self.stdscr.getch() if key == -1: # No input return elif key == ord("q") or key == ord("Q"): self.running = False return # Handle Tab to switch focus if key == ord("\t") or key == 9: focus_order = ["sidebar", "main", "footer"] idx = focus_order.index(self.focus) self.focus = focus_order[(idx + 1) % len(focus_order)] self._update_footer() return # Handle input based on focus if self.focus == "sidebar": if key == curses.KEY_UP or key == ord("k"): self._on_prev() return elif key == curses.KEY_DOWN or key == ord("j"): self._on_next() return elif self.focus == "main": if self._handle_page_input(key): return elif self.focus == "footer": if key == curses.KEY_LEFT or key == ord("h"): self.footer.move_selection(-1) elif key == curses.KEY_RIGHT or key == ord("l"): self.footer.move_selection(1) elif ( key == ord("\n") or key == curses.KEY_ENTER or key == 10 or key == 13 ): self.footer.activate_selected() return # Global keys (always active) if key == ord("n"): self._on_next() elif key == ord("p"): self._on_prev() except curses.error: pass def _handle_page_input(self, key: int) -> bool: """Handle page-specific keyboard input. Returns True if handled.""" page_id = ( self.pages[self.page_index][0] if self.page_index < len(self.pages) else "" ) if page_id == "network_scan": if key == ord("s") or key == ord("S") or key == ord("\n"): self._run_network_scan() return True return False if page_id == "deps_summary": if key == curses.KEY_UP or key == ord("k"): self.selected_dep_index = (self.selected_dep_index - 1) % min( len(self.dependencies), 10 ) return True elif key == curses.KEY_DOWN or key == ord("j"): self.selected_dep_index = (self.selected_dep_index + 1) % min( len(self.dependencies), 10 ) return True elif key == ord("v") or key == ord("V"): self._verify_dependencies() return True elif key == ord(" ") or key == ord("\n"): # In ncurses, we navigate to the dependency page to see details/install dep = self.dependencies[self.selected_dep_index] self.show_page(f"dep_{dep['id']}") return True elif page_id == "init_cluster": if key == curses.KEY_UP or key == ord("k"): self.selected_env_index = (self.selected_env_index - 1) % len( self.cluster_environments ) self.cluster_env = self.cluster_environments[self.selected_env_index] if self.cluster_env == "knoe-service-cluster": self._verify_k3s_services() return True elif key == curses.KEY_DOWN or key == ord("j"): self.selected_env_index = (self.selected_env_index + 1) % len( self.cluster_environments ) self.cluster_env = self.cluster_environments[self.selected_env_index] if self.cluster_env == "knoe-service-cluster": self._verify_k3s_services() return True selected_env = self.cluster_environments[self.selected_env_index] if selected_env == "knoe-dev-cluster": if key == ord(" "): # Toggle/Select context pass # We could implement sub-selection elif key == curses.KEY_NPAGE: # Page Down to scroll context list? self.selected_kubectx_index = ( self.selected_kubectx_index + 1 ) % len(self.kubectx_list) return True elif key == curses.KEY_PPAGE: # Page Up to scroll context list? self.selected_kubectx_index = ( self.selected_kubectx_index - 1 ) % len(self.kubectx_list) return True elif selected_env == "knoe-service-cluster": if key == ord("d") or key == ord("D"): self._deploy_k3s_services() return True elif selected_env == "knoe-prod-cluster": return self.prod_artifacts_path.handle_key(key) return False def _get_kubectx_list(self) -> list[str]: """Get list of kubernetes contexts.""" try: res = subprocess.run(["kubectx"], capture_output=True, text=True) if res.returncode == 0: return res.stdout.strip().split("\n") # Fallback to kubectl res = subprocess.run( ["kubectl", "config", "get-contexts", "-o", "name"], capture_output=True, text=True, ) if res.returncode == 0: return res.stdout.strip().split("\n") except Exception: pass return ["default"] def _write_k3s_kubeconfig(self, server: str, token: str) -> Path: """Delegate to the shared ``_write_k3s_kubeconfig`` helper.""" return _write_k3s_kubeconfig(server, token) def _verify_k3s_services(self): """Verify registry:2, OpenBao, and OpenTofu on remote k3s cluster. Delegates to the shared ``_verify_k3s_services_status`` so the same resolution logic (env → knoe.cfg → ansible) is used. """ def _verify(): self.k3s_services_status = { "registry": "Checking...", "openbao": "Checking...", "opentofu": "Checking...", } try: self.k3s_services_status = _verify_k3s_services_status( project_root=self.project_root, ) except Exception as e: self.status_message = f"K3s connection error: {str(e)}" self.k3s_services_status = { "registry": "Error", "openbao": "Error", "opentofu": "Error", } threading.Thread(target=_verify).start() def _deploy_k3s_services(self): """Deploy registry, OpenBao, and OpenTofu to remote k3s cluster. Uses the shared resolution (env → knoe.cfg → ansible) for the server URL / token and KUBECONFIG so behaviour matches the Tk UI. """ def _deploy(): self.status_message = "Deploying services to remote k3s..." server, token = _resolve_k3s_connection( project_root=self.project_root, ) if not server or not token: self.status_message = "Missing k3s server or token." return env = os.environ.copy() env["KNOE_HOME"] = str(self.project_root) env["KNOE_SERVICE"] = str(self.project_root) env["NAMESPACE"] = self.db_namespace or "default" env["DB_PASSWORD"] = self.db_password or "" env["OPENTOFU_ADMIN_PASSWORD"] = self.db_password or "" env["KNOE_MODE"] = "k3s" # Resolve KUBECONFIG: prefer existing cert-based file, # fall back to token-generated kubeconfig. kubeconfig = _find_kubeconfig_file(env) if not kubeconfig: try: kubeconfig = str(_write_k3s_kubeconfig(server, token)) except Exception: pass if kubeconfig: env["KUBECONFIG"] = kubeconfig try: _merge_kubeconfig(kubeconfig) except Exception: pass try: argocd_ns = env.get("ARGOCD_NAMESPACE") or "argocd" registry_ns = env.get("REGISTRY_NAMESPACE") or "default" env["ARGOCD_NAMESPACE"] = argocd_ns env["REGISTRY_NAMESPACE"] = registry_ns registry_ops.update( namespace=registry_ns, env=env, project_root=self.project_root, mode="k3s", ) openbao_ops.update( namespace=env["NAMESPACE"], env=env, project_root=self.project_root, mode="k3s", ) opentofu_ops.update( namespace=env["NAMESPACE"], env=env, project_root=self.project_root, mode="k3s", ) except Exception as e: self.status_message = f"Deploy error: {e}" return self._verify_k3s_services() self.status_message = "Deployment triggered on remote k3s." threading.Thread(target=_deploy).start() def _update_footer(self): """Update footer buttons based on current page.""" page_id = ( self.pages[self.page_index][0] if self.page_index < len(self.pages) else "" ) # Default buttons prev_enabled = self.page_index > 0 next_label = "Next" next_enabled = True # Last page customization if self.page_index == len(self.pages) - 1: next_label = "Finish" buttons = [ ("Previous", self._on_prev, prev_enabled), (next_label, self._on_next, next_enabled), ("Quit", lambda: setattr(self, "running", False), True), ] self.footer.set_buttons(buttons) # Set help text based on focus if self.focus == "sidebar": self.footer.set_help_text("TAB: Focus Main | UP/DN: Page") elif self.focus == "main": self.footer.set_help_text("TAB: Focus Footer | Arrows: Interact") else: self.footer.set_help_text("TAB: Focus Sidebar | Arrows: Select Button") def _on_next(self): """Navigate to next page.""" if self.page_index < len(self.pages) - 1: self.page_index += 1 self._update_footer() def _on_prev(self): """Navigate to previous page.""" if self.page_index > 0: self.page_index -= 1 self._update_footer() def show_page(self, page_id: str): """Navigate to a specific page by ID.""" for i, (pid, _) in enumerate(self.pages): if pid == page_id: self.page_index = i self._update_footer() return def _verify_dependencies(self): """Run dependency verification in background.""" if self.validation_running: return self.validation_running = True def _worker(): for dep in self.dependencies: did = dep["id"] try: res = inst_config.get_dep_info(dep) self.validation_results[did] = res except Exception: self.validation_results[did] = (False, None, None) time.sleep(0.05) # Feedback delay self.validation_running = False threading.Thread(target=_worker, daemon=True).start() def _run_network_scan(self): """Run network scan via network-agent.""" if getattr(self, "_scan_running", False): return self._scan_running = True self.scan_results_console.clear() self.scan_results_console.write("Starting network scan...\n") def _worker(): try: scan_binary = inst_config.get_resource_path("scan/network-agent") if not scan_binary.exists(): self.scan_results_console.write( f"Error: scan binary not found at {scan_binary}\n" ) self._scan_running = False return knoe_home = resolve_knoe_home() scan_dir = knoe_home / "scan" scan_dir.mkdir(parents=True, exist_ok=True) env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" pending_line = "" def _handle_line(line: str): if "KDC is:" in line: try: ip = ( line.split("KDC is:")[1] .strip() .split()[0] .strip("[]():,") ) if ip: self.kerberos_kdc = ip self.kerberos_enabled = True except Exception: pass elif "Active Directory" in line or "88" in line: match = re.search(r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", line) if match: self.kerberos_kdc = match.group(0) self.kerberos_enabled = True def _on_chunk(text: str): nonlocal pending_line self.scan_results_console.write(text) pending_line += text while "\n" in pending_line: line, pending_line = pending_line.split("\n", 1) _handle_line(line + "\n") rc = run_streaming_cmd( [str(scan_binary)], env=env, cwd=str(scan_dir), on_stdout=_on_chunk, on_stderr=_on_chunk, ) if pending_line: _handle_line(pending_line) if rc == 0: self.scan_results_console.write("\nScan complete.\n") else: self.scan_results_console.write(f"\nScan failed (exit {rc})\n") except Exception as e: self.scan_results_console.write(f"\nError: {str(e)}\n") finally: self._scan_running = False threading.Thread(target=_worker, daemon=True).start() # ======================================================================== # Screen Renderers (11 main screens + dependency screens) # ======================================================================== def _render_welcome(self): """Render the welcome screen.""" win = CursesWindow(self.main_content_win) win.render_title("Welcome to Knoe.DB Installer", y=2) win.render_paragraph( "This installer will guide you through setting up the Knoe database system, " "including Docker containers, dependencies, and deployment configuration.", y=5, ) win.render_paragraph( "Navigation: Use TAB to switch focus between the sidebar and main content.", y=10, ) win.render_paragraph( "Sidebar Focused: Use UP/DOWN to navigate between pages.", y=12 ) win.render_paragraph( "Main Content Focused: Use Arrow Keys or specific shortcuts for each page.", y=14, ) def _render_deps_summary(self): """Render dependencies summary screen.""" win = CursesWindow(self.main_content_win) win.render_title("Dependencies Summary", y=2) y = 5 win.render_text(y, 4, f"Total dependencies: {len(self.dependencies)}") if not self.validation_running and not self.validation_results: self._verify_dependencies() y += 2 for i, dep in enumerate(self.dependencies[:10]): # Show first 10 did = dep["id"] res = self.validation_results.get(did) if res: ok, _, version = res else: ok = dep.get("installed", False) version = dep.get("version", "") status_color = curses.color_pair(2) if ok else curses.color_pair(3) status_char = "v" if ok else "X" status_str = f"[{status_char}]" is_selected = self.focus == "main" and i == self.selected_dep_index attr = curses.A_REVERSE if is_selected else curses.A_NORMAL win.render_text(y, 4, status_str, status_color | attr) win.render_text(y, 8, dep["name"], attr) if ok and version: v_str = f" ({version})"[:20] win.render_text(y, 8 + len(dep["name"]), v_str, curses.A_DIM | attr) elif not ok: win.render_text( y, 8 + len(dep["name"]), " (Missing)", curses.A_DIM | attr ) y += 1 if self.validation_running: win.render_text(y + 1, 4, "Scanning system...", curses.color_pair(4)) else: win.render_text(y + 1, 4, "Press 'v' to re-verify all", curses.A_DIM) def _render_network_scan(self): """Render network scanning screen.""" win = CursesWindow(self.main_content_win) win.render_title("Network Configuration", y=2) win.render_paragraph( "Scanning network for existing services and configurations...", y=5 ) y = 8 h, w = self.main_content_win.getmaxyx() if not hasattr(self, "scan_results_console"): from .ncurses_ui import TerminalConsole # height = content_height - header - footer - titles self.scan_results_console = TerminalConsole( self.main_content_win, h - 12, w - 8, y, 4 ) self.scan_results_console.render() y_footer = h - 2 if getattr(self, "_scan_running", False): win.render_text( y_footer, 4, "Scanning in progress...", curses.color_pair(4) ) else: win.render_text( y_footer, 4, "Press 's' to start network scan", curses.A_BOLD ) def _render_env_setup(self): """Render environment setup screen.""" win = CursesWindow(self.main_content_win) win.render_title("System Environment Setup", y=2) win.render_paragraph( "Configure system environment variables and paths for Knoe installation.", y=5, ) y = 9 win.render_text(y, 4, f"KNOE_HOME: {self.project_root}") y += 2 win.render_text(y, 4, f"Cluster Environment: {self.cluster_env}") def _render_kerberos_config(self): """Render Kerberos configuration screen.""" win = CursesWindow(self.main_content_win) win.render_title("Kerberos Authentication", y=2) win.render_paragraph( "Configure Kerberos authentication for secure database access.", y=5 ) y = 9 enabled = "Enabled" if self.kerberos_enabled else "Disabled" win.render_text(y, 4, f"Kerberos: {enabled}") if self.kerberos_enabled: y += 2 win.render_text(y, 4, f"Realm: {self.kerberos_realm}") y += 1 win.render_text(y, 4, f"KDC: {self.kerberos_kdc}") def _render_init_password(self): """Render database password initialization screen.""" win = CursesWindow(self.main_content_win) win.render_title("Database Creation", y=2) win.render_paragraph( "Define the database namespace and root password for this deployment.", y=5 ) y = 9 win.render_text(y, 4, f"Database Name (Namespace): {self.db_namespace}") y += 2 win.render_text(y, 4, f"Owner (Local User): {self.db_username}") y += 2 win.render_text(y, 4, "Password: " + "*" * len(self.db_password)) y += 1 win.render_text(y, 4, "Confirm: " + "*" * len(self.db_password_confirm)) def _render_init_db_build(self): """Render Docker build screen.""" win = CursesWindow(self.main_content_win) win.render_title("Build Database Container", y=2) win.render_paragraph("Building Docker container for Knoe database...", y=5) win.render_text( 8, 4, "[Docker build process - console output would appear here]" ) def _render_init_cluster(self): """Render cluster initialization screen.""" win = CursesWindow(self.main_content_win) win.render_title("Cluster Environment", y=2) win.render_paragraph( "Select the target cluster environment and verify k3d/k3s/prod plus OpenBao are available.", y=5, ) y = 9 for i, env in enumerate(self.cluster_environments): selector = "(*)" if i == self.selected_env_index else "( )" win.render_text(y, 4, f"{selector} {env}") y += 1 y += 1 selected_env = self.cluster_environments[self.selected_env_index] if selected_env == "knoe-dev-cluster": win.render_text( y, 4, "Available kubectx contexts (Page Up/Down to scroll):", curses.A_BOLD, ) y += 1 start_idx = max(0, self.selected_kubectx_index - 2) for i, ctx in enumerate(self.kubectx_list[start_idx : start_idx + 5]): is_selected = (start_idx + i) == self.selected_kubectx_index if is_selected: attr = curses.A_REVERSE if self.focus == "main" else curses.A_BOLD else: attr = curses.A_NORMAL win.render_text(y, 6, f"- {ctx}", attr) y += 1 elif selected_env == "knoe-service-cluster": win.render_text(y, 4, "Remote K3s Service Status:", curses.A_BOLD) y += 1 for service, status in self.k3s_services_status.items(): color = ( curses.color_pair(2) if status == "Good" else ( curses.color_pair(3) if status == "Failing" else curses.A_NORMAL ) ) win.render_text(y, 6, f"{service}: ") win.render_text(y, 6 + len(service) + 2, status, color) y += 1 if any(s == "Failing" for s in self.k3s_services_status.values()): win.render_text( y + 1, 4, "Press 'd' to deploy missing services", curses.A_BOLD ) elif selected_env == "knoe-prod-cluster": win.render_text(y, 4, "Local Artifact Staging Directory:", curses.A_BOLD) y += 2 # Handle InputField render focus self.prod_artifacts_path.render(focused=(self.focus == "main")) win.render_text(y + 2, 4, "(Used by etc/deploy_pipeline.sh --mode gcp)") def _render_init_scripts(self): """Render initialization scripts screen.""" win = CursesWindow(self.main_content_win) win.render_title("Initialization Scripts", y=2) win.render_paragraph("Running database initialization scripts...", y=5) def _render_init_cnpg_deploy(self): """Render CNPG deployment screen.""" win = CursesWindow(self.main_content_win) win.render_title("Deploy Knoe Database", y=2) win.render_paragraph( "Deploying CloudNativePG operator and Knoe database cluster...", y=5 ) def _render_create_installer(self): """Render installer creation screen.""" win = CursesWindow(self.main_content_win) win.render_title("Create Installation Media", y=2) win.render_paragraph("Create a bootable installer or deployment package.", y=5) if self.selected_disk_type: y = 9 win.render_text(y, 4, f"Type: {self.selected_disk_type}") def _render_dependency(self, dep: dict): """Render individual dependency installation screen.""" win = CursesWindow(self.main_content_win) win.render_title(f"Install: {dep['name']}", y=2) if dep.get("description"): win.render_paragraph(dep["description"], y=5) y = 10 status = "Installed" if dep.get("installed") else "Not Installed" win.render_text(y, 4, f"Status: {status}") class NcursesUIAdapter: """Adapter skeleton that matches the Tkinter UI event interface.""" def __init__(self, runner, controller): self.runner = runner self.controller = controller self._bind_runner() def _bind_runner(self): 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 to bridge into curses rendering) def on_enter_milestone(self, milestone, state): pass def on_validation_error(self, milestone, state, errors): pass def on_progress(self, milestone, state, message, percent): pass def on_complete(self, milestone, state): pass def start(self): run_ncurses_installer(self.controller) def run_ncurses_installer(controller): """Entry point for ncurses installer.""" def _main(stdscr): installer = KnoeNcursesInstaller(stdscr, controller) knoe.run() try: curses.wrapper(_main) except KeyboardInterrupt: pass except Exception as e: print(f"Error running ncurses installer: {e}", file=sys.stderr) raise