Skip to content

openstb.simulator.plugin.util

Functions:

Name Description
find_config_loader

Try to find a config loader plugin.
load_config_plugins

Load configuration plugins.

find_config_loader 

find_config_loader(source)

Try to find a config loader plugin.

This is intended for use in a user interface. Given the source (e.g, the filename) of the desired configuration, it searches through the registered ConfigLoader plugins for one which says it might be able to handle it.

Note that the plugins may return false positives or false negatives when asked about their ability to handle a source (see the could_handle method of the ConfigLoader class for details). If a class is returned, there is no guarantee it will actually be able to handle a source.

Parameters:

Name Type Description Default
source str

The provided configuration source.

 required

Returns:

Name Type Description
name (str, None)

The name of the plugin to use, or None if no plugin was found.
cls (ConfigLoader, None)

The plugin to use, or None if no plugin was found.
Source code in openstb/simulator/plugin/util.py 
18
19
20
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 find_config_loader(
    source: str,
) -> tuple[str, type[abc.ConfigLoader]] | tuple[None, None]:
    """Try to find a config loader plugin.

    This is intended for use in a user interface. Given the source (e.g, the filename)
    of the desired configuration, it searches through the registered `ConfigLoader`
    plugins for one which says it might be able to handle it.

    Note that the plugins may return false positives or false negatives when asked about
    their ability to handle a source (see the `could_handle` method of the
    `ConfigLoader` class for details). If a class is returned, there is no guarantee it
    will actually be able to handle a source.

    Parameters
    ----------
    source : str
        The provided configuration source.

    Returns
    -------
    name : str, None
        The name of the plugin to use, or None if no plugin was found.
    cls : openstb.simulator.plugin.abc.ConfigLoader, None
        The plugin to use, or None if no plugin was found.

    """
    plugins = cast(
        list[tuple[str, str, type[abc.ConfigLoader]]],
        loader.registered_plugins("openstb.simulator.config_loader", load=True),
    )
    for name, _, cls in plugins:
        if cls.could_handle(source):
            return name, cls
    return None, None

load_config_plugins 

load_config_plugins(config_class, config, missing='error', extra='error', unhandled_type='error', **keys)

Load configuration plugins.

This uses a mapping of config keys to type annotations, either extracted from the configuration class or given as keyword arguments, to tell it what type of plugin is expected for each key. It then checks for a function registered with the openstb.simulator.plugin.loader.register_loader decorator to convert the entry in the configuration dictionary to a plugin instance.

Parameters:

Name Type Description Default
config_class type

The class (not an instance of the class) specifying the configuration structure. This would typically be a subclass of typing.TypedDict, but can be any class which has type-annotated properties corresponding to the configuration entries.

 required
config mapping

The current configuration. This will be modified in-place.

 required
missing (error, warn, ignore)

How to handle missing, extra (unknown) entries in config, and unhandled plugin types (no registered loader): raise an error, issue a warning or ignore them.

 "error"
extra (error, warn, ignore)

How to handle missing, extra (unknown) entries in config, and unhandled plugin types (no registered loader): raise an error, issue a warning or ignore them.

 "error"
unhandled_type (error, warn, ignore)

How to handle missing, extra (unknown) entries in config, and unhandled plugin types (no registered loader): raise an error, issue a warning or ignore them.

 "error"
**keys

Keyword arguments can be given to map the keys in config to the expected plugin type, e.g., load_config_plugins(..., signal=abc.Signal, trajectory=abc.Trajectory). If no keyword arguments are given, the type annotations of config_class are used as the keys.

 {}
Source code in openstb/simulator/plugin/util.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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def load_config_plugins(
    config_class: type[SimulationConfig] | None,
    config: MutableMapping[str, Any],
    missing: Literal["error", "warn", "ignore"] = "error",
    extra: Literal["error", "warn", "ignore"] = "error",
    unhandled_type: Literal["error", "warn", "ignore"] = "error",
    **keys,
):
    """Load configuration plugins.

    This uses a mapping of config keys to type annotations, either extracted from the
    configuration class or given as keyword arguments, to tell it what type of plugin
    is expected for each key. It then checks for a function registered with the
    `openstb.simulator.plugin.loader.register_loader` decorator to convert the entry in
    the configuration dictionary to a plugin instance.

    Parameters
    ----------
    config_class : type
        The class (not an instance of the class) specifying the configuration structure.
        This would typically be a subclass of `typing.TypedDict`, but can be any class
        which has type-annotated properties corresponding to the configuration entries.
    config : mapping
        The current configuration. This will be modified in-place.
    missing, extra, unhandled_type : {"error", "warn", "ignore"}
        How to handle missing, extra (unknown) entries in ``config``, and unhandled
        plugin types (no registered loader): raise an error, issue a warning or ignore
        them.
    **keys
        Keyword arguments can be given to map the keys in ``config`` to the expected
        plugin type, e.g., `load_config_plugins(..., signal=abc.Signal,
        trajectory=abc.Trajectory)`. If no keyword arguments are given, the type
        annotations of ``config_class`` are used as the keys.


    """
    # No keys: use the annotations of the configuration class.
    if not keys:
        if config_class is not None:
            keys = inspect.get_annotations(config_class)

    if not keys:
        raise ValueError(_("could not determine the configuration keys to process"))

    # Keys specified in the input. We will use this to track extra keys.
    config_keys = set(config.keys())

    for key, annotation in keys.items():
        required = True
        is_list = False

        # Get the unsubscripted type.
        origin = get_origin(annotation)

        # Entries of a TypedDict can be marked as not required. In this case, the first
        # subscripted argument is the type so pull that out, and then get its
        # unsubscripted version (we may have a NotRequired[list[...]] for example).
        if origin == NotRequired:
            required = False
            annotation = get_args(annotation)[0]
            origin = get_origin(annotation)

        # Expect a list. The first unsubscripted argument is the type within the list.
        if origin in (list, List):
            is_list = True
            annotation = get_args(annotation)[0]

        # Try to find a function to load the plugin.
        loader_func = loader.registered_loaders.get(annotation, None)
        if loader_func is None:
            if unhandled_type == "ignore":
                continue

            msg = _(
                "no loader available for plugin type {type} requested by section {key}"
            ).format(type=annotation.__name__, key=key)
            if unhandled_type == "warn":
                warnings.warn(msg, stacklevel=2)
                continue

            raise RuntimeError(msg)

        # No configuration entry for this key.
        if key not in config_keys:
            if not required or missing == "ignore":
                continue

            msg = _("section {key} missing from configuration").format(key=key)
            if missing == "warn":
                warnings.warn(msg, stacklevel=2)
                continue

            raise RuntimeError(msg)

        # A list is expected.
        if is_list:
            if not isinstance(config[key], Sequence):
                raise RuntimeError(
                    _("section {key} expects a list of plugins").format(key=key)
                )
            config[key] = [loader_func(c) for c in config[key]]

        # Single plugin.
        else:
            config[key] = loader_func(config[key])

        # Have handled this entry from the original configuration.
        config_keys.remove(key)

    # Configuration specified keys we don't know about.
    if config_keys and extra != "ignore":
        msg = _n(
            "extra section {keys} in configuration",
            "extra sections in configuration: {keys}",
            len(config_keys),
        ).format(keys=", ".join(config_keys))
        if extra == "error":
            raise RuntimeError(msg)
        warnings.warn(msg, stacklevel=2)