Coverage for steam_pysigma\comsol\BuildComsolModel.py: 88%

1# STEAM PySigma is a python wrapper of STEAM-SIGMA written in Java. 

2# Copyright (C) 2023, CERN, Switzerland. All rights reserved. 

3# 

4# This program is free software: you can redistribute it and/or modify 

5# it under the terms of the GNU General Public License as published by 

6# the Free Software Foundation, version 3 of the License. 

7# 

8# This program is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU General Public License for more details. 

12# 

13# You should have received a copy of the GNU General Public License 

14# along with this program. If not, see <https://www.gnu.org/licenses/>. 

15 

16import os 

17from pathlib import Path 

18 

19import numpy as np 

20 

21from steam_pysigma.comsol.BuildGlobalVariables import BuildGlobalVariables 

22from steam_pysigma.sigma import pysigma as ps 

23from steam_pysigma.domain_generator.GeometryMultipole import GeometryMultipole 

24 

25 

26class BuildComsolModel: 

27 """ 

28 Class take in domains and a configuration and builds a functioning comsol model saved as java. 

29 """ 

30 

31 def __init__(self, model_data=None, input_conductor_params=None, output_path=None, settings=None, 

32 path_to_results=None, input_coordinates_path=None, roxie_data=None, bh_curve_database=None): 

33 """ 

34 Class builds a comsol model. 

35 :param model_data: data dict from magnet_name.yaml 

36 :param input_conductor_params: magnet_name.set 

37 :param output_path: Output path of java files and mph model. 

38 :param settings: settings dict 

39 :param path_to_results: location of comsol-generated results 

40 :param input_coordinates_path: path to file with coordinates to evaluate B_field 

41 :param roxie_data: roxie data 

42 :param bh_curve_database: path to bh curve file 

43 """ 

44 self.g = ps.GatewaySIGMA() 

45 self.a = ps.ArraysSIGMA 

46 self.model_data = model_data 

47 self.output_path = output_path 

48 self.roxie_data = roxie_data.Roxie_Data 

49 self.settings = settings 

50 self.input_conductor_params = input_conductor_params 

51 self.funlib_map = ps.FunLibCSVReader().read_csv_funclib_create_hashmap(self.g) 

52 self.COMSOL_exe_path = Path(self.settings.comsolexe_path).parent 

53 self.java_jdk_path = self.settings.JAVA_jdk_path 

54 self.COMSOL_compile_path = os.path.join(self.COMSOL_exe_path, 'comsolcompile.exe') # Change os.join() 

55 self.COMSOL_batch_path = os.path.join(self.COMSOL_exe_path, 'comsolbatch.exe') 

56 self.COMSOL_version = os.path.basename(os.path.dirname(self.COMSOL_exe_path.parent.parent)).replace("COMSOL", 

57 "") 

58 self.path_to_results = path_to_results if path_to_results != None else self.output_path 

59 self.bh_curve_database = bh_curve_database 

60 if self.COMSOL_version != "60" and self.COMSOL_version != "53a": 

61 raise Exception("Not supporting any other versions than 6.0 and 5.3a") 

62 

63 self.model_name = f"{self.model_data.GeneralParameters.magnet_name}_Model" 

64 

65 self.model_java_file_path = f"{os.path.join(self.output_path, self.model_name)}.java" 

66 self.compile_batch_file_path = f"{os.path.join(self.output_path, self.model_name)}_Compile_and_Open.bat" 

67 self.model_class_file_path = f"{os.path.join(self.output_path, self.model_name)}.class" 

68 self.split_java_file_path = [] 

69 self.study_type = self.model_data.Options_SIGMA.simulation.study_type 

70 self.num_qh = self.model_data.Quench_Protection.Quench_Heaters.N_strips 

71 self.make_batch_mode_executable = self.model_data.Options_SIGMA.simulation.make_batch_mode_executable 

72 self.generate_study = self.model_data.Options_SIGMA.simulation.generate_study 

73 self.input_coordinates_path = input_coordinates_path 

74 

75 self.variables2DConverted = self.model_data.Options_SIGMA.postprocessing.out_2D_at_points.variables 

76 self.time2DConverted = self.model_data.Options_SIGMA.postprocessing.out_2D_at_points.time # List of all exported time 

77 self.variables1DvsTime = self.model_data.Options_SIGMA.postprocessing.out_1D_vs_times.variables 

78 self.time1DConverted = self.model_data.Options_SIGMA.postprocessing.out_1D_vs_times.time 

79 self.variables1DvsTimeVector = self.model_data.Options_SIGMA.postprocessing.out_1D_vs_times.variables 

80 self.timeRange = ", ".join(["range(" + ", ".join(map(str, lst)) + ")" for lst in 

81 self.model_data.Options_SIGMA.time_vector_solution.time_step]) 

82 

83 print(f"Comsol compile path {self.COMSOL_compile_path}") 

84 print(f"Comsol version {self.COMSOL_version}") 

85 bgv = BuildGlobalVariables(self.g, self.model_data) 

86 bgv.validate_sigma_model_data() 

87 self.cfg = self.setup_config_sigma() 

88 self.srv = self.g.TxtSigmaServer(self.cfg.getOutputModelPath(), self.cfg.getComsolBatchPath(), 

89 self.cfg.getComsolCompilePath()) 

90 

91 

92 geom_multipole = GeometryMultipole(self.g, self.roxie_data, self.model_data, self.bh_curve_database, self.input_conductor_params, 

93 self.settings, self.COMSOL_version) 

94 

95 self.domains = geom_multipole.build_magnet() 

96 self.wedge_areas = geom_multipole.wedge_areas 

97 self.iron_yoke_areas = geom_multipole.iron_yoke_areas 

98 #self.plot_magnet(self.iron_yoke_areas , self.wedge_areas, self.roxie_data) 

99 self.connect_create_MPH_model(self.domains) 

100 self.save_files_java() 

101 self.save_compile_and_open_bat() 

102 

103 def setup_config_sigma(self): 

104 cfg = self.g.ConfigSigma() 

105 cfg.setComsolVersion(self.COMSOL_version) # for sigma_60 

106 cfg.setOutputModelPath(self.model_java_file_path) 

107 cfg.setComsolBatchPath(self.COMSOL_batch_path) 

108 cfg.setComsolCompilePath(self.COMSOL_compile_path) 

109 cfg.setExternalCFunLibPath(self.settings.CFunLibPath) 

110 cfg.setRunStudy(self.make_batch_mode_executable) 

111 cfg.setNumOfQHs(self.num_qh) 

112 cfg.setStudyType(self.study_type) 

113 cfg.setSymFactor(1) 

114 if self.model_data.Options_SIGMA.quench_initialization.quench_init_HT is not None: 

115 array = self.a.convert_list_to_string_array(self.g.gateway, 

116 self.model_data.Options_SIGMA.quench_initialization.quench_init_HT) 

117 cfg.setQuenchInitHT(array) 

118 if self.model_data.Options_SIGMA.quench_initialization.quench_init_heat is not None: 

119 cfg.setQuenchInitHeat(float(self.model_data.Options_SIGMA.quench_initialization.quench_init_heat)) 

120 qh_positions = self.g.gateway.jvm.java.util.ArrayList() 

121 if self.model_data.Options_SIGMA.quench_heaters.quench_heater_positions is not None: 

122 for qh in self.model_data.Options_SIGMA.quench_heaters.quench_heater_positions: 

123 temp = self.g.gateway.jvm.java.util.ArrayList() 

124 for pos in qh: 

125 temp.add(pos) 

126 qh_positions.add(temp) 

127 cfg.setQHPositions(qh_positions) 

128 return cfg 

129 

130 def build_study(self): 

131 """ 

132 If valid study time defined function creates a COMSOL study 

133 :return: None 

134 """ 

135 if self.study_type is not None: 

136 if self.study_type == self.g.MPHC.LABEL_STATIONARY: 

137 # Add code to create and run study 

138 self.g.StudyAPI.setNewBasicStationaryStudy(self.srv, self.cfg, "sol1") 

139 elif self.study_type == self.g.MPHC.LABEL_TRANSIENT: 

140 self.g.StudyAPI.setNewMonolithicStudy(self.srv, self.cfg, "Default_study", self.timeRange) 

141 else: 

142 ValueError("Invaid study_type input") 

143 

144 def create_results(self, input_coordinates_path, path_to_results): 

145 """ 

146 Function creates SIGMA result objects to export 

147 :param input_coordinates_path: path to coordinate file to evaluate 2D variables in. 

148 :param path_to_results: path to result folder 

149 :return: 

150 """ 

151 if self.make_batch_mode_executable: 

152 for i in range(len(self.variables2DConverted)): 

153 time_vector_2D = ', '.join(str(x) for x in self.time2DConverted[i]) 

154 

155 self.g.ResultsAPI.create2DResultNode(self.srv, self.cfg, self.variables2DConverted[i], time_vector_2D, 

156 f"data {i}", 

157 input_coordinates_path, str(path_to_results)) 

158 if self.study_type == self.g.MPHC.LABEL_TRANSIENT: 

159 for j in range(len(self.variables1DvsTime)): 

160 self.g.ResultsAPI.create1DResultNodeAllTimes(self.srv, self.cfg, self.variables1DvsTime[j], 

161 f"1DExport_{j}", path_to_results) 

162 

163 for k in range(len(self.variables1DvsTimeVector)): 

164 time_vector_1D = ', '.join(str(x) for x in self.time1DConverted[k]) 

165 self.g.ResultsAPI.create1DResultNodeTimeVector(self.srv, self.cfg, self.variables1DvsTimeVector[k], 

166 time_vector_1D, 

167 f"data {i + 1 + k}", path_to_results) 

168 else: 

169 pass 

170 # print("No study run, no results to be exported.") 

171 

172 def connect_create_MPH_model(self, domains): 

173 """ 

174 This function connects to the COMSOL server and creates an MPH model by using the MagnetMPHBuilder. 

175 It then builds the MPH model with the specified domains and global variables. 

176 If the study type is "Transient" and there are Quench Heaters present, it also builds a Quench Heater MPH model. 

177 

178 :param domains: a tuple of domains to be included in the MPH model. 

179 :return: None 

180 """ 

181 

182 self.srv.connect(self.cfg.getComsolBatchPath()) 

183 self.model = self.g.MagnetMPHBuilder(self.cfg, self.srv) 

184 # Create map with I0. 

185 global_map = BuildGlobalVariables(self.g, self.model_data).build_global_variables() 

186 global_map.put(self.g.MPHC.LABEL_CLIQ_CURRENT_EXT_INITIAL, 

187 f"{self.model_data.Power_Supply.I_initial}") 

188 

189 self.model = self.g.MagnetMPHBuilder(self.cfg, self.srv) 

190 self.model.buildMPH(self.a.create_domain_array(self.g.gateway, tuple(domains)), global_map, 4, 1, 

191 self.funlib_map) 

192 

193 if self.study_type == self.g.MPHC.LABEL_TRANSIENT and self.cfg.getNumOfQHs() > 0: 

194 builder_qh = self.g.QuenchHeaterMPHBuilder(self.cfg, self.srv) 

195 self.model.connectToServer() 

196 builder_qh.buildQuenchHeaterMPH() 

197 if self.generate_study: 

198 self.build_study() 

199 # Define result nodes 

200 

201 self.create_results(self.input_coordinates_path, self.path_to_results) 

202 self.srv.build(self.cfg.getOutputModelPath()) 

203 self.model.save() 

204 

205 def save_files_java(self): 

206 with open(self.model_java_file_path) as java_file: 

207 

208 print("Java file splitting started.") 

209 max_no_lines = 6e4 

210 public_indexes = [] 

211 files = [] 

212 content = java_file.readlines() 

213 for i, line in enumerate(content): 

214 if "public static" in line: 

215 public_indexes += [i] 

216 

217 no_lines = public_indexes[-1] - public_indexes[0] 

218 no_files = int(np.ceil(no_lines / max_no_lines)) 

219 max_no_lines = round(no_lines / no_files) 

220 real_indexes = [public_indexes[i] - public_indexes[0] for i in range(len(public_indexes))] 

221 closest = [min(real_indexes, key=lambda x: abs(x - max_no_lines * (i + 1))) + public_indexes[0] 

222 for i in range(no_files)] 

223 no_run = [int(content[i][content[i].index('run') + 3:content[i].index('(')]) for i in closest[0:-1]] 

224 no_run += [int(content[public_indexes[-2]][content[public_indexes[-2]].index('run') 

225 + 3:content[public_indexes[-2]].index('(')]) + 1] 

226 additional_lines = {} 

227 for i in range(no_files): 

228 file_path = os.path.join(self.output_path, self.model_name) 

229 files += [open(file_path + '_%d.java' % i, 'w')] 

230 name = self.model_name + '_%d' % i 

231 self.split_java_file_path += [f"{os.path.join(self.output_path, self.model_name + '_%d' % i)}"] 

232 files[i].writelines(content[0:public_indexes[0] - 2] + ['public class ' + name + ' {\n', '\n']) 

233 if i == 0: 

234 files[i].writelines('\tpublic static Model run1(Model mph) {\n') 

235 files[i].writelines(content[public_indexes[0] + 2:closest[i]] + ['}\n']) 

236 additional_lines[name] = {'start': 2, 'end': no_run[i] - 1} 

237 elif i + 1 == no_files: 

238 files[i].writelines(content[closest[i - 1]:public_indexes[-1]] + ['}\n']) 

239 additional_lines[name] = {'start': no_run[i - 1], 'end': len(public_indexes) - 1} 

240 else: 

241 files[i].writelines(content[closest[i - 1]:closest[i]] + ['}\n']) 

242 additional_lines[name] = {'start': no_run[i - 1], 'end': no_run[i] - 1} 

243 files[i].close() 

244 

245 with open(self.model_java_file_path, 'w') as java_file: 

246 content = content[0:public_indexes[0] + 2] + ['\n'] + \ 

247 content[public_indexes[-1]:public_indexes[-1] + 2] + ['\t}\n'] + ['}\n'] 

248 content.insert(public_indexes[0] + 2, '\t\tmph = ' + self.model_name + '_0.run1(mph);\n') 

249 ll = 1 

250 for name, item in additional_lines.items(): 

251 for j in range(item['end'] - item['start'] + 1): 

252 content.insert(public_indexes[0] + 2 + ll + j, 

253 '\t\tmph = ' + name + '.run' + str(item['start'] + j) + '(mph);\n') 

254 ll += j + 1 

255 content.insert(public_indexes[0] + 2 + ll, '\t\treturn mph;\n') 

256 content.insert(public_indexes[0] + 3 + ll, '\t}\n') 

257 java_file.writelines(content) 

258 

259 if len(self.split_java_file_path) == 1: 

260 print(f"BuilderSIGMA successfully saved file: {self.split_java_file_path[0]}") 

261 else: 

262 print("BuilderSIGMA successfully saved files:") 

263 for file in self.split_java_file_path: 

264 print(file) 

265 

266 def save_compile_and_open_bat(self): 

267 """ 

268 Function writes a .bat file which has to be run to create a COMSOL model. 

269 :return: None 

270 """ 

271 script_lines = [] 

272 class_paths = [] 

273 network_drive = False 

274 for file in self.split_java_file_path: 

275 if file.startswith("\\\\"): # Check if the path is a UNC path 

276 network_drive = True 

277 print("Network drive identified: Configuring .bat file to support network paths.") 

278 script_lines.append(f'pushd "{self.output_path}"') 

279 script_lines.append( 

280 f'"{self.COMSOL_compile_path}" -jdkroot "{self.java_jdk_path}" "{os.path.basename(file)}.java"') 

281 script_lines.append( 

282 f'"{self.java_jdk_path}\\bin\\jar.exe" cf "{os.path.basename(file)}.jar" "{os.path.basename(file)}.class"') 

283 class_paths.append(f'"{os.path.basename(file)}.jar"') 

284 else: 

285 script_lines.append(f'"{self.COMSOL_compile_path}" -jdkroot "{self.java_jdk_path}" "{file}.java"') 

286 script_lines.append(f'"{self.java_jdk_path}\\bin\\jar.exe" cf "{file}.jar" "{file}.class"') 

287 class_paths.append(f'"{os.path.join(self.output_path, file)}.jar"') 

288 

289 script_lines.append( 

290 f'"{self.COMSOL_compile_path}" -jdkroot "{self.java_jdk_path}" -classpathadd {" ".join(class_paths)} "{self.model_java_file_path}"') 

291 script_lines.append(f'"{self.COMSOL_batch_path}" -inputfile "{self.model_class_file_path}" ' 

292 f'-outputfile "{os.path.join(self.output_path, self.model_data.GeneralParameters.magnet_name)}.mph"') 

293 if network_drive: 

294 script_lines.append('popd') 

295 

296 with open(self.compile_batch_file_path, "w") as outfile: 

297 outfile.write("\n".join(str(line) for line in script_lines)) 

298 

299 print(f'BuilderSIGMA successfully saved: {self.compile_batch_file_path}') 

300 os.chdir(self.output_path) 

301