Coverage for test/unit/elements/aggregation/test_consumer.py: 99%

1import unittest 

2 

3import networkx as nx 

4 

5import bim2sim.elements.aggregation.hvac_aggregations 

6from bim2sim.elements import aggregation 

7from bim2sim.elements import hvac_elements as hvac 

8from bim2sim.elements.graphs.hvac_graph import HvacGraph 

9from bim2sim.elements.mapping.units import ureg 

10from test.unit.elements.helper import SetupHelperHVAC 

11 

12 

13class ConsumerHelper(SetupHelperHVAC): 

14 

15 def get_setup_consumer1(self): 

16 """ Get consumer circuit made of 2 parallel pumps, space heater and 

17 pipes. 

18 """ 

19 flags = {} 

20 with self.flag_manager(flags): 

21 # generator circuit 

22 con_vl_a = [self.element_generator( 

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

24 fitting1 = self.element_generator( 

25 hvac.PipeFitting, flags=['con1'], n_ports=3, diameter=30, 

26 length=60) 

27 p_pump1_p = [ 

28 self.element_generator( 

29 hvac.Pipe, flags=['con1'], length=40, diameter=20), 

30 self.element_generator( 

31 hvac.Pump, flags=['con1'], rated_power=1, rated_height=8, 

32 rated_volume_flow=6, diameter=20), 

33 self.element_generator( 

34 hvac.Pipe, flags=['con1'], length=40, diameter=20), 

35 ] 

36 p_pump2_p = [ 

37 self.element_generator( 

38 hvac.Pipe, flags=['con1'], length=40, diameter=20), 

39 self.element_generator( 

40 hvac.Pump, flags=['con1'], rated_power=1, rated_height=8, 

41 rated_volume_flow=6, diameter=20), 

42 self.element_generator( 

43 hvac.Pipe, flags=['con1'], length=40, diameter=20), 

44 ] 

45 fitting2 = self.element_generator( 

46 hvac.PipeFitting, flags=['con2'], n_ports=3, diameter=30, 

47 length=60) 

48 con_vl_b = [self.element_generator( 

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

50 consumer = self.element_generator( 

51 hvac.SpaceHeater, flags=['spaceheater']) 

52 con_rl_a = [self.element_generator( 

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

54 

55 # connect 

56 self.connect_strait([*con_vl_a, fitting1]) 

57 self.connect_strait([fitting1, *p_pump1_p, fitting2]) 

58 self.connect_strait(p_pump2_p) 

59 fitting1.ports[2].connect(p_pump2_p[0].ports[0]) 

60 p_pump2_p[-1].ports[1].connect(fitting2.ports[2]) 

61 self.connect_strait([fitting2, *con_vl_b, consumer, *con_rl_a]) 

62 

63 # full system 

64 gen_circuit = [ 

65 *con_vl_a, fitting1, *p_pump1_p, *p_pump2_p, fitting2, 

66 *con_vl_b, consumer, *con_rl_a 

67 ] 

68 

69 flags['connect'] = [con_vl_a[0], con_rl_a[-1]] 

70 

71 graph = HvacGraph(gen_circuit) 

72 return graph, flags 

73 

74 def get_setup_consumer2(self): 

75 """ Get consumer circuit made of 2 parallel pumps , underfloorheating 

76 and pipes. 

77 """ 

78 flags = {} 

79 with self.flag_manager(flags): 

80 # generator circuit 

81 con_vl_a = [self.element_generator( 

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

83 fitting1 = self.element_generator( 

84 hvac.PipeFitting, flags=['con2'], n_ports=3, diameter=30, length=60) 

85 p_pump1_p = [ 

86 self.element_generator( 

87 hvac.Pipe, flags=['con2'], length=40, diameter=20), 

88 self.element_generator( 

89 hvac.Pump, flags=['con2'], rated_power=1, rated_height=8, 

90 rated_volume_flow=6, diameter=20), 

91 self.element_generator( 

92 hvac.Pipe, flags=['con2'], length=40, diameter=20), 

93 ] 

94 p_pump2_p = [ 

95 self.element_generator( 

96 hvac.Pipe, flags=['con2'], length=40, diameter=20), 

97 self.element_generator( 

98 hvac.Pump, flags=['con2'], rated_power=1, rated_height=8, 

99 rated_volume_flow=6, diameter=20), 

100 self.element_generator( 

101 hvac.Pipe, flags=['con2'], length=40, diameter=20), 

102 ] 

103 fitting2 = self.element_generator( 

104 hvac.PipeFitting, flags=['con2'], n_ports=3, diameter=30, 

105 length=60) 

106 con_vl_b = [self.element_generator( 

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

108 underfloor_pipes = [self.element_generator( 

109 hvac.Pipe, length=1000, diameter=10) for i in range(3)] 

110 con_rl_a = [self.element_generator( 

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

112 

113 # connect 

114 self.connect_strait([*con_vl_a, fitting1]) 

115 self.connect_strait([fitting1, *p_pump1_p, fitting2]) 

116 self.connect_strait(p_pump2_p) 

117 fitting1.ports[2].connect(p_pump2_p[0].ports[0]) 

118 p_pump2_p[-1].ports[1].connect(fitting2.ports[2]) 

119 self.connect_strait([fitting2, *con_vl_b, *underfloor_pipes, *con_rl_a]) 

120 

121 # full system 

122 gen_circuit = [ 

123 *con_vl_a, fitting1, *p_pump1_p, *p_pump2_p, fitting2, 

124 *con_vl_b, *underfloor_pipes, *con_rl_a 

125 ] 

126 

127 flags['connect'] = [con_vl_a[0], con_rl_a[-1]] 

128 

129 graph = HvacGraph(gen_circuit) 

130 

131 uf_ports = (port for pipe in underfloor_pipes for port in pipe.ports) 

132 subgraph = graph.subgraph(uf_ports) 

133 consumer = bim2sim.elements.aggregation.hvac_aggregations.UnderfloorHeating(graph, subgraph) 

134 flags['underfloor'] = [consumer] 

135 

136 graph.merge( 

137 mapping=consumer.get_replacement_mapping(), 

138 inner_connections=consumer.inner_connections 

139 ) 

140 # #ToDO: Workaround.... Hvac Graph.elements haben keine Portverknüpfungen ... vielleicht das problem 

141 # for port_a, port_b in consumer.get_replacement_mapping().items(): 

142 # if port_a and port_b: 

143 # port_a.connect(port_b) 

144 

145 return graph, flags 

146 

147 def get_setup_consumer3(self): 

148 """ Get consumer circuit made of 2 parallel pumps , 2x space heater and 

149 pipes. 

150 """ 

151 flags = {} 

152 with self.flag_manager(flags): 

153 # generator circuit 

154 con_vl_a = [self.element_generator( 

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

156 fitting1 = self.element_generator( 

157 hvac.PipeFitting, flags=['con3'], n_ports=3, diameter=30, length=60) 

158 p_pump1_p = [ 

159 self.element_generator( 

160 hvac.Pipe, flags=['con3'], length=40, diameter=20), 

161 self.element_generator( 

162 hvac.Pump, flags=['con3'], rated_power=1, rated_height=8, 

163 rated_volume_flow=6, diameter=20), 

164 self.element_generator( 

165 hvac.Pipe, flags=['con3'], length=40, diameter=20), 

166 ] 

167 p_pump2_p = [ 

168 self.element_generator( 

169 hvac.Pipe, flags=['con3'], length=40, diameter=20), 

170 self.element_generator( 

171 hvac.Pump, flags=['con3'], rated_power=1, rated_height=8, 

172 rated_volume_flow=6, diameter=20), 

173 self.element_generator( 

174 hvac.Pipe, flags=['con3'], length=40, diameter=20), 

175 ] 

176 fitting2 = self.element_generator( 

177 hvac.PipeFitting, flags=['con3'], n_ports=3, diameter=30, length=60) 

178 con_vl_b = [self.element_generator( 

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

180 consumer1 = self.element_generator( 

181 hvac.SpaceHeater, flags=['spaceheater']) 

182 con_mid = [self.element_generator( 

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

184 consumer2 = self.element_generator( 

185 hvac.SpaceHeater, flags=['spaceheater']) 

186 con_rl_a = [self.element_generator( 

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

188 

189 # connect 

190 self.connect_strait([*con_vl_a, fitting1]) 

191 self.connect_strait([fitting1, *p_pump1_p, fitting2]) 

192 self.connect_strait(p_pump2_p) 

193 fitting1.ports[2].connect(p_pump2_p[0].ports[0]) 

194 p_pump2_p[-1].ports[1].connect(fitting2.ports[2]) 

195 self.connect_strait([fitting2, *con_vl_b, consumer1, *con_mid, consumer2, *con_rl_a]) 

196 

197 # full system 

198 gen_circuit = [ 

199 *con_vl_a, fitting1, *p_pump1_p, *p_pump2_p, fitting2, 

200 *con_vl_b, consumer1, *con_mid, consumer2, *con_rl_a 

201 ] 

202 

203 flags['connect'] = [con_vl_a[0], con_rl_a[-1]] 

204 

205 graph = HvacGraph(gen_circuit) 

206 return graph, flags 

207 

208 def get_setup_system(self): 

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

210 distributor, consumer1(1xSpaceheater), consumer2 (1xUnderfloorheating) 

211 and pipes.""" 

212 graph1, flags1 = super().get_setup_simple_boiler() 

213 graph2, flags2 = self.get_setup_consumer1() 

214 graph3, flags3 = self.get_setup_consumer2() 

215 

216 distributor = flags1['distributor'][0] 

217 distributor_ports = self.fake_add_ports(distributor, 4) 

218 

219 graph = nx.compose(graph3, nx.compose(graph2, graph1)) 

220 

221 vl_p2, rl_p2 = flags3.pop('connect') 

222 graph.add_edge(rl_p2.ports[1], distributor_ports[3]) 

223 graph.add_edge(distributor_ports[2], vl_p2.ports[0]) 

224 

225 vl_p1, rl_p1 = flags2.pop('connect') 

226 graph.add_edge(rl_p1.ports[1], distributor_ports[1]) 

227 graph.add_edge(distributor_ports[0], vl_p1.ports[0]) 

228 

229 flags = {**flags1, **flags2, **flags3} 

230 return graph, flags 

231 

232 def get_setup_system2(self): 

233 """Simple generator system made of boiler, pump, expansion tank, distributor consumer3(2xSpaceheater) 

234 and pipes""" 

235 graph1, flags1 = super().get_setup_simple_boiler() 

236 graph2, flags2 = self.get_setup_consumer3() 

237 

238 distributor = flags1['distributor'][0] 

239 distributor_ports = self.fake_add_ports(distributor, 2) 

240 

241 graph = nx.compose(graph2, graph1) 

242 

243 vl_p1, rl_p1 = flags2.pop('connect') 

244 graph.add_edge(rl_p1.ports[1], distributor_ports[1]) 

245 graph.add_edge(distributor_ports[0], vl_p1.ports[0]) 

246 

247 flags = {**flags1, **flags2} 

248 return graph, flags 

249 

250 

251class TestConsumerAggregation(unittest.TestCase): 

252 helper = None 

253 

254 @classmethod 

255 def setUpClass(cls): 

256 cls.helper = ConsumerHelper() 

257 

258 def tearDown(self) -> None: 

259 self.helper.reset() 

260 

261 def test_find_matches(self): 

262 """ Test detection of consumer cycle in setup system.""" 

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

264 

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

266 

267 self.assertEqual( 

268 len(matches), 2, 

269 f"There are 2 cases for Consumer Cycles but 'find_matches' " 

270 f"returned {len(matches)}") 

271 

272 consumer = [item for item in flags['spaceheater'] + flags['underfloor']] 

273 all_elements = sum((list(match.elements) for match in matches), []) 

274 for item in consumer: 

275 self.assertIn(item, all_elements) 

276 

277 def test_aggregation_consumer1(self): 

278 """ Test aggregation of consumer cycle no 1.""" 

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

280 

281 matches, metas = bim2sim.elements.aggregation.hvac_aggregations.Consumer.find_matches(graph) 

282 

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

284 consumer = bim2sim.elements.aggregation.hvac_aggregations.Consumer(graph, match, **meta) 

285 if hvac.SpaceHeater in {type(ele) for ele in consumer.elements}: 

286 # we only want consumer with SpaceHeater 

287 break 

288 

289 graph.merge( 

290 mapping=consumer.get_replacement_mapping(), 

291 inner_connections=consumer.inner_connections 

292 ) 

293 

294 self.assertAlmostEqual( 

295 consumer.rated_volume_flow, 12 * ureg.meter ** 3 / ureg.hour) 

296 self.assertTrue(consumer.has_pump) 

297 # self.assertAlmostEqual(consumer.temperature_inlet, 1000) 

298 # self.assertAlmostEqual(consumer.temperature_outlet, 1000) 

299 # self.assertAlmostEqual(consumer.volume, 1000) 

300 # self.assertAlmostEqual(consumer.height, 1000) 

301 # list of all aggregated consumers description 

302 self.assertIn('SpaceHeater', consumer.description) 

303 

304 def test_aggregation_consumer2(self): 

305 """ Test aggregation of consumer cycle no 2.""" 

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

307 

308 matches, metas = bim2sim.elements.aggregation.hvac_aggregations.Consumer.find_matches(graph) 

309 

310 consumer2 = None 

311 

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

313 for ele in flags['con2']: 

314 if ele in match.elements: 

315 consumer2 = bim2sim.elements.aggregation.hvac_aggregations.Consumer(graph, match, **meta) 

316 break 

317 if consumer2: 

318 break 

319 else: 

320 self.assertTrue(False, 'Kein Consumer-Kreis identifiziert!') 

321 

322 graph.merge( 

323 mapping=consumer2.get_replacement_mapping(), 

324 inner_connections=consumer2.inner_connections 

325 ) 

326 

327 self.assertAlmostEqual(consumer2.rated_volume_flow, 12 * ureg.meter ** 3 / ureg.hour) 

328 self.assertTrue(consumer2.has_pump) 

329 # self.assertAlmostEqual(consumer.temperature_inlet, 1000) 

330 # self.assertAlmostEqual(consumer.temperature_outlet, 1000) 

331 # self.assertAlmostEqual(consumer.volume, 1000) Not Implemented 

332 # self.assertAlmostEqual(consumer.height, 1000) Not Implemented 

333 # list of all aggregated consumers description 

334 self.assertIn('UnderfloorHeating', consumer2.description) 

335 

336 def test_aggregation_consumer3(self): 

337 """ Test aggregation of consumer cycle no 2.""" 

338 graph, flags = self.helper.get_setup_system2() 

339 

340 # graph.plot() 

341 

342 matches, metas = bim2sim.elements.aggregation.hvac_aggregations.Consumer.find_matches(graph) 

343 

344 consumer = bim2sim.elements.aggregation.hvac_aggregations.Consumer(graph, matches[0], **metas[0]) 

345 

346 graph.merge( 

347 mapping=consumer.get_replacement_mapping(), 

348 inner_connections=consumer.inner_connections 

349 ) 

350 

351 self.assertAlmostEqual( 

352 consumer.rated_volume_flow, 12 * ureg.meter ** 3 / ureg.hour) 

353 self.assert_(consumer.has_pump) 

354 # self.assertAlmostEqual(consumer.temperature_inlet, 1000) 

355 # self.assertAlmostEqual(consumer.temperature_outlet, 1000) 

356 # self.assertAlmostEqual(consumer.volume, 1000) 

357 # self.assertAlmostEqual(consumer.height, 1000) 

358 # list of all aggregated consumers description 

359 self.assertIn('2 x SpaceHeater', consumer.description) 

360 

361 

362if __name__ == '__main__': 

363 unittest.main()