Coverage for physioblocks / configuration / aliases.py: 100%

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

26 

27""" 

28Declares functions to handle configuration aliases. 

29 

30**Aliases** are :class:`~physioblocks.configuration.base.Configuration` 

31instances representing a complete or incomplete type and that can be loaded 

32and completed or overwritten with the load function. 

33""" 

34 

35import functools 

36from copy import copy 

37from pathlib import Path 

38from typing import Any, TypeAlias 

39 

40from physioblocks.configuration.base import Configuration 

41from physioblocks.io.configuration import read_json 

42 

43ConfigurationAlias: TypeAlias = Configuration | dict[str, Any] 

44"""type alias for configuration aliases""" 

45 

46__aliases_register: dict[str, ConfigurationAlias] = {} 

47 

48 

49def add_alias(alias_key: str, alias: ConfigurationAlias) -> None: 

50 """ 

51 Add an alias to the global alias register 

52 

53 :param alias_key: the alias unique key 

54 :type alias_key: str 

55 

56 :param alias_key: the alias unique key 

57 :type alias_key: str 

58 """ 

59 if alias_key in __aliases_register: 

60 raise KeyError(str.format("Alias key {0} is already registered.", alias_key)) 

61 

62 if isinstance(alias, Configuration | dict) is False: 

63 raise TypeError( 

64 str.format( 

65 "Incorrect type for alias: {0}, expected {1}", 

66 type(alias).__name__, 

67 Configuration.__name__, 

68 ) 

69 ) 

70 

71 __aliases_register[alias_key] = alias 

72 

73 

74def has_alias(alias_key: str) -> bool: 

75 """ 

76 Get if an alias is defined for the given key. 

77 

78 :param alias_key: the key to test 

79 :type alias_key: str 

80 

81 :return: True if an alias is defined for the key, False otherwise 

82 :rtype: bool 

83 """ 

84 if alias_key in __aliases_register: 

85 return True 

86 

87 # if the alias is not registered, check if the given alias key 

88 # is a path to the alias configuration file. 

89 alias_path = Path(alias_key) 

90 return alias_path.exists() is True and alias_path.is_file() is True 

91 

92 

93@functools.singledispatch 

94def unwrap_aliases(item: Any) -> Any: 

95 """ 

96 Process the given item and create a new one with all its alias replaced with 

97 their matching configuration object recursivly. 

98 

99 :param item: the item to process 

100 :param item: Configuration | dict | list 

101 

102 :return: the item to process 

103 :rtype: Configuration | dict | list 

104 """ 

105 raise TypeError(str.format("Type {0} can not load aliases.", type(item).__name__)) 

106 

107 

108@unwrap_aliases.register 

109def _unwrap_aliases_config(configuration: Configuration) -> Any: 

110 new_config: ConfigurationAlias = configuration.copy() 

111 

112 # recursively unwrap the alias defined by the label 

113 while isinstance(new_config, Configuration) and has_alias(new_config.label): 

114 new_alias = get_alias(new_config.label) 

115 new_config = _update_configuration(new_config, new_alias) 

116 

117 # update the alias values with the current config values 

118 new_config = _update_configuration(configuration, new_config) 

119 

120 # The unwrap configuration values 

121 if isinstance(new_config, Configuration): 

122 values_config: dict[str, Any] = unwrap_aliases(new_config.configuration_items) 

123 new_config.configuration_items = values_config 

124 return new_config 

125 else: 

126 return unwrap_aliases(new_config) 

127 

128 

129@unwrap_aliases.register 

130def _unwrap_aliases_dict(items: dict) -> dict[str, Any]: # type: ignore 

131 return { 

132 key: value 

133 if isinstance(value, Configuration | dict | list) is False 

134 else unwrap_aliases(value) 

135 for key, value in items.items() 

136 } 

137 

138 

139@unwrap_aliases.register 

140def _unwrap_aliases_list(items: list) -> list[Any]: # type: ignore 

141 return [ 

142 value 

143 if isinstance(value, Configuration | dict | list) is False 

144 else unwrap_aliases(value) 

145 for value in items 

146 ] 

147 

148 

149def get_alias(alias_key: str) -> Any: 

150 """ 

151 Get an alias from the global alias register. 

152 

153 :param alias_key: the alias unique key 

154 :type alias_key: str 

155 

156 :return: the alias 

157 :rtype: Configuration | dict | list 

158 """ 

159 

160 alias_config = None 

161 if alias_key in __aliases_register: 

162 alias_config = copy(__aliases_register[alias_key]) 

163 else: 

164 alias_config = read_json(alias_key) 

165 

166 return alias_config 

167 

168 

169def _update_configuration( 

170 configuration: Configuration, alias: ConfigurationAlias 

171) -> ConfigurationAlias: 

172 if isinstance(alias, Configuration): 

173 new_values = alias.configuration_items.copy() 

174 new_values = _recursive_update(new_values, configuration.configuration_items) 

175 return Configuration(alias.label, new_values) 

176 elif isinstance(alias, dict): 

177 new_values = alias.copy() 

178 new_values = _recursive_update(new_values, configuration.configuration_items) 

179 return new_values 

180 raise TypeError(str.format("{0} is not a valid alias type.", type(alias).__name__)) 

181 

182 

183def _recursive_update( 

184 updated: dict[str, Any], updater: dict[str, Any] 

185) -> dict[str, Any]: 

186 new_values = updated.copy() 

187 for key, value in updater.items(): 

188 if key in updated and isinstance(value, dict): 

189 new_values[key] = _recursive_update(updated[key], value) 

190 else: 

191 new_values[key] = value 

192 

193 return new_values 

194 

195 

196def remove_alias(alias_key: str) -> None: 

197 """ 

198 Remove an alias from the global alias register 

199 

200 :param alias_key: the alias unique key 

201 :type alias_key: str 

202 """ 

203 __aliases_register.pop(alias_key)