Coverage for mlair/helpers/datastore.py: 100%

204 statements  

« prev     ^ index     » next       coverage.py v6.4.2, created at 2023-06-30 10:22 +0000

1"""Implementation of experiment's data store.""" 

2 

3__all__ = ['DataStoreByVariable', 'DataStoreByScope', 'NameNotFoundInDataStore', 'NameNotFoundInScope', 'EmptyScope', 

4 'AbstractDataStore'] 

5__author__ = 'Lukas Leufen' 

6__date__ = '2019-11-22' 

7 

8import inspect 

9import logging 

10import types 

11from abc import ABC 

12from functools import wraps 

13from typing import Any, List, Tuple, Dict 

14 

15 

16class NameNotFoundInDataStore(Exception): 

17 """Exception that get raised if given name is not found in the entire data store.""" 

18 

19 pass 

20 

21 

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.""" 

24 

25 pass 

26 

27 

28class EmptyScope(Exception): 

29 """Exception that get raised if given scope is not part of the data store.""" 

30 

31 pass 

32 

33 

34class CorrectScope: 

35 """ 

36 This class is used as decorator for all class methods, that have scope in parameters. 

37 

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 """ 

43 

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) 

50 

51 def __call__(self, *args, **kwargs): 

52 """ 

53 Call method of decorator. 

54 

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) 

66 

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) 

70 

71 @staticmethod 

72 def correct(arg: str): 

73 """ 

74 Add leading general prefix. 

75 

76 :param arg: string argument of scope to add prefix general if necessary 

77 

78 :return: corrected string 

79 """ 

80 if not arg.startswith("general"): 

81 arg = "general." + arg 

82 return arg 

83 

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. 

87 

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. 

91 

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) 

96 

97 :return: updated tuple 

98 """ 

99 t_new = (*t[:ind], self.correct(new), *t[ind + update:]) 

100 return t_new 

101 

102 

103class TrackParameter: 

104 """Hint: Tracking is not working for static methods.""" 

105 

106 def __init__(self, func): 

107 """Construct decorator.""" 

108 wraps(func)(self) 

109 

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 

123 

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) 

127 

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 

137 

138 @staticmethod 

139 def _decrypt_args(*args): 

140 if len(args) == 2: 

141 return args[0], None, args[1] 

142 else: 

143 return args 

144 

145 

146class AbstractDataStore(ABC): 

147 """ 

148 Abstract data store for all settings for the experiment workflow. 

149 

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 """ 

154 

155 tracker = [{}] 

156 

157 def __init__(self): 

158 """Initialise by creating empty data store.""" 

159 self._store: Dict = {} 

160 

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. 

164 

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 

171 

172 def get(self, name: str, scope: str) -> None: 

173 """ 

174 Abstract method to get an object from the data store. 

175 

176 :param name: Name to look for 

177 :param scope: scope to search the name for 

178 :return: the stored object 

179 """ 

180 pass 

181 

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. 

186 

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. 

191 

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 

195 

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 

202 

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. 

206 

207 :param name: Name to look for 

208 :return: search result 

209 """ 

210 pass 

211 

212 def search_scope(self, scope: str) -> None: 

213 """ 

214 Abstract method to search for all object names that are stored for given scope. 

215 

216 :param scope: scope to look for 

217 :return: search result 

218 """ 

219 pass 

220 

221 def list_all_scopes(self) -> None: 

222 """ 

223 Abstract method to list all scopes in data store. 

224 

225 :return: all found scopes 

226 """ 

227 pass 

228 

229 def list_all_names(self) -> None: 

230 """ 

231 Abstract method to list all names available in the data store. 

232 

233 :return: all names 

234 """ 

235 pass 

236 

237 def clear_data_store(self) -> None: 

238 """ 

239 Reset entire data store. 

240 

241 Warning: This will remove all entries of the data store without any exception. 

242 """ 

243 self._store = {} 

244 

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). 

249 

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. 

254 

255 :param arg_list: list with all elements to look for 

256 :param scope: the scope to search in 

257 

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 

268 

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`. 

273 

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. 

276 

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) 

283 

284 

285class DataStoreByVariable(AbstractDataStore): 

286 """ 

287 Data store for all settings for the experiment workflow. 

288 

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. 

292 

293 This implementation stores data as 

294 

295 .. code-block:: 

296 

297 <variable1> 

298 <scope1>: value 

299 <scope2>: value 

300 <variable2> 

301 <scope1>: value 

302 <scope3>: value 

303 

304 """ 

305 

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`. 

311 

312 In the current implementation, existing entries are overwritten. 

313 

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}") 

325 

326 @CorrectScope 

327 @TrackParameter 

328 def get(self, name: str, scope: str) -> Any: 

329 """ 

330 Retrieve an object with `name` from `scope`. 

331 

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). 

336 

337 :param name: Name to look for 

338 :param scope: scope to search the name for 

339 

340 :return: the stored object 

341 """ 

342 return self._stride_through_scopes(name, scope)[2] 

343 

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}") 

359 

360 def search_name(self, name: str) -> List[str]: 

361 """ 

362 Search for all occurrences of given `name` in the entire data store. 

363 

364 :param name: Name to look for 

365 

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 []) 

369 

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. 

374 

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`. 

377 

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 

382 

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) 

414 

415 def list_all_scopes(self) -> List[str]: 

416 """ 

417 List all available scopes in data store. 

418 

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) 

427 

428 def list_all_names(self) -> List[str]: 

429 """ 

430 List all names available in the data store. 

431 

432 :return: all names 

433 """ 

434 return sorted(self._store.keys()) 

435 

436 

437class DataStoreByScope(AbstractDataStore): 

438 """ 

439 Data store for all settings for the experiment workflow. 

440 

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. 

444 

445 This implementation stores data as 

446 

447 .. code-block:: 

448 

449 <scope1> 

450 <variable1>: value 

451 <variable2>: value 

452 <scope2> 

453 <variable1>: value 

454 <variable3>: value 

455 

456 """ 

457 

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`. 

463 

464 In the current implementation, existing entries are overwritten. 

465 

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}") 

476 

477 @CorrectScope 

478 @TrackParameter 

479 def get(self, name: str, scope: str) -> Any: 

480 """ 

481 Retrieve an object with `name` from `scope`. 

482 

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). 

487 

488 :param name: Name to look for 

489 :param scope: scope to search the name for 

490 

491 :return: the stored object 

492 """ 

493 return self._stride_through_scopes(name, scope)[2] 

494 

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}") 

510 

511 def search_name(self, name: str) -> List[str]: 

512 """ 

513 Search for all occurrences of given `name` in the entire data store. 

514 

515 :param name: Name to look for 

516 

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) 

524 

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. 

529 

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`. 

532 

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 

537 

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) 

565 

566 def list_all_scopes(self) -> List[str]: 

567 """ 

568 List all available scopes in data store. 

569 

570 :return: names of all stored objects 

571 """ 

572 return sorted(self._store.keys()) 

573 

574 def list_all_names(self) -> List[str]: 

575 """ 

576 List all names available in the data store. 

577 

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)