Coverage for mlair/helpers/datastore.py: 100%
204 statements
« prev ^ index » next coverage.py v6.4.2, created at 2022-12-02 15:24 +0000
1"""Implementation of experiment's data store."""
3__all__ = ['DataStoreByVariable', 'DataStoreByScope', 'NameNotFoundInDataStore', 'NameNotFoundInScope', 'EmptyScope',
4 'AbstractDataStore']
5__author__ = 'Lukas Leufen'
6__date__ = '2019-11-22'
8import inspect
9import logging
10import types
11from abc import ABC
12from functools import wraps
13from typing import Any, List, Tuple, Dict
16class NameNotFoundInDataStore(Exception):
17 """Exception that get raised if given name is not found in the entire data store."""
19 pass
22class NameNotFoundInScope(Exception):
23 """Exception that get raised if given name is not found in the provided scope, but can be found in other scopes."""
25 pass
28class EmptyScope(Exception):
29 """Exception that get raised if given scope is not part of the data store."""
31 pass
34class CorrectScope:
35 """
36 This class is used as decorator for all class methods, that have scope in parameters.
38 After decoration, the scope argument is not required on method call anymore. If no scope parameter is given, this
39 decorator automatically adds the default scope=`general` to the arguments. Furthermore, calls like
40 `scope=general.sub` are obsolete, because this decorator adds the prefix `general.` if not provided. Therefore, a
41 call like `scope=sub` will actually become `scope=general.sub` after passing this decorator.
42 """
44 def __init__(self, func):
45 """Construct decorator."""
46 setattr(self, "wrapper", func)
47 if hasattr(func, "__wrapped__"):
48 func = func.__wrapped__
49 wraps(func)(self)
51 def __call__(self, *args, **kwargs):
52 """
53 Call method of decorator.
55 Update tuple if scope argument does not start with `general` or slot `scope=general` into args if not provided
56 in neither args nor kwargs.
57 """
58 f_arg = inspect.getfullargspec(self.__wrapped__)
59 pos_scope = f_arg.args.index("scope")
60 if len(args) < (len(f_arg.args) - len(f_arg.defaults or "")):
61 new_arg = kwargs.pop("scope", "general") or "general"
62 args = self.update_tuple(args, new_arg, pos_scope)
63 else:
64 args = self.update_tuple(args, args[pos_scope], pos_scope, update=True)
65 return self.wrapper(*args, **kwargs)
67 def __get__(self, instance, cls):
68 """Create bound method object and supply self argument to the decorated method. <Python Cookbook, p.347>"""
69 return types.MethodType(self, instance)
71 @staticmethod
72 def correct(arg: str):
73 """
74 Add leading general prefix.
76 :param arg: string argument of scope to add prefix general if necessary
78 :return: corrected string
79 """
80 if not arg.startswith("general"):
81 arg = "general." + arg
82 return arg
84 def update_tuple(self, t: Tuple, new: Any, ind: int, update: bool = False):
85 """
86 Update single entry n given tuple or slot entry into given position.
88 Either update a entry in given tuple t (<old1>, <old2>, <old3>) --(ind=1)--> (<old1>, <new>, <old3>) or slot
89 entry into given position (<old1>, <old2>, <old3>) --(ind=1,update=True)--> (<old1>, <new>, <old2>, <old3>). In
90 the latter case, length of returned tuple is increased by 1 in comparison to given tuple.
92 :param t: tuple to update
93 :param new: new element to add to tuple
94 :param ind: position to add or slot in
95 :param update: updates entry if true, otherwise slot in (default: False)
97 :return: updated tuple
98 """
99 t_new = (*t[:ind], self.correct(new), *t[ind + update:])
100 return t_new
103class TrackParameter:
104 """Hint: Tracking is not working for static methods."""
106 def __init__(self, func):
107 """Construct decorator."""
108 wraps(func)(self)
110 def __call__(self, *args, **kwargs):
111 """
112 Call method of decorator.
113 """
114 name, obj, scope = self.track(*args)
115 f_name = self.__wrapped__.__name__
116 try:
117 res = self.__wrapped__(*args, **kwargs)
118 logging.debug(f"{f_name}: {name}({scope})={res if obj is None else obj}")
119 except Exception as e:
120 logging.debug(f"{f_name}: {name}({scope})={obj}")
121 raise
122 return res
124 def __get__(self, instance, cls):
125 """Create bound method object and supply self argument to the decorated method. <Python Cookbook, p.347>"""
126 return types.MethodType(self, instance)
128 def track(self, tracker_obj, *args):
129 name, obj, scope = self._decrypt_args(*args)
130 tracker = tracker_obj.tracker[-1]
131 new_entry = {"method": self.__wrapped__.__name__, "scope": scope}
132 if name in tracker:
133 tracker[name].append(new_entry)
134 else:
135 tracker[name] = [new_entry]
136 return name, obj, scope
138 @staticmethod
139 def _decrypt_args(*args):
140 if len(args) == 2:
141 return args[0], None, args[1]
142 else:
143 return args
146class AbstractDataStore(ABC):
147 """
148 Abstract data store for all settings for the experiment workflow.
150 Save experiment parameters for the proceeding run_modules and predefine parameters loaded during the experiment
151 setup phase. The data store is hierarchically structured, so that global settings can be overwritten by local
152 adjustments.
153 """
155 tracker = [{}]
157 def __init__(self):
158 """Initialise by creating empty data store."""
159 self._store: Dict = {}
161 def set(self, name: str, obj: Any, scope: str, log: bool = False) -> None:
162 """
163 Abstract method to add an object to the data store.
165 :param name: Name of object to store
166 :param obj: The object itself to be stored
167 :param scope: the scope / context of the object, under that the object is valid
168 :param log: log which objects are stored if enabled (default false)
169 """
170 pass
172 def get(self, name: str, scope: str) -> None:
173 """
174 Abstract method to get an object from the data store.
176 :param name: Name to look for
177 :param scope: scope to search the name for
178 :return: the stored object
179 """
180 pass
182 @CorrectScope
183 def get_default(self, name: str, scope: str, default: Any) -> Any:
184 """
185 Retrieve an object with `name` from `scope` and return given default if object wasn't found.
187 Same functionality like the standard get method. But this method adds a default argument that is returned if no
188 data was stored in the data store. Use this function with care, because it will not report any errors and just
189 return the given default value. Currently, there is no statement that reports, if the returned value comes from
190 the data store or the default value.
192 :param name: Name to look for
193 :param scope: scope to search the name for
194 :param default: default value that is return, if no data was found for given name and scope
196 :return: the stored object or the default value
197 """
198 try:
199 return self.get(name, scope)
200 except (NameNotFoundInDataStore, NameNotFoundInScope):
201 return default
203 def search_name(self, name: str) -> None:
204 """
205 Abstract method to search for all occurrences of given `name` in the entire data store.
207 :param name: Name to look for
208 :return: search result
209 """
210 pass
212 def search_scope(self, scope: str) -> None:
213 """
214 Abstract method to search for all object names that are stored for given scope.
216 :param scope: scope to look for
217 :return: search result
218 """
219 pass
221 def list_all_scopes(self) -> None:
222 """
223 Abstract method to list all scopes in data store.
225 :return: all found scopes
226 """
227 pass
229 def list_all_names(self) -> None:
230 """
231 Abstract method to list all names available in the data store.
233 :return: all names
234 """
235 pass
237 def clear_data_store(self) -> None:
238 """
239 Reset entire data store.
241 Warning: This will remove all entries of the data store without any exception.
242 """
243 self._store = {}
245 @CorrectScope
246 def create_args_dict(self, arg_list: List[str], scope: str) -> Dict:
247 """
248 Create dictionary from given argument list (as keys) and the stored data inside data store (as values).
250 Try to load all stored elements for `arg_list` and create an entry in return dictionary for each valid key
251 value pair. Not existing keys from arg_list are skipped. This method works on a single scope only and cannot
252 create a dictionary with values from different scopes. Depending on the implementation of the __get__ method,
253 all superior scopes are included in the parameter search, if no element is found for the given subscope.
255 :param arg_list: list with all elements to look for
256 :param scope: the scope to search in
258 :return: dictionary with all valid elements from given arg_list as key and the corresponding stored object as
259 value.
260 """
261 args = {}
262 for arg in arg_list:
263 try:
264 args[arg] = self.get(arg, scope)
265 except (NameNotFoundInDataStore, NameNotFoundInScope):
266 pass
267 return args
269 @CorrectScope
270 def set_from_dict(self, arg_dict: Dict, scope: str, log: bool = False) -> None:
271 """
272 Store multiple objects from dictionary under same `scope`.
274 Each object needs to be parsed as key value pair inside the given dictionary. All new entries are stored under
275 the same scope.
277 :param arg_dict: updates for the data store, provided as key value pairs
278 :param scope: scope to store updates
279 :param log: log which objects are stored if enabled (default false)
280 """
281 for (k, v) in arg_dict.items():
282 self.set(k, v, scope, log=log)
285class DataStoreByVariable(AbstractDataStore):
286 """
287 Data store for all settings for the experiment workflow.
289 Save experiment parameters for the proceeding run_modules and predefine parameters loaded during the experiment
290 setup phase. The data store is hierarchically structured, so that global settings can be overwritten by local
291 adjustments.
293 This implementation stores data as
295 .. code-block::
297 <variable1>
298 <scope1>: value
299 <scope2>: value
300 <variable2>
301 <scope1>: value
302 <scope3>: value
304 """
306 @CorrectScope
307 @TrackParameter
308 def set(self, name: str, obj: Any, scope: str, log: bool = False) -> None:
309 """
310 Store an object `obj` with given `name` under `scope`.
312 In the current implementation, existing entries are overwritten.
314 :param name: Name of object to store
315 :param obj: The object itself to be stored
316 :param scope: the scope / context of the object, under that the object is valid
317 :param log: log which objects are stored if enabled (default false)
318 """
319 # open new variable related store with `name` as key if not existing
320 if name not in self._store.keys():
321 self._store[name] = {}
322 self._store[name][scope] = obj
323 if log: # pragma: no cover
324 logging.debug(f"set: {name}({scope})={obj}")
326 @CorrectScope
327 @TrackParameter
328 def get(self, name: str, scope: str) -> Any:
329 """
330 Retrieve an object with `name` from `scope`.
332 If no object can be found in the exact scope, take an iterative look on the levels above. Raise a
333 NameNotFoundInDataStore error, if no object with given name can be found in the entire data store. Raise a
334 NameNotFoundInScope error, if the object is in the data store but not in the given scope and its levels above
335 (could be either included in another scope or a more detailed sub-scope).
337 :param name: Name to look for
338 :param scope: scope to search the name for
340 :return: the stored object
341 """
342 return self._stride_through_scopes(name, scope)[2]
344 @CorrectScope
345 def _stride_through_scopes(self, name, scope, depth=0):
346 if depth <= scope.count("."):
347 local_scope = scope.rsplit(".", maxsplit=depth)[0]
348 try:
349 return name, local_scope, self._store[name][local_scope]
350 except KeyError:
351 return self._stride_through_scopes(name, scope, depth + 1)
352 else:
353 occurrences = self.search_name(name)
354 if len(occurrences) == 0:
355 raise NameNotFoundInDataStore(f"Couldn't find {name} in data store")
356 else:
357 raise NameNotFoundInScope(f"Couldn't find {name} in scope {scope} . {name} is only defined in "
358 f"{occurrences}")
360 def search_name(self, name: str) -> List[str]:
361 """
362 Search for all occurrences of given `name` in the entire data store.
364 :param name: Name to look for
366 :return: list with all scopes and sub-scopes containing an object stored as `name`
367 """
368 return sorted(self._store[name] if name in self._store.keys() else [])
370 @CorrectScope
371 def search_scope(self, scope: str, current_scope_only=True, return_all=False) -> List[str or Tuple]:
372 """
373 Search for given `scope` and list all object names stored under this scope.
375 For an expanded search in all superior scopes, set `current_scope_only=False`. To return the scope and the
376 object's value too, set `return_all=True`.
378 :param scope: scope to look for
379 :param current_scope_only: look only for all names for given scope if true, else search for names from superior
380 scopes too.
381 :param return_all: return name, definition scope and value if True, else just the name
383 :return: list with all object names (if `return_all=False`) or list with tuple of object name, object scope and
384 object value ordered by name (if `return_all=True`)
385 """
386 if current_scope_only:
387 names = []
388 for (k, v) in self._store.items():
389 if scope in v.keys():
390 names.append(k)
391 if len(names) > 0:
392 if return_all:
393 return sorted([(name, scope, self._store[name][scope]) for name in names], key=lambda tup: tup[0])
394 else:
395 return sorted(names)
396 else:
397 raise EmptyScope(f"Given scope {scope} is not part of the data store. Available scopes are: "
398 f"{self.list_all_scopes()}")
399 else:
400 results = []
401 for name in self.list_all_names():
402 try:
403 res = self._stride_through_scopes(name, scope)
404 if return_all:
405 results.append(res)
406 else:
407 results.append(res[0])
408 except (NameNotFoundInDataStore, NameNotFoundInScope):
409 pass
410 if return_all:
411 return sorted(results, key=lambda tup: tup[0])
412 else:
413 return sorted(results)
415 def list_all_scopes(self) -> List[str]:
416 """
417 List all available scopes in data store.
419 :return: names of all stored objects
420 """
421 scopes = []
422 for v in self._store.values():
423 for scope in v.keys():
424 if scope not in scopes:
425 scopes.append(scope)
426 return sorted(scopes)
428 def list_all_names(self) -> List[str]:
429 """
430 List all names available in the data store.
432 :return: all names
433 """
434 return sorted(self._store.keys())
437class DataStoreByScope(AbstractDataStore):
438 """
439 Data store for all settings for the experiment workflow.
441 Save experiment parameters for the proceeding run_modules and predefine parameters loaded during the experiment
442 setup phase. The data store is hierarchically structured, so that global settings can be overwritten by local
443 adjustments.
445 This implementation stores data as
447 .. code-block::
449 <scope1>
450 <variable1>: value
451 <variable2>: value
452 <scope2>
453 <variable1>: value
454 <variable3>: value
456 """
458 @CorrectScope
459 @TrackParameter
460 def set(self, name: str, obj: Any, scope: str, log: bool = False) -> None:
461 """
462 Store an object `obj` with given `name` under `scope`.
464 In the current implementation, existing entries are overwritten.
466 :param name: Name of object to store
467 :param obj: The object itself to be stored
468 :param scope: the scope / context of the object, under that the object is valid
469 :param log: log which objects are stored if enabled (default false)
470 """
471 if scope not in self._store.keys():
472 self._store[scope] = {}
473 self._store[scope][name] = obj
474 if log: # pragma: no cover
475 logging.debug(f"set: {name}({scope})={obj}")
477 @CorrectScope
478 @TrackParameter
479 def get(self, name: str, scope: str) -> Any:
480 """
481 Retrieve an object with `name` from `scope`.
483 If no object can be found in the exact scope, take an iterative look on the levels above. Raise a
484 NameNotFoundInDataStore error, if no object with given name can be found in the entire data store. Raise a
485 NameNotFoundInScope error, if the object is in the data store but not in the given scope and its levels above
486 (could be either included in another scope or a more detailed sub-scope).
488 :param name: Name to look for
489 :param scope: scope to search the name for
491 :return: the stored object
492 """
493 return self._stride_through_scopes(name, scope)[2]
495 @CorrectScope
496 def _stride_through_scopes(self, name, scope, depth=0):
497 if depth <= scope.count("."):
498 local_scope = scope.rsplit(".", maxsplit=depth)[0]
499 try:
500 return name, local_scope, self._store[local_scope][name]
501 except KeyError:
502 return self._stride_through_scopes(name, scope, depth + 1)
503 else:
504 occurrences = self.search_name(name)
505 if len(occurrences) == 0:
506 raise NameNotFoundInDataStore(f"Couldn't find {name} in data store")
507 else:
508 raise NameNotFoundInScope(f"Couldn't find {name} in scope {scope} . {name} is only defined in "
509 f"{occurrences}")
511 def search_name(self, name: str) -> List[str]:
512 """
513 Search for all occurrences of given `name` in the entire data store.
515 :param name: Name to look for
517 :return: list with all scopes and sub-scopes containing an object stored as `name`
518 """
519 keys = []
520 for (key, val) in self._store.items():
521 if name in val.keys():
522 keys.append(key)
523 return sorted(keys)
525 @CorrectScope
526 def search_scope(self, scope: str, current_scope_only: bool = True, return_all: bool = False) -> List[str or Tuple]:
527 """
528 Search for given `scope` and list all object names stored under this scope.
530 For an expanded search in all superior scopes, set `current_scope_only=False`. To return the scope and the
531 object's value too, set `return_all=True`.
533 :param scope: scope to look for
534 :param current_scope_only: look only for all names for given scope if true, else search for names from superior
535 scopes too.
536 :param return_all: return name, definition scope and value if True, else just the name
538 :return: list with all object names (if `return_all=False`) or list with tuple of object name, object scope and
539 object value ordered by name (if `return_all=True`)
540 """
541 if current_scope_only:
542 try:
543 if return_all:
544 return [(name, scope, self._store[scope][name]) for name in sorted(self._store[scope].keys())]
545 else:
546 return sorted(self._store[scope].keys())
547 except KeyError:
548 raise EmptyScope(f"Given scope {scope} is not part of the data store. Available scopes are: "
549 f"{self.list_all_scopes()}")
550 else:
551 results = []
552 for name in self.list_all_names():
553 try:
554 res = self._stride_through_scopes(name, scope)
555 if return_all:
556 results.append(res)
557 else:
558 results.append(res[0])
559 except (NameNotFoundInDataStore, NameNotFoundInScope):
560 pass
561 if return_all:
562 return sorted(results, key=lambda tup: tup[0])
563 else:
564 return sorted(results)
566 def list_all_scopes(self) -> List[str]:
567 """
568 List all available scopes in data store.
570 :return: names of all stored objects
571 """
572 return sorted(self._store.keys())
574 def list_all_names(self) -> List[str]:
575 """
576 List all names available in the data store.
578 :return: all names
579 """
580 names = []
581 scopes = self.list_all_scopes()
582 for scope in scopes:
583 for name in self._store[scope].keys():
584 if name not in names:
585 names.append(name)
586 return sorted(names)