""" module_manager.py This module provides the ModuleManager class which allows for the management of dynamically loadable modules. The ModuleManager supports loading, unloading, and execution of modules, as well as handling module-specific commands and hooks. Classes: ModuleManager: Manages loading, unloading, and executing modules and their commands. Variables: hook_manager (HookManager): An instance of HookManager for managing hooks. """ import os import sys import importlib import logging import hashlib from hook_manager import HookManager hook_manager = HookManager() class ModuleManager: """ A manager for handling dynamically loadable modules, including loading, unloading, and executing module-specific commands and hooks. Attributes: module_dirs (list): A list of directories to search for modules. loaded_modules (dict): A dictionary of loaded modules. extra_commands (dict): A dictionary of additional commands provided by modules. module_hashes (dict): A dictionary storing file paths and their hashes. hook_manager (HookManager): An instance of HookManager for managing hooks. """ def __init__(self, module_dirs): """ Initializes a new instance of the ModuleManager class. Args: module_dirs (list): A list of directories to search for modules. """ self.module_dirs = module_dirs self.loaded_modules = {} self.extra_commands = {} self.module_hashes = {} # store file paths and hashes self._update_sys_path() self.hook_manager = hook_manager # Use the global hook_manager! def _update_sys_path(self): """ Updates the system path to include the directories in module_dirs. """ for dir in self.module_dirs: full_path = os.path.abspath(dir) if full_path not in sys.path: sys.path.append(full_path) logging.info(f"Added {full_path} to sys.path") def add_module_dir(self, new_dir): """ Adds a new directory to the list of module directories and updates the system path. Args: new_dir (str): The new directory to add. Returns: tuple: A tuple containing a boolean indicating success and a message. """ try: full_path = os.path.abspath(new_dir) if full_path not in self.module_dirs: self.module_dirs.append(full_path) self._update_sys_path() return True, f"Added module directory: {full_path}" return False, f"Module directory already exists: {full_path}" except Exception as e: logging.error(f"Error adding module directory: {str(e)}") return False, f"Error adding module directory: {str(e)}" def calculate_file_hash(self, file_path): """ Calculates the SHA-256 hash of a file. Args: file_path (str): The path to the file. Returns: str: The SHA-256 hash of the file. """ sha256_hash = hashlib.sha256() try: with open(file_path, "rb") as f: for byte_block in iter(lambda: f.read(4096), b""): sha256_hash.update(byte_block) return sha256_hash.hexdigest() except FileNotFoundError: logging.error(f"File not found: {file_path}") return None except Exception as e: logging.error(f"Error calculating hash for file '{file_path}': {str(e)}") return None def load_module(self, module_name): """ Loads a module by name, initializing it and registering its commands and hooks. Args: module_name (str): The name of the module to load. Returns: tuple: A tuple containing a boolean indicating success and a message. """ self.hook_manager.execute_hook('pre_module_load', module_name) for dir in self.module_dirs: try: module_path = os.path.join(dir, f"{module_name}.py") if not os.path.exists(module_path): continue module = importlib.import_module(f'{os.path.basename(dir)}.{module_name}') self.loaded_modules[module_name] = module # Calculate and store the hash of the module file module_hash = self.calculate_file_hash(module_path) if module_hash: self.module_hashes[module_path] = module_hash self._initialize_module(module) self.hook_manager.execute_hook('post_module_load', module_name) return True, f"Module '{module_name}' loaded and initialized successfully from {dir}." except ImportError as e: logging.error(f"ImportError for module '{module_name}' from {dir}: {str(e)}") continue except Exception as e: logging.error(f"Error loading module '{module_name}' from {dir}: {str(e)}") return False, f"Error loading module '{module_name}' from {dir}: {str(e)}" return False, f"Error: Unable to load module '{module_name}' from any of the module directories." def _initialize_module(self, module): """ Initializes a loaded module by calling its initialize method, registering commands, and registering hooks. Args: module: The loaded module. """ if hasattr(module, 'initialize'): module.initialize() if hasattr(module, 'get_commands'): new_commands = module.get_commands() self.extra_commands.update(new_commands) if hasattr(module, 'register_hooks'): module.register_hooks(self.hook_manager) def unload_module(self, module_name): """ Unloads a module by name, shutting it down and removing its commands and hooks. Args: module_name (str): The name of the module to unload. Returns: tuple: A tuple containing a boolean indicating success and a message. """ self.hook_manager.execute_hook('pre_module_unload', module_name) if module_name in self.loaded_modules: try: module = self.loaded_modules[module_name] self._shutdown_module(module) self.hook_manager.execute_hook('post_module_unload', module_name) # Remove the module's hash from our tracking module_path = module.__file__ self.module_hashes.pop(module_path, None) del self.loaded_modules[module_name] logging.info(f"Module '{module_name}' unloaded successfully.") return True, f"Module '{module_name}' unloaded and shut down." except Exception as e: logging.error(f"Error unloading module '{module_name}': {str(e)}") return False, f"Error unloading module '{module_name}': {str(e)}" return False, f"Module '{module_name}' is not loaded." def _shutdown_module(self, module): """ Shuts down a loaded module by calling its shutdown method and removing its commands and hooks. Args: module: The loaded module. """ if hasattr(module, 'shutdown'): module.shutdown() if hasattr(module, 'get_commands'): commands_to_remove = module.get_commands().keys() for cmd in commands_to_remove: self.extra_commands.pop(cmd, None) if hasattr(module, 'unregister_hooks'): module.unregister_hooks(self.hook_manager) def list_modules(self): """ Lists all currently loaded modules. Returns: list: A list of names of loaded modules. """ return list(self.loaded_modules.keys()) def list_commands(self): """ Lists all commands provided by the loaded modules. Returns: list: A list of command names. """ return list(self.extra_commands.keys()) def execute_command(self, command, args): """ Executes a command provided by a loaded module. Args: command (str): The name of the command to execute. args (list): The arguments to pass to the command. Returns: tuple: A tuple containing a boolean indicating success and the command result or an error message. """ if command in self.extra_commands: try: result = self.extra_commands[command](args) return True, result except Exception as e: logging.error(f"Error executing command '{command}': {str(e)}") return False, f"Error executing command '{command}': {str(e)}" return False, f"Command '{command}' not found" def check_modules_modified(self): """ Checks if any loaded modules have been modified on disk by comparing file hashes. Returns: list: A list of names of modified modules. """ modified_modules = [] for module_path, stored_hash in self.module_hashes.items(): current_hash = self.calculate_file_hash(module_path) if current_hash and current_hash != stored_hash: module_name = os.path.splitext(os.path.basename(module_path))[0] modified_modules.append(module_name) return modified_modules