Skip to content

openstb.simulator.plugin.loader

Functions:

Name Description
load_plugin

Load a plugin and create an instance of it.
load_plugin_class

Load a plugin class.
register_loader

Decorator to register a plugin loading function.
registered_plugins

List registered plugins.

load_plugin 

load_plugin(group, plugin_spec)

Load a plugin and create an instance of it.

In general, you should call one of the more specific functions in this module to load the particular type of plugin you want.

Parameters:

Name Type Description Default
group str

The name of the entry point group that the plugin belongs to, e.g., "openstb.simulator.trajectory".

 required
plugin_spec (dict, Plugin)

If a dictionary, this specifies the name and parameters of the plugin to load. Otherwise, it is assumed to be an instance of a compatible class and is returned unchanged.

 required

Returns:

Type Description
Plugin
Source code in openstb/simulator/plugin/loader.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def load_plugin(group: str, plugin_spec: PluginOrSpec[T_Plugin]) -> T_Plugin:
    """Load a plugin and create an instance of it.

    In general, you should call one of the more specific functions in this module to
    load the particular type of plugin you want.

    Parameters
    ----------
    group : str
        The name of the entry point group that the plugin belongs to, e.g.,
        "openstb.simulator.trajectory".
    plugin_spec : dict, Plugin
        If a dictionary, this specifies the name and parameters of the plugin to
        load. Otherwise, it is assumed to be an instance of a compatible class and is
        returned unchanged.

    Returns
    -------
    abc.Plugin

    """
    # Note: we cannot check isinstance(..., Mapping) here as plugins may implement the
    # Mapping interface.
    if isinstance(plugin_spec, dict):
        cls: type[T_Plugin] = load_plugin_class(group, plugin_spec["name"])

        # Start with parameters from the spec.
        params: dict[str, Any] = {}
        if "parameters" in plugin_spec:
            params.update(plugin_spec["parameters"])

        # If the plugin initialiser accepts a spec_source parameter, add that.
        cls_params = inspect.signature(cls).parameters
        if "spec_source" in cls_params and "spec_source" in plugin_spec:
            params["spec_source"] = plugin_spec["spec_source"]

        return cls(**params)

    return plugin_spec

load_plugin_class 

load_plugin_class(group, name)

Load a plugin class.

Three formats are supported for the name of the plugin:

  1. The name of an installed plugin registered under the entry point group.

  2. A reference to an attribute name and the importable module it belongs to. The name "FromDisk:my_plugins.trajectories" effectively corresponds to this function being from my_plugins.trajectories import FromDisk; return FromDisk.

  3. The name of an attribute in a Python file and the path of the file. The name "FromDisk:/path/to/my_plugins.py" results in the function dynamically importing the /path/to/my_plugins.py file and returning the FromDisk attribute from it. Note that relative paths will be resolved relative to the current working directory.

Parameters:

Name Type Description Default
group str

The name of the entry point group that the plugin belongs to, e.g., "openstb.simulator.trajectory".

 required
name str

The user-specified name of the plugin.

 required

Returns:

Name Type Description
plugin class
Source code in openstb/simulator/plugin/loader.py 
 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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def load_plugin_class(group: str, name: str) -> type[T_Plugin]:
    """Load a plugin class.

    Three formats are supported for the name of the plugin:

    1. The name of an installed plugin registered under the entry point group.

    2. A reference to an attribute name and the importable module it belongs to. The
       name "FromDisk:my_plugins.trajectories" effectively corresponds to this function
       being `from my_plugins.trajectories import FromDisk; return FromDisk`.

    3. The name of an attribute in a Python file and the path of the file. The name
       "FromDisk:/path/to/my_plugins.py" results in the function dynamically importing
       the `/path/to/my_plugins.py` file and returning the `FromDisk` attribute from it.
       Note that relative paths will be resolved relative to the current working
       directory.

    Parameters
    ----------
    group : str
        The name of the entry point group that the plugin belongs to, e.g.,
        "openstb.simulator.trajectory".
    name : str
        The user-specified name of the plugin.

    Returns
    -------
    plugin : class

    """
    classname, sep, modname_or_path = name.partition(":")

    # Separator not present: this is a registered plugin.
    if not sep:
        eps = importlib.metadata.entry_points().select(group=group, name=name)
        if not eps:
            raise ValueError(
                _("no {group} plugin named '{name}' is installed").format(
                    group=group, name=name
                )
            )
        if len(eps) > 1:
            raise ValueError(
                _("multiple {group} plugins named '{name}' are installed").format(
                    group=group, name=name
                )
            )
        return eps[name].load()

    # Not a registered plugin; does it refer to a file?
    path = Path(modname_or_path)
    if path.is_file():
        # Generate a unique module name. It is possible that two files at different
        # locations with the same stem could be requested; using just the stem could
        # result in the cached version of the first being used for the second.
        md5 = hashlib.md5(str(path.resolve()).encode("utf-8")).hexdigest()
        modname = f"openstb.simulator.dynamic_plugins.{md5}.{path.stem}"

        # Generate an importlib spec for this module.
        spec = importlib.util.spec_from_file_location(modname, path)
        if spec is None or spec.loader is None:
            raise ValueError(
                _(
                    "could not create import specification for plugin file {path}"
                ).format(path=path)
            )

        # Create and load the module.
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)

        # If the attribute is missing, raise a better error as the default will include
        # the hashed module name we generated.
        try:
            return getattr(mod, classname)
        except AttributeError:
            raise AttributeError(
                _("file {path} has no attribute {classname}").format(
                    path=path, classname=classname
                )
            )

    # Assume it is an installed module.
    mod = importlib.import_module(modname_or_path)
    return getattr(mod, classname)

register_loader 

register_loader(__loader: F_PluginLoader[T_Plugin]) -> F_PluginLoader[T_Plugin]

register_loader(*, docstring: bool | None = None, name: str | None = None) -> Callable[[F_PluginLoader[T_Plugin]], F_PluginLoader[T_Plugin]]

register_loader(__loader=None, *, docstring=None, name=None)

Decorator to register a plugin loading function.

This should be applied to a function which takes a openstb.simulator.types.PluginOrSpec parameter and returns a plugin instance.

Parameters:

Name Type Description Default
docstring Boolean

Whether to set a docstring for the class. If None (the default), then a docstring will only be set if one does not already exist.

 None
name str

The name of the plugin type to use in the docstring. If not specified, the class name will be used.

 None
Source code in openstb/simulator/plugin/loader.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def register_loader(
    __loader: F_PluginLoader[T_Plugin] | None = None,
    *,
    docstring: bool | None = None,
    name: str | None = None,
):
    """Decorator to register a plugin loading function.

    This should be applied to a function which takes a
    `openstb.simulator.types.PluginOrSpec` parameter and returns a plugin instance.

    Parameters
    ----------
    docstring : Boolean, optional
        Whether to set a docstring for the class. If None (the default), then a
        docstring will only be set if one does not already exist.
    name : str, optional
        The name of the plugin type to use in the docstring. If not specified, the
        class name will be used.

    """

    def registration(loader: F_PluginLoader[T_Plugin]) -> F_PluginLoader[T_Plugin]:
        # Use the return annotation to do the actual registration.
        cls = inspect.signature(loader).return_annotation
        registered_loaders[cls] = loader

        # Check if we need to set a doctring.
        if docstring is None:
            set_docstring = loader.__doc__ is None or len(loader.__doc__.strip()) == 0
        else:
            set_docstring = docstring
        if set_docstring:
            docname = name or cls.__name__
            loader.__doc__ = f"""Load a {docname} plugin.

Parameters
----------
plugin_spec : dict, {cls.__name__}
    If a dictionary, this specifies the name and parameters of the {docname} to
    load. Otherwise, it is assumed to be an instance of a compatible class and is
    returned unchanged.

Returns
-------
{cls.__name__}"""

        return loader

    # Directly used.
    if __loader is not None:
        return registration(__loader)

    # Used with parameters.
    return registration

registered_plugins 

registered_plugins(group: str, load: Literal[True]) -> list[tuple[str, str, type[Plugin]]]

registered_plugins(group: str, load: Literal[False]) -> list[tuple[str, str]]

registered_plugins(group: str, load: bool) -> list[tuple[str, str]] | list[tuple[str, str, type[Plugin]]]

registered_plugins(group, load=False)

List registered plugins.

Parameters:

Name Type Description Default
group str

The name of the entry point group to list plugins for, e.g., "openstb.simulator.trajectory".

 required
load Boolean

If True, include the loaded plugins in the return. If False, only report the name and source of each plugin.

 False

Returns:

Name Type Description
installed list

A list of (name, src) tuples where name is the name the plugin is registered as and src is the reference to the module and class implementing the plugin. If load was True, a third entry cls will be added to each tuple containing the loaded plugin class.
Source code in openstb/simulator/plugin/loader.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def registered_plugins(
    group: str, load: bool = False
) -> list[tuple[str, str]] | list[tuple[str, str, type[abc.Plugin]]]:
    """List registered plugins.

    Parameters
    ----------
    group : str
        The name of the entry point group to list plugins for, e.g.,
        "openstb.simulator.trajectory".
    load : Boolean
        If True, include the loaded plugins in the return. If False, only report the
        name and source of each plugin.

    Returns
    -------
    installed : list
        A list of (name, src) tuples where name is the name the plugin is registered as
        and src is the reference to the module and class implementing the plugin. If
        ``load`` was True, a third entry ``cls`` will be added to each tuple containing
        the loaded plugin class.

    """
    if load:
        return [
            (ep.name, ep.value, ep.load())
            for ep in importlib.metadata.entry_points(group=group)
        ]
    return [(ep.name, ep.value) for ep in importlib.metadata.entry_points(group=group)]