Coverage for fiqus/getdp_runners/RunGetdpCCT.py: 97%

1import os 

2import timeit 

3from pathlib import Path 

4import subprocess 

5 

6import gmsh 

7 

8from fiqus.utils.Utils import FilesAndFolders as uff 

9from fiqus.utils.Utils import GmshUtils 

10from fiqus.data.RegionsModelFiQuS import RegionsModel 

11from fiqus.pro_assemblers.ProAssembler import ASS_PRO as ass_pro 

12 

13 

14class RunGetdpCCT: 

15 def __init__(self, fdm, GetDP_path: dict = None, verbose=True): 

16 """ 

17 Class to preparing brep files by adding terminals. 

18 :param fdm: FiQuS data model 

19 :param verbose: If True more information is printed in python console. 

20 """ 

21 self.fdm = fdm 

22 self.cctdm = fdm.magnet 

23 self.GetDP_path = GetDP_path 

24 self.model_folder = os.path.join(os.getcwd()) 

25 self.magnet_name = fdm.general.magnet_name 

26 self.mesh_folder = Path(self.model_folder).parent 

27 self.verbose = verbose 

28 self.cctrm = uff.read_data_from_yaml(os.path.join(self.model_folder, f'{self.magnet_name}.regions'), RegionsModel) 

29 self.ap = ass_pro(os.path.join(self.model_folder, self.magnet_name)) 

30 self.gu = GmshUtils(self.model_folder, self.verbose) 

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

32 self.pos_names = [] 

33 self.call_method = 'subprocess' 

34 # for variable, volume, file_ext in zip(self.cctdm.solve.variables, self.cctdm.solve.volumes, self.cctdm.solve.file_exts): 

35 # self.pos_names.append(f'{variable}_{volume}.{file_ext}') 

36 

37 def assemble_pro(self): 

38 if self.verbose: 

39 print('Assembling Pro File Started') 

40 start_time = timeit.default_timer() 

41 if self.verbose: 

42 print('Assembling pro file') 

43 self.ap.assemble_combined_pro(template=self.cctdm.solve.pro_template, rm=self.cctrm, dm=self.cctdm, mf=f"{os.path.join(self.mesh_folder, self.magnet_name)}.msh") 

44 if self.verbose: 

45 print(f'Assembling Pro File Took {timeit.default_timer() - start_time:.2f} s') 

46 

47 def solve_and_postprocess(self, gui=False): 

48 command = "-solve -v2 -pos" 

49 model_file = 'Center_line.csv' 

50 self._run(command=command, model_file=model_file, gui=gui) 

51 

52 def postprocess(self, gui=False): 

53 command = "-v2 -pos" 

54 model_file = 'Test.dat' 

55 self._run(command=command, model_file=model_file, gui=gui) 

56 

57 def _run(self, command, model_file, gui=False): 

58 if self.verbose: 

59 print('Solving Started !!!') 

60 start_time = timeit.default_timer() 

61 mpi_prefix = [] 

62 res = 'MagDynAVComplex' 

63 post_op = 'Get_LocalFields' 

64 

65 if self.call_method == 'onelab': 

66 gmsh.onelab.run(f"{self.magnet_name}", f"{self.GetDP_path} {os.path.join(self.model_folder, self.magnet_name)}.pro {command} -msh {os.path.join(self.mesh_folder, self.magnet_name)}.msh") 

67 elif self.call_method == 'subprocess': 

68 command = ["-solve", res, "-v2", "-pos", post_op, "-verbose", str(self.fdm.run.verbosity_GetDP), '-sub_pc_type lu', '-pc_type lu', '-pc_factor_mat_solver_type mumps'] 

69 subprocess.call(mpi_prefix + [f"{self.GetDP_path}", f"{os.path.join(self.model_folder, self.magnet_name)}.pro"] + command + [f"-msh {os.path.join(self.mesh_folder, self.magnet_name)}.msh"]) 

70 

71 self.model_file = os.path.join(self.model_folder, model_file) 

72 if self.verbose: 

73 print('' 

74 f'Solving Took {timeit.default_timer() - start_time:.2f} s !!!' 

75 '') 

76 if gui: 

77 self.gu.launch_interactive_GUI() 

78 else: 

79 gmsh.clear() 

80 gmsh.finalize()