Coverage for test/unit/elements/helper.py: 68%

1from contextlib import contextmanager 

2from unittest import mock 

3 

4import networkx as nx 

5 

6from bim2sim.elements import bps_elements as bps 

7from bim2sim.elements import hvac_elements as hvac 

8from bim2sim.elements.aggregation import hvac_aggregations 

9from bim2sim.elements.aggregation.hvac_aggregations import \ 

10 ConsumerHeatingDistributorModule 

11from bim2sim.elements.hvac_elements import HVACPort 

12from bim2sim.elements.graphs.hvac_graph import HvacGraph 

13from bim2sim.elements.mapping.units import ureg 

14 

15 

16class SetupHelper: 

17 ifc = mock.Mock() 

18 ifc.Name = 'Test' 

19 ifc.HasAssignments = [] 

20 type(ifc).GlobalId = mock.PropertyMock(side_effect=range(100000), 

21 name='GlobalId') 

22 

23 def __init__(self): 

24 self._flags = None 

25 self.elements = [] 

26 

27 # self.setup, self.flags = self.get_setup() 

28 # self.setup.plot(r'c:\temp') 

29 

30 def reset(self) -> None: 

31 self.elements.clear() 

32 

33 @contextmanager 

34 def flag_manager(self, flags): 

35 self._flags = flags 

36 yield 

37 self._flags = None 

38 

39 def element_generator(self): 

40 raise NotImplementedError 

41 

42 

43class SetupHelperHVAC(SetupHelper): 

44 

45 @classmethod 

46 def fake_add_ports(cls, parent, n=2): 

47 new_ports = [HVACPort(parent=parent) for i in range(n)] 

48 parent.ports.extend(new_ports) 

49 if isinstance(parent, hvac.HVACProduct): 

50 parent.inner_connections.extend(parent.get_inner_connections()) 

51 return new_ports 

52 

53 @staticmethod 

54 def connect_strait(items): 

55 """Connects item[n].ports[0] with item[n+1].ports[1] for all items""" 

56 last = None 

57 for item in items: 

58 if last: 

59 last.ports[1].connect(item.ports[0]) 

60 last = item 

61 

62 def element_generator(self, element_cls, n_ports=2, flags=None, **kwargs): 

63 # instantiate 

64 with mock.patch.object(hvac.HVACProduct, 'get_ports', return_value=[]): 

65 element = element_cls(**kwargs) 

66 self.elements.append(element) 

67 # # set attributes 

68 # for name, value in kwargs.items(): 

69 # if name not in element.attributes.names: 

70 # raise AssertionError("Can't set attribute '%s' to %s. Choices are %s" % 

71 # (name, element_cls.__name__, list(element.attributes.names))) 

72 # setattr(element, name, value * getattr(element_cls, name).unit) 

73 

74 # add ports 

75 self.fake_add_ports(element, n_ports) 

76 

77 # assign flags 

78 if flags: 

79 if self._flags is None: 

80 raise AssertionError( 

81 "Use contextmanager .flag_manager when setting flags") 

82 for flag in flags: 

83 self._flags.setdefault(flag, []).append(element) 

84 

85 return element 

86 

87 def get_simple_pipe(self): 

88 pipe = self.element_generator( 

89 hvac.Pipe, 

90 length=1 * ureg.m 

91 ) 

92 pipe.ports[0].flow_direction = -1 

93 pipe.ports[1].flow_direction = 1 

94 return HvacGraph([pipe]), pipe 

95 

96 def get_simple_junction(self): 

97 junction = self.element_generator( 

98 hvac.Junction, 

99 volume=1 * ureg.m ** 3) 

100 return HvacGraph([junction]) 

101 

102 def get_simple_valve(self): 

103 valve = self.element_generator( 

104 hvac.Valve, 

105 nominal_pressure_difference=100 * ureg.pascal 

106 ) 

107 return HvacGraph([valve]) 

108 

109 def get_simple_pump(self): 

110 pump = self.element_generator( 

111 hvac.Pump, 

112 rated_volume_flow=1 * ureg.meter ** 3 / ureg.s, 

113 rated_pressure_difference=10000 * ureg.pascal, 

114 rated_height=10 * ureg.meter, 

115 rated_power=5 * ureg.kilowatt) 

116 return HvacGraph([pump]), pump 

117 

118 def get_simple_radiator(self): 

119 radiator = self.element_generator( 

120 hvac.SpaceHeater, 

121 rated_power=20 * ureg.kilowatt, 

122 flow_temperature=70 * ureg.celsius, 

123 return_temperature=50 * ureg.celsius, 

124 ) 

125 return HvacGraph([radiator]) 

126 

127 def get_simple_boiler(self): 

128 boiler = self.element_generator( 

129 hvac.Boiler, 

130 rated_power=100 * ureg.kilowatt, 

131 return_temperature=50 * ureg.celsius 

132 ) 

133 return HvacGraph([boiler]) 

134 

135 def get_simple_consumer(self): 

136 consumer = self.element_generator( 

137 hvac_aggregations.Consumer, 

138 rated_power=20 * ureg.kilowatt, 

139 base_graph=nx.Graph(), 

140 match_graph=nx.Graph() 

141 ) 

142 return HvacGraph([consumer]), consumer 

143 

144 def get_simple_three_way_valve(self): 

145 three_way_valve = self.element_generator( 

146 hvac.ThreeWayValve, 

147 nominal_pressure_difference=100 * ureg.pascal 

148 ) 

149 return HvacGraph([three_way_valve]) 

150 

151 def get_simple_heat_pump(self): 

152 heat_pump = self.element_generator( 

153 hvac.HeatPump, 

154 rated_power=100 * ureg.kilowatt 

155 ) 

156 return HvacGraph([heat_pump]) 

157 

158 def get_simple_chiller(self): 

159 chiller = self.element_generator( 

160 hvac.Chiller, 

161 rated_power=100 * ureg.kilowatt, 

162 nominal_power_consumption=25, 

163 nominal_COP=4 * ureg.dimensionless 

164 ) 

165 return HvacGraph([chiller]) 

166 

167 def get_simple_cooling_tower(self): 

168 cooling_tower = self.element_generator( 

169 hvac.CoolingTower, 

170 rated_power=100 * ureg.kilowatt, 

171 ) 

172 return HvacGraph([cooling_tower]) 

173 

174 def get_simple_space_heater(self): 

175 space_heater = self.element_generator( 

176 hvac.SpaceHeater, 

177 rated_power=50 * ureg.kilowatt, 

178 flow_temperature=70 * ureg.celsius, 

179 return_temperature=50 * ureg.celsius, 

180 ) 

181 space_heater.ports[0].flow_direction = -1 

182 space_heater.ports[1].flow_direction = 1 

183 return HvacGraph([space_heater]), space_heater 

184 

185 def get_simple_storage(self): 

186 storage = self.element_generator( 

187 hvac.Storage, 

188 volume=1 * ureg.meter ** 3, 

189 height=1 * ureg.meter, 

190 diameter=1 * ureg.meter 

191 ) 

192 return HvacGraph([storage]) 

193 

194 def get_simple_generator_one_fluid(self): 

195 generator_one_fluid = self.element_generator( 

196 hvac_aggregations.GeneratorOneFluid, 

197 rated_power=100 * ureg.kilowatt, 

198 return_temperature=50 * ureg.celsius, 

199 flow_temperature=70 * ureg.celsius, 

200 base_graph=nx.Graph(), 

201 match_graph=nx.Graph() 

202 ) 

203 return HvacGraph([generator_one_fluid]) 

204 

205 def get_simple_chp(self): 

206 chp = self.element_generator( 

207 hvac.CHP, 

208 rated_power=100 * ureg.kilowatt 

209 ) 

210 return HvacGraph([chp]) 

211 

212 def get_simple_distributor(self): 

213 distributor = self.element_generator( 

214 hvac.Distributor, 

215 n_ports=6, 

216 ) 

217 distributor.ports[4].flow_direction = 1 

218 distributor.ports[5].flow_direction = -1 

219 return HvacGraph([distributor]), distributor 

220 

221 def get_setup_simple_heating_distributor_module(self): 

222 _, space_heater1 = self.get_simple_space_heater() 

223 _, space_heater2 = self.get_simple_space_heater() 

224 _, distributor = self.get_simple_distributor() 

225 pipes = [] 

226 for _ in range(6): 

227 _, pipe = self.get_simple_pipe() 

228 pipe.diameter = 0.2 * ureg.meter 

229 pipes.append(pipe) 

230 self.connect_strait([pipes[0], space_heater1, pipes[1]]) 

231 self.connect_strait([pipes[2], space_heater2, pipes[3]]) 

232 distributor.ports[0].connect(pipes[0].ports[0]) 

233 distributor.ports[1].connect(pipes[1].ports[1]) 

234 distributor.ports[2].connect(pipes[2].ports[0]) 

235 distributor.ports[3].connect(pipes[3].ports[1]) 

236 distributor.ports[4].connect(pipes[4].ports[0]) 

237 distributor.ports[5].connect(pipes[5].ports[1]) 

238 circuit = [*pipes[0:4], space_heater1, space_heater2, 

239 distributor] 

240 return HvacGraph(circuit), 

241 

242 def get_simple_consumer_heating_distributor_module(self): 

243 graph, = self.get_setup_simple_heating_distributor_module() 

244 matches, metas = hvac_aggregations.Consumer.find_matches(graph) 

245 # Merge Consumer 

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

247 module = hvac_aggregations.Consumer(graph, match, **meta) 

248 graph.merge( 

249 mapping=module.get_replacement_mapping(), 

250 inner_connections=module.inner_connections 

251 ) 

252 # Merge ConsumerHeatingDistributorModule 

253 matches, metas = (hvac_aggregations.ConsumerHeatingDistributorModule. 

254 find_matches(graph)) 

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

256 module = hvac_aggregations.ConsumerHeatingDistributorModule( 

257 graph, match, **meta) 

258 graph.merge( 

259 mapping=module.get_replacement_mapping(), 

260 inner_connections=module.inner_connections 

261 ) 

262 return graph 

263 

264 def get_setup_simple_boiler(self): 

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

266 distributor and pipes""" 

267 

268 flags = {} 

269 with self.flag_manager(flags): 

270 # generator circuit 

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

272 gen_vl_a = [ 

273 self.element_generator(hvac.Pipe, length=100, diameter=40) for i 

274 in range(3)] 

275 h_pump = self.element_generator(hvac.Pump, rated_power=2.2, 

276 rated_height=12, 

277 rated_volume_flow=8) 

278 gen_vl_b = [ 

279 self.element_generator(hvac.Pipe, flags=['strand1'], length=100, 

280 diameter=40) for i in range(5)] 

281 distributor = self.element_generator(hvac.Distributor, flags=[ 

282 'distributor']) # , volume=80 

283 gen_rl_a = [ 

284 self.element_generator(hvac.Pipe, length=100, diameter=40) for i 

285 in range(4)] 

286 fitting = self.element_generator(hvac.PipeFitting, n_ports=3, 

287 diameter=40, length=60) 

288 gen_rl_b = [ 

289 self.element_generator(hvac.Pipe, length=100, diameter=40) for i 

290 in range(4)] 

291 gen_rl_c = [ 

292 self.element_generator(hvac.Pipe, flags=['strand2'], 

293 length=(1 + i) * 40, diameter=15) 

294 for i in range(3) 

295 ] 

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

297 

298 # connect 

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

300 self.connect_strait(gen_vl) 

301 

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

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

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

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

306 

307 # full system 

308 gen_circuit = [ 

309 boiler, *gen_vl_a, h_pump, *gen_vl_b, distributor, 

310 *gen_rl_a, fitting, *gen_rl_b, *gen_rl_c, tank 

311 ] 

312 

313 return HvacGraph(gen_circuit), flags 

314 

315 # def test_port_mapping(self): 

316 # self.assertTrue(self.test_aggregation) 

317 # 

318 # mapping = self.test_aggregation.get_replacement_mapping() 

319 # 

320 # self.assertIs(self.test_aggregation.ports[0], mapping[self.edge_ports[0]]) 

321 # self.assertIs(self.test_aggregation.ports[1], mapping[self.edge_ports[1]]) 

322 

323 @staticmethod 

324 def elements_in_agg(agg): 

325 if not agg: 

326 return False 

327 if not agg.elements: 

328 return False 

329 

330 for ele in agg.elements: 

331 if not isinstance(ele, hvac.HVACProduct): 

332 return False 

333 return True 

334 

335 

336class SetupHelperBPS(SetupHelper): 

337 def element_generator(self, element_cls, flags=None, **kwargs): 

338 # with mock.patch.object(bps.BPSProduct, 'get_ports', return_value=[]): 

339 orient = kwargs.pop('orientation', None) # TODO WORKAROUND, 

340 element = element_cls(**kwargs) 

341 element.teaser_orientation = orient 

342 return element 

343 

344 def get_thermalzone(self, **kwargs): 

345 tz = self.element_generator( 

346 bps.ThermalZone, 

347 **kwargs) 

348 return tz 

349 

350 def get_thermalzones_diff_usage( 

351 self, usages: list, gross_areas: list): 

352 """Returns a number of ThermalZones with different usage. 

353 

354 The first ThermalZone has a usage that can be identified by regular 

355 expressions ('Living'). The second and third ThermalZone can't be 

356 identified due to random names, but the third one is similar to the 

357 first). The fourth ThermalZone can again be identified. 

358 

359 Args: 

360 usages: list of usage strings 

361 gross_area: list of gross_areas 

362 Returns: 

363 list of the three ThermalZone elements 

364 """ 

365 tz_elements = [] 

366 for usage, gross_area in zip(usages, gross_areas): 

367 tz_elements.append(self.get_thermalzone( 

368 usage=usage, gross_area=gross_area)) 

369 

370 return tz_elements 

371 

372 def get_setup_simple_house(self): 

373 out_wall_1 = self.element_generator( 

374 bps.OuterWall, 

375 net_area=20, 

376 gross_area=21, 

377 width=0.2, 

378 orientation=90, 

379 guid='outerWall001' 

380 ) 

381 window_1 = self.element_generator( 

382 bps.Window, 

383 net_area=2, 

384 width=0.1, 

385 guid='window001') 

386 tz_1 = self.get_thermalzone( 

387 net_area=100, 

388 gross_area=110, 

389 usage='Living', 

390 guid='tz001') 

391 tz_1.bound_elements = [out_wall_1, window_1] 

392 build_1 = self.element_generator( 

393 bps.Building, 

394 bldg_name='simpleTestBuilding', 

395 year_of_construction=2010, 

396 guid='bldg001' 

397 ) 

398 # set relations 

399 build_1.thermal_zones.append(tz_1) 

400 tz_1.bound_elements.extend([out_wall_1, window_1]) 

401 elements = { 

402 out_wall_1.guid: out_wall_1, 

403 window_1.guid: window_1, 

404 tz_1.guid: tz_1, 

405 build_1.guid: build_1 

406 } 

407 return elements