core_daemon/module_manager.py

253 lines
9.5 KiB
Python
Raw Permalink Normal View History

"""
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