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

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 test.unit.elements.helper import SetupHelperHVAC 

8 

9 

10class ConsumerDistributorHelper(SetupHelperHVAC): 

11 

12 def get_setup_con_dist1(self): 

13 """ Get a simple setup to test ConsumerHeatingDistributorModule. 

14 

15 The setup includes the following components: 

16 * 1 Distributor with 4 ports 

17 * 1 Consumer connected with 3 pipe each (flow and return) to 

18 Distributor 

19 * 1 Boiler connected with 3 pipe each (flow and return) to Distributor 

20 """ 

21 flags = {} 

22 with self.flag_manager(flags): 

23 # generator circuit 

24 con_vl = [self.element_generator( 

25 hvac.Pipe, length=100, diameter=30, flags=['PipeCon']) 

26 for i in range(1)] 

27 con_rl = [self.element_generator( 

28 hvac.Pipe, length=100, diameter=30, flags=['PipeCon']) 

29 for i in range(1)] 

30 consumer = self.element_generator( 

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

32 boil_vl = [self.element_generator( 

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

34 boil_rl = [self.element_generator( 

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

36 boiler = self.element_generator( 

37 hvac.Boiler, n_ports=2, flags=['boiler'],) 

38 distributor = self.element_generator( 

39 hvac.Distributor, n_ports=4, flags=["distributor"]) 

40 

41 # connect elements 

42 self.connect_strait([*con_vl, consumer, *con_rl]) 

43 self.connect_strait([*boil_rl, boiler, *boil_vl]) 

44 boil_vl[-1].ports[-1].connect(distributor.ports[0]) 

45 distributor.ports[1].connect(boil_rl[0].ports[0]) 

46 distributor.ports[2].connect(con_vl[0].ports[0]) 

47 distributor.ports[3].connect(con_rl[-1].ports[-1]) 

48 

49 flags['edge_ports'] = [distributor.ports[0], distributor.ports[1]] 

50 

51 circuit = [ 

52 *con_vl, *con_rl, consumer, *boil_vl, *boil_rl, boiler, distributor] 

53 

54 graph = HvacGraph(circuit) 

55 return graph, flags 

56 

57 def get_setup_con_dist2(self): 

58 """ Get a simple setup to test ConsumerHeatingDistributorModule with 

59 undefined consumer ports. 

60 

61 The setup includes the following components: 

62 * 1 Distributor with 6 ports, 2 of them unconnected 

63 * 1 Consumer connected with 3 pipe each (flow and return) to 

64 Distributor 

65 * 1 Boiler connected with 3 pipe each (flow and return) to Distributor 

66 """ 

67 flags = {} 

68 with self.flag_manager(flags): 

69 # generator circuit 

70 boil_vl = [self.element_generator( 

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

72 boil_rl = [self.element_generator( 

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

74 boiler = self.element_generator( 

75 hvac.Boiler, n_ports=2, flags=['boiler'], ) 

76 # consumer circuit 

77 con_vl = [self.element_generator( 

78 hvac.Pipe, length=100, diameter=30, flags=['PipeCon']) 

79 for i in range(1)] 

80 con_rl = [self.element_generator( 

81 hvac.Pipe, length=100, diameter=30, flags=['PipeCon']) 

82 for i in range(1)] 

83 consumer = self.element_generator( 

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

85 distributor = self.element_generator( 

86 hvac.Distributor, n_ports=6, flags=["distributor"]) 

87 # undefined circuit 

88 undef_vl = [self.element_generator( 

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

90 undef_rl = [self.element_generator( 

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

92 

93 # connect elements 

94 self.connect_strait([*con_vl, consumer, *con_rl]) 

95 self.connect_strait([*boil_rl, boiler, *boil_vl]) 

96 boil_vl[-1].ports[-1].connect(distributor.ports[0]) 

97 distributor.ports[1].connect(boil_rl[0].ports[0]) 

98 distributor.ports[2].connect(con_vl[0].ports[0]) 

99 distributor.ports[3].connect(con_rl[-1].ports[-1]) 

100 distributor.ports[4].connect(undef_vl[0].ports[0]) 

101 distributor.ports[5].connect(undef_rl[-1].ports[-1]) 

102 

103 flags['edge_ports'] = [distributor.ports[0], distributor.ports[1]] 

104 flags['undefined_consumer_ports'] = [distributor.ports[4], 

105 distributor.ports[5]] 

106 

107 circuit = [ 

108 *con_vl, *con_rl, consumer, *boil_vl, *boil_rl, boiler, distributor] 

109 

110 graph = HvacGraph(circuit) 

111 return graph, flags 

112 

113 

114class TestConsumerDistributorModule(unittest.TestCase): 

115 helper = None 

116 

117 @classmethod 

118 def setUpClass(cls) -> None: 

119 cls.helper = ConsumerDistributorHelper() 

120 

121 def tearDown(self) -> None: 

122 self.helper.reset() 

123 

124 def test_find_matches(self): 

125 """ Test the find matches method.""" 

126 graph, flags = self.helper.get_setup_con_dist1() 

127 con_heat_dist_mod = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule 

128 matches, metas = con_heat_dist_mod.find_matches(graph) 

129 self.assertEqual( 

130 len(matches), 1, 

131 "There is 1 case for ConsumerDistrubtorModule Cycles but " 

132 "'find_matches' returned %d" % len(matches) 

133 ) 

134 

135 module = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule( 

136 graph, matches[0], **metas[0]) 

137 module_elements = [item for item in flags['spaceheater'] 

138 + flags['distributor'] + flags['PipeCon']] 

139 edge_ports_originals = [ 

140 edge_port.originals for edge_port in module.get_ports()] 

141 self.assertNotEqual(edge_ports_originals, flags['edge_ports']) 

142 self.assertCountEqual(module.elements, module_elements) 

143 

144 def test_aggregation(self): 

145 """ Test the aggregation of consumer heating distribution module.""" 

146 graph, flags = self.helper.get_setup_con_dist1() 

147 con_heat_dist_mod = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule 

148 matches, metas = con_heat_dist_mod.find_matches(graph) 

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

150 module = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule( 

151 graph, match, **meta) 

152 edge_ports = [edge_port.originals[0] for edge_port in 

153 module.get_ports()] 

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

155 graph.merge( 

156 mapping=module.get_replacement_mapping(), 

157 inner_connections=module.inner_connections 

158 ) 

159 

160 aggregated_con_heat_dis_mod = [ 

161 ele for ele in graph.elements 

162 if ele.__class__.__name__ == 'ConsumerHeatingDistributorModule'] 

163 self.assertEqual(len(aggregated_con_heat_dis_mod), 1) 

164 

165 def test_aggregation2(self): 

166 """ Test the aggregation of consumer heating distribution module and 

167 detection of undefined consumer ports.""" 

168 graph, flags = self.helper.get_setup_con_dist2() 

169 con_heat_dist_mod = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule 

170 matches, metas = con_heat_dist_mod.find_matches(graph) 

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

172 module = bim2sim.elements.aggregation.hvac_aggregations.ConsumerHeatingDistributorModule( 

173 graph, match, **meta) 

174 graph.merge( 

175 mapping=module.get_replacement_mapping(), 

176 inner_connections=module.inner_connections 

177 ) 

178 

179 aggregated_con_heat_dis_mod = [ 

180 ele for ele in graph.elements 

181 if ele.__class__.__name__ == 'ConsumerHeatingDistributorModule'] 

182 self.assertEqual(len(aggregated_con_heat_dis_mod), 1) 

183 self.assertEqual(flags['undefined_consumer_ports'], 

184 metas[0]['undefined_consumer_ports']) 

185 

186 

187if __name__ == '__main__': 

188 unittest.main()