Coverage for test/unit/elements/aggregation/test_pipestrand.py: 91%

1import unittest 

2 

3import bim2sim.elements.aggregation.hvac_aggregations 

4from bim2sim.elements import aggregation 

5from bim2sim.elements import hvac_elements as hvac 

6from bim2sim.elements.graphs.hvac_graph import HvacGraph 

7from bim2sim.elements.mapping.units import ureg 

8from test.unit.elements.helper import SetupHelperHVAC 

9 

10 

11class StrandHelper(SetupHelperHVAC): 

12 

13 def get_setup_strand1(self): 

14 """simple strait strand""" 

15 flags = {} 

16 with self.flag_manager(flags): 

17 # generator circuit 

18 strand = [self.element_generator( 

19 hvac.Pipe, length=100, diameter=30) for i in range(10)] 

20 

21 # connect 

22 self.connect_strait(strand) 

23 

24 # full system 

25 gen_circuit = [ 

26 *strand 

27 ] 

28 

29 flags['connect'] = [strand[0], strand[-1]] 

30 

31 graph = HvacGraph(gen_circuit) 

32 flags['edge_ports'] = [v for v, d in graph.degree() if d == 1] 

33 

34 return graph, flags 

35 

36 def get_setup_strand2(self): 

37 """simple strait strand with various diameters""" 

38 flags = {} 

39 with self.flag_manager(flags): 

40 # generator circuit 

41 strand1 = [self.element_generator( 

42 hvac.Pipe, length=100, diameter=30) for i in range(2)] 

43 strand2 = [self.element_generator( 

44 hvac.Pipe, length=200, diameter=50) for i in range(2)] 

45 strand3 = [self.element_generator( 

46 hvac.Pipe, length=100, diameter=30) for i in range(2)] 

47 strand4 = [self.element_generator( 

48 hvac.Pipe, length=50, diameter=15) for i in range(2)] 

49 

50 strand = [*strand1, *strand2, *strand3, *strand4] 

51 # connect 

52 self.connect_strait(strand) 

53 

54 # full system 

55 gen_circuit = [ 

56 *strand 

57 ] 

58 

59 flags['connect'] = [strand[0], strand[-1]] 

60 

61 graph = HvacGraph(gen_circuit) 

62 return graph, flags 

63 

64 def get_setup_strait_with_valve(self): 

65 """simple strait strand with valve""" 

66 flags = {} 

67 with self.flag_manager(flags): 

68 # generator circuit 

69 strand1 = [self.element_generator( 

70 hvac.Pipe, flags=['pipes'], length=100, diameter=30) for i in range(3)] 

71 strand2 = [self.element_generator( 

72 hvac.Pipe, flags=['pipes'], length=100, diameter=30) for i in range(3)] 

73 valve = self.element_generator(hvac.Valve, flags=['valve'], diameter=30) 

74 

75 strand = [*strand1, valve, *strand2] 

76 # connect 

77 self.connect_strait(strand) 

78 

79 # full system 

80 gen_circuit = [ 

81 *strand 

82 ] 

83 

84 flags['connect'] = [strand[0], strand[-1]] 

85 

86 graph = HvacGraph(gen_circuit) 

87 return graph, flags 

88 

89 def get_setup_straits_with_distributor(self): 

90 """distributor wit two connected straits""" 

91 flags = {} 

92 with self.flag_manager(flags): 

93 # generator circuit 

94 strand1 = [self.element_generator( 

95 hvac.Pipe, length=100, diameter=30) for i in range(2)] 

96 strand2 = [self.element_generator( 

97 hvac.Pipe, length=200, diameter=50) for i in range(2)] 

98 distributor = self.element_generator(hvac.Distributor, flags=['distributor']) 

99 

100 # connect 

101 self.connect_strait([*strand1, distributor, *strand2]) 

102 

103 # full system 

104 gen_circuit = [ 

105 *strand1, 

106 distributor, 

107 *strand2 

108 ] 

109 

110 graph = HvacGraph(gen_circuit) 

111 return graph, flags 

112 

113 def get_setup_cross(self): 

114 """two crossing strands""" 

115 flags = {} 

116 with self.flag_manager(flags): 

117 # generator circuit 

118 strand1 = [self.element_generator( 

119 hvac.Pipe, flags=['strand1'], length=100, diameter=30) for i in range(4)] 

120 strand2 = [self.element_generator( 

121 hvac.Pipe, flags=['strand2'], length=100, diameter=30) for i in range(4)] 

122 strand3 = [self.element_generator( 

123 hvac.Pipe, flags=['strand3'], length=100, diameter=30) for i in range(4)] 

124 strand4 = [self.element_generator( 

125 hvac.Pipe, flags=['strand4'], length=100, diameter=30) for i in range(4)] 

126 cross = self.element_generator(hvac.PipeFitting, n_ports=4, flags='cross') 

127 

128 # connect 

129 self.connect_strait(strand1) 

130 self.connect_strait(strand2) 

131 self.connect_strait(strand3) 

132 self.connect_strait(strand4) 

133 

134 cross.ports[0].connect(strand1[0].ports[0]) 

135 cross.ports[1].connect(strand2[0].ports[0]) 

136 cross.ports[2].connect(strand3[0].ports[0]) 

137 cross.ports[3].connect(strand4[0].ports[0]) 

138 

139 # full system 

140 gen_circuit = [ 

141 *strand1, 

142 *strand2, 

143 *strand3, 

144 *strand4, 

145 cross 

146 ] 

147 

148 graph = HvacGraph(gen_circuit) 

149 return graph, flags 

150 

151 def get_setup_system(self): 

152 """System (Kessel - Pumpe - Heizung - Absperrventil - T-Stück - Druckausgleichgefäß)""" 

153 flags = {} 

154 with self.flag_manager(flags): 

155 # generator circuit 

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

157 strand1 = [self.element_generator(hvac.Pipe, flags=['strand1'], length=100, diameter=40) for i in range(3)] 

158 h_pump = self.element_generator(hvac.Pump, rated_power=2.2, rated_height=12, rated_volume_flow=8) 

159 strand2 = [self.element_generator(hvac.Pipe, flags=['strand2'], length=100, diameter=40) for i in range(5)] 

160 spaceheater = self.element_generator(hvac.SpaceHeater, flags=['spaceheater']) # , volume=80 

161 strand3a = [self.element_generator(hvac.Pipe, flags=['strand3'], length=100, diameter=40) for i in range(4)] 

162 valve = self.element_generator(hvac.Valve, flags=['valve']) 

163 strand3b = [self.element_generator(hvac.Pipe, flags=['strand3'], length=100, diameter=40) for i in range(4)] 

164 fitting = self.element_generator(hvac.PipeFitting, n_ports=3, diameter=40, length=60) 

165 strand4 = [self.element_generator(hvac.Pipe, flags=['strand4'], length=100, diameter=40) for i in range(4)] 

166 strand5 = [ 

167 self.element_generator(hvac.Pipe, flags=['strand5'], length=100, diameter=40) for i in range(4)] 

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

169 

170 # connect 

171 circuit = [ 

172 boiler, *strand1, h_pump, *strand2, spaceheater, 

173 *strand3a, valve, *strand3b, fitting, *strand4, boiler 

174 ] 

175 self.connect_strait(circuit) 

176 self.connect_strait([*strand5, tank]) 

177 fitting.ports[2].connect(strand5[0].ports[0]) 

178 

179 # full system 

180 gen_circuit = [ 

181 *circuit, *strand5, tank 

182 ] 

183 

184 graph = HvacGraph(gen_circuit) 

185 return graph, flags 

186 

187 def get_setup_loop(self): 

188 """Circular strand with diagonal connected strand""" 

189 flags = {} 

190 with self.flag_manager(flags): 

191 # generator circuit 

192 strand1 = [self.element_generator( 

193 hvac.Pipe, flags=['strand1'], length=100, diameter=30) for i in range(6)] 

194 strand2 = [self.element_generator( 

195 hvac.Pipe, flags=['strand2'], length=100, diameter=30) for i in range(6)] 

196 strand3 = [self.element_generator( 

197 hvac.Pipe, flags=['strand3'], length=100, diameter=30) for i in range(4)] 

198 cross1 = self.element_generator(hvac.PipeFitting, n_ports=3, flags='cross') 

199 cross2 = self.element_generator(hvac.PipeFitting, n_ports=3, flags='cross') 

200 

201 # connect 

202 self.connect_strait([cross1, *strand1, cross2]) 

203 self.connect_strait([cross2, *strand2, cross1]) 

204 self.connect_strait(strand3) 

205 cross1.ports[2].connect(strand3[0].ports[0]) 

206 cross2.ports[2].connect(strand3[-1].ports[1]) 

207 

208 # full system 

209 gen_circuit = [ 

210 *strand1, 

211 *strand2, 

212 *strand3, 

213 cross1, 

214 cross2 

215 ] 

216 

217 graph = HvacGraph(gen_circuit) 

218 return graph, flags 

219 

220 

221class TestPipeStrand(unittest.TestCase): 

222 

223 helper = None 

224 

225 @classmethod 

226 def setUpClass(cls): 

227 cls.helper = StrandHelper() 

228 

229 def tearDown(self) -> None: 

230 self.helper.reset() 

231 

232 def test_strait_strand(self): 

233 """Test calculation of aggregated length""" 

234 graph, flags = self.helper.get_setup_strand1() 

235 ele = graph.elements 

236 

237 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

238 self.assertEqual(1, len(matches)) 

239 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, matches[0], **meta[0]) 

240 

241 exp_length = sum([e.length for e in ele]) 

242 self.assertAlmostEqual(exp_length, agg.length) 

243 

244 self.assertAlmostEqual(30 * ureg.millimeter, agg.diameter) 

245 

246 def test_strait_strand_variable(self): 

247 """Test calculation of aggregated length and diameter""" 

248 graph, flags = self.helper.get_setup_strand2() 

249 ele = graph.elements 

250 

251 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

252 self.assertEqual(1, len(matches)) 

253 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, matches[0], **meta[0]) 

254 

255 exp_length = sum([e.length for e in ele]) 

256 self.assertAlmostEqual(exp_length, agg.length) 

257 

258 exp_diameter = sum([e.length * e.diameter for e in ele]) / exp_length 

259 self.assertAlmostEqual(exp_diameter, agg.diameter) 

260 

261 def test_distributor_with_strands(self): 

262 """ Test calculation of aggregated length and diameter.""" 

263 graph, flags = self.helper.get_setup_straits_with_distributor() 

264 

265 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

266 self.assertEqual(2, len(matches)) 

267 

268 with self.assertRaises(AssertionError, 

269 msg="Pipestrand aggregation over a distributor" 

270 " should fail"): 

271 # pass full graph 

272 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, graph, **{}) 

273 

274 @unittest.skip( 

275 "PipeStrand aggregation with inert elements not implemented") 

276 def test_strait_strand_valve(self): 

277 """ Test calculation of aggregated length and diameter.""" 

278 graph, flags = self.helper.get_setup_strait_with_valve() 

279 

280 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

281 self.assertEqual(1, len(matches)) 

282 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, matches[0], **meta[0]) 

283 

284 exp_length = sum([e.length for e in flags['pipes']]) 

285 self.assertAlmostEqual(exp_length, agg.length) 

286 

287 self.assertAlmostEqual(30 * ureg.millimeter, agg.diameter) 

288 

289 def test_filter_strand(self): 

290 """Test filter for strait strand""" 

291 graph, flags = self.helper.get_setup_strand1() 

292 ele = graph.elements 

293 

294 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

295 self.assertEqual(1, len(matches)) 

296 

297 self.assertSetEqual(set(ele), set(matches[0].elements)) 

298 

299 def test_filter_cross(self): 

300 """Test filter for crossing strands""" 

301 graph, flags = self.helper.get_setup_cross() 

302 ele = (flags['strand1'] + flags['strand2'] + flags['strand3'] 

303 + flags['strand4']) 

304 

305 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

306 self.assertEqual(4, len(matches)) 

307 

308 self.assertSetEqual(set(ele), 

309 set(sum( 

310 [list(match.elements) for match in matches], 

311 []))) 

312 

313 def test_filter_system(self): 

314 """Test filter for crossing strands""" 

315 graph, flags = self.helper.get_setup_system() 

316 ele = (flags['strand1'] + flags['strand2'] + flags['strand3'] 

317 + flags['strand4'] + flags['strand5'] + flags['valve']) 

318 

319 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

320 self.assertEqual(5, len(matches)) 

321 

322 self.assertSetEqual(set(ele), 

323 set(sum( 

324 [list(match.elements) for match in matches], 

325 []))) 

326 

327 def test_filter_circular(self): 

328 """Test filter for crossing strands""" 

329 graph, flags = self.helper.get_setup_loop() 

330 ele = flags['strand1'] + flags['strand2'] + flags['strand3'] 

331 

332 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

333 self.assertEqual(len(matches), 3) 

334 

335 self.assertSetEqual(set(ele), 

336 set(sum( 

337 [list(match.elements) for match in matches], 

338 []))) 

339 

340 def test_pipestrand1(self): 

341 """Test calculation of aggregated length and diameter""" 

342 graph, flags = self.helper.get_setup_simple_boiler() 

343 elements = flags['strand1'] 

344 match_graph = graph.subgraph_from_elements(elements) 

345 

346 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(match_graph) 

347 self.assertEqual(1, len(matches)) 

348 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, matches[0], **meta[0]) 

349 

350 exp_length = sum([e.length for e in elements]) 

351 self.assertAlmostEqual(agg.length, exp_length) 

352 

353 self.assertAlmostEqual(40 * ureg.millimeter, agg.diameter) 

354 

355 def test_pipestrand2(self): 

356 """Test calculation of aggregated length and diameter""" 

357 

358 graph, flags = self.helper.get_setup_simple_boiler() 

359 elements = flags['strand2'] 

360 match_graph = graph.subgraph_from_elements(elements) 

361 

362 matches, metas = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(match_graph) 

363 self.assertEqual(1, len(matches)) 

364 aggregations = [] 

365 for match, meta in zip(matches, metas): 

366 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, match, **meta) 

367 aggregations.append(agg) 

368 graph.merge( 

369 mapping=agg.get_replacement_mapping(), 

370 inner_connections=agg.inner_connections) 

371 exp_length = sum([e.length for e in elements]) 

372 self.assertAlmostEqual(exp_length, agg.length) 

373 

374 self.assertAlmostEqual(15 * ureg.millimeter, agg.diameter) 

375 

376 def test_basics(self): 

377 graph, flags = self.helper.get_setup_simple_boiler() 

378 elements = flags['strand1'] 

379 # match_graph = graph.element_graph.subgraph(elements) 

380 # match_graph = graph.subgraph( 

381 # (port for ele in elements for port in ele.ports)) 

382 match = graph.subgraph_from_elements(elements) 

383 

384 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, match) 

385 

386 self.assertTrue(self.helper.elements_in_agg(agg)) 

387 

388 def test_detection(self): 

389 graph, flags = self.helper.get_setup_simple_boiler() 

390 

391 matches, meta = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

392 

393 self.assertEqual( 

394 len(matches), 5, 

395 "There are 5 cases for PipeStrand but 'find_matches' returned %d" 

396 % len(matches)) 

397 

398 def test_get_edge_ports_pipe_strand(self): 

399 """ Test the get_edge_ports method for a pipe strand.""" 

400 graph, flags = self.helper.get_setup_strand1() 

401 

402 matches, metas = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand.find_matches(graph) 

403 agg = bim2sim.elements.aggregation.hvac_aggregations.PipeStrand(graph, matches[0], metas[0]) 

404 edge_ports = [edge_port.originals[0] for edge_port in agg.get_ports()] 

405 self.assertEqual(set(flags['edge_ports']), set(edge_ports)) 

406 

407 

408if __name__ == '__main__': 

409 unittest.main()