Coverage for test/unit/elements/graphs/test_hvacgraph.py: 100%

1import unittest 

2 

3import networkx as nx 

4 

5from bim2sim.elements import hvac_elements as hvac 

6from bim2sim.elements.hvac_elements import HVACPort 

7from bim2sim.elements.graphs import hvac_graph 

8from test.unit.elements.helper import SetupHelperHVAC 

9 

10 

11class GraphHelper(SetupHelperHVAC): 

12 

13 def get_system_elements(self): 

14 """ Simple generator system made of boiler, pump, expansion tank, 

15 distributor and pipes. 

16 """ 

17 flags = {} 

18 with self.flag_manager(flags): 

19 # generator circuit 

20 boiler = self.element_generator(hvac.Boiler, rated_power=200) 

21 gen_vl_a = [ 

22 self.element_generator(hvac.Pipe, length=100, diameter=40) 

23 for i in range(3)] 

24 h_pump = self.element_generator( 

25 hvac.Pump, rated_power=2.2, rated_height=12, 

26 rated_volume_flow=8) 

27 gen_vl_b = [ 

28 self.element_generator( 

29 hvac.Pipe, flags=['strand1'], length=100, diameter=40) 

30 for i in range(5)] 

31 distributor = self.element_generator( 

32 hvac.Distributor, flags=['distributor']) # , volume=80 

33 gen_rl_a = [ 

34 self.element_generator(hvac.Pipe, length=100, diameter=40) 

35 for i in range(4)] 

36 fitting = self.element_generator( 

37 hvac.PipeFitting, n_ports=3, diameter=40, length=60) 

38 gen_rl_b = [ 

39 self.element_generator(hvac.Pipe, length=100, diameter=40) 

40 for i in range(4)] 

41 gen_rl_c = [ 

42 self.element_generator( 

43 hvac.Pipe, flags=['strand2'], length=(1 + i) * 40, 

44 diameter=15) 

45 for i in range(3) 

46 ] 

47 tank = self.element_generator(hvac.Storage, n_ports=1) 

48 

49 # connect 

50 gen_vl = [boiler, *gen_vl_a, h_pump, *gen_vl_b, distributor] 

51 self.connect_strait(gen_vl) 

52 

53 self.connect_strait([distributor, *gen_rl_a, fitting]) 

54 self.connect_strait([fitting, *gen_rl_b, boiler]) 

55 self.connect_strait([*gen_rl_c, tank]) 

56 fitting.ports[2].connect(gen_rl_c[0].ports[0]) 

57 

58 # full system 

59 gen_circuit = [ 

60 boiler, *gen_vl_a, h_pump, *gen_vl_b, distributor, 

61 *gen_rl_a, fitting, *gen_rl_b, *gen_rl_c, tank 

62 ] 

63 

64 return gen_circuit, flags 

65 

66 

67def generate_element_strait(number=5, prefix=""): 

68 """ Generate 2 port elements and connect them in line.""" 

69 elements = [] 

70 

71 # create elements 

72 for i in range(number): 

73 ele = hvac.HVACProduct() 

74 ele.name = prefix + str(i) 

75 ele.ports.append(hvac.HVACPort(ele)) 

76 ele.ports.append(hvac.HVACPort(ele)) 

77 ele.inner_connections.extend(ele.get_inner_connections()) 

78 elements.append(ele) 

79 # connect 

80 if i > 0: 

81 elements[i - 1].ports[1].connect(elements[i].ports[0]) 

82 return elements 

83 

84 

85def attach(element1, element2, use_existing=False): 

86 """ Connect elements. 

87 

88 If use_existing is True free ports (if any) are used else new ports 

89 are created. 

90 """ 

91 if use_existing: 

92 free_ports1 = [port for port in element1.ports if not port.connection] 

93 free_ports2 = [port for port in element2.ports if not port.connection] 

94 else: 

95 free_ports1 = [] 

96 free_ports2 = [] 

97 

98 if free_ports1: 

99 port_e1 = free_ports1[0] 

100 else: 

101 port_e1 = HVACPort(element1) 

102 element1.ports.append(port_e1) 

103 

104 if free_ports2: 

105 port_e2 = free_ports2[0] 

106 else: 

107 port_e2 = HVACPort(element2) 

108 element2.ports.append(port_e2) 

109 

110 element1.inner_connections = element1.get_inner_connections() 

111 element2.inner_connections = element2.get_inner_connections() 

112 

113 port_e1.connect(port_e2) 

114 

115 

116# @patch.object(element.BaseElement, '__repr__', lambda self: self.name) 

117class TestHVACGraph(unittest.TestCase): 

118 helper = None 

119 

120 @classmethod 

121 def setUpClass(cls): 

122 cls.helper = GraphHelper() 

123 

124 def tearDown(self) -> None: 

125 self.helper.reset() 

126 

127 def test_create(self): 

128 """ Test instantiating and basic behaviour.""" 

129 strait = generate_element_strait() 

130 single_loop = generate_element_strait() 

131 attach(single_loop[0], single_loop[-1]) 

132 

133 graph_strait = hvac_graph.HvacGraph(strait) 

134 self.assertSetEqual(set(strait), set(graph_strait.elements)) 

135 self.assertEqual(len(graph_strait.get_connections()), len(strait) - 1) 

136 

137 graph_loop = hvac_graph.HvacGraph(single_loop) 

138 self.assertSetEqual(set(single_loop), set(graph_loop.elements)) 

139 self.assertEqual(len(graph_loop.get_connections()), len(single_loop)) 

140 

141 def test_merge(self): 

142 """ Test merging elements into simpler representation.""" 

143 strait = generate_element_strait(10, "strait") 

144 to_replace = strait[2:-3] 

145 replacement = generate_element_strait(1, "replacement")[0] 

146 

147 graph = hvac_graph.HvacGraph(strait) 

148 mapping = {port: None for ele in to_replace for port in ele.ports} 

149 mapping[to_replace[0].ports[0]] = replacement.ports[0] 

150 mapping[to_replace[-1].ports[1]] = replacement.ports[1] 

151 

152 graph.merge(mapping, [(replacement.ports[0], replacement.ports[1])]) 

153 

154 for ele in graph.elements: 

155 self.assertNotIn(ele, to_replace) 

156 

157 path_port = nx.shortest_path( 

158 graph, strait[0].ports[0], strait[-1].ports[1]) 

159 self.assertIn(replacement.ports[0], path_port) 

160 self.assertIn(replacement.ports[-1], path_port) 

161 

162 path_element = nx.shortest_path( 

163 graph.element_graph, strait[0], strait[-1]) 

164 self.assertIn(replacement, path_element) 

165 

166 def test_type_chain(self): 

167 """ Test chain detection.""" 

168 elements, flags = self.helper.get_system_elements() 

169 port_graph = hvac_graph.HvacGraph(elements) 

170 ele_graph = port_graph.element_graph 

171 

172 wanted = [hvac.Pipe] 

173 chains = hvac_graph.HvacGraph.get_type_chains(ele_graph, wanted) 

174 self.assertEqual(5, len(chains), "Unexpected number of chains found!") 

175 

176 wanted = [hvac.Pipe, hvac.Pump] 

177 chains2 = hvac_graph.HvacGraph.get_type_chains(ele_graph, wanted) 

178 self.assertEqual(4, len(chains2), "Unexpected number of chains found!") 

179 

180 def test_cycles(self): 

181 """ Test cycle detection.""" 

182 # generate single cycle 

183 core_cycle = generate_element_strait(10, 'core') 

184 attach(core_cycle[0], core_cycle[-1], True) 

185 

186 # test single cycle 

187 graph1 = hvac_graph.HvacGraph(core_cycle) 

188 cyc1 = graph1.get_cycles() 

189 self.assertEqual(len(cyc1), 1) 

190 self.assertSetEqual( 

191 set(cyc1[0]), 

192 {port for ele in core_cycle for port in ele.ports}) 

193 

194 # generate second cycle and attach it with loose end 

195 attached1 = generate_element_strait(prefix='attached') 

196 attach(core_cycle[0], attached1[0]) 

197 attach(core_cycle[5], attached1[-2]) 

198 

199 # test double cycle 

200 graph2 = hvac_graph.HvacGraph(core_cycle + attached1) 

201 # graph2.plot(ports=True) 

202 # graph2.plot(ports=False) 

203 cyc2 = graph2.get_cycles() 

204 self.assertEqual(len(cyc2), 2) 

205 cyc2_elements = {port.parent for port in cyc2[0] + cyc2[1]} 

206 ref_elements = set(core_cycle + attached1[:-1]) 

207 self.assertSetEqual(cyc2_elements, ref_elements) 

208 

209 def test_nodes(self): 

210 """ Element and port graph nodes.""" 

211 elements, flags = self.helper.get_system_elements() 

212 all_elements = set(elements) 

213 all_ports = {port for ele in elements for port in ele.ports} 

214 

215 port_graph = hvac_graph.HvacGraph(elements) 

216 self.assertSetEqual(set(port_graph.nodes), all_ports) 

217 self.assertSetEqual(set(port_graph.elements), all_elements) 

218 

219 ele_graph = port_graph.element_graph 

220 self.assertSetEqual(set(ele_graph.nodes), all_elements) 

221 

222 def test_subgraph_from_elements(self): 

223 """ Test creating a subgraph of the port_graph with given elements.""" 

224 elements, flags = self.helper.get_system_elements() 

225 port_graph = hvac_graph.HvacGraph(elements) 

226 

227 # test subgraph creation 

228 sub_graph = port_graph.subgraph_from_elements(flags['strand1']) 

229 self.assertEqual(set(sub_graph.elements), set(flags['strand1'])) 

230 

231 # test if AssertionError is raised if elements are not in port_graph 

232 classes_to_remove = {hvac.Pipe} 

233 new_port_graph = hvac_graph.HvacGraph.remove_classes_from( 

234 port_graph, classes_to_remove) 

235 with self.assertRaises(AssertionError): 

236 new_port_graph.subgraph_from_elements(flags['strand1']) 

237 

238 def test_remove_classes(self): 

239 """ Test remove elements of given classes from port graph and 

240 element graph.""" 

241 elements, flags = self.helper.get_system_elements() 

242 port_graph = hvac_graph.HvacGraph(elements) 

243 ele_graph = port_graph.element_graph 

244 

245 classes_to_remove = {hvac.Boiler, hvac.Distributor} 

246 new_port_graph = hvac_graph.HvacGraph.remove_classes_from( 

247 port_graph, classes_to_remove) 

248 new_ele_graph = hvac_graph.HvacGraph.remove_classes_from( 

249 ele_graph, classes_to_remove 

250 ) 

251 

252 self.assertFalse(hvac.Boiler and hvac.Distributor 

253 in {ele.__class__ for ele in new_port_graph.elements}) 

254 self.assertFalse(hvac.Boiler and hvac.Distributor 

255 in {node.__class__ for node in new_ele_graph.nodes})