Source code for genericroboticarm.robo_APIs.dynamic_loader

"""
Utility module for loading RobotInterface implementations.
"""

import importlib
import inspect
import logging
import pkgutil
from types import ModuleType
from .robo_interface import RoboInterface




def collect_robo_implementations(package: str = __package__) -> set[type[RoboInterface]]:
    """
    Recursively find all subclasses of RoboInterface in a package.

    Args:
        package: Dotted path of the package

    Returns:
        list of RoboInterface subclasses
    """
    classes: set[type[RoboInterface]] = set()

    pkg = importlib.import_module(package)

    def walk_modules(module: ModuleType):
        for _, mod_name, is_pkg in pkgutil.iter_modules(module.__path__):
            full_name = f"{module.__name__}.{mod_name}"
            try:
                mod = importlib.import_module(full_name)

                for _, obj in inspect.getmembers(mod, inspect.isclass):
                    if (
                        issubclass(obj, RoboInterface)
                        and obj is not RoboInterface
                        and not inspect.isabstract(obj)
                    ):
                        classes.add(obj)
                if is_pkg:
                    walk_modules(mod)
            except ModuleNotFoundError as error:
                logging.warning(f"Could not import {module.__name__} because of missing module: {error}")


    walk_modules(pkg)
    return classes





def get_arm_impl(impl_str: str) -> RoboInterface | None:
    """Dynamically imports the specified robotic arm implementation class.

    Args:
        impl_str (str): The robotic arm implementation class to use, formatted as <module>:<class>
    """
    try:
        module_name, class_name = impl_str.split(":")
        module = importlib.import_module(module_name)
        implementation = getattr(module, class_name)
        if not issubclass(implementation, RoboInterface):
            logging.error(f"{implementation} is no subclass of RoboInterface")
        logging.info(f"Successfully imported {class_name} from {module_name}")
        return implementation

    except (ValueError, ImportError, AttributeError) as e:
        logging.error(f"Failed to import {impl_str}: {e}")
        return None