Coverage for biobb_gromacs/gromacs/mdrun_plumed.py: 57%

1#!/usr/bin/env python3 

2 

3"""Module containing the MDrun class and the command line interface.""" 

4import os 

5import shutil 

6from pathlib import PurePath 

7from typing import Optional 

8from biobb_common.tools import file_utils as fu 

9from biobb_common.tools.file_utils import launchlogger 

10from biobb_gromacs.gromacs.mdrun_base import MdrunBase 

11 

12 

13class MdrunPlumed(MdrunBase): 

14 """ 

15 | biobb_gromacs MdrunPlumed 

16 | Wrapper of the `GROMACS mdrun <http://manual.gromacs.org/current/onlinehelp/gmx-mdrun.html>`_ module. 

17 | MDRun is the main computational chemistry engine within GROMACS. It performs Molecular Dynamics simulations, but it can also perform Stochastic Dynamics, Energy Minimization, test particle insertion or (re)calculation of energies. 

18 

19 Args: 

20 input_tpr_path (str): Path to the portable binary run input file TPR. File type: input. `Sample file <https://github.com/bioexcel/biobb_gromacs/raw/master/biobb_gromacs/test/data/gromacs/mdrun.tpr>`_. Accepted formats: tpr (edam:format_2333). 

21 output_gro_path (str): Path to the output GROMACS structure GRO file. File type: output. `Sample file <https://github.com/bioexcel/biobb_gromacs/raw/master/biobb_gromacs/test/reference/gromacs/ref_mdrun.gro>`_. Accepted formats: gro (edam:format_2033). 

22 output_edr_path (str): Path to the output GROMACS portable energy file EDR. File type: output. `Sample file <https://github.com/bioexcel/biobb_gromacs/raw/master/biobb_gromacs/test/reference/gromacs/ref_mdrun.edr>`_. Accepted formats: edr (edam:format_2330). 

23 output_log_path (str): Path to the output GROMACS trajectory log file LOG. File type: output. `Sample file <https://github.com/bioexcel/biobb_gromacs/raw/master/biobb_gromacs/test/reference/gromacs/ref_mdrun.log>`_. Accepted formats: log (edam:format_2330). 

24 output_trr_path (str) (Optional): Path to the GROMACS uncompressed raw trajectory file TRR. File type: output. `Sample file <https://github.com/bioexcel/biobb_gromacs/raw/master/biobb_gromacs/test/reference/gromacs/ref_mdrun.trr>`_. Accepted formats: trr (edam:format_3910). 

25 input_cpt_path (str) (Optional): Path to the input GROMACS checkpoint file CPT. File type: input. Accepted formats: cpt (edam:format_2333). 

26 output_xtc_path (str) (Optional): Path to the GROMACS compressed trajectory file XTC. File type: output. Accepted formats: xtc (edam:format_3875). 

27 output_cpt_path (str) (Optional): Path to the output GROMACS checkpoint file CPT. File type: output. Accepted formats: cpt (edam:format_2333). 

28 output_dhdl_path (str) (Optional): Path to the output dhdl.xvg file only used when free energy calculation is turned on. File type: output. Accepted formats: xvg (edam:format_2033). 

29 input_plumed_path (str) (Optional): Path to the main PLUMED input file. If provided, PLUMED will be used during the simulation. All files used by the main PLUMED input file must exist in the input_plumed_folder and be called with just their name. Make sure to provide a GROMACS version with the PLUMED patch. File type: input. Accepted formats: dat (edam:format_2330). 

30 input_plumed_folder (dir) (Optional): Path to the folder with all files needed by the main PLUMED input file, see input_plumed_path. File type: input. Accepted formats: directory (edam:format_1915) 

31 output_plumed_folder (dir) (Optional): Folder where PLUMED generated output files will be saved. File type: output. Accepted formats: directory (edam:format_1915) 

32 properties (dict - Python dictionary object containing the tool parameters, not input/output files): 

33 * **mpi_bin** (*str*) - (None) Path to the MPI runner. Usually "mpirun" or "srun". 

34 * **mpi_np** (*int*) - (0) [0~1000|1] Number of MPI processes. Usually an integer bigger than 1. 

35 * **mpi_flags** (*str*) - (None) Path to the MPI hostlist file. 

36 * **checkpoint_time** (*int*) - (15) [0~1000|1] Checkpoint writing interval in minutes. Only enabled if an output_cpt_path is provided. 

37 * **noappend** (*bool*) - (False) Include the noappend flag to open new output files and add the simulation part number to all output file names 

38 * **num_threads** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of threads that are going to be used. 

39 * **num_threads_mpi** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of GROMACS MPI threads that are going to be used. 

40 * **num_threads_omp** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of GROMACS OPENMP threads that are going to be used. 

41 * **num_threads_omp_pme** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of GROMACS OPENMP_PME threads that are going to be used. 

42 * **use_gpu** (*bool*) - (False) Use settings appropriate for GPU. Adds: -nb gpu -pme gpu 

43 * **gpu_id** (*str*) - (None) list of unique GPU device IDs available to use. 

44 * **gpu_tasks** (*str*) - (None) list of GPU device IDs, mapping each PP task on each node to a device. 

45 * **gmx_lib** (*str*) - (None) Path set GROMACS GMXLIB environment variable. 

46 * **binary_path** (*str*) - ("gmx") Path to the GROMACS executable binary. 

47 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files. 

48 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist. 

49 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory. 

50 * **container_path** (*str*) - (None) Path to the binary executable of your container. 

51 * **container_image** (*str*) - (None) Container Image identifier. 

52 * **container_volume_path** (*str*) - ("/data") Path to an internal directory in the container. 

53 * **container_working_dir** (*str*) - (None) Path to the internal CWD in the container. 

54 * **container_user_id** (*str*) - (None) User number id to be mapped inside the container. 

55 * **container_shell_path** (*str*) - ("/bin/bash") Path to the binary executable of the container shell. 

56 

57 Examples: 

58 This is a use example of how to use the building block from Python:: 

59 

60 from biobb_gromacs.gromacs.mdrun_plumed import mdrun_plumed 

61 prop = { 'num_threads': 0, 

62 'binary_path': 'gmx' } 

63 mdrun_plumed(input_tpr_path='/path/to/myPortableBinaryRunInputFile.tpr', 

64 output_trr_path='/path/to/newTrajectory.trr', 

65 output_gro_path='/path/to/newStructure.gro', 

66 output_edr_path='/path/to/newEnergy.edr', 

67 output_log_path='/path/to/newSimulationLog.log', 

68 properties=prop) 

69 

70 Info: 

71 * wrapped_software: 

72 * name: GROMACS Mdrun with PLUMED 

73 * version: 2025.2 

74 * license: LGPL 2.1 

75 * multinode: mpi 

76 * ontology: 

77 * name: EDAM 

78 * schema: http://edamontology.org/EDAM.owl 

79 """ 

80 

81 def __init__(self, input_tpr_path: str, output_gro_path: str, output_edr_path: str, 

82 output_log_path: str, output_trr_path: Optional[str] = None, input_cpt_path: Optional[str] = None, 

83 output_xtc_path: Optional[str] = None, output_cpt_path: Optional[str] = None, 

84 output_dhdl_path: Optional[str] = None, input_plumed_path: Optional[str] = None, 

85 input_plumed_folder: Optional[str] = None, output_plumed_folder: Optional[str] = None, 

86 properties: Optional[dict] = None, **kwargs) -> None: 

87 properties = properties or {} 

88 

89 # Call parent class constructor 

90 super().__init__(properties) 

91 self.locals_var_dict = locals().copy() 

92 

93 # Input/Output files 

94 self.io_dict = { 

95 "in": {"input_tpr_path": input_tpr_path, "input_cpt_path": input_cpt_path, 

96 "input_plumed_path": input_plumed_path, "input_plumed_folder": input_plumed_folder}, 

97 "out": {"output_trr_path": output_trr_path, "output_gro_path": output_gro_path, 

98 "output_edr_path": output_edr_path, "output_log_path": output_log_path, 

99 "output_xtc_path": output_xtc_path, "output_cpt_path": output_cpt_path, 

100 "output_dhdl_path": output_dhdl_path, "output_plumed_folder": output_plumed_folder} 

101 } 

102 

103 # Properties specific for BB 

104 self._init_common_properties(properties) 

105 

106 # Check the properties 

107 self.check_properties(properties) 

108 self.check_arguments() 

109 

110 @launchlogger 

111 def launch(self) -> int: 

112 """Execute the :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` object.""" 

113 

114 # Setup Biobb 

115 if self.check_restart(): 

116 return 0 

117 

118 self.stage_files() 

119 

120 working_dir = self._get_working_dir() 

121 

122 self.cmd = [self.binary_path, 'mdrun', 

123 '-s', PurePath(self.stage_io_dict["in"]["input_tpr_path"]).name, 

124 '-c', PurePath(self.stage_io_dict["out"]["output_gro_path"]).name, 

125 '-e', PurePath(self.stage_io_dict["out"]["output_edr_path"]).name, 

126 '-g', PurePath(self.stage_io_dict["out"]["output_log_path"]).name] 

127 

128 if self.stage_io_dict["out"].get("output_trr_path"): 

129 self.cmd.append('-o') 

130 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_trr_path"]).name) 

131 

132 if self.stage_io_dict["in"].get("input_plumed_path"): 

133 self.cmd.append('-plumed') 

134 self.cmd.append(PurePath(self.stage_io_dict["in"]["input_plumed_path"]).name) 

135 

136 if self.stage_io_dict["in"].get("input_cpt_path"): 

137 self.cmd.append('-cpi') 

138 self.cmd.append(PurePath(self.stage_io_dict["in"]["input_cpt_path"]).name) 

139 if self.stage_io_dict["out"].get("output_xtc_path"): 

140 self.cmd.append('-x') 

141 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_xtc_path"]).name) 

142 else: 

143 self.tmp_files.append('traj_comp.xtc') 

144 if self.stage_io_dict["out"].get("output_cpt_path"): 

145 self.cmd.append('-cpo') 

146 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_cpt_path"]).name) 

147 if self.checkpoint_time: 

148 self.cmd.append('-cpt') 

149 self.cmd.append(str(self.checkpoint_time)) 

150 if self.stage_io_dict["out"].get("output_dhdl_path"): 

151 self.cmd.append('-dhdl') 

152 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_dhdl_path"]).name) 

153 

154 # Shared mpi / working-directory / runtime flags 

155 self._prepend_mpi_runner() 

156 self.cmd = ["cd", working_dir, ";"] + self.cmd 

157 self._append_gmx_runtime_flags() 

158 

159 # Run Biobb block 

160 self.run_biobb() 

161 

162 # Copy files to host 

163 self.copy_to_host() 

164 

165 # Remove temporal files 

166 self.remove_tmp_files() 

167 

168 self.check_arguments(output_files_created=True, raise_exception=False) 

169 return self.return_code 

170 

171 def stage_files(self): 

172 """ 

173 Stage the input/output files in a temporal unique directory aka sandbox. 

174 

175 Overwrite the parent class method to handle PLUMED input files. 

176 """ 

177 

178 # If PLUMED is requested, change the working directory to the sandbox 

179 if self.io_dict["in"].get("input_plumed_path"): 

180 fu.log("PLUMED detected: Enabling chdir_sandbox to ensure relative paths work.", self.out_log) 

181 self.chdir_sandbox = True 

182 

183 super().stage_files() 

184 

185 # If plumed folder is provided, flatten its contents into the sandbox 

186 if self.stage_io_dict["in"].get("input_plumed_folder"): 

187 plumed_folder = self.stage_io_dict["in"]["input_plumed_folder"] 

188 for item in os.listdir(plumed_folder): 

189 s = os.path.join(plumed_folder, item) 

190 d = os.path.join(self.stage_io_dict["unique_dir"], item) 

191 if os.path.isdir(s): 

192 shutil.copytree(s, d, dirs_exist_ok=True) 

193 else: 

194 shutil.copy2(s, d) 

195 

196 def _extra_copy_to_host(self) -> None: 

197 """Copy PLUMED-generated output files from the sandbox to the host.""" 

198 # Bulk Copy PLUMED outputs 

199 if self.io_dict["out"].get("output_plumed_folder"): 

200 dest_folder = self.io_dict["out"]["output_plumed_folder"] 

201 os.makedirs(dest_folder, exist_ok=True) 

202 

203 unique_dir = self.stage_io_dict["unique_dir"] 

204 # We ignore files that were inputs 

205 input_filenames = [os.path.basename(f) for f in self.io_dict["in"].values() if f] 

206 # We ignore standard GMX outputs already copied 

207 gmx_output_filenames = [os.path.basename(f) for f in self.stage_io_dict["out"].values() if f and isinstance(f, str)] 

208 

209 fu.log(f"Searching for PLUMED outputs in {unique_dir}...", self.out_log) 

210 for item in os.listdir(unique_dir): 

211 if item not in input_filenames and item not in gmx_output_filenames: 

212 if os.path.isdir(os.path.join(unique_dir, item)): 

213 # Skip directories 

214 continue 

215 # NOTE: Here we could list specific PLUMED output patterns or skip files contained in the input_plumed_folder 

216 src = os.path.join(unique_dir, item) 

217 dst = os.path.join(dest_folder, item) 

218 fu.log(f"Copying PLUMED output: {item} --> {dest_folder}", self.out_log) 

219 shutil.copy2(src, dst) 

220 

221 

222def mdrun_plumed(input_tpr_path: str, output_gro_path: str, output_edr_path: str, 

223 output_log_path: str, output_trr_path: Optional[str] = None, input_cpt_path: Optional[str] = None, 

224 output_xtc_path: Optional[str] = None, output_cpt_path: Optional[str] = None, 

225 output_dhdl_path: Optional[str] = None, input_plumed_path: Optional[str] = None, 

226 input_plumed_folder: Optional[str] = None, output_plumed_folder: Optional[str] = None, 

227 properties: Optional[dict] = None, **kwargs) -> int: 

228 """Create :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` class and 

229 execute the :meth:`launch() <gromacs.mdrun_plumed.MdrunPlumed.launch>` method.""" 

230 return MdrunPlumed(**dict(locals())).launch() 

231 

232 

233mdrun_plumed.__doc__ = MdrunPlumed.__doc__ 

234main = MdrunPlumed.get_main(mdrun_plumed, "Wrapper for the GROMACS mdrun with PLUMED module.") 

235 

236 

237if __name__ == '__main__': 

238 main()