""" 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 sys import threading from pathlib import Path from typing import Optional, Callable from . import config as inst_config from .ncurses_ui import CursesWindow, TerminalConsole, NavFooter, InputField, Checkbox 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 = "development" self.db_username = "postgres" 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() # Register all screens in order self._register_screens() # Setup footer self.footer = NavFooter(self.footer_win, 3, self.width, self.height - 3) 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), ("kerberos_config", self._render_kerberos_config), ("init_password", self._render_init_password), ("init_db_build", self._render_init_db_build), ("init_cluster", self._render_init_cluster), ("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.""" try: # Clear all windows self.stdscr.clear() self.header_win.clear() self.content_win.clear() self.sidebar_win.clear() self.main_content_win.clear() self.footer_win.clear() # Render header self._render_header() # Render sidebar self._render_sidebar() # Render current page if 0 <= self.page_index < len(self.pages): page_id, render_func = self.pages[self.page_index] render_func() # Render footer self.footer.render() # Refresh all windows self.header_win.refresh() self.sidebar_win.refresh() self.main_content_win.refresh() self.footer_win.refresh() self.stdscr.refresh() 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"), ("Kerberos", "kerberos_config"), ("Password", "init_password"), ("Build", "init_db_build"), ("Cluster", "init_cluster"), ("Scripts", "init_scripts"), ("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 elif 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 _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 Password Setup", y=2) win.render_paragraph( "Set the initial database administrator password.", y=5 ) y = 9 win.render_text(y, 4, f"Username: {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("Initialize Database Cluster", y=2) win.render_paragraph( "Initializing PostgreSQL cluster and configuration...", y=5 ) 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}") 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