dqm_ml_core.utils.registry

Plugin registry for dynamically loading DQM components.

This module contains the PluginLoadedRegistry class and load_registered_plugins function for discovering and loading metric processors, data loaders, and output writers via Python entry points.

logger = logging.getLogger(__name__) module-attribute

PluginLoadedRegistry

Singleton registry that provides lazy access to all registered DQM components.

Components include: - Metrics (DatametricProcessor) - DataLoaders - OutputWriters

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py

class PluginLoadedRegistry:
    """
    Singleton registry that provides lazy access to all registered DQM components.

    Components include:
    - Metrics (DatametricProcessor)
    - DataLoaders
    - OutputWriters
    """

    _metrics_registry: dict[str, type[DatametricProcessor]] | None = None
    _dataloaders_registry: dict[str, Any] | None = None
    _outputwriter_registry: dict[str, Any] | None = None

    @classmethod
    def get_metrics_registry(cls) -> dict[str, type[DatametricProcessor]]:
        """Return the registry of available metric processors.

        Returns:
            A dictionary mapping metric processor names to their classes.
        """
        if not cls._metrics_registry:
            cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", DatametricProcessor)

        return cls._metrics_registry

    @classmethod
    def get_dataloaders_registry(cls) -> dict[str, Any]:
        """Return the registry of available data loaders.

        Returns:
            A dictionary mapping data loader names to their classes.
        """
        if not cls._dataloaders_registry:
            cls._dataloaders_registry = load_registered_plugins("dqm_ml.dataloaders", None)  # TODO add base class
        return cls._dataloaders_registry

    @classmethod
    def get_outputwriter_registry(cls) -> dict[str, Any]:
        """Return the registry of available output writers.

        Returns:
            A dictionary mapping output writer names to their classes.
        """
        if not cls._outputwriter_registry:
            cls._outputwriter_registry = load_registered_plugins("dqm_ml.outputwriter", None)  # TODO add base class

        return cls._outputwriter_registry

get_dataloaders_registry() -> dict[str, Any] classmethod

Return the registry of available data loaders.

Returns:

Type	Description
dict[str, Any]	A dictionary mapping data loader names to their classes.

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py

@classmethod
def get_dataloaders_registry(cls) -> dict[str, Any]:
    """Return the registry of available data loaders.

    Returns:
        A dictionary mapping data loader names to their classes.
    """
    if not cls._dataloaders_registry:
        cls._dataloaders_registry = load_registered_plugins("dqm_ml.dataloaders", None)  # TODO add base class
    return cls._dataloaders_registry

get_metrics_registry() -> dict[str, type[DatametricProcessor]] classmethod

Return the registry of available metric processors.

Returns:

Type	Description
dict[str, type[DatametricProcessor]]	A dictionary mapping metric processor names to their classes.

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py

@classmethod
def get_metrics_registry(cls) -> dict[str, type[DatametricProcessor]]:
    """Return the registry of available metric processors.

    Returns:
        A dictionary mapping metric processor names to their classes.
    """
    if not cls._metrics_registry:
        cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", DatametricProcessor)

    return cls._metrics_registry

get_outputwriter_registry() -> dict[str, Any] classmethod

Return the registry of available output writers.

Returns:

Type	Description
dict[str, Any]	A dictionary mapping output writer names to their classes.

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py

@classmethod
def get_outputwriter_registry(cls) -> dict[str, Any]:
    """Return the registry of available output writers.

    Returns:
        A dictionary mapping output writer names to their classes.
    """
    if not cls._outputwriter_registry:
        cls._outputwriter_registry = load_registered_plugins("dqm_ml.outputwriter", None)  # TODO add base class

    return cls._outputwriter_registry

load_registered_plugins(plugin_group: str, base_class: Any, base_name: str = 'default') -> dict[str, Any]

Discover and load plugins registered via Python entry points.

Parameters:

Name	Type	Description	Default
plugin_group	str	The entry point group name (e.g., 'dqm_ml.metrics').	required
base_class	Any	Optional base class to verify plugin type safety.	required
base_name	str	Name of the base class to ignore during discovery.	'default'

Returns:

Type	Description
dict[str, Any]	A dictionary mapping plugin names to their loaded classes.

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py

def load_registered_plugins(plugin_group: str, base_class: Any, base_name: str = "default") -> dict[str, Any]:
    """Discover and load plugins registered via Python entry points.

    Args:
        plugin_group: The entry point group name (e.g., 'dqm_ml.metrics').
        base_class: Optional base class to verify plugin type safety.
        base_name: Name of the base class to ignore during discovery.

    Returns:
        A dictionary mapping plugin names to their loaded classes.
    """
    try:
        # python 3.10+
        plugin_entry_points: EntryPoints = entry_points(group=plugin_group)
    except TypeError:
        # Old version for older python version
        logger.warning(f"Old python version not supported: {sys.version_info}")
        return {}

    registry = {}
    for v in plugin_entry_points:
        # Filter base class registry (not callable)
        if v.name != base_name:
            obj = v.load()
            if base_class is None or issubclass(obj, base_class):
                logger.debug(f"Referencing {plugin_group} - {v.name} class {obj} from {base_class}")
                registry[v.name] = obj
            else:
                logger.error(f"Entry point {plugin_group} - {v.name} class {obj} not derived from {base_class} ignored")

    # return a dict to class builder registry
    return registry