Skip to content

Utilities

LUME-torch provides utility functions for working with variables, paths, imports, and model data.

Variable Utilities

Functions for serializing, deserializing, and managing variables.

Serialization

lume_torch.utils.variables_as_yaml(input_variables, output_variables, file=None)

Returns and optionally saves YAML formatted string defining the in- and output variables.

Parameters

input_variables : list of ScalarVariable List of input variables. output_variables : list of ScalarVariable List of output variables. file : str or os.PathLike, optional If not None, YAML formatted string is saved to given file path.

Returns

str YAML formatted string defining the in- and output variables.

Source code in lume_torch/utils.py 
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
def variables_as_yaml(
    input_variables: list[ScalarVariable],
    output_variables: list[ScalarVariable],
    file: Union[str, os.PathLike] = None,
) -> str:
    """Returns and optionally saves YAML formatted string defining the in- and output variables.

    Parameters
    ----------
    input_variables : list of ScalarVariable
        List of input variables.
    output_variables : list of ScalarVariable
        List of output variables.
    file : str or os.PathLike, optional
        If not None, YAML formatted string is saved to given file path.

    Returns
    -------
    str
        YAML formatted string defining the in- and output variables.

    """
    logger.debug(
        f"Creating YAML for {len(input_variables)} input and {len(output_variables)} output variables"
    )
    for variables in [input_variables, output_variables]:
        verify_unique_variable_names(variables)
    v = {
        "input_variables": [var.model_dump() for var in input_variables],
        "output_variables": [var.model_dump() for var in output_variables],
    }
    s = yaml.safe_dump(serialize_variables(v), default_flow_style=None, sort_keys=False)
    if file is not None:
        logger.info(f"Saving variables to file: {file}")
        with open(file, "w") as f:
            f.write(s)
    return s

lume_torch.utils.serialize_variables(v)

Performs custom serialization for in- and output variables.

Parameters

v : dict Object to serialize.

Returns

dict Dictionary with serialized in- and output variables.

Source code in lume_torch/utils.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
def serialize_variables(v: dict):
    """Performs custom serialization for in- and output variables.

    Parameters
    ----------
    v : dict
        Object to serialize.

    Returns
    -------
    dict
        Dictionary with serialized in- and output variables.

    """
    logger.debug("Serializing variables")
    for key, value in v.items():
        if key in ["input_variables", "output_variables"] and isinstance(value, list):
            # Call model_dump with mode='python' to ensure proper serialization
            serialized_vars = [
                var.model_dump(mode="python") if hasattr(var, "model_dump") else var
                for var in value
            ]
            v[key] = {
                var_dict["name"]: {
                    var_k: var_v
                    for var_k, var_v in var_dict.items()
                    if not (var_k == "name" or var_v is None)
                }
                for var_dict in serialized_vars
            }
    return v

Deserialization

lume_torch.utils.variables_from_yaml(yaml_obj)

Parses YAML object and returns in- and output variable lists.

Parameters

yaml_obj : str or os.PathLike YAML formatted string or file path.

Returns

tuple of (list of ScalarVariable, list of ScalarVariable) In- and output variable lists.

Source code in lume_torch/utils.py 
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
def variables_from_yaml(
    yaml_obj: Union[str, os.PathLike],
) -> tuple[list[ScalarVariable], list[ScalarVariable]]:
    """Parses YAML object and returns in- and output variable lists.

    Parameters
    ----------
    yaml_obj : str or os.PathLike
        YAML formatted string or file path.

    Returns
    -------
    tuple of (list of ScalarVariable, list of ScalarVariable)
        In- and output variable lists.

    """
    if os.path.exists(yaml_obj):
        logger.debug(f"Loading variables from file: {yaml_obj}")
        with open(yaml_obj) as f:
            yaml_str = f.read()
    else:
        logger.debug("Parsing variables from YAML string")
        yaml_str = yaml_obj
    config = deserialize_variables(yaml.safe_load(yaml_str))
    return variables_from_dict(config)

lume_torch.utils.variables_from_dict(config)

Parses given config and returns in- and output variable lists.

Parameters

config : dict Variable configuration.

Returns

tuple of (list of ScalarVariable, list of ScalarVariable) In- and output variable lists.

Source code in lume_torch/utils.py 
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
def variables_from_dict(
    config: dict,
) -> tuple[list[ScalarVariable], list[ScalarVariable]]:
    """Parses given config and returns in- and output variable lists.

    Parameters
    ----------
    config : dict
        Variable configuration.

    Returns
    -------
    tuple of (list of ScalarVariable, list of ScalarVariable)
        In- and output variable lists.

    """
    logger.debug("Parsing variables from dictionary")
    input_variables, output_variables = [], []
    for key, value in {**config}.items():
        if key in ["input_variables", "output_variables"]:
            for var in value:
                variable_class = get_variable(var["variable_class"])
                if key == "input_variables":
                    input_variables.append(variable_class(**var))
                elif key == "output_variables":
                    output_variables.append(variable_class(**var))
    for variables in [input_variables, output_variables]:
        verify_unique_variable_names(variables)
    logger.debug(
        f"Parsed {len(input_variables)} input and {len(output_variables)} output variables"
    )
    return input_variables, output_variables

lume_torch.utils.deserialize_variables(v)

Performs custom deserialization for in- and output variables.

Parameters

v : dict Object to deserialize.

Returns

dict Dictionary with deserialized in- and output variables.

Source code in lume_torch/utils.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def deserialize_variables(v):
    """Performs custom deserialization for in- and output variables.

    Parameters
    ----------
    v : dict
        Object to deserialize.

    Returns
    -------
    dict
        Dictionary with deserialized in- and output variables.

    """
    logger.debug("Deserializing variables")
    for key, value in v.items():
        if key in ["input_variables", "output_variables"] and isinstance(value, dict):
            v[key] = [
                var_dict | {"name": var_name} for var_name, var_dict in value.items()
            ]
    return v

Validation

lume_torch.utils.verify_unique_variable_names(variables)

Verifies that variable names are unique.

Raises a ValueError if any reoccurring variable names are found.

Parameters

variables : list of ScalarVariable List of scalar variables.

Raises

ValueError If any variable names are not unique.

Source code in lume_torch/utils.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def verify_unique_variable_names(variables: list[ScalarVariable]):
    """Verifies that variable names are unique.

    Raises a ValueError if any reoccurring variable names are found.

    Parameters
    ----------
    variables : list of ScalarVariable
        List of scalar variables.

    Raises
    ------
    ValueError
        If any variable names are not unique.

    """
    names = [var.name for var in variables]
    non_unique_names = [name for name in set(names) if names.count(name) > 1]
    if non_unique_names:
        logger.error(f"Variable names {non_unique_names} are not unique")
        raise ValueError(f"Variable names {non_unique_names} are not unique.")

Path Utilities

Functions for handling file paths and path resolution.

lume_torch.utils.get_valid_path(path, directory='')

Validates path exists either as relative or absolute path and returns the first valid option.

Parameters

path : str or os.PathLike Path to validate. directory : str or os.PathLike, optional Directory against which relative paths are checked.

Returns

str or os.PathLike The first valid path option as an absolute path.

Raises

OSError If file is not found.

Source code in lume_torch/utils.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def get_valid_path(
    path: Union[str, os.PathLike],
    directory: Union[str, os.PathLike] = "",
) -> Union[str, os.PathLike]:
    """Validates path exists either as relative or absolute path and returns the first valid option.

    Parameters
    ----------
    path : str or os.PathLike
        Path to validate.
    directory : str or os.PathLike, optional
        Directory against which relative paths are checked.

    Returns
    -------
    str or os.PathLike
        The first valid path option as an absolute path.

    Raises
    ------
    OSError
        If file is not found.

    """
    relative_path = os.path.join(directory, path)
    if os.path.exists(relative_path):
        logger.debug(f"Found relative path: {relative_path}")
        return os.path.abspath(relative_path)
    elif os.path.exists(path):
        logger.debug(f"Found absolute path: {path}")
        return os.path.abspath(path)
    else:
        logger.error(
            f"File {path} not found in directory {directory} or as absolute path"
        )
        raise OSError(f"File {path} is not found.")

lume_torch.utils.replace_relative_paths(d, model_fields=None, directory='')

Replaces dictionary entries with absolute paths where the model field annotation is not string or path-like.

Parameters

d : dict Dictionary to process. model_fields : dict, optional Model fields dictionary used to check expected type. directory : str or os.PathLike, optional Directory against which relative paths are checked.

Returns

dict Dictionary with replaced paths.

Source code in lume_torch/utils.py 
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
def replace_relative_paths(
    d: dict,
    model_fields: dict = None,
    directory: Union[str, os.PathLike] = "",
) -> dict:
    """Replaces dictionary entries with absolute paths where the model field annotation is not string or path-like.

    Parameters
    ----------
    d : dict
        Dictionary to process.
    model_fields : dict, optional
        Model fields dictionary used to check expected type.
    directory : str or os.PathLike, optional
        Directory against which relative paths are checked.

    Returns
    -------
    dict
        Dictionary with replaced paths.

    """
    logger.debug(f"Replacing relative paths with directory base: {directory}")
    if model_fields is None:
        model_fields = {}
    for k, v in d.items():
        if isinstance(v, (str, os.PathLike)):
            if k in model_fields.keys():
                field_types = [model_fields[k].annotation]
                if get_origin(model_fields[k].annotation) is Union:
                    field_types = list(get_args(model_fields[k].annotation))
                if all([t not in field_types for t in [str, os.PathLike]]):
                    d[k] = get_valid_path(v, directory)
        elif isinstance(v, list):
            if k in model_fields.keys():
                field_types = []
                for i, field_type in enumerate(get_args(model_fields[k].annotation)):
                    if get_origin(field_type) is Union:
                        field_types.extend(list(get_args(field_type)))
                    else:
                        field_types.append(field_type)
                for i, ele in enumerate(v):
                    if isinstance(ele, (str, os.PathLike)) and all(
                        [t not in field_types for t in [str, os.PathLike]]
                    ):
                        v[i] = get_valid_path(ele, directory)
        elif isinstance(v, dict):
            model_subfields = {
                ".".join(key.split(".")[1:]): value
                for key, value in model_fields.items()
                if key.startswith(f"{k}.")
            }
            d[k] = replace_relative_paths(v, model_subfields, directory)
    return d

Import Utilities

lume_torch.utils.try_import_module(name)

Tries to import module if required.

Parameters

name : str Module name.

Returns

module or None Imported module if successful, None otherwise.

Source code in lume_torch/utils.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def try_import_module(name: str):
    """Tries to import module if required.

    Parameters
    ----------
    name : str
        Module name.

    Returns
    -------
    module or None
        Imported module if successful, None otherwise.

    """
    if name not in sys.modules:
        try:
            module = importlib.import_module(name)
            logger.debug(f"Successfully imported module: {name}")
        except ImportError as e:
            logger.debug(f"Failed to import module {name}: {e}")
            module = None
    else:
        module = sys.modules[name]
        logger.debug(f"Module {name} already in sys.modules")
    return module

Model Utilities

Utilities for working with model inputs and outputs.

lume_torch.models.utils.itemize_dict(d)

Itemizes the given in-/output dictionary.

Parameters

d : dict of str to float, torch.Tensor, or Distribution Dictionary to itemize.

Returns

list of dict List of in-/output dictionaries, each containing only a single value per in-/output.

Source code in lume_torch/models/utils.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def itemize_dict(
    d: dict[str, Union[float, torch.Tensor, Distribution]],
) -> list[dict[str, Union[float, torch.Tensor]]]:
    """Itemizes the given in-/output dictionary.

    Parameters
    ----------
    d : dict of str to float, torch.Tensor, or Distribution
        Dictionary to itemize.

    Returns
    -------
    list of dict
        List of in-/output dictionaries, each containing only a single value per in-/output.

    """
    has_tensors = any([isinstance(value, torch.Tensor) for value in d.values()])
    itemized_dicts = []
    if has_tensors:
        for k, v in d.items():
            for i, ele in enumerate(v.flatten()):
                if i >= len(itemized_dicts):
                    itemized_dicts.append({k: ele.item()})
                else:
                    itemized_dicts[i][k] = ele.item()
    else:
        itemized_dicts = [d]
    return itemized_dicts

lume_torch.models.utils.format_inputs(input_dict)

Formats values of the input dictionary as tensors.

Parameters

input_dict : dict of str to float or torch.Tensor Dictionary of input variable names to values.

Returns

dict of str to torch.Tensor Dictionary of input variable names to tensors.

Source code in lume_torch/models/utils.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def format_inputs(
    input_dict: dict[str, Union[float, torch.Tensor]],
) -> dict[str, torch.Tensor]:
    """Formats values of the input dictionary as tensors.

    Parameters
    ----------
    input_dict : dict of str to float or torch.Tensor
        Dictionary of input variable names to values.

    Returns
    -------
    dict of str to torch.Tensor
        Dictionary of input variable names to tensors.

    """
    formatted_inputs = {}
    for var_name, value in input_dict.items():
        v = value if isinstance(value, torch.Tensor) else torch.tensor(value)
        formatted_inputs[var_name] = v
    return formatted_inputs

lume_torch.models.utils.InputDictModel

Bases: BaseModel

Pydantic model for input dictionary validation.

Attributes

input_dict : dict of str to torch.Tensor or float Input dictionary to validate.

Source code in lume_torch/models/utils.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
class InputDictModel(BaseModel):
    """Pydantic model for input dictionary validation.

    Attributes
    ----------
    input_dict : dict of str to torch.Tensor or float
        Input dictionary to validate.

    """

    input_dict: Dict[str, Union[torch.Tensor, float]]

    model_config = ConfigDict(arbitrary_types_allowed=True, strict=True)

Usage Examples

Working with Variables

from lume_torch.variables import ScalarVariable
from lume_torch.utils import (
    variables_as_yaml,
    variables_from_yaml,
    serialize_variables,
    deserialize_variables
)

# Create variables
variables = [
    ScalarVariable(name="x", value_range=[0, 10]),
    ScalarVariable(name="y", value_range=[0, 10]),
]

# Serialize to YAML string
yaml_str = variables_as_yaml(variables)
print(yaml_str)

# Deserialize from YAML
loaded_vars = variables_from_yaml(yaml_str)

# Serialize to dict
var_dict = serialize_variables(variables)

# Deserialize from dict
restored_vars = deserialize_variables(var_dict)

Path Resolution

from lume_torch.utils import get_valid_path, replace_relative_paths
import os

# Get valid path (resolves relative paths)
config_path = get_valid_path("models/model.yml")

# Replace relative paths in a dictionary
config = {
    "model": "model.pt",
    "transformers": ["transform1.pt", "transform2.pt"]
}

# Convert relative to absolute paths
base_path = "/path/to/models"
absolute_config = replace_relative_paths(config, base_path)

Dynamic Imports

from lume_torch.utils import try_import_module

# Safely import optional dependencies
mlflow = try_import_module("mlflow")
if mlflow is not None:
    # Use mlflow
    mlflow.log_metric("loss", 0.05)
else:
    print("MLflow not installed")

# Import custom model class
model_module = try_import_module("my_models.custom_model")
if model_module:
    ModelClass = getattr(model_module, "CustomModel")

Itemizing Dictionaries

from lume_torch.models.utils import itemize_dict
import torch

# Batched input dictionary
batch_input = {
    "x": torch.tensor([1.0, 2.0, 3.0]),
    "y": torch.tensor([4.0, 5.0, 6.0])
}

# Convert to list of individual dictionaries
individual_inputs = itemize_dict(batch_input)
# Result: [{"x": 1.0, "y": 4.0}, {"x": 2.0, "y": 5.0}, {"x": 3.0, "y": 6.0}]

Formatting Inputs

from lume_torch.models.utils import format_inputs
import torch

# Convert dictionary to tensor
input_dict = {"x": 1.0, "y": 2.0}
input_order = ["x", "y"]

input_tensor = format_inputs(input_dict, input_order)
# Result: torch.tensor([[1.0, 2.0]])

See Also

  • Variables - Variable types and usage
  • Models - Using utilities with models