Coverage for test/unit/tasks/common/test_inspect.py: 100%

1import tempfile 

2import unittest 

3from pathlib import Path 

4from unittest.mock import MagicMock 

5 

6import numpy as np 

7 

8import bim2sim.tasks.common.create_elements 

9import bim2sim.tasks.common.load_ifc 

10import bim2sim.tasks.hvac.connect_elements 

11from bim2sim.kernel.decision.decisionhandler import DebugDecisionHandler 

12from bim2sim.elements.base_elements import Port, ProductBased 

13from bim2sim.elements.hvac_elements import HeatExchanger, Pipe 

14from bim2sim.plugins import Plugin 

15from bim2sim.project import Project 

16from bim2sim.tasks.hvac import ConnectElements 

17from bim2sim.sim_settings import PlantSimSettings 

18from bim2sim.utilities.types import IFCDomain 

19 

20 

21class PluginDummy(Plugin): 

22 name = 'test' 

23 sim_settings = PlantSimSettings 

24 default_tasks = [ 

25 bim2sim.tasks.common.load_ifc.LoadIFC, 

26 bim2sim.tasks.common.create_elements.CreateElementsOnIfcTypes, 

27 bim2sim.tasks.hvac.connect_elements.ConnectElements 

28 ] 

29 

30test_rsrc_path = Path(__file__).parent.parent.parent.parent / 'resources' 

31 

32 

33class TestInspect(unittest.TestCase): 

34 """Basic scenario for connection tests with HeatExchanger (IfcPipeFitting) 

35 and four Pipes (IfcPipeSegment)""" 

36 

37 def tearDown(self): 

38 self.project.finalize(True) 

39 self.test_dir.cleanup() 

40 

41 def test_case_1(self): 

42 """HeatExchange with 4 (semantically) connected pipes""" 

43 self.test_dir = tempfile.TemporaryDirectory() 

44 ifc_paths = { 

45 IFCDomain.hydraulic: test_rsrc_path / 

46 'hydraulic/ifc/B01_2_HeatExchanger_Pipes.ifc'} 

47 self.project = Project.create(self.test_dir.name, ifc_paths, 

48 plugin=PluginDummy, ) 

49 self.project.sim_settings.weather_file_path = ( 

50 test_rsrc_path / 'weather_files/DEU_NW_Aachen.105010_TMYx.mos') 

51 handler = DebugDecisionHandler([HeatExchanger.key]) 

52 handler.handle(self.project.run(cleanup=False)) 

53 

54 elements = self.project.playground.state['elements'] 

55 heat_exchanger = elements.get('0qeZDHlQRzcKJYopY4$fEf') 

56 self.assertEqual( 

57 4, len([port for port in heat_exchanger.ports if port.connection])) 

58 

59 def test_case_2(self): 

60 """HeatExchange and Pipes are exported without ports""" 

61 self.test_dir = tempfile.TemporaryDirectory() 

62 ifc_paths = { 

63 IFCDomain.hydraulic: test_rsrc_path / 

64 'hydraulic/ifc/' 

65 'B01_3_HeatExchanger_noPorts.ifc'} 

66 self.project = Project.create(self.test_dir.name, ifc_paths, 

67 plugin=PluginDummy, ) 

68 self.project.sim_settings.weather_file_path = ( 

69 test_rsrc_path / 'weather_files/DEU_NW_Aachen.105010_TMYx.mos') 

70 handler = DebugDecisionHandler([HeatExchanger.key, 

71 *(Pipe.key,) * 4]) 

72 handler.handle(self.project.run(cleanup=False)) 

73 elements = self.project.playground.state['elements'] 

74 heat_exchanger = elements.get('0qeZDHlQRzcKJYopY4$fEf') 

75 self.assertEqual(0, len([port for port in heat_exchanger.ports 

76 if port.connection])) 

77 

78 # assert warnings ?? 

79 

80 def test_case_3(self): 

81 """No connections but ports are less than 10 mm apart""" 

82 self.test_dir = tempfile.TemporaryDirectory() 

83 ifc_paths = { 

84 IFCDomain.hydraulic: test_rsrc_path / 

85 'hydraulic/ifc/' 

86 'B01_4_HeatExchanger_noConnection.ifc'} 

87 self.project = Project.create(self.test_dir.name, ifc_paths, 

88 plugin=PluginDummy, ) 

89 self.project.sim_settings.weather_file_path = ( 

90 test_rsrc_path / 'weather_files/DEU_NW_Aachen.105010_TMYx.epw') 

91 handler = DebugDecisionHandler([HeatExchanger.key]) 

92 handler.handle(self.project.run(cleanup=False)) 

93 

94 elements = self.project.playground.state['elements'] 

95 heat_exchanger = elements.get('3FQzmSvzrgbaIM6zA4FX8S') 

96 self.assertEqual(4, len([port for port in heat_exchanger.ports 

97 if port.connection])) 

98 

99 def test_case_4(self): 

100 """Mix of case 1 and 3""" 

101 self.test_dir = tempfile.TemporaryDirectory() 

102 ifc_paths = { 

103 IFCDomain.hydraulic: test_rsrc_path / 

104 'hydraulic/ifc/' 

105 'B01_5_HeatExchanger_mixConnection.ifc'} 

106 self.project = Project.create(self.test_dir.name, ifc_paths, 

107 plugin=PluginDummy, ) 

108 self.project.sim_settings.weather_file_path = ( 

109 test_rsrc_path / 'weather_files/DEU_NW_Aachen.105010_TMYx.mos') 

110 handler = DebugDecisionHandler([HeatExchanger.key]) 

111 handler.handle(self.project.run(cleanup=False)) 

112 

113 elements = self.project.playground.state['elements'] 

114 heat_exchanger = elements.get('3FQzmSvzrgbaIM6zA4FX8S') 

115 self.assertEqual(4, len([port for port in heat_exchanger.ports 

116 if port.connection])) 

117 

118 

119class TestInspectMethods(unittest.TestCase): 

120 

121 @staticmethod 

122 def create_element(positions): 

123 parent = ProductBased() 

124 for pos in positions: 

125 port = Port(parent) 

126 port._calc_position = MagicMock(return_value=np.array(pos)) 

127 parent.ports.append(port) 

128 return parent 

129 

130 def test_connect_by_position(self): 

131 """Test Inspect.connect_by_position by various scenarios""" 

132 parent1 = self.create_element([[0, 0, 0], [0, 0, 20]]) 

133 parent2 = self.create_element([[0, 0, 20], [0, 0, 40]]) 

134 parent3 = self.create_element([[0, 5, 40], [0, 0, 60]]) 

135 parent4 = self.create_element([[0, 0, 0], [0, 0, 0], [0, 0, 5]]) 

136 parent5 = self.create_element([[0, 0, 25], [0, 20, 0]]) 

137 

138 # no distance 

139 connections = ConnectElements.connections_by_position( 

140 parent1.ports + parent2.ports, eps=10) 

141 self.assertEqual(1, len(connections)) 

142 self.assertSetEqual( 

143 {parent1.ports[1], parent2.ports[0]}, set(connections[0])) 

144 

145 # accepted distance 

146 connections = ConnectElements.connections_by_position( 

147 parent2.ports + parent3.ports, eps=10) 

148 self.assertEqual(1, len(connections), "One valid connection") 

149 self.assertSetEqual( 

150 {parent2.ports[1], parent3.ports[0]}, set(connections[0])) 

151 

152 # not accepted distance 

153 connections = ConnectElements.connections_by_position( 

154 parent1.ports + parent3.ports, eps=10) 

155 self.assertEqual(0, len(connections), "Not accepted distance") 

156 

157 # no connections within element 

158 connections = ConnectElements.connections_by_position( 

159 parent4.ports, eps=10) 

160 self.assertEqual(0, len(connections), "No connections within element") 

161 

162 # multiple possibilities 

163 connections = ConnectElements.connections_by_position( 

164 parent1.ports + parent2.ports + parent5.ports, eps=10) 

165 self.assertEqual(1, len(connections), 

166 "Only one connection per port allowed") 

167 self.assertSetEqual( 

168 {parent1.ports[1], parent2.ports[0]}, set(connections[0]))