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

1import timeit 

2import logging 

3from enum import Enum 

4import os 

5import subprocess 

6import re 

7import pandas as pd 

8 

9import gmsh 

10import numpy as np 

11 

12#from fiqus.data import RegionsModelFiQuS 

13from fiqus.utils.Utils import GmshUtils, FilesAndFolders 

14from fiqus.data.RegionsModelFiQuS import RegionsModel 

15# import fiqus.data.DataConductorACGeom as geom 

16 

17from fiqus.pro_assemblers.ProAssembler import ASS_PRO 

18 

19logger = logging.getLogger('FiQuS') 

20 

21class Solve: 

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

23 self.fdm = fdm 

24 self.cacdm = fdm.magnet 

25 self.GetDP_path = GetDP_path 

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

27 self.magnet_name = fdm.general.magnet_name 

28 self.geometry_folder = geometry_folder 

29 self.mesh_folder = mesh_folder 

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

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

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

33 

34 self.verbose = verbose 

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

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

37 

38 self.ass_pro = ASS_PRO(os.path.join(self.solution_folder, self.magnet_name)) 

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

40 self.material_properties_model = None 

41 

42 self.ed = {} # excitation dictionary 

43 

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

45 

46 def assemble_pro(self): 

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

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

49 

50 def read_excitation(self, inputs_folder_path): 

51 """ 

52 Function for reading a CSV file for the 'piecewise' excitation case. 

53 

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

55 :type inputs_folder_path: str 

56 """ 

57 if self.cacdm.solve.source_parameters.source_type == 'piecewise' and self.cacdm.solve.source_parameters.piecewise.source_csv_file: 

58 input_file = os.path.join(inputs_folder_path, self.cacdm.solve.source_parameters.piecewise.source_csv_file) 

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

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

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

62 self.ed['time'] = excitation_time 

63 excitation_b = df['b'].to_numpy(dtype='float').tolist() 

64 self.ed['b'] = excitation_b 

65 excitation_I = df['I'].to_numpy(dtype='float').tolist() 

66 self.ed['I'] = excitation_I 

67 

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

69 command = ["-v2", "-verbose", "3"] 

70 if solve: 

71 command += ["-solve", "MagDyn", "-mat_mumps_icntl_14","100"] # icntl for mumps just by precaution 

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

73 

74 logger.info(f"Running GetDP with command: {command}") 

75 startTime = timeit.default_timer() 

76 

77 if self.cacdm.solve.general_parameters.noOfMPITasks: 

78 mpi_prefix = ["mpiexec", "-np", str(self.cacdm.solve.general_parameters.noOfMPITasks)] 

79 else: 

80 mpi_prefix = [] 

81 

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

83 

84 with getdpProcess.stdout: 

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

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

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

88 if not "Test" in line: 

89 if line.startswith("Info"): 

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

91 logger.info(parsedLine) 

92 elif line.startswith("Warning"): 

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

94 logger.warning(parsedLine) 

95 elif line.startswith("Error"): 

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

97 logger.error(parsedLine) 

98 logger.error("Solving CAC failed.") 

99 # raise Exception(parsedLine) 

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

101 logger.critical(line) 

102 else: 

103 logger.info(line) 

104 

105 simulation_time = timeit.default_timer()-startTime 

106 

107 if gui and ((postOperation and not solve) or (solve and postOperation)): # and self.cacdm.postproc.generate_pos_files 

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

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

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

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

112 posFiles = [ 

113 fileName 

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

115 if fileName.endswith(".pos") 

116 ] 

117 for posFile in posFiles: 

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

119 self.gu.launch_interactive_GUI() 

120 else: 

121 if gmsh.isInitialized(): 

122 gmsh.clear() 

123 gmsh.finalize()