Skip to content

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 (Processor) - DataLoaders - OutputWriters

Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class PluginLoadedRegistry:
    """
    Singleton registry that provides lazy access to all registered DQM components.

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

    _metrics_registry: dict[str, type[Processor]] | None = None
    _features_registry: dict[str, type[Processor]] | None = None
    _gap_registry: dict[str, type[Processor]] | 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[Processor]]:
        """Return the registry of available metric processors.

        Returns:
            A dictionary mapping metric processor names to their classes.
        """
        if not cls._metrics_registry:
            from dqm_ml_core.api.metrics_processor import MetricsProcessor

            cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", MetricsProcessor)

        return cls._metrics_registry

    @classmethod
    def get_features_registry(cls) -> dict[str, type[Processor]]:
        """Return the registry of available feature extraction processors.

        Returns:
            A dictionary mapping feature processor names to their classes.
        """
        if not cls._features_registry:
            from dqm_ml_core.api.features_processor import FeaturesProcessor

            cls._features_registry = load_registered_plugins("dqm_ml.features", FeaturesProcessor)

        return cls._features_registry

    @classmethod
    def get_gap_registry(cls) -> dict[str, type[Processor]]:
        """Return the registry of available gap processors.

        Returns:
            A dictionary mapping gap processor names to their classes.
        """
        if not cls._gap_registry:
            from dqm_ml_core.api.gap_processor import GapProcessor

            cls._gap_registry = load_registered_plugins("dqm_ml.gap", GapProcessor)

        return cls._gap_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 
113
114
115
116
117
118
119
120
121
122
@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_features_registry() -> dict[str, type[Processor]] classmethod

Return the registry of available feature extraction processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping feature processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def get_features_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available feature extraction processors.

    Returns:
        A dictionary mapping feature processor names to their classes.
    """
    if not cls._features_registry:
        from dqm_ml_core.api.features_processor import FeaturesProcessor

        cls._features_registry = load_registered_plugins("dqm_ml.features", FeaturesProcessor)

    return cls._features_registry

get_gap_registry() -> dict[str, type[Processor]] classmethod

Return the registry of available gap processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping gap processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
@classmethod
def get_gap_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available gap processors.

    Returns:
        A dictionary mapping gap processor names to their classes.
    """
    if not cls._gap_registry:
        from dqm_ml_core.api.gap_processor import GapProcessor

        cls._gap_registry = load_registered_plugins("dqm_ml.gap", GapProcessor)

    return cls._gap_registry

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

Return the registry of available metric processors.

Returns:

Type Description
dict[str, type[Processor]]

A dictionary mapping metric processor names to their classes.
Source code in packages/dqm-ml-core/src/dqm_ml_core/utils/registry.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
@classmethod
def get_metrics_registry(cls) -> dict[str, type[Processor]]:
    """Return the registry of available metric processors.

    Returns:
        A dictionary mapping metric processor names to their classes.
    """
    if not cls._metrics_registry:
        from dqm_ml_core.api.metrics_processor import MetricsProcessor

        cls._metrics_registry = load_registered_plugins("dqm_ml.metrics", MetricsProcessor)

    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 
124
125
126
127
128
129
130
131
132
133
134
@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 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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