Coverage for physioblocks / configuration / simulation / simulations.py: 96%

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

28Define the configuration of simulations 

29""" 

30 

31from collections.abc import Iterable 

32from os import linesep 

33from typing import Any 

34 

35from physioblocks.configuration.base import Configuration, ConfigurationError 

36from physioblocks.configuration.constants import ( 

37 INIT_VARIABLES_ID, 

38 MAGNITUDES, 

39 NET_ID, 

40 OUTPUTS_FUNCTIONS_ID, 

41 PARAMETERS_ID, 

42 SOLVER_ID, 

43 TIME_MANAGER_ID, 

44 VARIABLES_MAGNITUDES, 

45) 

46from physioblocks.configuration.functions import load, save 

47from physioblocks.registers.load_function_register import loads 

48from physioblocks.registers.save_function_register import saves 

49from physioblocks.registers.type_register import get_registered_type_id 

50from physioblocks.simulation import AbstractFunction 

51from physioblocks.simulation.functions import ( 

52 is_state_function, 

53 is_time_function, 

54) 

55from physioblocks.simulation.runtime import AbstractSimulation 

56from physioblocks.simulation.setup import SimulationFactory 

57from physioblocks.simulation.state import STATE_NAME_ID 

58from physioblocks.simulation.time_manager import TIME_QUANTITY_ID 

59 

60 

61@loads(AbstractSimulation) # type: ignore 

62def load_simulation_config( 

63 config: Configuration, 

64 configuration_type: type, 

65 configuration_object: AbstractSimulation | None = None, 

66 *args: Any, 

67 **kwargs: Any, 

68) -> AbstractSimulation: 

69 """ 

70 Load a simulation from a configuration. 

71 

72 :param config: the configuration 

73 :type config: Configuration 

74 

75 :return: the simulation 

76 :rtype: AbstractSimulation 

77 """ 

78 

79 if configuration_object is None: 

80 net = None 

81 if NET_ID in config: 

82 net = load(config[NET_ID]) 

83 

84 # Solver 

85 solver = None 

86 if SOLVER_ID in config: 

87 solver = load(config[SOLVER_ID]) 

88 

89 # magnitudes 

90 magnitudes = None 

91 if VARIABLES_MAGNITUDES in config: 

92 magnitudes = load(config[VARIABLES_MAGNITUDES]) 

93 

94 sim_factory = SimulationFactory( 

95 configuration_type, 

96 solver, 

97 net, 

98 simulation_options={MAGNITUDES: magnitudes}, 

99 ) 

100 

101 configuration_object = sim_factory.create_simulation() 

102 

103 _configure_simulation(config, configuration_object) 

104 

105 return configuration_object 

106 

107 

108def _save_sim_factory(factory: SimulationFactory) -> Configuration: 

109 simulation_type_id = get_registered_type_id(factory.simulation_type) 

110 sim_factory_config_item = Configuration(simulation_type_id) 

111 sim_factory_config_item[NET_ID] = save(factory.net) 

112 

113 return sim_factory_config_item 

114 

115 

116@saves(AbstractSimulation) 

117def save_simulation_config( 

118 simulation: AbstractSimulation, *args: Any, **kwargs: Any 

119) -> Configuration: 

120 """ 

121 Save a simulation in a configuration. 

122 

123 :param sim: the simulation to save 

124 :type sim: AbstractSimulation 

125 

126 :param simulation_type_id: the simulation type id 

127 :type simulation_type_id: str 

128 

129 :return: the configuration 

130 :rtype: Configuration 

131 """ 

132 

133 sim_config = _save_sim_factory(simulation.factory) 

134 

135 sim_config[TIME_MANAGER_ID] = save(simulation.time_manager) 

136 sim_config[SOLVER_ID] = save(simulation.solver) 

137 

138 # State 

139 variable_init_values = save(simulation.state.variables) 

140 if isinstance(variable_init_values, dict): 

141 sim_config[INIT_VARIABLES_ID] = variable_init_values 

142 else: 

143 raise ConfigurationError( 

144 str.format( 

145 "Expected a dict for {0} configuration, got {1}.", 

146 INIT_VARIABLES_ID, 

147 type(variable_init_values).__name__, 

148 ) 

149 ) 

150 sim_config[VARIABLES_MAGNITUDES] = save(simulation.magnitudes) 

151 

152 # Parameters 

153 # Get quantities 

154 parameters: dict[str, Any] = { 

155 key: qty 

156 for key, qty in simulation.parameters.items() 

157 if key not in simulation.update_functions 

158 } 

159 references: dict[str, Any] = simulation.models.copy() 

160 parameters_config: dict[str, Any] = save( 

161 parameters, configuration_references=references 

162 ) 

163 references.update(parameters) 

164 

165 # update with fonctions 

166 function_config = save( 

167 simulation.update_functions, configuration_references=references 

168 ) 

169 parameters_config.update(function_config) 

170 sim_config[PARAMETERS_ID] = parameters_config 

171 

172 # All the quantities (with update function in the references) 

173 references.update(simulation.parameters) 

174 

175 if len(simulation.outputs_functions) > 0: 

176 sim_config[OUTPUTS_FUNCTIONS_ID] = save( 

177 simulation.outputs_functions, 

178 configuration_references=references, 

179 ) 

180 

181 return sim_config 

182 

183 

184def _configure_simulation( 

185 config: Configuration, simulation: AbstractSimulation 

186) -> None: 

187 # Set initial values 

188 

189 # ParameterRegister 

190 _check_exising_config(PARAMETERS_ID, config) 

191 _check_missing_keys(PARAMETERS_ID, config[PARAMETERS_ID], simulation.parameters) 

192 

193 constants_config = { 

194 key: value 

195 for key, value in config[PARAMETERS_ID].items() 

196 if isinstance(value, Configuration) is False 

197 } 

198 functions_config = { 

199 key: value 

200 for key, value in config[PARAMETERS_ID].items() 

201 if isinstance(value, Configuration) is True 

202 } 

203 

204 # first load the constants 

205 load(constants_config, configuration_object=simulation.parameters) 

206 

207 # then load functions 

208 references: dict[str, Any] = simulation.quantities 

209 references.update(simulation.models) 

210 loaded_functions = load( 

211 functions_config, configuration_references=references, configuration_sort=True 

212 ) 

213 __initialize_functions(simulation, loaded_functions) 

214 

215 # Time Manager 

216 _check_exising_config(TIME_MANAGER_ID, config) 

217 load(config[TIME_MANAGER_ID], configuration_object=simulation.time_manager) 

218 

219 # State 

220 _check_exising_config(INIT_VARIABLES_ID, config) 

221 _check_missing_keys(INIT_VARIABLES_ID, config[INIT_VARIABLES_ID], simulation.state) 

222 

223 load( 

224 config[INIT_VARIABLES_ID], 

225 configuration_object=simulation.state, 

226 configuration_references=simulation.quantities, 

227 ) 

228 

229 references.update(simulation.quantities) 

230 references.update(simulation.models) 

231 

232 if OUTPUTS_FUNCTIONS_ID in config: 

233 outputs = load( 

234 config[OUTPUTS_FUNCTIONS_ID], configuration_references=references 

235 ) 

236 __configure_outputs(simulation, outputs) 

237 

238 

239def __initialize_functions( 

240 sim: AbstractSimulation, functions_parameters: dict[str, AbstractFunction] 

241) -> None: 

242 for param_id, param_value in functions_parameters.items(): 

243 if param_id in sim.parameters: 

244 arguments: dict[str, Any] = {} 

245 if is_time_function(param_value): 

246 sim.register_timed_parameter_update(param_id, param_value) 

247 arguments[TIME_QUANTITY_ID] = sim.time_manager.time.current 

248 if is_state_function(param_value): 

249 arguments[STATE_NAME_ID] = sim.state 

250 sim.parameters[param_id].initialize(param_value.eval(**arguments)) 

251 

252 

253def __configure_outputs( 

254 sim: AbstractSimulation, config: dict[str, AbstractFunction] 

255) -> None: 

256 for output_id, update_func in config.items(): 

257 sim.register_output_function(output_id, update_func) 

258 

259 

260def _check_exising_config(key: str, config: Configuration) -> None: 

261 if key not in config: 

262 raise ConfigurationError( 

263 str.format( 

264 "Missing key {0} in {1} configuration.", 

265 key, 

266 config.label, 

267 ) 

268 ) 

269 

270 

271def _check_missing_keys( 

272 name: str, tested: Iterable[str], reference: Iterable[str] 

273) -> None: 

274 missing_keys = [key for key in reference if key not in tested] 

275 if len(missing_keys) > 0: 

276 raise ConfigurationError( 

277 str.format( 

278 "Missing keys in {0} configuration:{2}{1}", name, missing_keys, linesep 

279 ) 

280 )