Skip to content

OTEAPI DLite Plugin Strategies

This page provides documentation for the oteapi_dlite.strategies submodule, where all the OTEAPI DLite Plugin strategies are located.

These strategies will be available when setting up a server in an environment with oteapi-dlite installed.

convert

Generic function strategy that converts zero or more input instances to zero or more new output instances.

DLiteConvertConfig

Bases: FunctionConfig

DLite convert strategy resource config.

Source code in oteapi_dlite/strategies/convert.py 
128
129
130
131
132
133
134
class DLiteConvertConfig(FunctionConfig):
    """DLite convert strategy resource config."""

    configuration: Annotated[
        DLiteConvertStrategyConfig,
        Field(description="DLite convert strategy-specific configuration."),
    ]

configuration: Annotated[DLiteConvertStrategyConfig, Field(description='DLite convert strategy-specific configuration.')] instance-attribute

DLiteConvertInputConfig

Bases: AttrDict

Configuration for input instance to generic DLite converter.

At least one of label or datamodel should be given.

Source code in oteapi_dlite/strategies/convert.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class DLiteConvertInputConfig(AttrDict):
    """Configuration for input instance to generic DLite converter.

    At least one of `label` or `datamodel` should be given.
    """

    label: Annotated[
        Optional[str],
        Field(
            description="Label of the instance.",
        ),
    ] = None
    datamodel: Annotated[
        Optional[str],
        Field(
            description="URI of data model.",
        ),
    ] = None
    property_mappings: Annotated[
        bool,
        Field(
            description="Whether to infer instance from property mappings.",
        ),
    ] = False

datamodel: Annotated[Optional[str], Field(description='URI of data model.')] = None class-attribute instance-attribute

label: Annotated[Optional[str], Field(description='Label of the instance.')] = None class-attribute instance-attribute

property_mappings: Annotated[bool, Field(description='Whether to infer instance from property mappings.')] = False class-attribute instance-attribute

DLiteConvertOutputConfig

Bases: AttrDict

Configuration for output instance to generic DLite converter.

Source code in oteapi_dlite/strategies/convert.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class DLiteConvertOutputConfig(AttrDict):
    """Configuration for output instance to generic DLite converter."""

    label: Annotated[
        Optional[str],
        Field(
            description="Label to use when storing the instance.",
        ),
    ] = None
    datamodel: Annotated[
        Optional[str],
        Field(
            description="URI of data model.  Used for documentation.",
        ),
    ] = None

datamodel: Annotated[Optional[str], Field(description='URI of data model. Used for documentation.')] = None class-attribute instance-attribute

label: Annotated[Optional[str], Field(description='Label to use when storing the instance.')] = None class-attribute instance-attribute

DLiteConvertStrategy

Generic DLite convert strategy for converting zero or more input instances to zero or more output instances.

Registers strategies:

  • ("functionType", "application/vnd.dlite-convert")
Source code in oteapi_dlite/strategies/convert.py 
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
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
@dataclass
class DLiteConvertStrategy:
    """Generic DLite convert strategy for converting zero or more input
    instances to zero or more output instances.

    **Registers strategies**:

    - `("functionType", "application/vnd.dlite-convert")`

    """

    convert_config: DLiteConvertConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific endpoint
        of the OTE-API Services.

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            SessionUpdate instance.
        """
        config = self.convert_config.configuration
        module = importlib.import_module(config.module_name, config.package)
        function = getattr(module, config.function_name)
        kwargs = config.kwargs

        coll = get_collection(session)

        instances = []
        for i, input_config in enumerate(config.inputs):
            if input_config.label:
                instances.append(
                    coll.get(input_config.label, input_config.datamodel)
                )
            elif input_config.datamodel:
                inst = coll.get_instances(
                    metaid=input_config.datamodel,
                    property_mappings=input_config.property_mappings,
                    # More to do: add more arguments...
                )
                instances.append(inst)
            else:
                raise ValueError(
                    "either `label` or `datamodel` must be specified in "
                    f"inputs[{i}]"
                )
        outputs = function(*instances, **kwargs)
        if isinstance(outputs, dlite.Instance):
            outputs = [outputs]

        for inst, output_config in zip(outputs, config.outputs):
            coll.add(output_config.label, inst)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

convert_config: DLiteConvertConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services.

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteSessionUpdate

SessionUpdate instance.
Source code in oteapi_dlite/strategies/convert.py 
157
158
159
160
161
162
163
164
165
166
167
168
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
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific endpoint
    of the OTE-API Services.

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        SessionUpdate instance.
    """
    config = self.convert_config.configuration
    module = importlib.import_module(config.module_name, config.package)
    function = getattr(module, config.function_name)
    kwargs = config.kwargs

    coll = get_collection(session)

    instances = []
    for i, input_config in enumerate(config.inputs):
        if input_config.label:
            instances.append(
                coll.get(input_config.label, input_config.datamodel)
            )
        elif input_config.datamodel:
            inst = coll.get_instances(
                metaid=input_config.datamodel,
                property_mappings=input_config.property_mappings,
                # More to do: add more arguments...
            )
            instances.append(inst)
        else:
            raise ValueError(
                "either `label` or `datamodel` must be specified in "
                f"inputs[{i}]"
            )
    outputs = function(*instances, **kwargs)
    if isinstance(outputs, dlite.Instance):
        outputs = [outputs]

    for inst, output_config in zip(outputs, config.outputs):
        coll.add(output_config.label, inst)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/convert.py 
150
151
152
153
154
155
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteConvertStrategyConfig

Bases: AttrDict

Configuration for generic DLite converter.

Source code in oteapi_dlite/strategies/convert.py 
 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
class DLiteConvertStrategyConfig(AttrDict):
    """Configuration for generic DLite converter."""

    function_name: Annotated[
        str,
        Field(
            description="Name of convert function.  It will be pased the input "
            "instances as arguments and should return a sequence of output "
            "instances.",
        ),
    ]
    module_name: Annotated[
        str,
        Field(
            description=(
                "Name of Python module containing the convertion function."
            ),
        ),
    ]
    package: Annotated[
        Optional[str],
        Field(
            description=(
                "Used when performing a relative import of the converter "
                "function.  It specifies the package to use as the anchor "
                "point from which to resolve the relative import to an absolute"
                " import."
            ),
        ),
    ] = None
    pypi_package: Annotated[
        Optional[str],
        Field(
            description=(
                "Package name on PyPI.  This field is currently only "
                "informative, but might be used in the future for automatic "
                "package installation."
            ),
        ),
    ] = None
    inputs: Annotated[
        Sequence[DLiteConvertInputConfig],
        Field(
            description="Input instances.",
        ),
    ] = []
    outputs: Annotated[
        Sequence[DLiteConvertOutputConfig],
        Field(
            description="Output instances.",
        ),
    ] = []
    kwargs: Annotated[
        Optional[dict],
        Field(
            description="Additional keyword arguments passed "
            "to the convert function.",
        ),
    ] = {}  # noqa: RUF012

function_name: Annotated[str, Field(description='Name of convert function. It will be pased the input instances as arguments and should return a sequence of output instances.')] instance-attribute

inputs: Annotated[Sequence[DLiteConvertInputConfig], Field(description='Input instances.')] = [] class-attribute instance-attribute

kwargs: Annotated[Optional[dict], Field(description='Additional keyword arguments passed to the convert function.')] = {} class-attribute instance-attribute

module_name: Annotated[str, Field(description='Name of Python module containing the convertion function.')] instance-attribute

outputs: Annotated[Sequence[DLiteConvertOutputConfig], Field(description='Output instances.')] = [] class-attribute instance-attribute

package: Annotated[Optional[str], Field(description='Used when performing a relative import of the converter function. It specifies the package to use as the anchor point from which to resolve the relative import to an absolute import.')] = None class-attribute instance-attribute

pypi_package: Annotated[Optional[str], Field(description='Package name on PyPI. This field is currently only informative, but might be used in the future for automatic package installation.')] = None class-attribute instance-attribute

filter

Filter that removes all but specified instances in the collection.

DLiteFilterConfig

Bases: FilterConfig

DLite generate strategy config.

Source code in oteapi_dlite/strategies/filter.py 
82
83
84
85
86
87
88
class DLiteFilterConfig(FilterConfig):
    """DLite generate strategy config."""

    configuration: Annotated[
        DLiteQueryConfig,
        Field(description="DLite filter strategy-specific configuration."),
    ]

configuration: Annotated[DLiteQueryConfig, Field(description='DLite filter strategy-specific configuration.')] instance-attribute

DLiteFilterStrategy

Filter that removes all but specified instances in the collection.

The query configuration should be a regular expression matching labels to keep in the collection. All other labels will be removed.

Registers strategies:

  • ("filterType", "dlite/filter")
Source code in oteapi_dlite/strategies/filter.py 
 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
@dataclass
class DLiteFilterStrategy:
    """Filter that removes all but specified instances in the collection.

    The `query` configuration should be a regular expression matching labels
    to keep in the collection.  All other labels will be removed.

    **Registers strategies**:

    - `("filterType", "dlite/filter")`

    """

    filter_config: DLiteFilterConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy."""
        config = self.filter_config.configuration

        # Alias for query configuration
        keep_label = (
            config.keep_label if config.keep_label else self.filter_config.query
        )

        instdict = {}  # Map instance labels to [uuid, metaURI]
        coll = get_collection(session)
        for s, _, o in coll.get_relations(p="_has-uuid"):
            instdict[s] = [o]
        for s, _, o in coll.get_relations(p="_has-meta"):
            instdict[s].append(o)

        removal = set()  # Labels marked for removal

        # 1: remove_label, remove_datamodel
        if config.remove_label or config.remove_datamodel:
            for label, (_, metauri) in instdict.items():
                if config.remove_label and re.match(config.remove_label, label):
                    removal.add(label)

                if config.remove_datamodel and re.match(
                    config.remove_datamodel, metauri
                ):
                    removal.add(label)
        else:
            removal.update(instdict.keys())

        # 2: keep_label, keep_datamodel
        for label in set(removal):
            if keep_label and re.match(keep_label, label):
                removal.remove(label)

            _, metauri = instdict[label]
            if config.keep_datamodel and re.match(
                config.keep_datamodel, metauri
            ):
                removal.remove(label)

        # 3: keep_referred
        if config.keep_referred:
            labels = {uuid: label for label, (uuid, _) in instdict.items()}
            kept = set(instdict.keys()).difference(removal)
            for label in kept:
                removal.difference_update(
                    labels[inst.uuid]
                    for inst in get_referred_instances(coll.get(label))
                    if inst.uuid in labels
                )

        # 4: remove from collection
        for label in removal:
            coll.remove(label)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

filter_config: DLiteFilterConfig instance-attribute

get(session=None)

Execute the strategy.

Source code in oteapi_dlite/strategies/filter.py 
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 get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy."""
    config = self.filter_config.configuration

    # Alias for query configuration
    keep_label = (
        config.keep_label if config.keep_label else self.filter_config.query
    )

    instdict = {}  # Map instance labels to [uuid, metaURI]
    coll = get_collection(session)
    for s, _, o in coll.get_relations(p="_has-uuid"):
        instdict[s] = [o]
    for s, _, o in coll.get_relations(p="_has-meta"):
        instdict[s].append(o)

    removal = set()  # Labels marked for removal

    # 1: remove_label, remove_datamodel
    if config.remove_label or config.remove_datamodel:
        for label, (_, metauri) in instdict.items():
            if config.remove_label and re.match(config.remove_label, label):
                removal.add(label)

            if config.remove_datamodel and re.match(
                config.remove_datamodel, metauri
            ):
                removal.add(label)
    else:
        removal.update(instdict.keys())

    # 2: keep_label, keep_datamodel
    for label in set(removal):
        if keep_label and re.match(keep_label, label):
            removal.remove(label)

        _, metauri = instdict[label]
        if config.keep_datamodel and re.match(
            config.keep_datamodel, metauri
        ):
            removal.remove(label)

    # 3: keep_referred
    if config.keep_referred:
        labels = {uuid: label for label, (uuid, _) in instdict.items()}
        kept = set(instdict.keys()).difference(removal)
        for label in kept:
            removal.difference_update(
                labels[inst.uuid]
                for inst in get_referred_instances(coll.get(label))
                if inst.uuid in labels
            )

    # 4: remove from collection
    for label in removal:
        coll.remove(label)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/filter.py 
106
107
108
109
110
111
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteQueryConfig

Bases: AttrDict

Configuration for the DLite filter strategy.

First the remove_label and remove_datamodel configurations are used to mark matching instances for removal. If neither remove_label or remove_datamodel are given, all instances are marked for removal.

Then instances matching keep_label and keep_datamodel are unmarked for removal.

If keep_referred is true, any instance that is referred to by an instance not marked for removal is also unmarked for removal.

Finally, the instances that are still marked for removal are removed from the collection.

Source code in oteapi_dlite/strategies/filter.py 
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
53
54
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
class DLiteQueryConfig(AttrDict):
    """Configuration for the DLite filter strategy.

    First the `remove_label` and `remove_datamodel` configurations are
    used to mark matching instances for removal.  If neither
    `remove_label` or `remove_datamodel` are given, all instances are
    marked for removal.

    Then instances matching `keep_label` and `keep_datamodel` are unmarked
    for removal.

    If `keep_referred` is true, any instance that is referred to by
    an instance not marked for removal is also unmarked for removal.

    Finally, the instances that are still marked for removal are removed
    from the collection.
    """

    remove_label: Annotated[
        Optional[str],
        Field(description="Regular expression matching labels to remove."),
    ] = None
    remove_datamodel: Annotated[
        Optional[str],
        Field(
            description="Regular expression matching datamodel URIs to remove.",
        ),
    ] = None
    keep_label: Annotated[
        Optional[str],
        Field(
            description=(
                "Regular expression matching labels to keep. This "
                "configuration overrides `remove_label` and "
                "`remove_datamodel`. Alias for the FilterStrategy `query` "
                "configuration, that is inherited from the oteapi-core Filter "
                "data model."
            ),
        ),
    ] = None
    keep_datamodel: Annotated[
        Optional[str],
        Field(
            description=(
                "Regular expression matching datamodel URIs to keep in "
                "collection. This configuration overrides `remove_label` and "
                "`remove_datamodel`."
            ),
        ),
    ] = None
    keep_referred: Annotated[
        bool,
        Field(
            description=(
                "Whether to keep all instances in the collection that are "
                "directly or indirectly referred to (via ref-types or "
                "collections) by kept instances."
            ),
        ),
    ] = True

keep_datamodel: Annotated[Optional[str], Field(description='Regular expression matching datamodel URIs to keep in collection. This configuration overrides `remove_label` and `remove_datamodel`.')] = None class-attribute instance-attribute

keep_label: Annotated[Optional[str], Field(description='Regular expression matching labels to keep. This configuration overrides `remove_label` and `remove_datamodel`. Alias for the FilterStrategy `query` configuration, that is inherited from the oteapi-core Filter data model.')] = None class-attribute instance-attribute

keep_referred: Annotated[bool, Field(description='Whether to keep all instances in the collection that are directly or indirectly referred to (via ref-types or collections) by kept instances.')] = True class-attribute instance-attribute

remove_datamodel: Annotated[Optional[str], Field(description='Regular expression matching datamodel URIs to remove.')] = None class-attribute instance-attribute

remove_label: Annotated[Optional[str], Field(description='Regular expression matching labels to remove.')] = None class-attribute instance-attribute

generate

Generic generate strategy using DLite storage plugin.

hasInput = 'https://w3id.org/emmo#EMMO_36e69413_8c59_4799_946c_10b05d266e22' module-attribute

hasOutput = 'https://w3id.org/emmo#EMMO_c4bace1d_4db0_4cd3_87e9_18122bae2840' module-attribute

DLiteGenerateConfig

Bases: FunctionConfig

DLite generate strategy config.

Source code in oteapi_dlite/strategies/generate.py 
245
246
247
248
249
250
251
class DLiteGenerateConfig(FunctionConfig):
    """DLite generate strategy config."""

    configuration: Annotated[
        DLiteStorageConfig,
        Field(description="DLite generate strategy-specific configuration."),
    ]

configuration: Annotated[DLiteStorageConfig, Field(description='DLite generate strategy-specific configuration.')] instance-attribute

DLiteGenerateStrategy

Generic DLite generate strategy utilising DLite storage plugins.

Registers strategies:

  • ("mediaType", "application/vnd.dlite-generate")
Source code in oteapi_dlite/strategies/generate.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
@dataclass
class DLiteGenerateStrategy:
    """Generic DLite generate strategy utilising DLite storage plugins.

    **Registers strategies**:

    - `("mediaType", "application/vnd.dlite-generate")`

    """

    generate_config: DLiteGenerateConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific endpoint
        of the OTE-API Services.

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            SessionUpdate instance.
        """
        config = self.generate_config.configuration
        cacheconfig = config.datacache_config

        if session is None:
            session = {}

        driver = (
            config.driver
            if config.driver
            else get_driver(mediaType=config.mediaType)
        )

        coll = get_collection(session, config.collection_id)

        if config.label:
            inst = coll[config.label]
        elif config.datamodel:
            instances = coll.get_instances(
                metaid=config.datamodel,
                property_mappings=True,
                allow_incomplete=config.allow_incomplete,
            )
            inst = next(instances)
        elif config.store_collection:
            if config.store_collection_id:
                inst = coll.copy(newid=config.store_collection_id)
            else:
                inst = coll
        else:  # fail if there are more instances
            raise ValueError(
                "One of `label` or `datamodel` configurations should be given."
            )

        # Save instance
        if config.location:
            inst.save(driver, config.location, config.options)
        else:  # missing test
            if cacheconfig and cacheconfig.accessKey:
                key = cacheconfig.accessKey
            else:  # missing test
                key = "generate_data"
            cache = DataCache()
            with tempfile.TemporaryDirectory() as tmpdir:
                inst.save(driver, f"{tmpdir}/data", config.options)
                with Path(f"{tmpdir}/data").open("rb") as f:
                    cache.add(f.read(), key=key)

        # Store documentation of this instance in the knowledge base
        if config.kb_document_class:

            # Import here to avoid hard dependencies on tripper.
            from tripper import RDF
            from tripper.convert import save_container

            # kb_settings = get_settings(session, "tripper.triplestore")
            # if not kb_settings:
            #     raise KeyError(
            #         "The `kb_document_class` configuration requires that a "
            #         "'tripper.triplestore' settings has been added using the "
            #         "application/vnd.dlite-settings strategy."
            #     )
            # IRI of new individual
            iri = individual_iri(
                class_iri=config.kb_document_class,
                base_iri=config.kb_document_base_iri,
            )

            triples = [(iri, RDF.type, config.kb_document_class)]
            if config.kb_document_context:
                for prop, val in config.kb_document_context.items():
                    triples.append((iri, prop, val))

            ts = get_triplestore(session)
            try:
                if config.kb_document_computation:
                    comput = individual_iri(
                        class_iri=config.kb_document_computation,
                        base_iri=config.kb_document_base_iri,
                    )
                    triples.extend(
                        [
                            (comput, RDF.type, config.kb_document_computation),
                            (comput, hasOutput, iri),
                        ]
                    )

                    # Relate computation individual `comput` to its
                    # input individuals.
                    #
                    # This simple implementation works against KB.  It
                    # assumes that the input of
                    # `kb_document_computation` is documented in the
                    # KB and that there only exists one individual of each
                    # input class.
                    #
                    # In the case of multiple individuals of the input
                    # classes, the workflow executer must be involded
                    # in the documentation.  It can either do the
                    # documentation itself or provide a callback
                    # providing the needed info, which can be called
                    # from this strategy.

                    # Relate to input dataset individuals
                    restrictions = ts.restrictions(
                        config.kb_document_computation, hasInput
                    )
                    for r in restrictions:
                        input_class = r["value"]
                        indv = ts.value(predicate=RDF.type, object=input_class)
                        triples.append((comput, r["property"], indv))

                    # Add output dataset individuals
                    restrictions = ts.restrictions(
                        config.kb_document_computation, hasOutput
                    )
                    for r in restrictions:
                        output_class = r["value"]
                        indv = ts.value(
                            predicate=RDF.type,
                            object=output_class,
                            default=None,
                        )
                        if indv and indv != iri:
                            triples.append((comput, r["property"], indv))

                # Document data source
                resource = {
                    "dataresource": {
                        "type": config.kb_document_class,
                        "downloadUrl": config.location,
                        "mediaType": (
                            config.mediaType
                            if config.mediaType
                            else "application/vnd.dlite-parse"
                        ),
                        "configuration": {
                            "datamodel": (
                                config.datamodel
                                if config.datamodel
                                else inst.meta.uri
                            ),
                            "driver": config.driver,
                            "options": (  # Trying to be clever here...
                                config.options.replace("mode=w", "mode=r")
                                if config.options
                                else config.options
                            ),
                        },
                    },
                    # "parse": {},  # No supported by OTEAPI yet...
                    "mapping": {
                        "mappingType": "mappings",
                        # __TODO__
                        # Populate prefixes and triples from mapping
                        # strategy in current partial pipeline
                        # "prefixes": {},
                        # "triples": [],
                    },
                }
                update_dict(resource, config.kb_document_update)

                save_container(
                    ts,
                    resource,
                    iri,
                    recognised_keys="basic",
                )
                ts.add_triples(triples)

            finally:
                ts.close()

        # __TODO__
        # Can we safely assume that all strategies in a pipeline will be
        # executed in the same Python interpreter?  If not, we should write
        # the collection to a storage, such that it can be shared with the
        # other strategies.

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

generate_config: DLiteGenerateConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services.

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteSessionUpdate

SessionUpdate instance.
Source code in oteapi_dlite/strategies/generate.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific endpoint
    of the OTE-API Services.

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        SessionUpdate instance.
    """
    config = self.generate_config.configuration
    cacheconfig = config.datacache_config

    if session is None:
        session = {}

    driver = (
        config.driver
        if config.driver
        else get_driver(mediaType=config.mediaType)
    )

    coll = get_collection(session, config.collection_id)

    if config.label:
        inst = coll[config.label]
    elif config.datamodel:
        instances = coll.get_instances(
            metaid=config.datamodel,
            property_mappings=True,
            allow_incomplete=config.allow_incomplete,
        )
        inst = next(instances)
    elif config.store_collection:
        if config.store_collection_id:
            inst = coll.copy(newid=config.store_collection_id)
        else:
            inst = coll
    else:  # fail if there are more instances
        raise ValueError(
            "One of `label` or `datamodel` configurations should be given."
        )

    # Save instance
    if config.location:
        inst.save(driver, config.location, config.options)
    else:  # missing test
        if cacheconfig and cacheconfig.accessKey:
            key = cacheconfig.accessKey
        else:  # missing test
            key = "generate_data"
        cache = DataCache()
        with tempfile.TemporaryDirectory() as tmpdir:
            inst.save(driver, f"{tmpdir}/data", config.options)
            with Path(f"{tmpdir}/data").open("rb") as f:
                cache.add(f.read(), key=key)

    # Store documentation of this instance in the knowledge base
    if config.kb_document_class:

        # Import here to avoid hard dependencies on tripper.
        from tripper import RDF
        from tripper.convert import save_container

        # kb_settings = get_settings(session, "tripper.triplestore")
        # if not kb_settings:
        #     raise KeyError(
        #         "The `kb_document_class` configuration requires that a "
        #         "'tripper.triplestore' settings has been added using the "
        #         "application/vnd.dlite-settings strategy."
        #     )
        # IRI of new individual
        iri = individual_iri(
            class_iri=config.kb_document_class,
            base_iri=config.kb_document_base_iri,
        )

        triples = [(iri, RDF.type, config.kb_document_class)]
        if config.kb_document_context:
            for prop, val in config.kb_document_context.items():
                triples.append((iri, prop, val))

        ts = get_triplestore(session)
        try:
            if config.kb_document_computation:
                comput = individual_iri(
                    class_iri=config.kb_document_computation,
                    base_iri=config.kb_document_base_iri,
                )
                triples.extend(
                    [
                        (comput, RDF.type, config.kb_document_computation),
                        (comput, hasOutput, iri),
                    ]
                )

                # Relate computation individual `comput` to its
                # input individuals.
                #
                # This simple implementation works against KB.  It
                # assumes that the input of
                # `kb_document_computation` is documented in the
                # KB and that there only exists one individual of each
                # input class.
                #
                # In the case of multiple individuals of the input
                # classes, the workflow executer must be involded
                # in the documentation.  It can either do the
                # documentation itself or provide a callback
                # providing the needed info, which can be called
                # from this strategy.

                # Relate to input dataset individuals
                restrictions = ts.restrictions(
                    config.kb_document_computation, hasInput
                )
                for r in restrictions:
                    input_class = r["value"]
                    indv = ts.value(predicate=RDF.type, object=input_class)
                    triples.append((comput, r["property"], indv))

                # Add output dataset individuals
                restrictions = ts.restrictions(
                    config.kb_document_computation, hasOutput
                )
                for r in restrictions:
                    output_class = r["value"]
                    indv = ts.value(
                        predicate=RDF.type,
                        object=output_class,
                        default=None,
                    )
                    if indv and indv != iri:
                        triples.append((comput, r["property"], indv))

            # Document data source
            resource = {
                "dataresource": {
                    "type": config.kb_document_class,
                    "downloadUrl": config.location,
                    "mediaType": (
                        config.mediaType
                        if config.mediaType
                        else "application/vnd.dlite-parse"
                    ),
                    "configuration": {
                        "datamodel": (
                            config.datamodel
                            if config.datamodel
                            else inst.meta.uri
                        ),
                        "driver": config.driver,
                        "options": (  # Trying to be clever here...
                            config.options.replace("mode=w", "mode=r")
                            if config.options
                            else config.options
                        ),
                    },
                },
                # "parse": {},  # No supported by OTEAPI yet...
                "mapping": {
                    "mappingType": "mappings",
                    # __TODO__
                    # Populate prefixes and triples from mapping
                    # strategy in current partial pipeline
                    # "prefixes": {},
                    # "triples": [],
                },
            }
            update_dict(resource, config.kb_document_update)

            save_container(
                ts,
                resource,
                iri,
                recognised_keys="basic",
            )
            ts.add_triples(triples)

        finally:
            ts.close()

    # __TODO__
    # Can we safely assume that all strategies in a pipeline will be
    # executed in the same Python interpreter?  If not, we should write
    # the collection to a storage, such that it can be shared with the
    # other strategies.

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/generate.py 
266
267
268
269
270
271
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteStorageConfig

Bases: AttrDict

Configuration for a generic DLite storage filter.

The DLite storage driver to can be specified using either the driver or mediaType field.

Where the output should be written, is specified using either the location or datacache_config.accessKey field.

Either label or datamodel should be provided.

Source code in oteapi_dlite/strategies/generate.py 
 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
 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
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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
class DLiteStorageConfig(AttrDict):
    """Configuration for a generic DLite storage filter.

    The DLite storage driver to can be specified using either the `driver`
    or `mediaType` field.

    Where the output should be written, is specified using either the
    `location` or `datacache_config.accessKey` field.

    Either `label` or `datamodel` should be provided.
    """

    driver: Annotated[
        Optional[str],
        Field(
            description='Name of DLite driver (ex: "json").',
        ),
    ] = None
    mediaType: Annotated[
        Optional[str],
        Field(
            description='Media type for DLite driver (ex: "application/json").',
        ),
    ] = None
    options: Annotated[
        Optional[str],
        Field(
            description=(
                "Comma-separated list of options passed to the DLite "
                "storage plugin."
            ),
        ),
    ] = None
    location: Annotated[
        Optional[str],
        Field(
            description=(
                "Location of storage to write to.  If unset to store in data "
                "cache using the key provided with "
                "`datacache_config.accessKey` (defaults to 'generate_data')."
            ),
        ),
    ] = None
    label: Annotated[
        Optional[str],
        Field(
            description=(
                "Label of DLite instance in the collection to serialise."
            ),
        ),
    ] = None
    datamodel: Annotated[
        Optional[str],
        Field(
            description=(
                "URI to the datamodel of the new instance.  Needed when "
                "generating the instance from mappings.  Cannot be combined "
                "with `label`"
            ),
        ),
    ] = None
    store_collection: Annotated[
        bool,
        Field(
            description="Whether to store the entire collection in the session "
            "instead of a single instance.  Cannot be combined with `label` or "
            "`datamodel`.",
        ),
    ] = False
    store_collection_id: Annotated[
        Optional[str],
        Field(
            description="Used together with `store_collection` If given, store "
            "a copy of the collection with this id.",
        ),
    ] = None
    allow_incomplete: Annotated[
        Optional[bool],
        Field(
            description="Whether to allow incomplete property mappings.",
        ),
    ] = False
    collection_id: Annotated[
        Optional[str],
        Field(
            description=("ID of the collection to use."),
        ),
    ] = None
    datacache_config: Annotated[
        Optional[DataCacheConfig],
        Field(
            description="Configuration options for the local data cache.",
        ),
    ] = None
    kb_document_class: Annotated[
        Optional[str],
        Field(
            description=(
                "IRI of a class in the ontology."
                "\n\n"
                "If given, the generated DLite instance is documented in the "
                "knowledge base as an instance of this class."
                "\n\n"
                "Expects that the 'tripper.triplestore' setting has been "
                "set using the SettingsStrategy (vnd.dlite-settings). "
                "This settings should be a dict that can be passed "
                "as keyword arguments to `tripper.Triplestore()`."
                "\n\n"
                "Example of adding expected settings using OTELib:\n"
                "\n\n"
                ">>> kb_settings = client.create_filter(\n"
                "...     filterType='application/vnd.dlite-settings',\n"
                "...     configuration={\n"
                "...         'label': 'tripper.triplestore',\n"
                "...         'settings': {\n"
                "...             'backend': 'rdflib',\n"
                "...             'triplestore_url': '/path/to/local/kb.ttl',\n"
                "...         },\n"
                "...     },\n"
                "... )\n"
                ">>> generate = client.create_function(\n"
                "...     functionType='application/vnd.dlite-generate'\n"
                "...     configuration={\n"
                "...         kb_document_class='http://example.com#MyClass'\n"
                "...         ...\n"
                "...     },\n"
                "... )\n"
                ">>> pipeline = ... >> generate >> kb_settings\n"
                ">>> pipeline.get()\n"
            ),
        ),
    ] = None
    kb_document_update: Annotated[
        Optional[dict],
        Field(
            description=(
                "Dict updating the documentation (partial pipeline) created "
                "with `kb_document_class`."
                "\n\n"
                "This dict should be structured as follows: "
                "\n\n"
                "    {\n"
                '      "dataresource": {...},\n'
                '      "parse": {...}\n'
                '      "mapping": {...}\n'
                "    }\n"
                "\n"
                "where the provided items will override the the default "
                "configurations in respective partial pipeline created by "
                '`kb_document_class`.  Any of the items "dataresource", '
                '"parse" and "mapping" are optional.',
            ),
        ),
    ] = None
    kb_document_base_iri: Annotated[
        str, Field(description="Base IRI or prefix for created individuals.")
    ] = ":"
    kb_document_context: Annotated[
        Optional[dict],
        Field(
            description=(
                "If `kb_document_class` is given, this configuration will add "
                "additional context to the documentation of the generated "
                "individual."
                "\n\n"
                "This might be useful to make it easy to later access the "
                "generated individual."
                "\n\n"
                "This configuration should be a dict mapping providing the "
                "additional documentation of the driver. It should map OWL "
                "properties to either tripper literals or IRIs."
                "\n\n"
                "Example: `{RDF.type: ONTO.MyDataSet, "
                "EMMO.isDescriptionFor: ONTO.MyMaterial}`"
            ),
        ),
    ] = None
    kb_document_computation: Annotated[
        Optional[str],
        Field(
            description=(
                "IRI of a computation subclass."
                "\n\n"
                "Requires `kb_document_class`, and is used to "
                "document the computation (model) that the "
                "individual (of `kb_document_class`) to be documented "
                "is output of."
                "When `kb_document_computation` is given a new individual of "
                "the computation subclass is created. Input and "
                "output datasets are documented using the relation "
                " `emmo:hasInput` and `emmo:hasOutput`, "
                "respectively.  The individual of `kb_document_class` is "
                "one of the output individuals."
                "\n\n"
                "Note: This configuration relies on several assumptions:\n"
                "  - The `kb_document_computation` class exists in the "
                "knowledge base and is related to its input and output "
                "dataset classes via `emmo:hasInput` and `emmo:hasOutput` "
                "restrictions, respectively.\n"
                "  - There exists only one individual of each input dataset "
                "class.\n"
                "  - There exists at most one individual of each output "
                "dataset class.\n"
            ),
        ),
    ] = None

allow_incomplete: Annotated[Optional[bool], Field(description='Whether to allow incomplete property mappings.')] = False class-attribute instance-attribute

collection_id: Annotated[Optional[str], Field(description='ID of the collection to use.')] = None class-attribute instance-attribute

datacache_config: Annotated[Optional[DataCacheConfig], Field(description='Configuration options for the local data cache.')] = None class-attribute instance-attribute

datamodel: Annotated[Optional[str], Field(description='URI to the datamodel of the new instance. Needed when generating the instance from mappings. Cannot be combined with `label`')] = None class-attribute instance-attribute

driver: Annotated[Optional[str], Field(description='Name of DLite driver (ex: "json").')] = None class-attribute instance-attribute

kb_document_base_iri: Annotated[str, Field(description='Base IRI or prefix for created individuals.')] = ':' class-attribute instance-attribute

kb_document_class: Annotated[Optional[str], Field(description="IRI of a class in the ontology.\n\nIf given, the generated DLite instance is documented in the knowledge base as an instance of this class.\n\nExpects that the 'tripper.triplestore' setting has been set using the SettingsStrategy (vnd.dlite-settings). This settings should be a dict that can be passed as keyword arguments to `tripper.Triplestore()`.\n\nExample of adding expected settings using OTELib:\n\n\n>>> kb_settings = client.create_filter(\n... filterType='application/vnd.dlite-settings',\n... configuration={\n... 'label': 'tripper.triplestore',\n... 'settings': {\n... 'backend': 'rdflib',\n... 'triplestore_url': '/path/to/local/kb.ttl',\n... },\n... },\n... )\n>>> generate = client.create_function(\n... functionType='application/vnd.dlite-generate'\n... configuration={\n... kb_document_class='http://example.com#MyClass'\n... ...\n... },\n... )\n>>> pipeline = ... >> generate >> kb_settings\n>>> pipeline.get()\n")] = None class-attribute instance-attribute

kb_document_computation: Annotated[Optional[str], Field(description='IRI of a computation subclass.\n\nRequires `kb_document_class`, and is used to document the computation (model) that the individual (of `kb_document_class`) to be documented is output of.When `kb_document_computation` is given a new individual of the computation subclass is created. Input and output datasets are documented using the relation `emmo:hasInput` and `emmo:hasOutput`, respectively. The individual of `kb_document_class` is one of the output individuals.\n\nNote: This configuration relies on several assumptions:\n - The `kb_document_computation` class exists in the knowledge base and is related to its input and output dataset classes via `emmo:hasInput` and `emmo:hasOutput` restrictions, respectively.\n - There exists only one individual of each input dataset class.\n - There exists at most one individual of each output dataset class.\n')] = None class-attribute instance-attribute

kb_document_context: Annotated[Optional[dict], Field(description='If `kb_document_class` is given, this configuration will add additional context to the documentation of the generated individual.\n\nThis might be useful to make it easy to later access the generated individual.\n\nThis configuration should be a dict mapping providing the additional documentation of the driver. It should map OWL properties to either tripper literals or IRIs.\n\nExample: `{RDF.type: ONTO.MyDataSet, EMMO.isDescriptionFor: ONTO.MyMaterial}`')] = None class-attribute instance-attribute

kb_document_update: Annotated[Optional[dict], Field(description=('Dict updating the documentation (partial pipeline) created with `kb_document_class`.\n\nThis dict should be structured as follows: \n\n {\n "dataresource": {...},\n "parse": {...}\n "mapping": {...}\n }\n\nwhere the provided items will override the the default configurations in respective partial pipeline created by `kb_document_class`. Any of the items "dataresource", "parse" and "mapping" are optional.'))] = None class-attribute instance-attribute

label: Annotated[Optional[str], Field(description='Label of DLite instance in the collection to serialise.')] = None class-attribute instance-attribute

location: Annotated[Optional[str], Field(description="Location of storage to write to. If unset to store in data cache using the key provided with `datacache_config.accessKey` (defaults to 'generate_data').")] = None class-attribute instance-attribute

mediaType: Annotated[Optional[str], Field(description='Media type for DLite driver (ex: "application/json").')] = None class-attribute instance-attribute

options: Annotated[Optional[str], Field(description='Comma-separated list of options passed to the DLite storage plugin.')] = None class-attribute instance-attribute

store_collection: Annotated[bool, Field(description='Whether to store the entire collection in the session instead of a single instance. Cannot be combined with `label` or `datamodel`.')] = False class-attribute instance-attribute

store_collection_id: Annotated[Optional[str], Field(description='Used together with `store_collection` If given, store a copy of the collection with this id.')] = None class-attribute instance-attribute

KBError

Bases: ValueError

Invalid data in knowledge base.

Source code in oteapi_dlite/strategies/generate.py 
33
34
class KBError(ValueError):
    """Invalid data in knowledge base."""

individual_iri(class_iri, base_iri=':', randbytes=6)

Return an IRI for an individual of a class.

Parameters:

Name Type Description Default
class_iri str

IRI of the class to create an individual of.

 required
base_iri str

Base IRI of the created individual.

 ':'
randbytes int

Number of random bytes to include in the returned IRI.

 6

Returns:

Type Description
str

IRI of a new individual.
Source code in oteapi_dlite/strategies/generate.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
def individual_iri(
    class_iri: str, base_iri: str = ":", randbytes: int = 6
) -> str:
    """Return an IRI for an individual of a class.

    Arguments:
        class_iri: IRI of the class to create an individual of.
        base_iri: Base IRI of the created individual.
        randbytes: Number of random bytes to include in the returned IRI.

    Returns:
        IRI of a new individual.

    """
    basename = (
        class_iri.split(":", 1)[-1]
        .rsplit("/", 1)[-1]
        .rsplit("#", 1)[-1]
        .lower()
    )
    return f"{base_iri}{basename}-{os.urandom(randbytes).hex()}"

mapping

Mapping filter strategy.

DLiteMappingConfig

Bases: MappingConfig

DLite mapping strategy config.

Source code in oteapi_dlite/strategies/mapping.py 
33
34
35
36
37
38
39
40
41
class DLiteMappingConfig(MappingConfig):
    """DLite mapping strategy config."""

    configuration: Annotated[
        DLiteMappingStrategyConfig,
        Field(
            description="DLite mapping strategy-specific configuration.",
        ),
    ] = DLiteMappingStrategyConfig()

configuration: Annotated[DLiteMappingStrategyConfig, Field(description='DLite mapping strategy-specific configuration.')] = DLiteMappingStrategyConfig() class-attribute instance-attribute

DLiteMappingStrategy

Strategy for a mapping.

Registers strategies:

  • ("mappingType", "mappings")
Source code in oteapi_dlite/strategies/mapping.py 
44
45
46
47
48
49
50
51
52
53
54
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
@dataclass
class DLiteMappingStrategy:
    """Strategy for a mapping.

    **Registers strategies**:

    - `("mappingType", "mappings")`

    """

    mapping_config: DLiteMappingConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize strategy."""
        if session is None:
            session = {}

        ts = get_triplestore(session)

        if self.mapping_config.prefixes:
            for prefix, iri in self.mapping_config.prefixes.items():
                ts.bind(prefix, iri)

        if self.mapping_config.triples:
            ts.add_triples(
                [
                    [
                        ts.expand_iri(t) if isinstance(t, str) else t
                        for t in triple
                    ]
                    for triple in self.mapping_config.triples
                ]
            )

        coll = get_collection(session)
        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute strategy and return a dictionary."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

mapping_config: DLiteMappingConfig instance-attribute

get(session=None)

Execute strategy and return a dictionary.

Source code in oteapi_dlite/strategies/mapping.py 
84
85
86
87
88
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute strategy and return a dictionary."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

initialize(session=None)

Initialize strategy.

Source code in oteapi_dlite/strategies/mapping.py 
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
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize strategy."""
    if session is None:
        session = {}

    ts = get_triplestore(session)

    if self.mapping_config.prefixes:
        for prefix, iri in self.mapping_config.prefixes.items():
            ts.bind(prefix, iri)

    if self.mapping_config.triples:
        ts.add_triples(
            [
                [
                    ts.expand_iri(t) if isinstance(t, str) else t
                    for t in triple
                ]
                for triple in self.mapping_config.triples
            ]
        )

    coll = get_collection(session)
    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

DLiteMappingStrategyConfig

Bases: AttrDict

Configuration for a DLite mapping filter.

Source code in oteapi_dlite/strategies/mapping.py 
22
23
24
25
26
27
28
29
30
class DLiteMappingStrategyConfig(AttrDict):
    """Configuration for a DLite mapping filter."""

    datamodel: Annotated[
        Optional[AnyUrl],
        Field(
            description="URI of the datamodel that is mapped.",
        ),
    ] = None

datamodel: Annotated[Optional[AnyUrl], Field(description='URI of the datamodel that is mapped.')] = None class-attribute instance-attribute

parse

Generic parse strategy using DLite storage plugin.

DLiteParseConfig

Bases: AttrDict

Configuration for generic DLite parser.

Source code in oteapi_dlite/strategies/parse.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
53
54
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
class DLiteParseConfig(AttrDict):
    """Configuration for generic DLite parser."""

    driver: Annotated[
        str,
        Field(
            description='Name of DLite driver (ex: "json").',
        ),
    ]
    location: Annotated[
        Optional[str],
        Field(
            description=(
                "Explicit location of storage.  Normally data is read from the "
                "data cache using `datacache_config.accessKey` (default: "
                "'key')."
            ),
        ),
    ] = None
    options: Annotated[
        Optional[str],
        Field(
            description=(
                "Comma-separated list of options passed to the DLite storage "
                "plugin."
            ),
        ),
    ] = None
    id: Annotated[
        Optional[str],
        Field(
            description="If given, the id of the instance in the storage.",
        ),
    ] = None
    label: Annotated[
        Optional[str],
        Field(
            description=(
                "Optional label of the new DLite instance in the collection."
            ),
        ),
    ] = None
    datamodel: Annotated[
        Optional[str],
        Field(
            description=(
                "DLite datamodel documenting the structure of the data set. "
                "Often unused, since the datamodel is implicitly defined in "
                "the DLite driver (DLite plugin), but for a documentation "
                "point of view this is a very important field."
            ),
        ),
    ] = None
    datacache_config: Annotated[
        Optional[DataCacheConfig],
        Field(
            description="Configuration options for the local data cache.",
        ),
    ] = None

datacache_config: Annotated[Optional[DataCacheConfig], Field(description='Configuration options for the local data cache.')] = None class-attribute instance-attribute

datamodel: Annotated[Optional[str], Field(description='DLite datamodel documenting the structure of the data set. Often unused, since the datamodel is implicitly defined in the DLite driver (DLite plugin), but for a documentation point of view this is a very important field.')] = None class-attribute instance-attribute

driver: Annotated[str, Field(description='Name of DLite driver (ex: "json").')] instance-attribute

id: Annotated[Optional[str], Field(description='If given, the id of the instance in the storage.')] = None class-attribute instance-attribute

label: Annotated[Optional[str], Field(description='Optional label of the new DLite instance in the collection.')] = None class-attribute instance-attribute

location: Annotated[Optional[str], Field(description="Explicit location of storage. Normally data is read from the data cache using `datacache_config.accessKey` (default: 'key').")] = None class-attribute instance-attribute

options: Annotated[Optional[str], Field(description='Comma-separated list of options passed to the DLite storage plugin.')] = None class-attribute instance-attribute

DLiteParseResourceConfig

Bases: ResourceConfig

DLite parse strategy resource config.

Source code in oteapi_dlite/strategies/parse.py 
82
83
84
85
86
87
88
class DLiteParseResourceConfig(ResourceConfig):
    """DLite parse strategy resource config."""

    configuration: Annotated[
        DLiteParseConfig,
        Field(description="DLite parse strategy-specific configuration."),
    ]

configuration: Annotated[DLiteParseConfig, Field(description='DLite parse strategy-specific configuration.')] instance-attribute

DLiteParseStrategy

Generic DLite parse strategy utilising DLite storage plugins.

Registers strategies:

  • ("mediaType", "application/vnd.dlite-parse")
Source code in oteapi_dlite/strategies/parse.py 
 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
174
175
176
177
178
179
180
181
182
183
184
185
@dataclass
class DLiteParseStrategy:
    """Generic DLite parse strategy utilising DLite storage plugins.

    **Registers strategies**:

    - `("mediaType", "application/vnd.dlite-parse")`

    """

    parse_config: DLiteParseResourceConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific endpoint
        of the OTE-API Services.

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            SessionUpdate instance.
        """
        config = self.parse_config.configuration
        cacheconfig = config.datacache_config

        driver = (
            config.driver
            if config.driver
            else get_driver(
                mediaType=self.parse_config.mediaType,
            )
        )

        # Create instance
        if config.location:
            inst = dlite.Instance.from_location(
                driver=driver,
                location=config.location,
                options=config.options,
                id=config.id,
            )
        else:
            if cacheconfig and cacheconfig.accessKey:
                key = cacheconfig.accessKey
            elif session and "key" in session:
                key = session["key"]
            else:
                raise ValueError(
                    "either `location` or `datacache_config.accessKey` must be "
                    "provided"
                )

            # See if we can extract file suffix from downloadUrl
            if self.parse_config.downloadUrl:
                suffix = Path(str(self.parse_config.downloadUrl)).suffix
            else:
                suffix = None

            cache = DataCache()
            with cache.getfile(key, suffix=suffix) as location:
                inst = dlite.Instance.from_location(
                    driver=driver,
                    location=str(location),
                    options=config.options,
                    id=config.id,
                )

        # Insert inst into collection
        coll = get_collection(session)
        label = config.label if config.label else inst.uuid
        coll.add(label, inst)

        # __TODO__
        # See
        # https://github.com/EMMC-ASBL/oteapi-dlite/pull/84#discussion_r1050437185
        # and following comments.
        #
        # Since we cannot safely assume that all strategies in a
        # pipeline will be executed in the same Python interpreter,
        # the collection should be written to a storage, such that it
        # can be shared with the other strategies.

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

parse_config: DLiteParseResourceConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services.

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteSessionUpdate

SessionUpdate instance.
Source code in oteapi_dlite/strategies/parse.py 
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
174
175
176
177
178
179
180
181
182
183
184
185
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific endpoint
    of the OTE-API Services.

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        SessionUpdate instance.
    """
    config = self.parse_config.configuration
    cacheconfig = config.datacache_config

    driver = (
        config.driver
        if config.driver
        else get_driver(
            mediaType=self.parse_config.mediaType,
        )
    )

    # Create instance
    if config.location:
        inst = dlite.Instance.from_location(
            driver=driver,
            location=config.location,
            options=config.options,
            id=config.id,
        )
    else:
        if cacheconfig and cacheconfig.accessKey:
            key = cacheconfig.accessKey
        elif session and "key" in session:
            key = session["key"]
        else:
            raise ValueError(
                "either `location` or `datacache_config.accessKey` must be "
                "provided"
            )

        # See if we can extract file suffix from downloadUrl
        if self.parse_config.downloadUrl:
            suffix = Path(str(self.parse_config.downloadUrl)).suffix
        else:
            suffix = None

        cache = DataCache()
        with cache.getfile(key, suffix=suffix) as location:
            inst = dlite.Instance.from_location(
                driver=driver,
                location=str(location),
                options=config.options,
                id=config.id,
            )

    # Insert inst into collection
    coll = get_collection(session)
    label = config.label if config.label else inst.uuid
    coll.add(label, inst)

    # __TODO__
    # See
    # https://github.com/EMMC-ASBL/oteapi-dlite/pull/84#discussion_r1050437185
    # and following comments.
    #
    # Since we cannot safely assume that all strategies in a
    # pipeline will be executed in the same Python interpreter,
    # the collection should be written to a storage, such that it
    # can be shared with the other strategies.

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/parse.py 
103
104
105
106
107
108
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

parse_excel

Strategy for parsing an Excel spreadsheet to a DLite instance.

DLiteExcelParseConfig

Bases: AttrDict

Configuration for DLite Excel parser.

Source code in oteapi_dlite/strategies/parse_excel.py 
29
30
31
32
33
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
63
64
class DLiteExcelParseConfig(AttrDict):
    """Configuration for DLite Excel parser."""

    metadata: Annotated[
        Optional[HttpUrl],
        Field(
            description=(
                "URI of DLite metadata to return.  If not provided, the "
                "metadata will be inferred from the excel file."
            ),
        ),
    ] = None

    id: Annotated[
        Optional[str], Field(description="Optional id on new instance.")
    ] = None

    label: Annotated[
        Optional[str],
        Field(
            description="Optional label for new instance in collection.",
        ),
    ] = "excel-data"

    excel_config: Annotated[
        XLSXParseConfig,
        Field(
            description="DLite-specific excel configurations.",
        ),
    ]
    storage_path: Annotated[
        Optional[str],
        Field(
            description="Path to metadata storage",
        ),
    ] = None

excel_config: Annotated[XLSXParseConfig, Field(description='DLite-specific excel configurations.')] instance-attribute

id: Annotated[Optional[str], Field(description='Optional id on new instance.')] = None class-attribute instance-attribute

label: Annotated[Optional[str], Field(description='Optional label for new instance in collection.')] = 'excel-data' class-attribute instance-attribute

metadata: Annotated[Optional[HttpUrl], Field(description='URI of DLite metadata to return. If not provided, the metadata will be inferred from the excel file.')] = None class-attribute instance-attribute

storage_path: Annotated[Optional[str], Field(description='Path to metadata storage')] = None class-attribute instance-attribute

DLiteExcelParseResourceConfig

Bases: ResourceConfig

DLite excel parse strategy resource config.

Source code in oteapi_dlite/strategies/parse_excel.py 
67
68
69
70
71
72
73
class DLiteExcelParseResourceConfig(ResourceConfig):
    """DLite excel parse strategy resource config."""

    configuration: Annotated[
        DLiteExcelParseConfig,
        Field(description="DLite excel parse strategy-specific configuration."),
    ]

configuration: Annotated[DLiteExcelParseConfig, Field(description='DLite excel parse strategy-specific configuration.')] instance-attribute

DLiteExcelSessionUpdate

Bases: DLiteSessionUpdate

Class for returning values from DLite excel parser.

Source code in oteapi_dlite/strategies/parse_excel.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class DLiteExcelSessionUpdate(DLiteSessionUpdate):
    """Class for returning values from DLite excel parser."""

    inst_uuid: Annotated[
        str,
        Field(
            description="UUID of new instance.",
        ),
    ]
    label: Annotated[
        str,
        Field(
            description="Label of the new instance in the collection.",
        ),
    ]

inst_uuid: Annotated[str, Field(description='UUID of new instance.')] instance-attribute

label: Annotated[str, Field(description='Label of the new instance in the collection.')] instance-attribute

DLiteExcelStrategy

Parse strategy for Excel files.

Registers strategies:

  • ("mediaType", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
Source code in oteapi_dlite/strategies/parse_excel.py 
 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
@dataclass
class DLiteExcelStrategy:
    """Parse strategy for Excel files.

    **Registers strategies**:

    - `("mediaType",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")`

    """

    parse_config: DLiteExcelParseResourceConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteExcelSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific endpoint
        of the OTE-API Services.

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            DLite instance.

        """
        config = self.parse_config.configuration

        xlsx_config = self.parse_config.model_dump()
        xlsx_config["configuration"] = config.excel_config
        xlsx_config["mediaType"] = (
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
        )
        parser: IParseStrategy = XLSXParseStrategy(xlsx_config)
        columns: dict[str, Any] = parser.get(session)["data"]

        names, units = zip(*[split_column_name(column) for column in columns])
        rec = dict2recarray(columns, names=names)

        if not isinstance(units, (list, tuple)):
            # This check is to satisfy mypy for the `infer_metadata` call below.
            raise TypeError(
                f"units must be a list or tuple, instead it was {type(units)}"
            )

        if config.metadata:
            if config.storage_path is not None:
                for storage_path in config.storage_path.split("|"):
                    dlite.storage_path.append(storage_path)
            meta = dlite.get_instance(config.metadata)
            # check the metadata config would go here
        else:
            meta = infer_metadata(rec, units=units)

        inst = meta(dimensions=[len(rec)], id=config.id)
        for name in names:
            inst[name] = rec[name]

        # Insert inst into collection
        coll = get_collection(session)
        coll.add(config.label, inst)

        update_collection(coll)
        return DLiteExcelSessionUpdate(
            collection_id=coll.uuid,
            inst_uuid=inst.uuid,
            label=config.label,
        )

parse_config: DLiteExcelParseResourceConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services.

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteExcelSessionUpdate

DLite instance.
Source code in oteapi_dlite/strategies/parse_excel.py 
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
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteExcelSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific endpoint
    of the OTE-API Services.

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        DLite instance.

    """
    config = self.parse_config.configuration

    xlsx_config = self.parse_config.model_dump()
    xlsx_config["configuration"] = config.excel_config
    xlsx_config["mediaType"] = (
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    )
    parser: IParseStrategy = XLSXParseStrategy(xlsx_config)
    columns: dict[str, Any] = parser.get(session)["data"]

    names, units = zip(*[split_column_name(column) for column in columns])
    rec = dict2recarray(columns, names=names)

    if not isinstance(units, (list, tuple)):
        # This check is to satisfy mypy for the `infer_metadata` call below.
        raise TypeError(
            f"units must be a list or tuple, instead it was {type(units)}"
        )

    if config.metadata:
        if config.storage_path is not None:
            for storage_path in config.storage_path.split("|"):
                dlite.storage_path.append(storage_path)
        meta = dlite.get_instance(config.metadata)
        # check the metadata config would go here
    else:
        meta = infer_metadata(rec, units=units)

    inst = meta(dimensions=[len(rec)], id=config.id)
    for name in names:
        inst[name] = rec[name]

    # Insert inst into collection
    coll = get_collection(session)
    coll.add(config.label, inst)

    update_collection(coll)
    return DLiteExcelSessionUpdate(
        collection_id=coll.uuid,
        inst_uuid=inst.uuid,
        label=config.label,
    )

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/parse_excel.py 
106
107
108
109
110
111
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

infer_metadata(rec, units)

Infer dlite metadata from recarray rec.

Source code in oteapi_dlite/strategies/parse_excel.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def infer_metadata(rec: np.recarray, units: tuple[str, ...]) -> dlite.Instance:
    """Infer dlite metadata from recarray `rec`."""
    rnd = getrandbits(128)
    uri = f"http://onto-ns.com/meta/1.0/generated_from_excel_{rnd:0x}"
    metadata = DataModel(
        uri,
        description="Generated datamodel from excel file.",
    )
    metadata.add_dimension("nrows", "Number of rows.")
    for i, name in enumerate(rec.dtype.names):
        dtype = rec[name].dtype
        ptype = "string" if dtype.kind == "U" else dtype.name
        metadata.add_property(name, type=ptype, shape=["nrows"], unit=units[i])
    return metadata.get()

split_column_name(column)

Split column name into a (name, unit) tuple.

Source code in oteapi_dlite/strategies/parse_excel.py 
172
173
174
175
176
177
178
def split_column_name(column: str) -> tuple[str, str]:
    """Split column name into a (name, unit) tuple."""
    match = re.match(r"\s*([^ ([<]+)\s*[([<]?([^] )>]*)[])>]?", column)
    if not match:
        return column, ""
    name, unit = match.groups()
    return name, unit

parse_image

Strategy class for parsing an image to a DLite instance.

LOGGER = logging.getLogger('oteapi_dlite.strategies') module-attribute

DLiteImageConfig

Bases: ImageParserConfig

Configuration for DLite image parser.

Source code in oteapi_dlite/strategies/parse_image.py 
31
32
33
34
35
36
37
38
39
class DLiteImageConfig(ImageParserConfig):
    """Configuration for DLite image parser."""

    image_label: Annotated[
        str,
        Field(
            description="Label to assign to the image in the collection.",
        ),
    ] = "image"

image_label: Annotated[str, Field(description='Label to assign to the image in the collection.')] = 'image' class-attribute instance-attribute

DLiteImageParseStrategy

Parse strategy for image files.

Registers strategies:

  • ("mediaType", "image/vnd.dlite-gif")
  • ("mediaType", "image/vnd.dlite-jpeg")
  • ("mediaType", "image/vnd.dlite-jpg")
  • ("mediaType", "image/vnd.dlite-jp2")
  • ("mediaType", "image/vnd.dlite-png")
  • ("mediaType", "image/vnd.dlite-tiff")
Source code in oteapi_dlite/strategies/parse_image.py 
 53
 54
 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
@dataclass
class DLiteImageParseStrategy:
    """Parse strategy for image files.

    **Registers strategies**:

    - `("mediaType", "image/vnd.dlite-gif")`
    - `("mediaType", "image/vnd.dlite-jpeg")`
    - `("mediaType", "image/vnd.dlite-jpg")`
    - `("mediaType", "image/vnd.dlite-jp2")`
    - `("mediaType", "image/vnd.dlite-png")`
    - `("mediaType", "image/vnd.dlite-tiff")`

    """

    parse_config: DLiteImageResourceConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy.

        This method will be called through the strategy-specific
        endpoint of the OTE-API Services.  It assumes that the image to
        parse is stored in a data cache, and can be retrieved via a key
        that is supplied in either the session (highest priority)
        or in the parser configuration (lowest priority).

        Parameters:
            session: A session-specific dictionary context.

        Returns:
            DLite instance.
        """
        config = self.parse_config.configuration

        # Configuration for ImageDataParseStrategy in oteapi-core
        conf = self.parse_config.model_dump()
        conf["configuration"] = ImageParserConfig(
            **config.model_dump(), extra="ignore"
        )
        conf["mediaType"] = "image/" + conf["mediaType"].split("-")[-1]
        core_config = ImageParserResourceConfig(**conf)

        parse_strategy_session = ImageDataParseStrategy(core_config).initialize(
            session
        )
        output = ImageDataParseStrategy(core_config).get(parse_strategy_session)

        cache = DataCache()
        data = cache.get(output["image_key"])
        if isinstance(data, bytes):
            data = np.asarray(
                Image.frombytes(
                    data=data,
                    mode=output["image_mode"],
                    size=output["image_size"],
                )
            )
        if not isinstance(data, np.ndarray):
            raise TypeError(
                "Expected image data to be a numpy array, instead it was "
                f"{type(data)}."
            )

        meta = get_meta("http://onto-ns.com/meta/1.0/Image")
        inst = meta(dimensions=data.shape)
        inst["data"] = data

        LOGGER.info("session: %s", session)

        coll = get_collection(session)
        coll.add(config.image_label, inst)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

parse_config: DLiteImageResourceConfig instance-attribute

get(session=None)

Execute the strategy.

This method will be called through the strategy-specific endpoint of the OTE-API Services. It assumes that the image to parse is stored in a data cache, and can be retrieved via a key that is supplied in either the session (highest priority) or in the parser configuration (lowest priority).

Parameters:

Name Type Description Default
session Optional[dict[str, Any]]

A session-specific dictionary context.

 None

Returns:

Type Description
DLiteSessionUpdate

DLite instance.
Source code in oteapi_dlite/strategies/parse_image.py 
 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
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy.

    This method will be called through the strategy-specific
    endpoint of the OTE-API Services.  It assumes that the image to
    parse is stored in a data cache, and can be retrieved via a key
    that is supplied in either the session (highest priority)
    or in the parser configuration (lowest priority).

    Parameters:
        session: A session-specific dictionary context.

    Returns:
        DLite instance.
    """
    config = self.parse_config.configuration

    # Configuration for ImageDataParseStrategy in oteapi-core
    conf = self.parse_config.model_dump()
    conf["configuration"] = ImageParserConfig(
        **config.model_dump(), extra="ignore"
    )
    conf["mediaType"] = "image/" + conf["mediaType"].split("-")[-1]
    core_config = ImageParserResourceConfig(**conf)

    parse_strategy_session = ImageDataParseStrategy(core_config).initialize(
        session
    )
    output = ImageDataParseStrategy(core_config).get(parse_strategy_session)

    cache = DataCache()
    data = cache.get(output["image_key"])
    if isinstance(data, bytes):
        data = np.asarray(
            Image.frombytes(
                data=data,
                mode=output["image_mode"],
                size=output["image_size"],
            )
        )
    if not isinstance(data, np.ndarray):
        raise TypeError(
            "Expected image data to be a numpy array, instead it was "
            f"{type(data)}."
        )

    meta = get_meta("http://onto-ns.com/meta/1.0/Image")
    inst = meta(dimensions=data.shape)
    inst["data"] = data

    LOGGER.info("session: %s", session)

    coll = get_collection(session)
    coll.add(config.image_label, inst)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/parse_image.py 
70
71
72
73
74
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

DLiteImageResourceConfig

Bases: ResourceConfig

Resource config for DLite image parser.

Source code in oteapi_dlite/strategies/parse_image.py 
42
43
44
45
46
47
48
49
50
class DLiteImageResourceConfig(ResourceConfig):
    """Resource config for DLite image parser."""

    configuration: Annotated[
        DLiteImageConfig,
        Field(
            description="Image parse strategy-specific configuration.",
        ),
    ] = DLiteImageConfig()

configuration: Annotated[DLiteImageConfig, Field(description='Image parse strategy-specific configuration.')] = DLiteImageConfig() class-attribute instance-attribute

serialise

Filter for serialisation using DLite.

SerialiseConfig

Bases: AttrDict

DLite serialise-specific configurations.

Source code in oteapi_dlite/strategies/serialise.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
class SerialiseConfig(AttrDict):
    """DLite serialise-specific configurations."""

    driver: Annotated[
        str,
        Field(
            description="Name of DLite plugin used for serialisation.",
        ),
    ]
    location: Annotated[
        Path,
        Field(
            description="Path or URL to serialise to.",
        ),
    ]
    options: Annotated[
        Optional[str],
        Field(
            description="Options passed to the driver.",
        ),
    ] = ""
    labels: Annotated[
        Optional[Sequence[str]],
        Field(
            None,
            description=(
                "Optional sequence of labels in the collection to serialise.  "
                "The default is to serialise the entire collection."
            ),
        ),
    ] = None

driver: Annotated[str, Field(description='Name of DLite plugin used for serialisation.')] instance-attribute

labels: Annotated[Optional[Sequence[str]], Field(None, description='Optional sequence of labels in the collection to serialise. The default is to serialise the entire collection.')] = None class-attribute instance-attribute

location: Annotated[Path, Field(description='Path or URL to serialise to.')] instance-attribute

options: Annotated[Optional[str], Field(description='Options passed to the driver.')] = '' class-attribute instance-attribute

SerialiseFilterConfig

Bases: FilterConfig

Filter config for serialise.

Source code in oteapi_dlite/strategies/serialise.py 
54
55
56
57
58
59
60
61
62
class SerialiseFilterConfig(FilterConfig):
    """Filter config for serialise."""

    configuration: Annotated[
        SerialiseConfig,
        Field(
            description="Serialise-specific configurations.",
        ),
    ]

configuration: Annotated[SerialiseConfig, Field(description='Serialise-specific configurations.')] instance-attribute

SerialiseStrategy

Filter for serialisation using DLite.

Registers strategies:

  • ("filterType", "dlite_serialise")
Source code in oteapi_dlite/strategies/serialise.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
@dataclass
class SerialiseStrategy:
    """Filter for serialisation using DLite.

    **Registers strategies**:

    - `("filterType", "dlite_serialise")`

    """

    filter_config: SerialiseFilterConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy."""
        config = self.filter_config.configuration

        coll = get_collection(session)

        storage = dlite.Storage(
            driver_or_url=config.driver,
            location=str(config.location),
            options=config.options,
        )
        if config.labels is None:
            coll.save_to_storage(storage)
        else:
            for label in config.labels:
                inst = coll.get(label)
                inst.save_to_storage(storage)

        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

filter_config: SerialiseFilterConfig instance-attribute

get(session=None)

Execute the strategy.

Source code in oteapi_dlite/strategies/serialise.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy."""
    config = self.filter_config.configuration

    coll = get_collection(session)

    storage = dlite.Storage(
        driver_or_url=config.driver,
        location=str(config.location),
        options=config.options,
    )
    if config.labels is None:
        coll.save_to_storage(storage)
    else:
        for label in config.labels:
            inst = coll.get(label)
            inst.save_to_storage(storage)

    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/serialise.py 
77
78
79
80
81
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

settings

Generic strategy for adding configurations to the session.

JSONSerialisable = Union[dict, list, str, int, float, bool, NoneType] module-attribute

NoneType = type(None) module-attribute

SettingsConfig

Bases: AttrDict

Configuration for a generic "settings" filter.

This strategy stores settings in the session such that they are available for other strategies. For this to work, this strategy should be added to the end of the pipeline (since it uses the initiate() method).

The settings are stored as a JSON string which can be accessed by its label.

Source code in oteapi_dlite/strategies/settings.py 
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
53
54
55
class SettingsConfig(AttrDict):
    """Configuration for a generic "settings" filter.

    This strategy stores settings in the session such that they are
    available for other strategies.  For this to work, this strategy
    should be added to the end of the pipeline (since it uses the
    `initiate()` method).

    The settings are stored as a JSON string which can be accessed
    by its label.

    """

    label: Annotated[
        str,
        Field(
            description=(
                "Label for accessing this configuration.  "
                "It should be unique."
            ),
        ),
    ]
    settings: Annotated[
        JSONSerialisable,
        Field(
            description=(
                "The configurations to be stored, represented as a Python "
                "object that can be serialised to JSON."
            ),
        ),
    ]

label: Annotated[str, Field(description='Label for accessing this configuration. It should be unique.')] instance-attribute

settings: Annotated[JSONSerialisable, Field(description='The configurations to be stored, represented as a Python object that can be serialised to JSON.')] instance-attribute

SettingsFilterConfig

Bases: FilterConfig

Settings strategy config.

Source code in oteapi_dlite/strategies/settings.py 
58
59
60
61
62
63
64
class SettingsFilterConfig(FilterConfig):
    """Settings strategy config."""

    configuration: Annotated[
        SettingsConfig,
        Field(description="Settings strategy-specific configuration."),
    ]

configuration: Annotated[SettingsConfig, Field(description='Settings strategy-specific configuration.')] instance-attribute

SettingsStrategy

Generic settings strategy for storing settings for other strategies.

Registers strategies:

  • ("mediaType", "application/vnd.dlite-settings")
Source code in oteapi_dlite/strategies/settings.py 
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
@dataclass
class SettingsStrategy:
    """Generic settings strategy for storing settings for other
    strategies.

    **Registers strategies**:

    - `("mediaType", "application/vnd.dlite-settings")`

    """

    settings_config: SettingsFilterConfig

    def initialize(
        self,
        session: Optional[dict[str, Any]] = None,
    ) -> SessionUpdate:
        """Store settings."""
        config = self.settings_config.configuration
        return add_settings(session, config.label, config.settings)

    def get(
        self, session: Optional[dict[str, Any]] = None  # noqa: ARG002
    ) -> SessionUpdate:
        """Do nothing."""
        return SessionUpdate()

settings_config: SettingsFilterConfig instance-attribute

get(session=None)

Do nothing.

Source code in oteapi_dlite/strategies/settings.py 
88
89
90
91
92
def get(
    self, session: Optional[dict[str, Any]] = None  # noqa: ARG002
) -> SessionUpdate:
    """Do nothing."""
    return SessionUpdate()

initialize(session=None)

Store settings.

Source code in oteapi_dlite/strategies/settings.py 
80
81
82
83
84
85
86
def initialize(
    self,
    session: Optional[dict[str, Any]] = None,
) -> SessionUpdate:
    """Store settings."""
    config = self.settings_config.configuration
    return add_settings(session, config.label, config.settings)