Coverage for src/navdict/navdict.py: 75%
382 statements
« prev ^ index » next coverage.py v7.8.2, created at 2026-07-17 16:04 +0200
1"""
2NavDict: A navigable dictionary with dot notation access and automatic file loading.
4NavDict extends Python's built-in dictionary to support convenient dot notation
5access (data.user.name) alongside traditional key access (data["user"]["name"]).
6It automatically loads data files and can instantiate classes dynamically based
7on configuration.
9Features:
10 - Dot notation access for nested data structures
11 - Automatic file loading (CSV, YAML, JSON, etc.)
12 - Dynamic class instantiation from configuration
13 - Full backward compatibility with standard dictionaries
15Example:
16 >>> from navdict import navdict
17 >>> data = navdict({"user": {"name": "Alice", "config_file": "yaml//settings.yaml"}})
18 >>> data.user.name # "Alice"
19 >>> data.user.config_file # Automatically loads and parses settings.yaml
20 >>> data["user"]["name"] # Still works with traditional access
22Author: Rik Huygen
23License: MIT
24"""
26from __future__ import annotations
28__all__ = [
29 "navdict", # noqa: ignore typo
30 "NavDict",
31 "NavigableDict",
32 "expand_env_vars",
33 "get_resource_location",
34 "load_class",
35 "load_csv",
36 "load_yaml",
37]
39import csv
40import datetime
41import importlib
42import itertools
43import logging
44import os
45import re
46import textwrap
47import warnings
48from enum import IntEnum
49from pathlib import Path
50from typing import Any
51from typing import Callable
52from typing import TypeVar
53from typing import cast
55from rich.text import Text
56from rich.tree import Tree
57from ruamel.yaml import YAML
58from ruamel.yaml.scanner import ScannerError
60from navdict.directive import is_directive
61from navdict.directive import unravel_directive
62from navdict.directive import get_directive_plugin
64logger = logging.getLogger("navdict")
66T = TypeVar("T")
69def load_class(class_name: str):
70 """
71 Find and returns a class based on the fully qualified name.
73 A class name can be preceded with the string `class//` or `factory//`. This is used in YAML
74 files where the class is then instantiated on load.
76 Args:
77 class_name (str): a fully qualified name for the class
78 """
79 if class_name.startswith("class//"): 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true
80 class_name = class_name[7:]
81 elif class_name.startswith("factory//"): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 class_name = class_name[9:]
84 module_name, class_name = class_name.rsplit(".", 1)
85 module = importlib.import_module(module_name)
86 return getattr(module, class_name)
89def get_resource_location(parent_location: Path | None, in_dir: str | None) -> Path:
90 """
91 Returns the resource location.
93 The resource location is the path to the file that is provided in a directive
94 such as `yaml//` or `csv//`. The location of the file can be given as an absolute
95 path or can be relative in which case there are two possibilities:
97 1. the parent location is not None.
98 In this case the resource location will be relative to the parent's location.
99 2. the parent location is None.
100 In this case the resource location is taken to be relative to the current working directory '.'
101 unless the environment variable NAVDICT_DEFAULT_RESOURCE_LOCATION is provided in which case
102 it is taken from that variable.
103 3. when both arguments are None, the resource location will be the current working directory '.'
104 unless the environment variable NAVDICT_DEFAULT_RESOURCE_LOCATION is provided in which case
105 it is taken from that variable.
107 In all cases, the returned path is expanded, i.e. `~` is expanded to the user's home directory,
108 and resolved, but not checked for existence.
110 Args:
111 parent_location: the location of the parent navdict, or None
112 in_dir: a location extracted from the directive's value.
114 Returns:
115 A Path object with the resource location.
117 """
119 match (parent_location, in_dir):
120 case (_, str()) if in_dir.startswith("~") or Path(in_dir).is_absolute():
121 location = Path(in_dir)
122 case (None, str()):
123 location = Path(os.getenv("NAVDICT_DEFAULT_RESOURCE_LOCATION", ".")) / in_dir
124 case (Path(), str()):
125 location = parent_location / in_dir
126 case (Path(), None):
127 location = parent_location
128 case _:
129 location = Path(os.getenv("NAVDICT_DEFAULT_RESOURCE_LOCATION", "."))
131 # logger.debug(f"{location=}, {fn=}")
133 return location.expanduser()
136ENV_VAR_PATTERN = re.compile(r"ENV\[\s*['\"]?(\w+)['\"]?\s*]")
139def expand_env_vars(value: str) -> str:
140 """
141 Expand `ENV[VARNAME]` references in `value` with the value of the corresponding environment
142 variable.
144 The variable name may optionally be wrapped in single or double quotes, i.e. `ENV[VARNAME]`,
145 `ENV['VARNAME']` and `ENV["VARNAME"]` are all equivalent.
147 Args:
148 value: a string that may contain zero or more `ENV[VARNAME]` references.
150 Returns:
151 The string with all `ENV[VARNAME]` references expanded.
153 Raises:
154 ValueError: when a referenced environment variable is not set.
155 """
157 def _replace(match: re.Match) -> str:
158 var_name = match[1]
159 try:
160 return os.environ[var_name]
161 except KeyError:
162 raise ValueError(f"Environment variable '{var_name}' referenced in '{value}' is not set.") from None
164 return ENV_VAR_PATTERN.sub(_replace, value)
167def load_csv(resource_name: str, parent_location: Path | None, *args, **kwargs) -> list[list[str]]:
168 """
169 Find and return the content of a CSV file.
171 If the `resource_name` argument starts with the directive (`csv//`), it will be split off automatically.
173 The `kwargs` dictionary can contain the following keys:
175 - `header_rows`: the number of header rows to skip when processing the file.
176 - `delimiter`: the delimiter to use when parsing the CSV file. The default is a comma (`,`).
177 - `expand_env`: when `True` (the default), `ENV[VARNAME]` references in `resource_name` are expanded
178 with the value of the corresponding environment variable before the file is located. Set to `False`
179 to disable this expansion.
181 Returns:
182 A list of the split lines, i.e. a list of lists of strings.
183 """
185 # logger.debug(f"{resource_name=}, {parent_location=}")
187 if resource_name.startswith("csv//"): 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 resource_name = resource_name[5:]
190 if not resource_name: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true
191 raise ValueError("Resource name should not be empty, but contain a valid filename.")
193 if kwargs.get("expand_env", True):
194 resource_name = expand_env_vars(resource_name)
196 parts = resource_name.rsplit("/", 1)
197 in_dir, fn = parts if len(parts) > 1 else (None, parts[0]) # use a tuple here to make Mypy happy
199 try:
200 n_header_rows = int(kwargs["header_rows"])
201 except KeyError:
202 n_header_rows = 0
204 csv_location = get_resource_location(parent_location, in_dir)
206 def filter_lines(file_obj, n_skip):
207 """
208 Generator that filters out comment lines and skips header lines.
209 The standard library csv module cannot handle this functionality.
210 """
212 for line in itertools.islice(file_obj, n_skip, None):
213 if not line.strip().startswith("#"):
214 yield line
216 try:
217 with open(csv_location / fn, "r", encoding="utf-8") as file:
218 filtered_lines = filter_lines(file, n_header_rows)
219 csv_reader = csv.reader(filtered_lines, delimiter=kwargs.get("delimiter", ","))
220 data = list(csv_reader)
221 except FileNotFoundError:
222 logger.error(f"Couldn't load resource '{resource_name}', file not found", exc_info=True)
223 raise
225 return data
228def _load_int_enum(enum_name: str, enum_content) -> IntEnum:
229 """Dynamically build (and return) and IntEnum.
231 In the YAML file this will look like below.
232 The IntEnum directive (where <name> is the class name):
234 enum: int_enum//<name>
236 The IntEnum content:
238 content:
239 E:
240 alias: ['E_SIDE', 'RIGHT_SIDE']
241 value: 1
242 F:
243 alias: ['F_SIDE', 'LEFT_SIDE']
244 value: 0
246 Args:
247 - enum_name: Enumeration name (potentially prepended with "int_enum//").
248 - enum_content: Content of the enumeration, as read from the navdict field.
250 Returns:
251 An IntEnum class with the given name and content.
252 """
253 if enum_name.startswith("int_enum//"): 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 enum_name = enum_name[10:]
256 definition = {}
257 for side_name, side_definition in enum_content.items():
258 if "alias" in side_definition: 258 ↛ 261line 258 didn't jump to line 261 because the condition on line 258 was always true
259 aliases = side_definition["alias"]
260 else:
261 aliases = []
262 value = side_definition["value"]
264 definition[side_name] = value
266 for alias in aliases:
267 definition[alias] = value
269 return IntEnum(enum_name, definition)
272def load_yaml(resource_name: str, parent_location: Path | None, *args, **kwargs) -> NavigableDict:
273 """Find and return the content of a YAML file.
275 The `kwargs` dictionary can contain the following keys:
277 - `expand_env`: when `True` (the default), `ENV[VARNAME]` references in `resource_name` are expanded
278 with the value of the corresponding environment variable before the file is located. Set to `False`
279 to disable this expansion.
281 Args:
282 resource_name: the name of the YAML file to load, optionally prepended with `yaml//`.
283 parent_location: the location of the parent navdict, or None.
284 *args: additional positional arguments (currently unused).
285 **kwargs: additional keyword arguments.
287 Returns:
288 A NavigableDict containing the content of the YAML file.
289 """
291 # logger.debug(f"{resource_name=}, {parent_location=}")
293 if resource_name.startswith("yaml//"): 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 resource_name = resource_name[6:]
296 if not resource_name: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 raise ValueError("Resource name should not be empty, but contain a valid filename.")
299 if kwargs.get("expand_env", True): 299 ↛ 302line 299 didn't jump to line 302 because the condition on line 299 was always true
300 resource_name = expand_env_vars(resource_name)
302 parts = resource_name.rsplit("/", 1)
303 in_dir, fn = parts if len(parts) > 1 else (None, parts[0]) # use a tuple here to make Mypy happy
305 yaml_location = get_resource_location(parent_location, in_dir)
307 try:
308 yaml = YAML(typ="safe")
309 with open(yaml_location / fn, "r") as file:
310 data = yaml.load(file)
312 except FileNotFoundError:
313 logger.error(f"Couldn't load resource '{resource_name}', file not found", exc_info=True)
314 raise
315 except IsADirectoryError:
316 logger.error(
317 f"Couldn't load resource '{resource_name}', file seems to be a directory",
318 exc_info=True,
319 )
320 raise
321 except ScannerError as exc:
322 msg = f"A error occurred while scanning the YAML file: {yaml_location / fn}."
323 logger.error(msg, exc_info=True)
324 raise IOError(msg) from exc
326 data = NavigableDict(data, _filename=yaml_location / fn)
328 # logger.debug(f"{data.get_private_attribute('_filename')=}")
330 return data
333def _get_attribute(self, name, default):
334 """
335 Safely retrieve an attribute from the object, returning a default if not found.
337 This method uses object.__getattribute__() to bypass any custom __getattr__
338 or __getattribute__ implementations on the class, accessing attributes directly
339 from the object's internal dictionary.
341 Args:
342 name (str): The name of the attribute to retrieve.
343 default: The value to return if the attribute does not exist.
345 Returns:
346 The attribute value if it exists, otherwise the default value.
348 Note:
349 This is typically used internally to avoid infinite recursion when
350 implementing custom attribute access methods.
351 """
352 try:
353 attr = object.__getattribute__(self, name)
354 except AttributeError:
355 attr = default
356 return attr
359class NavigableDict(dict):
360 """
361 A NavigableDict is a dictionary where all keys in the original dictionary are also accessible
362 as attributes to the class instance. So, if the original dictionary (setup) has a key
363 "site_id" which is accessible as `setup['site_id']`, it will also be accessible as
364 `setup.site_id`.
366 Args:
367 head (dict): the original dictionary
368 label (str): a label or name that is used when printing the navdict
370 Examples:
371 >>> setup = NavigableDict({'site_id': 'KU Leuven', 'version': "0.1.0"})
372 >>> assert setup['site_id'] == setup.site_id
373 >>> assert setup['version'] == setup.version
375 Note:
376 We always want **all** keys to be accessible as attributes, or none. That means all
377 keys of the original dictionary shall be of type `str`.
379 """
381 def __init__(
382 self,
383 head: dict | None = None,
384 label: str | None = None,
385 _filename: str | Path | None = None,
386 ):
387 head = head or {}
388 super().__init__(head)
389 self.__dict__["_memoized"] = {}
390 self.__dict__["_label"] = label
391 self.__dict__["_filename"] = Path(_filename) if _filename is not None else None
393 # TODO:
394 # if _filename was not given as an argument, we might want to check if the `head` has a `_filename` and do
395 # something like:
396 #
397 # if _filename is None and isinstance(head, navdict):
398 # _filename = head.__dict__["_filename"]
399 # self.__dict__["_filename"] = _filename
401 # By agreement, we only want the keys to be set as attributes if all keys are strings.
402 # That way we enforce that always all keys are navigable, or none.
404 if any(True for k in head.keys() if not isinstance(k, str)):
405 # invalid_keys = list(k for k in head.keys() if not isinstance(k, str))
406 # logger.warning(f"Dictionary will not be dot-navigable, not all keys are strings [{invalid_keys=}].")
407 return
409 for key, value in head.items():
410 if isinstance(value, dict):
411 value = NavigableDict(head.__getitem__(key), _filename=_filename, label=key)
412 setattr(self, key, value)
413 else:
414 setattr(self, key, super().__getitem__(key))
416 def get_label(self) -> str | None:
417 return self.__dict__["_label"]
419 def set_label(self, value: str):
420 self.__dict__["_label"] = value
422 def add(self, key: str, value: Any):
423 """Set a value for the given key.
425 If the value is a dictionary, it will be converted into a NavigableDict and the keys
426 will become available as attributes provided that all the keys are strings.
428 Args:
429 key (str): the name of the key / attribute to access the value
430 value (Any): the value to assign to the key
431 """
432 if isinstance(value, dict) and not isinstance(value, NavigableDict):
433 value = NavigableDict(value, label=key)
434 setattr(self, key, value)
436 def clear(self) -> None:
437 for key in list(self.keys()):
438 self.__delitem__(key)
440 def __repr__(self):
441 return f"{self.__class__.__name__}({super()!r}) [id={id(self)}]"
443 def __delitem__(self, key):
444 dict.__delitem__(self, key)
445 object.__delattr__(self, key)
447 def __setattr__(self, key, value):
448 # logger.info(f"called __setattr__({self!r}, {key}, {value})")
449 if isinstance(value, dict) and not isinstance(value, NavigableDict): 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true
450 value = NavigableDict(value, label=key)
451 self.__dict__[key] = value
452 super().__setitem__(key, value)
453 try:
454 del self.__dict__["_memoized"][key]
455 except KeyError:
456 pass
458 @staticmethod
459 def _alias_hook(key: str) -> str:
460 raise NotImplementedError
462 def set_alias_hook(self, hook: Callable[[str], str]):
463 """
464 Sets an alias (hook) function that maps the given argument, an attribute
465 or a dict key, to a valid attribute or key.
467 The `hook` function accepts a string argument and return a string for
468 which the argument is an alias. The returned argument is expected to
469 be a valid attribute or key for this navdict.
470 """
471 # The setattr() function will add the attribute to the dictionary
472 # which is not what we want.
473 # setattr(self, "_alias_hook", hook)
474 self.__dict__["_alias_hook"] = hook
476 def as_typed(self, schema_type: type[T]) -> T:
477 """Return this instance typed as `schema_type` for static analysis.
479 This helper does not transform data at runtime. It returns `self` and
480 exists to let type checkers and editors infer attribute completion when
481 callers provide a Protocol, TypedDict wrapper, or class describing the
482 expected structure.
483 """
484 return cast(T, self)
486 def __dir__(self):
487 """Include navigable keys in `dir()` for better interactive discovery."""
488 base = set(super().__dir__())
489 key_names = {key for key in self.keys() if isinstance(key, str) and key.isidentifier()}
490 for key in self.keys():
491 if isinstance(key, str) and not key.isidentifier():
492 base.discard(key)
493 return sorted(base | key_names)
495 # This method is called:
496 # - for *every* single attribute access on an object using dot notation.
497 # - when using the `getattr(obj, 'name') function
498 # - accessing any kind of attributes, e.g. instance or class variables,
499 # methods, properties, dunder methods, ...
500 #
501 # Note: `__getattr__` is only called when an attribute cannot be found
502 # through normal means.
503 def __getattribute__(self, key):
504 # logger.info(f"called __getattribute__({key}) ...")
505 try:
506 value = object.__getattribute__(self, key)
507 except AttributeError:
508 try:
509 alias = self._alias_hook(key)
510 value = object.__getattribute__(self, alias)
511 except (NotImplementedError, AttributeError):
512 # Either the alias hook is not implemented or the alias hook just returned a key that still doesn't exist.
513 # In both cases we want to raise an AttributeError to indicate that the attribute does not exist.
514 raise AttributeError(f"{type(self).__name__!r} object has no attribute {key!r}")
515 except Exception as exc:
516 logger.error(
517 f"Alias hook function {self._alias_hook.__name__!r}({key}) raised {type(exc).__name__!r}: {exc}"
518 )
519 raise AttributeError(f"{type(self).__name__!r} object has no attribute {key!r}")
521 if key.startswith("__"): # small optimization
522 return value
523 # We can not directly call the `_handle_directive` function here due to infinite recursion
524 if is_directive(value):
525 m = object.__getattribute__(self, "_handle_directive")
526 return m(key, value)
527 else:
528 return value
530 def __delattr__(self, item):
531 # logger.info(f"called __delattr__({self!r}, {item})")
532 object.__delattr__(self, item)
533 dict.__delitem__(self, item)
535 def __setitem__(self, key, value):
536 # logger.debug(f"called __setitem__({self!r}, {key}, {value})")
537 if isinstance(value, dict) and not isinstance(value, NavigableDict):
538 value = NavigableDict(value, label=key)
539 super().__setitem__(key, value)
540 self.__dict__[key] = value
541 try:
542 del self.__dict__["_memoized"][key]
543 except KeyError:
544 pass
546 # This method is called:
547 # - whenever square brackets `[]` are used on an object, e.g. indexing or slicing.
548 # - during iteration, if an object doesn't have __iter__ defined, Python will try
549 # to iterate using __getitem__ with successive integer indices starting from 0.
550 def __getitem__(self, key):
551 # logger.info(f"called __getitem__({self!r}, {key})")
552 try:
553 value = super().__getitem__(key)
554 except KeyError:
555 try:
556 alias = self._alias_hook(key)
557 value = super().__getitem__(alias)
558 except NotImplementedError:
559 raise KeyError(f"{type(self).__name__!r} has no key {key!r}")
560 except Exception as exc:
561 logger.error(
562 f"Alias hook function {self._alias_hook.__name__!r}({key}) raised {type(exc).__name__!r}: {exc}"
563 )
564 raise KeyError(f"{type(self).__name__!r} has no key {key!r}")
566 if isinstance(key, str) and key.startswith("__"): 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 return value
568 # no danger for recursion here, so we can directly call the function
569 if is_directive(value): 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true
570 return self._handle_directive(key, value)
571 else:
572 return value
574 def _handle_directive(self, key, value) -> Any:
575 """
576 This method will handle the available directives. This may be builtin directives
577 like `class/` or `factory//`, or it may be external directives that were provided
578 as a plugin. Some builtin directives have also been provided as a plugin, e.g.
579 'yaml//' and 'csv//'.
581 Args:
582 key: the key of the field that might contain a directive
583 value: the value which might be a directive
585 Returns:
586 This function will return the value, either the original value or the result of
587 evaluating and executing a directive.
588 """
589 # logger.debug(f"called _handle_directive({key}, {value!r}) [id={id(self)}]")
591 directive_key, directive_value = unravel_directive(value)
592 # logger.debug(f"{directive_key=}, {directive_value=}")
594 if directive := get_directive_plugin(directive_key):
595 # logger.debug(f"{directive.name=}")
597 if key in self.__dict__["_memoized"]:
598 return self.__dict__["_memoized"][key]
600 args, kwargs = self._get_args_and_kwargs(key)
601 parent_location = self._get_location()
602 result = directive.func(directive_value, parent_location, *args, **kwargs)
604 self.__dict__["_memoized"][key] = result
605 return result
607 match directive_key:
608 case "class":
609 args, kwargs = self._get_args_and_kwargs(key)
610 return load_class(directive_value)(*args, **kwargs)
612 case "factory": 612 ↛ 613line 612 didn't jump to line 613 because the pattern on line 612 never matched
613 factory_args = _get_attribute(self, f"{key}_args", {})
614 return load_class(directive_value)().create(**factory_args)
616 case "int_enum":
617 content = object.__getattribute__(self, "content")
618 return _load_int_enum(directive_value, content)
620 case _:
621 return value
623 def _get_location(self):
624 """Returns the location of the file from which this NavDict was loaded or None if no location exists."""
625 try:
626 filename = self.__dict__["_filename"]
627 return filename.parent if filename else None
628 except KeyError:
629 return None
631 def _get_args_and_kwargs(self, key):
632 """
633 Read the args and kwargs that are associated with the key of a directive.
635 An example of such a directive:
637 hexapod:
638 device: class//egse.hexapod.PunaProxy
639 device_args: [PUNA_01]
640 device_kwargs:
641 sim: true
643 There might not be any positional nor keyword arguments provided in which
644 case and empty tuple and/or dictionary is returned.
646 Returns:
647 A tuple containing any positional arguments and a dictionary containing
648 keyword arguments.
649 """
650 try:
651 args = object.__getattribute__(self, f"{key}_args")
652 except AttributeError:
653 args = ()
654 try:
655 kwargs = object.__getattribute__(self, f"{key}_kwargs")
656 except AttributeError:
657 kwargs = {}
659 return args, kwargs
661 def set_private_attribute(self, key: str, value: Any) -> None:
662 """Sets a private attribute for this object.
664 The name in key will be accessible as an attribute for this object, but the key will not
665 be added to the dictionary and not be returned by methods like keys().
667 The idea behind this private attribute is to have the possibility to add status information
668 or identifiers to this classes object that can be used by save() or load() methods.
670 Args:
671 key (str): the name of the private attribute (must start with an underscore character).
672 value: the value for this private attribute
674 Examples:
675 >>> setup = NavigableDict({'a': 1, 'b': 2, 'c': 3})
676 >>> setup.set_private_attribute("_loaded_from_dict", True)
677 >>> assert "c" in setup
678 >>> assert "_loaded_from_dict" not in setup
679 >>> assert setup.get_private_attribute("_loaded_from_dict") == True
681 """
682 if key in self: 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true
683 raise ValueError(f"Invalid argument key='{key}', this key already exists in the dictionary.")
684 if not key.startswith("_"): 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true
685 raise ValueError(f"Invalid argument key='{key}', must start with underscore character '_'.")
686 self.__dict__[key] = value
688 def get_private_attribute(self, key: str) -> Any:
689 """Returns the value of the given private attribute.
691 Args:
692 key (str): the name of the private attribute (must start with an underscore character).
694 Returns:
695 the value of the private attribute given in `key` or None if the attribute doesn't exist.
697 Note:
698 Because of the implementation, this private attribute can also be accessed as a 'normal'
699 attribute of the object. This use is however discouraged as it will make your code less
700 understandable. Use the methods to access these 'private' attributes.
701 """
702 if not key.startswith("_"): 702 ↛ 703line 702 didn't jump to line 703 because the condition on line 702 was never true
703 raise ValueError(f"Invalid argument key='{key}', must start with underscore character '_'.")
704 try:
705 return self.__dict__[key]
706 except KeyError:
707 return None
709 def has_private_attribute(self, key) -> bool:
710 """
711 Check if the given key is defined as a private attribute.
713 Args:
714 key (str): the name of a private attribute (must start with an underscore)
715 Returns:
716 True if the given key is a known private attribute.
717 Raises:
718 ValueError: when the key doesn't start with an underscore.
719 """
720 if not key.startswith("_"):
721 raise ValueError(f"Invalid argument key='{key}', must start with underscore character '_'.")
723 # logger.debug(f"{self.__dict__.keys()} for [id={id(self)}]")
725 try:
726 _ = self.__dict__[key]
727 return True
728 except KeyError:
729 return False
731 def get_raw_value(self, key):
732 """
733 Returns the raw value of the given key.
735 Some keys have special values that are interpreted by the NavigableDict class. An example is
736 a value that starts with 'class//'. When you access these values, they are first converted
737 from their raw value into their expected value, e.g. the instantiated object in the above
738 example. This method allows you to access the raw value before conversion.
739 """
740 try:
741 return object.__getattribute__(self, key)
742 except AttributeError:
743 raise KeyError(f"The key '{key}' is not defined.")
745 def __str__(self):
746 return self._pretty_str()
748 def _pretty_str(self, indent: int = 0):
749 msg = ""
751 for k, v in self.items():
752 if isinstance(v, NavigableDict):
753 msg += f"{' ' * indent}{k}:\n"
754 msg += v._pretty_str(indent + 1)
755 else:
756 msg += f"{' ' * indent}{k}: {v}\n"
758 return msg
760 def __rich__(self) -> Tree:
761 tree = Tree(self.__dict__["_label"] or "NavigableDict", guide_style="dim")
762 _walk_dict_tree(self, tree, text_style="dark grey")
763 return tree
765 def _save(self, fd, indent: int = 0):
766 """
767 Recursive method to write the dictionary to the file descriptor.
769 Indentation is done in steps of four spaces, i.e. `' '*indent`.
771 Args:
772 fd: a file descriptor as returned by the open() function
773 indent (int): indentation level of each line [default = 0]
775 """
777 # Note that the .items() method returns the actual values of the keys and doesn't use the
778 # __getattribute__ or __getitem__ methods. So the raw value is returned and not the
779 # _processed_ value.
781 for k, v in self.items():
782 # history shall be saved last, skip it for now
784 if k == "history": 784 ↛ 785line 784 didn't jump to line 785 because the condition on line 784 was never true
785 continue
787 # make sure to escape a colon in the key name
789 if isinstance(k, str) and ":" in k: 789 ↛ 790line 789 didn't jump to line 790 because the condition on line 789 was never true
790 k = '"' + k + '"'
792 if isinstance(v, NavigableDict):
793 fd.write(f"{' ' * indent}{k}:\n")
794 v._save(fd, indent + 1)
795 fd.flush()
796 continue
798 if isinstance(v, float): 798 ↛ 799line 798 didn't jump to line 799 because the condition on line 798 was never true
799 v = f"{v:.6E}"
800 fd.write(f"{' ' * indent}{k}: {v}\n")
801 fd.flush()
803 # now save the history as the last item
805 if "history" in self: 805 ↛ 806line 805 didn't jump to line 806 because the condition on line 805 was never true
806 fd.write(f"{' ' * indent}history:\n")
807 self.history._save(fd, indent + 1) # noqa
809 def get_memoized_keys(self):
810 return list(self.__dict__["_memoized"].keys())
812 def del_memoized_key(self, key: str):
813 try:
814 del self.__dict__["_memoized"][key]
815 return True
816 except KeyError:
817 return False
819 @staticmethod
820 def from_dict(my_dict: dict, label: str | None = None) -> NavigableDict:
821 """Create a NavigableDict from a given dictionary.
823 Remember that all keys in the given dictionary shall be of type 'str' in order to be
824 accessible as attributes.
826 Args:
827 my_dict: a Python dictionary
828 label: a label that will be attached to this navdict
830 Examples:
831 >>> setup = navdict.from_dict({"ID": "my-setup-001", "version": "0.1.0"}, label="Setup")
832 >>> assert setup["ID"] == setup.ID == "my-setup-001"
834 """
835 return NavigableDict(my_dict, label=label)
837 @staticmethod
838 def from_yaml_string(yaml_content: str | None = None, label: str | None = None) -> NavigableDict:
839 """Creates a NavigableDict from the given YAML string.
841 This method is mainly used for easy creation of a navdict from strings during unit tests.
843 Args:
844 yaml_content: a string containing YAML
845 label: a label that will be attached to this navdict
847 Returns:
848 a navdict that was loaded from the content of the given string.
849 """
851 if not yaml_content:
852 raise ValueError("Invalid argument to function: No input string or None given.")
854 yaml = YAML(typ="safe")
855 try:
856 data = yaml.load(yaml_content)
857 except ScannerError as exc:
858 raise ValueError(f"Invalid YAML string: {exc}")
860 return NavigableDict(data, label=label)
862 @staticmethod
863 def from_yaml_file(filename: str | Path | None = None) -> NavigableDict:
864 """Creates a navigable dictionary from the given YAML file.
866 Args:
867 filename (str): the path of the YAML file to be loaded
869 Returns:
870 a navdict that was loaded from the given location.
872 Raises:
873 ValueError: when no filename is given.
874 """
876 # logger.debug(f"{filename=}")
878 if not filename: 878 ↛ 879line 878 didn't jump to line 879 because the condition on line 878 was never true
879 raise ValueError("Invalid argument to function: No filename or None given.")
881 # Make sure the filename exists and is a regular file
882 filename = Path(filename).expanduser().resolve()
883 if not filename.is_file():
884 raise ValueError(f"Invalid argument to function, filename does not exist: {filename!s}")
886 data = load_yaml(str(filename), parent_location=None)
888 if data == {}: 888 ↛ 889line 888 didn't jump to line 889 because the condition on line 888 was never true
889 warnings.warn(f"Empty YAML file: {filename!s}")
891 return data
893 def to_yaml_file(
894 self,
895 filename: str | Path | None = None,
896 header: str | None = None,
897 top_level_group: str | None = None,
898 ) -> None:
899 """Saves a NavigableDict to a YAML file.
901 When no filename is provided, this method will look for a 'private' attribute
902 `_filename` and use that to save the data.
904 Args:
905 filename (str|Path): the path of the YAML file where to save the data
906 header (str): Custom header for this navdict
907 top_level_group (str): name of the optional top-level group
909 Note:
910 This method will **overwrite** the original or given YAML file and therefore you might
911 lose proper formatting and/or comments.
913 """
914 if filename is None and self.get_private_attribute("_filename") is None:
915 raise ValueError("No filename given or known, can not save navdict.")
917 if header is None: 917 ↛ 929line 917 didn't jump to line 929 because the condition on line 917 was always true
918 header = textwrap.dedent(
919 f"""
920 # This YAML file is generated by:
921 #
922 # navdict.to_yaml_file(setup, filename="{filename}')
923 #
924 # Created on {datetime.datetime.now(tz=datetime.timezone.utc).isoformat()}
926 """
927 )
929 target = Path(filename) if filename is not None else Path(self.get_private_attribute("_filename"))
931 with target.open("w") as fd:
932 fd.write(header)
933 indent = 0
934 if top_level_group: 934 ↛ 935line 934 didn't jump to line 935 because the condition on line 934 was never true
935 fd.write(f"{top_level_group}:\n")
936 indent = 1
938 self._save(fd, indent=indent)
940 self.set_private_attribute("_filename", target)
942 def get_filename(self) -> str | None:
943 """Returns the filename for this navdict or None when no filename could be determined."""
944 return self.get_private_attribute("_filename")
947navdict = NavigableDict
948NavDict = NavigableDict
949"""Shortcuts for NavigableDict and more Pythonic."""
952def _walk_dict_tree(dictionary: dict, tree: Tree, text_style: str = "green"):
953 for k, v in dictionary.items():
954 if isinstance(v, dict):
955 branch = tree.add(f"[purple]{k}", style="", guide_style="dim")
956 _walk_dict_tree(v, branch, text_style=text_style)
957 else:
958 text = Text.assemble((str(k), "medium_purple1"), ": ", (str(v), text_style))
959 tree.add(text)