""" System Bootloader This bootloader initializes the system components, manages the boot process, and provides a boot menu interface. It integrates with the enhanced CPU and manages initial resource allocation. """ import os import sys import time import json import asyncio import threading from typing import List, Dict, Optional from dataclasses import dataclass import queue import duckdb from dotenv import load_dotenv from pathlib import Path # Import system components sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from cpu.enhanced_cpu import CPUVRAMInterface from virtual_gpu_driver.src.driver_api import VirtualGPUDriver from display_server import display_server @dataclass class BootConfig: """Boot configuration settings""" cpu_cores: int = 50 cpu_threads: int = 100 vram_size_gb: int = 32 enable_gpu: bool = True boot_timeout: int = 5 safe_mode: bool = False os_image_dir: str = "os_img" # Directory containing OS images boot_partition: str = "/dev/sda1" kernel_params: str = "" class SystemBootloader: def __init__(self): self.config = BootConfig() self.boot_queue = queue.Queue() self.cpu_interface = None self.gpu_driver = None self.boot_status = "initialized" self.error_state = None self.display_server = display_server # Initialize DuckDB connection self._init_duckdb_storage() # Start display server in a separate thread self.start_display_server() def _init_duckdb_storage(self): """Initialize DuckDB with remote storage support""" # Load environment variables load_dotenv(Path(__file__).parent / '.env') # Get DuckDB configuration self.remote_url = os.getenv('DUCKDB_REMOTE_URL') self.local_path = os.getenv('DUCKDB_LOCAL_PATH', './bootloader_db.duck') # Initialize DuckDB connection self.db = duckdb.connect(self.local_path) # Configure S3 settings if remote storage is enabled if self.remote_url and self.remote_url.startswith('s3://'): self.db.execute(""" INSTALL httpfs; LOAD httpfs; SET s3_region='{}'; SET s3_access_key_id='{}'; SET s3_secret_access_key='{}'; """.format( os.getenv('DUCKDB_S3_REGION'), os.getenv('DUCKDB_S3_ACCESS_KEY'), os.getenv('DUCKDB_S3_SECRET_KEY') )) # Initialize boot storage tables self._init_storage_tables() def start_display_server(self): """Start the display server in a separate thread""" display_thread = threading.Thread(target=lambda: asyncio.run(self.display_server.start_server())) display_thread.daemon = True display_thread.start() async def log_message(self, message: str): """Log a message to the console and display server""" print(message) await self.display_server.console_log(message) def get_available_os_images(self) -> List[Dict[str, str]]: """Get list of available OS images from the os_img directory""" os_image_dir = Path(__file__).parent / self.config.os_image_dir if not os_image_dir.exists(): os_image_dir.mkdir(parents=True) images = [] for img_file in os_image_dir.glob('*.img'): size = img_file.stat().st_size images.append({ 'name': img_file.name, 'path': str(img_file), 'size': f"{size / (1024*1024):.2f} MB" }) return images async def load_os_image(self, image_path: str) -> bool: """Load an OS image from the local filesystem""" try: img_path = Path(image_path) if not img_path.exists(): await self.log_message(f"Error: OS image not found: {image_path}") return False # Update display with OS image info size = img_path.stat().st_size await self.display_server.update_os_image( name=img_path.name, size=f"{size / (1024*1024):.2f} MB", status="loading" ) # Simulate loading progress for progress in range(0, 101, 10): await self.display_server.update_boot_progress(progress) await asyncio.sleep(0.2) # Simulate loading time await self.display_server.update_os_image( name=img_path.name, size=f"{size / (1024*1024):.2f} MB", status="loaded" ) return True except Exception as e: await self.log_message(f"Error loading OS image: {str(e)}") return False async def initialize_system(self): """Initialize system components""" try: # Initialize CPU interface self.cpu_interface = CPUVRAMInterface( cores=self.config.cpu_cores, threads=self.config.cpu_threads, vram_size=self.config.vram_size_gb ) # Update display with CPU status await self.display_server.update_cpu_status( cores=self.config.cpu_cores, usage=0, temp=45 # Initial temperature ) # Initialize GPU driver if enabled if self.config.enable_gpu: self.gpu_driver = VirtualGPUDriver() # Initialize memory status await self.display_server.update_memory_status( total=f"{self.config.vram_size_gb} GB", used="0 GB", available=f"{self.config.vram_size_gb} GB" ) return True except Exception as e: await self.log_message(f"Error initializing system: {str(e)}") return False async def boot_system(self, os_image: str = None): """Boot the system with the specified OS image""" try: # Initialize system components if not await self.initialize_system(): return False # If no OS image specified, use the first available one if os_image is None: available_images = self.get_available_os_images() if not available_images: await self.log_message("Error: No OS images found in os_img directory") return False os_image = available_images[0]['path'] # Load the OS image if not await self.load_os_image(os_image): return False await self.log_message("System boot completed successfully") return True except Exception as e: await self.log_message(f"Error during system boot: {str(e)}") return False def _init_duckdb_storage(self): """Initialize DuckDB with local storage""" self.local_path = './bootloader_db.duck' self.db = duckdb.connect(self.local_path) self._init_storage_tables() def _init_storage_tables(self): """Initialize storage tables for boot configuration and state""" # Create boot configuration table self.db.execute(""" CREATE TABLE IF NOT EXISTS boot_config ( id INTEGER PRIMARY KEY, cpu_cores INTEGER, cpu_threads INTEGER, vram_size_gb INTEGER, enable_gpu BOOLEAN, boot_timeout INTEGER, safe_mode BOOLEAN, os_image_dir VARCHAR, boot_partition VARCHAR, kernel_params VARCHAR, last_modified TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Create boot history table self.db.execute(""" CREATE TABLE IF NOT EXISTS boot_history ( id INTEGER PRIMARY KEY, boot_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, boot_status VARCHAR, error_state VARCHAR, diagnostic_results JSON ) """) def clear_screen(self): """Clear the terminal screen""" os.system('cls' if os.name == 'nt' else 'clear') def display_boot_menu(self): """Display the boot menu interface""" self.clear_screen() print("=== System Boot Menu ===") print("1. Normal Boot") print("2. Safe Mode") print("3. Configure Boot Options") print("4. Hardware Diagnostics") print("5. Exit") print("\nAutomatic boot in {} seconds...".format(self.config.boot_timeout)) def initialize_hardware(self): """Initialize core hardware components""" try: # Initialize CPU-VRAM interface self.cpu_interface = CPUVRAMInterface(vram_size_gb=self.config.vram_size_gb) # Initialize GPU driver if enabled if self.config.enable_gpu: self.gpu_driver = VirtualGPUDriver() return True except Exception as e: self.error_state = str(e) return False def run_diagnostics(self): """Run system diagnostics""" tests = [ ("CPU Interface", self._test_cpu_interface), ("Memory Allocation", self._test_memory_allocation), ("GPU Driver", self._test_gpu_driver), ] results = {} for test_name, test_func in tests: print(f"Running {test_name}...", end=" ") try: result = test_func() results[test_name] = "PASS" if result else "FAIL" print(results[test_name]) except Exception as e: results[test_name] = f"ERROR: {str(e)}" print(results[test_name]) return results def _test_cpu_interface(self): """Test CPU interface functionality""" if not self.cpu_interface: return False try: # Test memory allocation block_id = self.cpu_interface.allocate_memory(1024) self.cpu_interface.free_memory(block_id) return True except: return False def _test_memory_allocation(self): """Test memory allocation and mapping""" if not self.cpu_interface: return False try: block_id = self.cpu_interface.allocate_memory(1024) result = self.cpu_interface.map_memory(block_id, 0x1000) self.cpu_interface.free_memory(block_id) return result except: return False def _test_gpu_driver(self): """Test GPU driver initialization""" return bool(self.gpu_driver) if self.config.enable_gpu else True def load_os_image(self) -> bool: """Load operating system image into memory""" try: print(f"\nLoading OS image from {self.config.os_image_path}") # Verify OS image exists if not os.path.exists(self.config.os_image_path): raise Exception("OS image not found") # Allocate memory for OS image image_size = os.path.getsize(self.config.os_image_path) os_block_id = self.cpu_interface.allocate_memory(image_size) # Load image into memory (simulated for this example) print("Loading image into memory...") with open(self.config.os_image_path, 'rb') as f: # In a real bootloader, this would use direct memory access image_data = f.read() # Map OS image to memory self.cpu_interface.map_memory(os_block_id, 0x100000) # Map to 1MB address print("OS image loaded successfully") return True except Exception as e: print(f"Failed to load OS image: {str(e)}") return False def setup_boot_environment(self): """Prepare the boot environment""" print("\nSetting up boot environment...") # Mount boot partition (simulated) print(f"Mounting boot partition: {self.config.boot_partition}") # Parse kernel parameters if self.config.kernel_params: print(f"Kernel parameters: {self.config.kernel_params}") return True def boot_system(self): """Main boot sequence""" try: # Display boot menu self.display_boot_menu() # Initialize hardware if not self.initialize_hardware(): raise Exception(f"Hardware initialization failed: {self.error_state}") # Run basic diagnostics diag_results = self.run_diagnostics() if any(result.startswith("ERROR") for result in diag_results.values()): raise Exception("System diagnostics failed") # Setup boot environment if not self.setup_boot_environment(): raise Exception("Failed to setup boot environment") # Load OS image if not self.load_os_image(): raise Exception("Failed to load OS image") # Start system components self.boot_status = "booting" print("\nTransferring control to OS...") time.sleep(1) # Simulate boot time self.boot_status = "ready" print("System boot completed successfully!") return True except Exception as e: self.boot_status = "failed" print(f"\nBoot Error: {str(e)}") return False def save_boot_config(self): """Save current boot configuration to storage""" self.db.execute(""" INSERT INTO boot_config ( cpu_cores, cpu_threads, vram_size_gb, enable_gpu, boot_timeout, safe_mode, os_image_path, boot_partition, kernel_params ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( self.config.cpu_cores, self.config.cpu_threads, self.config.vram_size_gb, self.config.enable_gpu, self.config.boot_timeout, self.config.safe_mode, self.config.os_image_path, self.config.boot_partition, self.config.kernel_params )) # Sync with remote storage if configured if self.remote_url: self.db.execute(f"EXPORT DATABASE '{self.remote_url}' (FORMAT PARQUET)") def load_boot_config(self): """Load last boot configuration from storage""" result = self.db.execute(""" SELECT * FROM boot_config ORDER BY last_modified DESC LIMIT 1 """).fetchone() if result: self.config.cpu_cores = result[1] self.config.cpu_threads = result[2] self.config.vram_size_gb = result[3] self.config.enable_gpu = result[4] self.config.boot_timeout = result[5] self.config.safe_mode = result[6] self.config.os_image_path = result[7] self.config.boot_partition = result[8] self.config.kernel_params = result[9] def log_boot_history(self, diagnostic_results: Dict): """Log boot attempt to history""" self.db.execute(""" INSERT INTO boot_history ( boot_status, error_state, diagnostic_results ) VALUES (?, ?, ?) """, ( self.boot_status, self.error_state, json.dumps(diagnostic_results) )) # Sync with remote storage if configured if self.remote_url: self.db.execute(f"EXPORT DATABASE '{self.remote_url}' (FORMAT PARQUET)") def shutdown(self): """Clean shutdown of system components""" if self.gpu_driver: # Cleanup GPU resources self.gpu_driver = None if self.cpu_interface: # Cleanup CPU-VRAM resources self.cpu_interface = None # Save final state and close database connection try: self.save_boot_config() self.db.close() except: pass print("System shutdown complete.") if __name__ == "__main__": bootloader = SystemBootloader() try: if bootloader.boot_system(): # Keep system running until interrupted while bootloader.boot_status == "ready": time.sleep(1) except KeyboardInterrupt: print("\nInitiating shutdown...") finally: bootloader.shutdown()