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 
120
121
122
123
124
125
126
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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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 
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
@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)

        coll = get_collection(session)

        instances = []
        for i, input_config in enumerate(config.inputs):
            input_config = config.inputs[i]
            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...
                )
            else:
                raise ValueError(
                    "either `label` or `datamodel` must be specified in "
                    f"inputs[{i}]"
                )

        outputs = function(*instances)
        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 
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
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)

    coll = get_collection(session)

    instances = []
    for i, input_config in enumerate(config.inputs):
        input_config = config.inputs[i]
        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...
            )
        else:
            raise ValueError(
                "either `label` or `datamodel` must be specified in "
                f"inputs[{i}]"
            )

    outputs = function(*instances)
    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 
142
143
144
145
146
147
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 
 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
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.",
        ),
    ] = []

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

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 
81
82
83
84
85
86
87
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 
 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
@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."""
        # pylint: disable=too-many-branches
        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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def get(
    self, session: "Optional[dict[str, Any]]" = None
) -> DLiteSessionUpdate:
    """Execute the strategy."""
    # pylint: disable=too-many-branches
    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 
105
106
107
108
109
110
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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
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.

DLiteGenerateConfig

Bases: FunctionConfig

DLite generate strategy config.

Source code in oteapi_dlite/strategies/generate.py 
115
116
117
118
119
120
121
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 
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
@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

        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:
            if cacheconfig and cacheconfig.accessKey:
                key = cacheconfig.accessKey
            elif "key" in session:  # type: ignore
                key = "generate_data"

            cache = DataCache()
            with tempfile.TemporaryDirectory() as tmpdir:
                inst.save(driver, "{tmpdir}/data", config.options)
                with open(f"{tmpdir}/data", "rb") as f:
                    cache.add(f.read(), key=key)

        # __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 
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
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

    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:
        if cacheconfig and cacheconfig.accessKey:
            key = cacheconfig.accessKey
        elif "key" in session:  # type: ignore
            key = "generate_data"

        cache = DataCache()
        with tempfile.TemporaryDirectory() as tmpdir:
            inst.save(driver, "{tmpdir}/data", config.options)
            with open(f"{tmpdir}/data", "rb") as f:
                cache.add(f.read(), key=key)

    # __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 
136
137
138
139
140
141
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 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 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
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

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

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

mapping

Mapping filter strategy.

DLiteMappingConfig

Bases: MappingConfig

DLite mapping strategy config.

Source code in oteapi_dlite/strategies/mapping.py 
29
30
31
32
33
34
35
36
37
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 
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
@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."""
        coll = get_collection(session)
        ts = Triplestore(backend="collection", collection=coll)

        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  # pylint: disable=not-an-iterable
                ]
            )

        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 
77
78
79
80
81
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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def initialize(
    self, session: Optional[dict[str, "Any"]] = None
) -> DLiteSessionUpdate:
    """Initialize strategy."""
    coll = get_collection(session)
    ts = Triplestore(backend="collection", collection=coll)

    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  # pylint: disable=not-an-iterable
            ]
        )

    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 
18
19
20
21
22
23
24
25
26
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 
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
class DLiteParseConfig(AttrDict):
    """Configuration for generic DLite parser."""

    driver: Annotated[
        Optional[str],
        Field(
            description='Name of DLite driver (ex: "json").',
        ),
    ] = None
    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[
        str,
        Field(
            description="Label of the new DLite instance in the collection.",
        ),
    ]
    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

driver: Annotated[Optional[str], Field(description='Name of DLite driver (ex: "json").')] = None class-attribute 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[str, Field(description='Label of the new DLite instance in the collection.')] 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 
68
69
70
71
72
73
74
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 
 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
@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)
        coll.add(config.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 
 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
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)
    coll.add(config.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 
89
90
91
92
93
94
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 
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
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 
66
67
68
69
70
71
72
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 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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 
 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
@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.keys()]
        )
        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 
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
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.keys()]
    )
    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 
105
106
107
108
109
110
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 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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 
173
174
175
176
177
178
179
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 
29
30
31
32
33
34
35
36
37
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 
 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
@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 
 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
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 
68
69
70
71
72
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 
40
41
42
43
44
45
46
47
48
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 
52
53
54
55
56
57
58
59
60
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 
 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
@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 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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 
75
76
77
78
79
def initialize(
    self, session: Optional[dict[str, "Any"]] = None
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)