""" Ncurses-based Prole 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 subprocess import sys import threading from pathlib import Path from . import config as inst_config from .ncurses_ui import CursesWindow, NavFooter, InputField class ProleNcursesInstaller: """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-prole-dev-cluster" self.cluster_environments = ["prole-dev-cluster", "prole-service-cluster", "prole-prod-cluster"] self.selected_env_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 self.k3s_server = os.environ.get("PROLE_K3S_SERVER", "https://pi.prole.org:6443") self.k3s_token = os.environ.get("PROLE_K3S_TOKEN", "") try: self.namespace_owner = getpass.getuser() except Exception: self.namespace_owner = "prole" 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 = {} # 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 = 25 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() # 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 = "Prole Database 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 attr = curses.A_REVERSE if page_id == current_page_id else 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 # Delegate to page-specific handler first if self._handle_page_input(key): return 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() elif key == curses.KEY_UP or key == ord('k'): self._on_prev() elif key == curses.KEY_DOWN or key == ord('j'): self._on_next() elif 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 == "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 == "prole-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 == "prole-service-cluster": self._verify_k3s_services() return True selected_env = self.cluster_environments[self.selected_env_index] if selected_env == "prole-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 == "prole-service-cluster": if key == ord('d') or key == ord('D'): self._deploy_k3s_services() return True elif selected_env == "prole-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: if not server.startswith("http"): server = f"https://{server}" cfg = ( "apiVersion: v1\n" "kind: Config\n" "clusters:\n" "- cluster:\n" f" server: {server}\n" " insecure-skip-tls-verify: true\n" " name: prole-k3s\n" "contexts:\n" "- context:\n" " cluster: prole-k3s\n" " user: prole-k3s\n" " name: prole-k3s\n" "current-context: prole-k3s\n" "users:\n" "- name: prole-k3s\n" " user:\n" f" token: {token}\n" ) prole_service = os.environ.get("PROLE_SERVICE", "").strip() if prole_service: path = Path(prole_service).expanduser() / "secrets" / "k3s.kubeconfig" else: prole_home = os.environ.get("PROLE_HOME", "").strip() if prole_home: path = Path(prole_home).expanduser() / "prole-k3s.kubeconfig" else: path = Path(self.project_root) / "prole-k3s.kubeconfig" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(cfg, encoding="utf-8") os.chmod(path, 0o600) return path def _verify_k3s_services(self): """Verify registry:2, OpenBao, and OpenTofu on remote k3s cluster.""" def _verify(): self.k3s_services_status = {"registry": "Checking...", "openbao": "Checking...", "opentofu": "Checking..."} if not self.k3s_token: self.k3s_services_status = {"registry": "Missing token", "openbao": "Missing token", "opentofu": "Missing token"} return base_cmd = [ "kubectl", "--server=" + self.k3s_server, "--token=" + self.k3s_token, "--insecure-skip-tls-verify=true" ] try: # Check registry res = subprocess.run(base_cmd + ["get", "service", "-A"], capture_output=True, text=True) if "registry" in res.stdout.lower(): self.k3s_services_status["registry"] = "Good" else: self.k3s_services_status["registry"] = "Failing" # In a real scenario, we might trigger deployment here # Check openbao if "openbao" in res.stdout.lower() or "bao" in res.stdout.lower(): self.k3s_services_status["openbao"] = "Good" else: self.k3s_services_status["openbao"] = "Failing" # Check opentofu if "opentofu" in res.stdout.lower(): self.k3s_services_status["opentofu"] = "Good" else: self.k3s_services_status["opentofu"] = "Failing" 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.""" def _deploy(): self.status_message = "Deploying services to remote k3s..." if not self.k3s_server or not self.k3s_token: self.status_message = "Missing k3s server or token." return env = os.environ.copy() env["PROLE_HOME"] = str(self.project_root) env["PROLE_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["PROLE_MODE"] = "k3s" try: kubeconfig_path = self._write_k3s_kubeconfig(self.k3s_server, self.k3s_token) env["KUBECONFIG"] = str(kubeconfig_path) 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 self.controller.run_script( "init_registry.sh", args=["-n", argocd_ns, "--registry-namespace", registry_ns, "update"], env=env, ) self.controller.run_script("init_openbao.sh", args=["-n", env["NAMESPACE"], "update"], env=env) self.controller.run_script("init_opentofu.sh", args=["-n", env["NAMESPACE"], "update"], env=env) finally: pass 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) 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 # ======================================================================== # 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 Prole Database Installer", y=2) win.render_paragraph( "This installer will guide you through setting up the Prole database system, " "including Docker containers, dependencies, and deployment configuration.", y=5 ) win.render_paragraph( "Navigation: Use arrow keys or h/j/k/l, Enter to select, 'q' to quit.", y=10 ) 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)}") y += 2 for dep in self.dependencies[:10]: # Show first 10 status = "[X]" if dep.get('installed') else "[ ]" win.render_text(y, 4, f"{status} {dep['name']}") y += 1 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 ) win.render_text(8, 4, "[Network scan functionality to be implemented]") 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 Prole installation.", y=5 ) y = 9 win.render_text(y, 4, f"PROLE_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 Prole 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 == "prole-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]): attr = curses.A_REVERSE if (start_idx + i) == self.selected_kubectx_index else curses.A_NORMAL win.render_text(y, 6, f"- {ctx}", attr) y += 1 elif selected_env == "prole-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 == "prole-prod-cluster": win.render_text(y, 4, "Local Artifact Staging Directory:", curses.A_BOLD) y += 2 self.prod_artifacts_path.render() 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 Prole Database", y=2) win.render_paragraph( "Deploying CloudNativePG operator and Prole 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 = ProleNcursesInstaller(stdscr, controller) installer.run() try: curses.wrapper(_main) except KeyboardInterrupt: pass except Exception as e: print(f"Error running ncurses installer: {e}", file=sys.stderr) raise