Coverage for src/meshpy/four_c/run_four_c.py: 57%

1# The MIT License (MIT) 

2# 

3# Copyright (c) 2018-2025 MeshPy Authors 

4# 

5# Permission is hereby granted, free of charge, to any person obtaining a copy 

6# of this software and associated documentation files (the "Software"), to deal 

7# in the Software without restriction, including without limitation the rights 

8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 

9# copies of the Software, and to permit persons to whom the Software is 

10# furnished to do so, subject to the following conditions: 

11# 

12# The above copyright notice and this permission notice shall be included in 

13# all copies or substantial portions of the Software. 

14# 

15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 

16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 

17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 

18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 

19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 

20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 

21# THE SOFTWARE. 

22"""Provide a function that allows to run 4C.""" 

23 

24import os as _os 

25import shutil as _shutil 

26import subprocess as _subprocess # nosec B404 

27import sys as _sys 

28from pathlib import Path as _Path 

29 

30from meshpy.utils.environment import get_env_variable as _get_env_variable 

31 

32 

33def run_four_c( 

34 input_file, 

35 output_dir, 

36 *, 

37 four_c_exe=None, 

38 mpi_command=None, 

39 n_proc=None, 

40 output_name="xxx", 

41 restart_step=None, 

42 restart_from=None, 

43 log_to_console=False, 

44): 

45 """Run a 4C simulation and return the exit code of the run. 

46 

47 This function looks into the environment variables for some parameters: 

48 "MESHPY_FOUR_C_EXE" 

49 "MESHPY_MPI_COMMAND" 

50 "MESHPY_MPI_NUM_PROC" 

51 If the corresponding keyword arguments are set, they overwrite the environment 

52 variable. 

53 

54 Args 

55 ---- 

56 input_file: str 

57 Path to the input file on the filesystem 

58 output_dir: str 

59 Directory where the simulation should be performed (will be created if 

60 it does not exist) 

61 four_c_exe: str 

62 Optionally explicitly specify path to the 4C executable 

63 mpi_command: str 

64 Command to launch MPI, defaults to "mpirun" 

65 n_proc: int 

66 Number of process used with MPI, defaults to 1 

67 output_name: str 

68 Base name of the output files 

69 restart_step: int 

70 Time step to restart from 

71 restart_from: str 

72 Path to initial simulation (relative to output_dir) 

73 log_to_console: bool 

74 If the 4C simulation output should be shown in the console. 

75 

76 Return 

77 ---- 

78 return_code: int 

79 Return code of 4C run 

80 """ 

81 

82 # Fist get all needed parameters 

83 if four_c_exe is None: 

84 four_c_exe = _get_env_variable("MESHPY_FOUR_C_EXE") 

85 if mpi_command is None: 

86 mpi_command = _get_env_variable("MESHPY_MPI_COMMAND", default="mpirun") 

87 if n_proc is None: 

88 n_proc = _get_env_variable("MESHPY_MPI_NUM_PROC", default="1") 

89 

90 # Setup paths and actual command to run 

91 _os.makedirs(output_dir, exist_ok=True) 

92 log_file = _os.path.join(output_dir, output_name + ".log") 

93 error_file = _os.path.join(output_dir, output_name + ".err") 

94 command = mpi_command.split(" ") + [ 

95 "-np", 

96 str(n_proc), 

97 four_c_exe, 

98 input_file, 

99 output_name, 

100 ] 

101 if restart_step is None and restart_from is None: 

102 pass 

103 elif restart_step is not None and restart_from is not None: 

104 command.extend([f"restart={restart_step}", f"restartfrom={restart_from}"]) 

105 else: 

106 raise ValueError( 

107 "Provide either both or no argument of [restart_step, restart_from]" 

108 ) 

109 

110 # Actually run the command 

111 with open(log_file, "w") as stdout_file, open(error_file, "w") as stderr_file: 

112 process = _subprocess.Popen( 

113 command, # nosec B603 

114 stdout=_subprocess.PIPE, 

115 stderr=_subprocess.PIPE, 

116 cwd=output_dir, 

117 text=True, 

118 ) 

119 

120 for stdout_line in process.stdout: 

121 if log_to_console: 

122 _sys.stdout.write(stdout_line) 

123 stdout_file.write(stdout_line) 

124 

125 for stderr_line in process.stderr: 

126 if log_to_console: 

127 _sys.stderr.write(stderr_line) 

128 stderr_file.write(stderr_line) 

129 

130 process.stdout.close() 

131 process.stderr.close() 

132 return_code = process.wait() 

133 return return_code 

134 

135 

136def clean_simulation_directory(sim_dir, *, ask_before_clean=False): 

137 """Clear the simulation directory. If it does not exist, it is created. 

138 Optionally the user can be asked before a deletion of files. 

139 

140 Args 

141 ---- 

142 sim_dir: 

143 Path to a directory 

144 ask_before_clean: bool 

145 Flag which indicates whether the user must confirm removal of files and directories 

146 """ 

147 

148 # Check if simulation directory exists. 

149 if _os.path.exists(sim_dir): 

150 if ask_before_clean: 

151 print(f'Path "{sim_dir}" already exists') 

152 while True: 

153 if ask_before_clean: 

154 answer = input("DELETE all contents? (y/n): ") 

155 else: 

156 answer = "y" 

157 if answer.lower() == "y": 

158 for filename in _os.listdir(sim_dir): 

159 file_path = _os.path.join(sim_dir, filename) 

160 try: 

161 if _os.path.isfile(file_path) or _os.path.islink(file_path): 

162 _os.unlink(file_path) 

163 elif _os.path.isdir(file_path): 

164 _shutil.rmtree(file_path) 

165 except Exception as e: 

166 raise ValueError(f"Failed to delete {file_path}. Reason: {e}") 

167 return 

168 elif answer.lower() == "n": 

169 raise ValueError("Directory is not deleted!") 

170 else: 

171 _Path(sim_dir).mkdir(parents=True, exist_ok=True)