# radian/components/component_manager.py import sys import importlib.util import inspect from pathlib import Path from typing import Dict, Optional, Type import yaml from radian.utils import logger from radian.components.base_component import BaseComponent log = logger.get_logger(__name__) # The expected name of the file containing the adapter class within a component's project. ADAPTER_FILENAME = "adapter.py" # The expected name of the adapter class. This can be made more flexible later if needed. ADAPTER_CLASS_NAME = "RadianAdapter" class ComponentManager: """ Discovers, loads, and manages all RADIAN components (plugins). It reads a configuration file to find components and dynamically loads them, making them available to the rest of the application. """ def __init__(self, config_path: Path): self.config_path = config_path self.components: Dict[str, BaseComponent] = {} log.info(f"ComponentManager initialized with config path: {config_path}") def load_components(self): """ Loads all components specified in the configuration file. This is the main entry point for the manager. """ log.info("Starting component discovery and loading...") try: with open(self.config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) except FileNotFoundError: log.error(f"Component configuration file not found at: {self.config_path}") return except yaml.YAMLError as e: log.error(f"Error parsing component YAML file: {e}") return if not config or 'components' not in config: log.warning("Component configuration file is empty or missing 'components' key.") return for name, comp_config in config['components'].items(): if 'path' in comp_config: log.info(f"Found component '{name}' configured with a local path.") self._load_component_from_path(name, Path(comp_config['path'])) else: log.warning(f"Component '{name}' is missing a 'path' key. Skipping.") log.info(f"Component loading complete. {len(self.components)} components loaded.") def _load_component_from_path(self, name: str, component_path: Path): """Dynamically loads a single component from a local directory path.""" adapter_file_path = component_path / ADAPTER_FILENAME if not adapter_file_path.is_file(): log.error(f"Component '{name}': Adapter file '{ADAPTER_FILENAME}' not found in {component_path}. Skipping.") return try: # Create a unique module name to avoid conflicts module_name = f"radian.external_components.{name}" # Create a module spec from the file path spec = importlib.util.spec_from_file_location(module_name, adapter_file_path) if not spec or not spec.loader: log.error(f"Component '{name}': Could not create module spec from {adapter_file_path}.") return # Create a new module object module = importlib.util.module_from_spec(spec) # Add the component's root directory to sys.path TEMPORARILY # so that its internal imports (e.g., from . import logic) work. sys.path.insert(0, str(component_path)) # Execute the module's code spec.loader.exec_module(module) # IMPORTANT: Remove the path from sys.path to avoid side effects sys.path.pop(0) # Find the adapter class within the loaded module AdapterClass = self._find_adapter_class(module) if AdapterClass: instance = AdapterClass() self.components[name] = instance log.info(f"Successfully loaded and instantiated component: '{name}'") else: log.error(f"Component '{name}': No valid adapter class inheriting from BaseComponent found in {adapter_file_path}.") except Exception as e: # Clean up path if an error occurred during exec_module if str(component_path) in sys.path: sys.path.remove(str(component_path)) log.error(f"Failed to load component '{name}' from {component_path}. Error: {e}", exc_info=True) def _find_adapter_class(self, module) -> Optional[Type[BaseComponent]]: """Inspects a module to find a class that inherits from BaseComponent.""" for name, obj in inspect.getmembers(module, inspect.isclass): # Check if it's a class defined in this module (not imported) # and if it's a subclass of BaseComponent (but not BaseComponent itself) if obj is not BaseComponent and issubclass(obj, BaseComponent): return obj return None def get_component(self, name: str) -> Optional[BaseComponent]: """Returns the loaded instance of a component by its name.""" return self.components.get(name) def get_all_components(self) -> Dict[str, BaseComponent]: """Returns a dictionary of all loaded components.""" return self.components