Coverage for fiqus/getdp_runners/RunGetdpHomogenizedConductor.py: 76%

1import os, re, subprocess, logging, timeit 

2import pandas as pd 

3import gmsh 

4 

5from fiqus.data.DataFiQuSConductor import Conductor, SolutionParameters 

6from fiqus.data.RegionsModelFiQuS import RegionsModel 

7from fiqus.utils.Utils import GmshUtils, FilesAndFolders 

8 

9from fiqus.pro_assemblers.ProAssembler import ASS_PRO 

10 

11logger = logging.getLogger('FiQuS') 

12 

13class Solve: 

14 def __init__(self, fdm, GetDP_path, geometry_folder, mesh_folder, verbose=True): 

15 self.fdm = fdm 

16 self.GetDP_path = GetDP_path 

17 

18 self.solution_folder = os.path.join(os.getcwd()) 

19 self.geometry_folder = geometry_folder 

20 self.mesh_folder = mesh_folder 

21 

22 self.mesh_file = os.path.join(self.mesh_folder, f"{self.fdm.general.magnet_name}.msh") 

23 self.pro_file = os.path.join(self.solution_folder, f"{self.fdm.general.magnet_name}.pro") 

24 self.regions_file = os.path.join(mesh_folder, f"{self.fdm.general.magnet_name}.regions") 

25 

26 self.verbose = verbose 

27 self.gu = GmshUtils(self.solution_folder, self.verbose) 

28 self.gu.initialize(verbosity_Gmsh=fdm.run.verbosity_Gmsh) 

29 

30 self.ass_pro = ASS_PRO(os.path.join(self.solution_folder, self.fdm.general.magnet_name)) 

31 self.regions_model = FilesAndFolders.read_data_from_yaml(self.regions_file, RegionsModel) 

32 self.material_properties_model = None 

33 

34 self.ed = {} # excitation dictionary 

35 self.rohm = {} # rohm parameter dictionary 

36 self.rohf = {} # rohf parameter dictionary 

37 

38 gmsh.option.setNumber("General.Terminal", verbose) 

39 

40 def read_ro_parameters(self, inputs_folder_path): 

41 """ 

42 Function for reading the CSV files containing the reduced order parameters for ROHF and ROHM. 

43 The function expects CSV files with an descriptive first header row (i.e. 'alphas,kappas,taus') and following rows with comma seperated parameter values for each cell starting from cell 1.  

44 

45 :param inputs_folder_path: The full path to the folder with input files 

46 """ 

47 class reduced_order_params: 

48 " This structure is used to access the parameter dictionaries inside the pro template with a convenient syntax 'mp.rohf' or 'mp.rohm'." 

49 def __init__(self, rohf_dict=None, rohm_dict=None): 

50 self.rohf = rohf_dict 

51 self.rohm = rohm_dict 

52 

53 if self.fdm.magnet.solve.rohm.enable and self.fdm.magnet.solve.rohm.parameter_csv_file: 

54 parameter_file = os.path.join(inputs_folder_path, self.fdm.magnet.solve.rohm.parameter_csv_file) 

55 df = pd.read_csv(parameter_file, delimiter=',', engine='python') 

56 df.index += 1 

57 logger.info(f'Using ROHM parameters from file: {parameter_file}:') 

58 logger.info(df) 

59 self.rohm = df.to_dict(orient='dict') 

60 

61 if self.fdm.magnet.solve.rohf.enable and self.fdm.magnet.solve.rohf.parameter_csv_file: 

62 parameter_file = os.path.join(inputs_folder_path, self.fdm.magnet.solve.rohf.parameter_csv_file) 

63 df = pd.read_csv(parameter_file, delimiter=',', engine='python') 

64 df.index += 1 

65 logger.info(f'Using ROHF parameters from file: {parameter_file}:') 

66 logger.info(df) 

67 self.rohf = df.to_dict(orient='dict') 

68 

69 # For the moment the reduced order parameters are treated as material properties mp structure. 

70 self.material_properties_model = reduced_order_params(rohf_dict=self.rohf, rohm_dict=self.rohm) 

71 

72 def read_excitation(self, inputs_folder_path): 

73 """ 

74 Function for reading a CSV file for the 'from_file' excitation case. 

75 

76 :param inputs_folder_path: The full path to the folder with input files. 

77 :type inputs_folder_path: str 

78 """ 

79 if self.fdm.magnet.solve.source_parameters.source_type == 'piecewise' and self.fdm.magnet.solve.source_parameters.piecewise.source_csv_file: 

80 input_file = os.path.join(inputs_folder_path, self.fdm.magnet.solve.source_parameters.piecewise.source_csv_file) 

81 logger.info(f'Using excitation from file: {input_file}') 

82 df = pd.read_csv(input_file, delimiter=',', engine='python') 

83 excitation_time = df['time'].to_numpy(dtype='float').tolist() 

84 self.ed['time'] = excitation_time 

85 excitation_value = df['value'].to_numpy(dtype='float').tolist() 

86 self.ed['value'] = excitation_value 

87 

88 def assemble_pro(self): 

89 logger.info("Assembling .pro file") 

90 self.ass_pro.assemble_combined_pro(template = self.fdm.magnet.solve.pro_template, rm = self.regions_model, dm = self.fdm, ed=self.ed, mp=self.material_properties_model) 

91 

92 def run_getdp(self, solve = True, postOperation = True, gui = False): 

93 

94 command = ["-v2", "-verbose", "3", "-mat_mumps_icntl_14","100"] 

95 if solve: 

96 command += ["-solve", "MagDyn"] 

97 command += ["-pos", "MagDyn"] 

98 

99 if self.fdm.magnet.solve.general_parameters.noOfMPITasks: 

100 mpi_prefix = ["mpiexec", "-np", str(self.fdm.magnet.solve.general_parameters.noOfMPITasks)] 

101 else: 

102 mpi_prefix = [] 

103 

104 startTime = timeit.default_timer() 

105 

106 getdpProcess = subprocess.Popen(mpi_prefix + [self.GetDP_path, self.pro_file, "-msh", self.mesh_file] + command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) 

107 

108 with getdpProcess.stdout: 

109 for line in iter(getdpProcess.stdout.readline, b""): 

110 line = line.decode("utf-8").rstrip() 

111 line = line.split("\r")[-1] 

112 if not "Test" in line: 

113 if line.startswith("Info"): 

114 parsedLine = re.sub(r"Info\s+:\s+", "", line) 

115 logger.info(parsedLine) 

116 elif line.startswith("Warning"): 

117 parsedLine = re.sub(r"Warning\s+:\s+", "", line) 

118 logger.warning(parsedLine) 

119 elif line.startswith("Error"): 

120 parsedLine = re.sub(r"Error\s+:\s+", "", line) 

121 logger.error(parsedLine) 

122 logger.error("Solving HomogenizedConductor failed.") 

123 # raise Exception(parsedLine) 

124 elif re.match("##", line): 

125 logger.critical(line) 

126 else: 

127 logger.info(line) 

128 

129 simulation_time = timeit.default_timer()-startTime 

130 # Save simulation time: 

131 if solve: 

132 logger.info(f"Solving HomogenizedConductor has finished in {round(simulation_time, 3)} seconds.") 

133 with open(os.path.join(self.solution_folder, 'txt_files', 'simulation_time.txt'), 'w') as file: 

134 file.write(str(simulation_time)) 

135 

136 

137 if gui and ((postOperation and not solve) or (solve and postOperation and self.fdm.magnet.postproc.generate_pos_files)): 

138 # gmsh.option.setNumber("Geometry.Volumes", 1) 

139 # gmsh.option.setNumber("Geometry.Surfaces", 1) 

140 # gmsh.option.setNumber("Geometry.Curves", 1) 

141 # gmsh.option.setNumber("Geometry.Points", 0) 

142 posFiles = [ 

143 fileName 

144 for fileName in os.listdir(self.solution_folder) 

145 if fileName.endswith(".pos") 

146 ] 

147 for posFile in reversed(posFiles): 

148 gmsh.open(os.path.join(self.solution_folder, posFile)) 

149 self.gu.launch_interactive_GUI() 

150 else: 

151 if gmsh.isInitialized(): 

152 gmsh.clear() 

153 gmsh.finalize() 

154 

155 def cleanup(self): 

156 """ 

157 This funtion is used to remove .msh, .pre and .res files from the solution folder, as they may be large and not needed. 

158 """ 

159 magnet_name = self.fdm.general.magnet_name 

160 cleanup = self.fdm.magnet.postproc.cleanup 

161 

162 if cleanup.remove_res_file: 

163 res_file_path = os.path.join(self.solution_folder, f"{magnet_name}.res") 

164 if os.path.exists(res_file_path): 

165 os.remove(res_file_path) 

166 logger.info(f"Removed {magnet_name}.res") 

167 

168 if cleanup.remove_pre_file: 

169 pre_file_path = os.path.join(self.solution_folder, f"{magnet_name}.pre") 

170 if os.path.exists(pre_file_path): 

171 os.remove(pre_file_path) 

172 logger.info(f"Removed {magnet_name}.pre") 

173 

174 if cleanup.remove_msh_file: 

175 msh_file_path = os.path.join(self.mesh_folder, f"{magnet_name}.msh") 

176 if os.path.exists(msh_file_path): 

177 os.remove(msh_file_path) 

178 logger.info(f"Removed {magnet_name}.msh")