Coverage for physioblocks / configuration / description / nets.py: 98%

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 

27from typing import Any 

28 

29from physioblocks.configuration.base import Configuration, ConfigurationError 

30from physioblocks.configuration.constants import ( 

31 BLOCKS_ITEM_ID, 

32 BOUNDARIES_ID, 

33 FLUX_DOF_DEFINITION_ID, 

34 NET_ID, 

35 NODES_ITEM_ID, 

36) 

37from physioblocks.configuration.functions import load, save 

38from physioblocks.description.blocks import BlockDescription 

39from physioblocks.description.flux import get_flux_dof_register 

40from physioblocks.description.nets import BoundaryCondition, Net 

41from physioblocks.registers.load_function_register import loads 

42from physioblocks.registers.save_function_register import saves 

43 

44 

45@saves(Net) 

46def save_net_config(net: Net, *args: Any, **kwargs: Any) -> Configuration: 

47 """ 

48 Create a Configuration for a net 

49 

50 :param net: the net 

51 :type net: Net 

52 

53 :return: the configuration 

54 :rtype: Configuration 

55 """ 

56 # save blocks 

57 blocks_config = save(net.blocks, *args, **kwargs) 

58 if isinstance(blocks_config, dict) is False: 

59 raise ConfigurationError( 

60 str.format( 

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

62 BLOCKS_ITEM_ID, 

63 type(net.blocks).__name__, 

64 ) 

65 ) 

66 

67 # add node informations 

68 for block_id, block in net.blocks.items(): 

69 block_connections = { 

70 str(node_index): net.local_to_global_node_id(block_id, node_index) 

71 for node_index in block.described_type.nodes 

72 } 

73 

74 blocks_config[block_id][NODES_ITEM_ID] = block_connections 

75 

76 node_ids = list(net.nodes.keys()) 

77 flux_dof_register = get_flux_dof_register() 

78 

79 net_config = Configuration(NET_ID) 

80 net_config[FLUX_DOF_DEFINITION_ID] = save(flux_dof_register.flux_dof_couples) 

81 net_config[NODES_ITEM_ID] = node_ids 

82 net_config[BLOCKS_ITEM_ID] = blocks_config 

83 net_config[BOUNDARIES_ID] = save(net.boundary_conditions, *args, **kwargs) 

84 

85 return net_config 

86 

87 

88@loads(Net) 

89def load_net_config( 

90 configuration: Configuration, 

91 configuration_type: type[Net], 

92 configurable_object: Net | None = None, 

93 *args: Any, 

94 **kwargs: Any, 

95) -> Net: 

96 """ 

97 Load a net from a configuration. 

98 

99 :param configuration: the configuration 

100 :type configuration: Configuration 

101 

102 :param boundaries: the boundaries of the net at each node 

103 :type boundaries: dict[str, list[BoundaryCondition]] 

104 

105 :return: the net 

106 :rtype: Net 

107 """ 

108 

109 configurable_net = ( 

110 configuration_type() if configurable_object is None else configurable_object 

111 ) 

112 nodes_ids = configuration[NODES_ITEM_ID] 

113 

114 # Fill the flux type singleton from the configured value 

115 if FLUX_DOF_DEFINITION_ID in configuration: 

116 load( 

117 configuration[FLUX_DOF_DEFINITION_ID], 

118 configuration_object=get_flux_dof_register(), 

119 ) 

120 

121 for node_id in nodes_ids: 

122 configurable_net.add_node(node_id) 

123 

124 blocks_config: dict[str, Configuration] = configuration[BLOCKS_ITEM_ID].copy() 

125 connections_config: dict[str, dict[str, str]] = { 

126 block_key: block_config.configuration_items.pop(NODES_ITEM_ID) 

127 for block_key, block_config in blocks_config.items() 

128 } 

129 

130 connections: dict[str, dict[int, str]] = { 

131 block_id: { 

132 int(node_index): global_node 

133 for node_index, global_node in connection_config.items() 

134 } 

135 for block_id, connection_config in connections_config.items() 

136 } 

137 blocks: dict[str, BlockDescription] = load(blocks_config) 

138 

139 for block_id, block in blocks.items(): 

140 # remove the node key (on a block configuration copy) before loading the block 

141 configurable_net.add_block(block_id, block, connections[block_id]) 

142 

143 boundaries: dict[str, list[BoundaryCondition]] | None = ( 

144 load(configuration[BOUNDARIES_ID]) if BOUNDARIES_ID in configuration else None 

145 ) 

146 

147 # Add boundaries conditions 

148 if boundaries is not None: 

149 for node_id, conditions in boundaries.items(): 

150 for bc in conditions: 

151 configurable_net.set_boundary( 

152 node_id, bc.condition_type, bc.condition_id 

153 ) 

154 

155 return configurable_net