Coverage for physioblocks / configuration / functions.py: 99%
183 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-09 16:40 +0100
1# SPDX-FileCopyrightText: Copyright INRIA
2#
3# SPDX-License-Identifier: LGPL-3.0-only
4#
5# Copyright INRIA
6#
7# This file is part of PhysioBlocks, a library mostly developed by the
8# [Ananke project-team](https://team.inria.fr/ananke) at INRIA.
9#
10# Authors:
11# - Colin Drieu
12# - Dominique Chapelle
13# - François Kimmig
14# - Philippe Moireau
15#
16# PhysioBlocks is free software: you can redistribute it and/or modify it under the
17# terms of the GNU Lesser General Public License as published by the Free Software
18# Foundation, version 3 of the License.
19#
20# PhysioBlocks is distributed in the hope that it will be useful, but WITHOUT ANY
21# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
22# PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
23#
24# You should have received a copy of the GNU Lesser General Public License along with
25# PhysioBlocks. If not, see <https://www.gnu.org/licenses/>.
27"""Declares functions to load and save PhysioBlocks object to generic
28:class:`~physioblocks.configuration.base.Configuration` objects.
30Before using generic :func:`load` and :func:`save` functions on a PhysioBlocks object,
31the object type must be registered with the
32:func:`~physioblocks.registers.type_register.register_type` decorator.
34To define a specific behavior when saving or loading an registered object with the
35generic :func:`load` or :func:`save` functions, declare a function decorated with
36:func:`~physioblocks.registers.load_function_register.loads` or
37:func:`~physioblocks.registers.save_function_register.saves`.
39.. note::
41 If you want to create a **Configurable Item** for a dataclass type,
42 you will not have to register a specific save or load function.
44 Registering the type the :func:`~physioblocks.registers.type_register.register_type`
45 decorator will suffice to create a configuration item that needs the same parameters
46 as the dataclass.
48See :doc:`register module <./registers>` to for decorators documentation to
49:func:`~physioblocks.registers.type_register.register_type` as well as
50:func:`~physioblocks.registers.load_function_register.loads` and
51:func:`~physioblocks.registers.save_function_register.saves`.
52"""
54import functools
55from collections.abc import Iterable, Mapping, Sequence
56from inspect import signature
57from typing import Any, TypeAlias, TypeVar
59import numpy as np
60from numpy.typing import NDArray
62from physioblocks.configuration.base import Configuration, ConfigurationError
63from physioblocks.registers.load_function_register import get_load_function, loads
64from physioblocks.registers.save_function_register import (
65 get_save_function,
66 has_save_function,
67)
68from physioblocks.registers.type_register import (
69 get_registered_type,
70 get_registered_type_id,
71 is_registered,
72)
74BaseTypes: TypeAlias = float | int | bool | str
75"""Type alias for basic types usable in a Configuration File"""
78def load(
79 configuration: Any,
80 configuration_key: str | None = None,
81 configuration_object: Any | None = None,
82 configuration_type: type[Any] | None = None,
83 configuration_references: dict[str, Any] | None = None,
84 configuration_sort: bool = False,
85) -> Any:
86 """
87 Generic load an object from the given configuration item.
89 The method can load:
90 - :class:`~physioblocks.configuration.base.Configuration`: Use the matching
91 registered load function.
92 - `dict` and `list`: recursivly load values in the collection
94 :param configuration: the configuration to load
95 :type configuration: Any
97 :param configuration_key: (optional) key of the configuration in the parent
98 configuration item.
99 :type configuration_key: str
101 :param configuration_object: (optional) The object to configure.
102 If empty, a the object is first instanciated then configured.
103 :type configuration_object: Any
105 :param configuration_type: (optional) the type of the object to configure.
106 If empty, it is determined from the configuration object.
107 :type configuration_type: Any
109 :param configuration_references: (optional) mapping of configuration item keys
110 with already configured objects to use in the current configured object.
111 :type configuration_references: dict[str, Any]
113 :param configuration_sort: (optional) flag to signal that configuration items
114 should be sorted be sorted before they are loaded. Default is False.
115 :type configuration_sort: dict[str, Any]
117 :return: the configured object
118 :rtype: Any
119 """
120 if (
121 isinstance(configuration, str)
122 and configuration_references is not None
123 and configuration in configuration_references
124 ):
125 # the value is already in the references:
126 return (
127 configuration_references[configuration]
128 if configuration_type is None
129 else configuration_type(configuration_references[configuration])
130 )
132 elif configuration_type is not None:
133 load_func = get_load_function(configuration_type)
134 return load_func(
135 configuration,
136 configuration_key=configuration_key,
137 configuration_object=configuration_object,
138 configuration_type=configuration_type,
139 configuration_references=configuration_references,
140 configuration_sort=configuration_sort,
141 )
142 elif isinstance(configuration, BaseTypes):
143 # No load function required
144 return configuration
146 elif isinstance(configuration, Configuration):
147 return load_configuration(
148 configuration,
149 configuration_key=configuration_key,
150 configuration_object=configuration_object,
151 configuration_type=configuration_type,
152 configuration_references=configuration_references,
153 configuration_sort=configuration_sort,
154 )
155 elif isinstance(configuration, Mapping):
156 return load_dict(
157 configuration,
158 configuration_key=configuration_key,
159 configuration_object=configuration_object,
160 configuration_type=configuration_type,
161 configuration_references=configuration_references,
162 configuration_sort=configuration_sort,
163 )
164 elif isinstance(configuration, Sequence):
165 return load_list(
166 configuration,
167 configuration_key=configuration_key,
168 configuration_object=configuration_object,
169 configuration_type=configuration_type,
170 configuration_references=configuration_references,
171 configuration_sort=configuration_sort,
172 )
174 raise TypeError(
175 str.format(
176 "Type {0} can not be loaded as a configuration.",
177 type(configuration).__name__,
178 )
179 )
182def load_configuration(
183 configuration: Configuration,
184 configuration_key: str | None = None,
185 configuration_object: Any | None = None,
186 configuration_references: dict[str, Any] | None = None,
187 configuration_sort: bool = False,
188 *args: Any,
189 **kwargs: Any,
190) -> Any:
191 """
192 Specific load function for a
193 :class:`~physioblocks.configuration.base.Configuration`: configuration item.
195 It recursivly loads any configuration item used in the ``configuration`` parameter.
197 :param configuration: the configuration item to load
198 :type configuration: Configuration
200 :param configuration_key: (optional) key of the configuration in the parent
201 configuration item.
202 :type configuration_key: str
204 :param configuration_object: (optional) The object to configure.
205 If empty, a the object is first instanciated then configured.
206 :type configuration_object: Any
208 :param configuration_type: (optional) the type of the object to configure.
209 If empty, it is determined from the configuration object.
210 :type configuration_type: Any
212 :param configuration_references: (optional) mapping of configuration item keys
213 with already configured objects to use in the current configured object.
214 :type configuration_references: dict[str, Any]
216 :param configuration_sort: (optional) flag to signal that configuration items
217 should be sorted be sorted before they are loaded. Default is False.
218 :type configuration_sort: dict[str, Any]
220 :return: the configured object
221 :rtype: Any
222 """
223 configuration = (
224 configuration
225 if configuration_sort is False
226 else __sort_configuration(configuration)
227 )
229 new_configuration_type = get_registered_type(configuration.label)
230 load_func = get_load_function(new_configuration_type)
232 return load_func(
233 configuration,
234 configuration_key=configuration_key,
235 configuration_object=configuration_object,
236 configuration_type=new_configuration_type,
237 configuration_references=configuration_references,
238 configuration_sort=configuration_sort,
239 )
242def load_dict(
243 configuration: Mapping[str, Any],
244 configuration_object: dict[str, Any] | None = None,
245 configuration_references: dict[str, Any] | None = None,
246 configuration_sort: bool = False,
247 *args: Any,
248 **kwargs: Any,
249) -> dict[str, Any]:
250 """
251 Specific load function for a `Mapping` configuration item.
253 It recursivly loads any configuration item used in the mapping values.
255 :param configuration: the configuration item to load
256 :type configuration: Configuration
258 :param configuration_key: (optional) key of the configuration in the parent
259 configuration item.
260 :type configuration_key: str
262 :param configuration_object: (optional) The object to configure.
263 If empty, a the object is first instanciated then configured.
264 :type configuration_object: Any
266 :param configuration_type: (optional) the type of the object to configure.
267 If empty, it is determined from the configuration object.
268 :type configuration_type: Any
270 :param configuration_references: (optional) mapping of configuration item keys
271 with already configured objects to use in the current configured object.
272 :type configuration_references: dict[str, Any]
274 :param configuration_sort: (optional) flag to signal that configuration items
275 should be sorted be sorted before they are loaded. Default is False.
276 :type configuration_sort: dict[str, Any]
278 :return: the configured object
279 :rtype: Any
280 """
281 configuration = (
282 configuration
283 if configuration_sort is False
284 else __sort_configuration(configuration)
285 )
287 if configuration_object is None:
288 configuration_object = {}
290 updated_references = (
291 configuration_references.copy() if configuration_references is not None else {}
292 )
293 configuration_values = {}
295 for key, value in configuration.items():
296 loaded_obj = load(
297 value,
298 configuration_key=key,
299 configuration_object=configuration_object.get(key),
300 configuration_type=type(configuration_object.get(key))
301 if configuration_object.get(key) is not None
302 and not isinstance(configuration_object.get(key), BaseTypes)
303 else None,
304 configuration_references=updated_references,
305 configuration_sort=configuration_sort,
306 )
307 configuration_values[key] = loaded_obj
308 updated_references[key] = loaded_obj
309 if isinstance(loaded_obj, Mapping):
310 updated_references.update(loaded_obj)
312 configuration_object.update(configuration_values)
314 return configuration_object
317def load_list(
318 configuration: Sequence[Any],
319 configuration_object: list[Any] | None = None,
320 configuration_references: dict[str, Any] | None = None,
321 configuration_sort: bool = False,
322 *args: Any,
323 **kwargs: Any,
324) -> Sequence[Any]:
325 """
326 Specific load function for a `Sequence` configuration item.
328 It recursivly loads any configuration item used in the sequence values.
330 :param configuration: the configuration item to load
331 :type configuration: Configuration
333 :param configuration_key: (optional) key of the configuration in the parent
334 configuration item.
335 :type configuration_key: str
337 :param configuration_object: (optional) The object to configure.
338 If empty, a the object is first instanciated then configured.
339 :type configuration_object: Any
341 :param configuration_type: (optional) the type of the object to configure.
342 If empty, it is determined from the configuration object.
343 :type configuration_type: Any
345 :param configuration_references: (optional) mapping of configuration item keys
346 with already configured objects to use in the current configured object.
347 :type configuration_references: dict[str, Any]
349 :param configuration_sort: (optional) flag to signal that configuration items
350 should be sorted be sorted before they are loaded. Default is False.
351 :type configuration_sort: dict[str, Any]
353 :return: the configured object
354 :rtype: Any
355 """
357 if configuration_object is None:
358 configuration_object = []
360 configuration_values = [
361 load(
362 configuration[index],
363 configuration_object=configuration_object[index]
364 if index < len(configuration_object)
365 else None,
366 configuration_type=type(configuration_object[index])
367 if index < len(configuration_object)
368 and not isinstance(configuration[index], BaseTypes)
369 else None,
370 configuration_references=configuration_references,
371 configuration_sort=configuration_sort,
372 )
373 for index in range(0, len(configuration))
374 ]
376 return configuration_values
379@loads(bool)
380def _bool_load(
381 configuration: Any,
382 *args: Any,
383 **kwargs: Any,
384) -> bool:
385 if isinstance(configuration, str):
386 return configuration.lower() == str(True)
387 return bool(configuration)
390T = TypeVar("T")
393@loads(object)
394def _base_load(
395 configuration: Any,
396 configuration_key: str | None = None,
397 configuration_object: T | None = None,
398 configuration_type: type[T] | None = None,
399 configuration_references: dict[str, Any] | None = None,
400 configuration_sort: bool = False,
401 *args: Any,
402 **kwargs: Any,
403) -> T:
404 """
405 Load function called when no other load function is defined.
407 :param configuration: the configuration to load
408 :type configuration: Configuration
410 :param configuration_object: The object to configure.
411 If empty, a new object is created and configured.
412 :type configuration_object: Any
414 :return: the configured object
415 :rtype: Any
416 """
417 if configuration_type is None and configuration_object is not None:
418 configuration_type = type(configuration_object)
419 elif configuration_type is None:
420 raise ConfigurationError(
421 str.format("Missing configuration type for {0}:{1}.", configuration_key)
422 )
424 config_args: list[Any] = []
425 config_kwargs: dict[str, Any] = {}
427 if isinstance(configuration, Configuration | Mapping):
428 config_kwargs.update(configuration)
429 elif isinstance(configuration, str | float | int | bool):
430 config_args.append(configuration)
431 elif isinstance(configuration, Sequence):
432 config_args.extend(configuration)
434 # load the values in the provided arguments
435 config_args = load(
436 config_args,
437 configuration_key=configuration_key,
438 configuration_references=configuration_references,
439 configuration_sort=configuration_sort,
440 )
441 config_kwargs = load(
442 config_kwargs,
443 configuration_key=configuration_key,
444 configuration_references=configuration_references,
445 configuration_sort=configuration_sort,
446 )
448 if configuration_object is None:
449 try:
450 configuration_object = configuration_type(*config_args, **config_kwargs)
451 except Exception as exception:
452 raise ConfigurationError(
453 str.format("Error while initialising key {0}", configuration_key)
454 ) from exception
455 else:
456 if len(config_args) == 0:
457 for key, value in config_kwargs.items():
458 setattr(configuration_object, key, value)
459 else:
460 raise ConfigurationError(
461 str.format(
462 "Can not set arguments {0} to existing object {1}. "
463 "Missing attribute keys.",
464 config_args,
465 configuration_object,
466 )
467 )
469 return configuration_object
472@functools.singledispatch
473def save(
474 obj: Any,
475 configuration_references: dict[str, Any] | None = None,
476 *args: Any,
477 **kwargs: Any,
478) -> Any:
479 """
480 Save an object to a configuration.
482 It first check if the object has a ``save`` function registered and use it.
484 When no specific ``save`` function is registered for a object type,
485 it saves recursivly every annotated parameters of the object class.
487 :param obj: the object to save
488 :type obj: Any
490 :return: the object configuration
491 :rtype: Any
493 :raise ConfigurationError: raise a Configuration Error when the
494 object can not be saved to a configuration.
495 """
497 # try to replace the object with a reference when possible.
498 if configuration_references is not None:
499 obj_reference = next(
500 (key for key, item in configuration_references.items() if item is obj), None
501 )
502 if obj_reference is not None:
503 return obj_reference
505 obj_type = type(obj)
507 if has_save_function(obj_type) is True:
508 save_func = get_save_function(obj_type)
509 return save_func(
510 obj, *args, configuration_references=configuration_references, **kwargs
511 )
512 elif is_registered(obj_type):
513 return _base_save_obj(
514 obj, *args, configuration_references=configuration_references, **kwargs
515 )
516 raise ConfigurationError(str.format("Can not configure object {0}.", obj))
519def _base_save_obj(obj: Any, *args: Any, **kwargs: Any) -> Configuration:
520 obj_type = type(obj)
521 type_id = get_registered_type_id(obj_type)
523 # get parameters of the constructor
524 parameters_ids = _get_init_parameters(obj_type)
525 config_parameters = {
526 key: save(getattr(obj, key), *args, **kwargs) for key in parameters_ids
527 }
528 return Configuration(type_id, config_parameters)
531@save.register
532def _save_dict(obj: Mapping, *args: Any, **kwargs: Any) -> dict[str, Any]: # type: ignore
533 """Save specific function for mappings."""
534 return {
535 key: save(
536 value,
537 *args,
538 **kwargs,
539 )
540 for key, value in obj.items()
541 }
544@save.register
545def _save_list(obj: Sequence, *args: Any, **kwargs: Any) -> list[Any]: # type: ignore
546 return [
547 save(
548 value,
549 *args,
550 **kwargs,
551 )
552 for value in obj
553 ]
556@save.register(float)
557@save.register(str)
558@save.register(int)
559def _save_base_types(
560 obj: Any,
561 configuration_references: dict[str, Any] | None = None,
562 *args: Any,
563 **kwargs: Any,
564) -> Any:
565 if configuration_references is not None:
566 obj_reference = next(
567 (
568 key
569 for key, item in configuration_references.items()
570 if isinstance(item, type(obj)) and item == obj
571 ),
572 None,
573 )
574 if obj_reference is not None:
575 return obj_reference
576 return obj
579@save.register
580def _save_bool(obj: bool, *args: Any, **kwargs: Any) -> str:
581 return str(obj)
584@save.register(np.ndarray)
585def _save_array(obj: NDArray[Any], *args: Any, **kwargs: Any) -> Any:
586 return float(obj) if obj.size == 1 else obj.tolist()
589def _get_init_parameters(obj_type: type[T]) -> list[str]:
590 # get parameters of the constructor
591 obj_cstr_sig = signature(obj_type.__init__)
592 # remove first argument (self)
593 parameters_ids = list(obj_cstr_sig.parameters.keys())[1:]
594 return parameters_ids
597def __sort_configuration(configuration: Mapping[str, Any]) -> Any:
598 # Sort dict entries based on on their dependencies to initialize most
599 # required arguments first
600 dependencies_score = __build_dependencies_sorting_score(configuration)
601 sorted_values = dict(
602 sorted(
603 configuration.items(),
604 key=lambda item: dependencies_score[item[0]],
605 )
606 )
607 if isinstance(configuration, Configuration):
608 return Configuration(configuration.label, sorted_values)
609 elif isinstance(configuration, Mapping):
610 return sorted_values
613def __build_dependencies_sorting_score(
614 configuration: Mapping[str, Any],
615) -> dict[str, int]:
616 dependencies = __build_dependencies(configuration)
617 for key, item in dependencies.items():
618 __check_dependencies(key, item, dependencies)
619 return {key: __get_score(key, dependencies) for key in configuration}
622def __check_dependencies(
623 key: str,
624 dependencies: set[str],
625 all_dependencies: Mapping[str, set[str]],
626 dependency_chain: list[str] | None = None,
627) -> None:
628 dependency_chain_copy = (
629 dependency_chain.copy() if dependency_chain is not None else []
630 )
631 if key in dependency_chain_copy:
632 raise ConfigurationError(
633 str.format("Item {0} is referencing itself: {1}", key, dependency_chain)
634 )
635 dependency_chain_copy.append(key)
636 for dependency_item in dependencies:
637 __check_dependencies(
638 dependency_item,
639 all_dependencies[dependency_item],
640 all_dependencies,
641 dependency_chain_copy,
642 )
645def __build_dependencies(configuration: Mapping[str, Any]) -> Mapping[str, set[str]]:
646 dependencies: dict[str, set[str]] = {}
647 for key, item in configuration.items():
648 all_values = __get_all_strings(item)
649 dependencies[key] = set.intersection(all_values, configuration.keys())
650 for new_key, new_item in configuration.items():
651 recursive_keys = __get_recursives_keys(new_item)
652 if new_key != key and any(
653 [
654 value in recursive_keys
655 for value in all_values
656 if value not in dependencies[key]
657 ]
658 ):
659 dependencies[key].add(new_key)
661 return dependencies
664def __get_score(item_key: str, dependencies: Mapping[str, set[str]]) -> int:
665 score = 1
666 for dependencies_key in dependencies[item_key]:
667 score += __get_score(dependencies_key, dependencies)
668 return score
671def __get_recursives_keys(entry: Any) -> set[str]:
672 result: set[str] = set()
673 if isinstance(entry, str):
674 return result
675 elif isinstance(entry, Mapping):
676 result = result.union(entry.keys())
677 result = result.union(__get_recursives_keys(entry.values()))
678 elif isinstance(entry, Iterable):
679 for value in entry:
680 result = result.union(__get_recursives_keys(value))
681 return result
684def __get_all_strings(entry: Any) -> set[str]:
685 result = set()
686 if isinstance(entry, str):
687 result.add(entry)
688 elif isinstance(entry, Mapping):
689 result = result.union(__get_all_strings(entry.values()))
690 elif isinstance(entry, Iterable):
691 for sub_entry in entry:
692 entry_set = __get_all_strings(sub_entry)
693 result = result.union(entry_set)
695 return result