Coverage for biobb_gromacs / gromacs / mdrun_plumed.py: 49%
170 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-05 08:26 +0000
1#!/usr/bin/env python3
3"""Module containing the MDrun class and the command line interface."""
4import os
5import shutil
6from typing import Optional
7from biobb_common.generic.biobb_object import BiobbObject
8from biobb_common.tools import file_utils as fu
9from biobb_common.tools.file_utils import launchlogger
10from biobb_gromacs.gromacs.common import get_gromacs_version
13class MdrunPlumed(BiobbObject):
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.
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.
57 Examples:
58 This is a use example of how to use the building block from Python::
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)
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 """
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 {}
89 # Call parent class constructor
90 super().__init__(properties)
91 self.locals_var_dict = locals().copy()
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 }
103 # Properties specific for BB
104 # general mpi properties
105 self.mpi_bin = properties.get('mpi_bin')
106 self.mpi_np = properties.get('mpi_np')
107 self.mpi_flags = properties.get('mpi_flags')
108 # gromacs cpu mpi/openmp properties
109 self.num_threads = str(properties.get('num_threads', ''))
110 self.num_threads_mpi = str(properties.get('num_threads_mpi', ''))
111 self.num_threads_omp = str(properties.get('num_threads_omp', ''))
112 self.num_threads_omp_pme = str(
113 properties.get('num_threads_omp_pme', ''))
114 # gromacs gpus
115 self.use_gpu = properties.get(
116 'use_gpu', False) # Adds: -nb gpu -pme gpu
117 self.gpu_id = str(properties.get('gpu_id', ''))
118 self.gpu_tasks = str(properties.get('gpu_tasks', ''))
119 # gromacs
120 self.checkpoint_time = properties.get('checkpoint_time')
121 self.noappend = properties.get('noappend', False)
123 # Properties common in all GROMACS BB
124 self.gmx_lib = properties.get('gmx_lib', None)
125 self.binary_path: str = properties.get('binary_path', 'gmx')
126 self.gmx_nobackup = properties.get('gmx_nobackup', True)
127 self.gmx_nocopyright = properties.get('gmx_nocopyright', True)
128 if self.gmx_nobackup:
129 self.binary_path += ' -nobackup'
130 if self.gmx_nocopyright:
131 self.binary_path += ' -nocopyright'
132 if (not self.mpi_bin) and (not self.container_path):
133 self.gmx_version = get_gromacs_version(self.binary_path)
135 # Check the properties
136 self.check_properties(properties)
137 self.check_arguments()
139 @launchlogger
140 def launch(self) -> int:
141 """Execute the :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` object."""
143 # Setup Biobb
144 if self.check_restart():
145 return 0
147 # Optional output files (if not added mrun will create them using a generic name)
148 if not self.stage_io_dict["out"].get("output_trr_path"):
149 self.stage_io_dict["out"]["output_trr_path"] = fu.create_name(
150 prefix=self.prefix, step=self.step, name='trajectory.trr')
151 self.tmp_files.append(self.stage_io_dict["out"]["output_trr_path"])
153 self.stage_files()
155 self.cmd = [self.binary_path, 'mdrun',
156 '-o', self.stage_io_dict["out"]["output_trr_path"],
157 '-s', self.stage_io_dict["in"]["input_tpr_path"],
158 '-c', self.stage_io_dict["out"]["output_gro_path"],
159 '-e', self.stage_io_dict["out"]["output_edr_path"],
160 '-g', self.stage_io_dict["out"]["output_log_path"]]
162 if self.stage_io_dict["in"].get("input_plumed_path"):
163 self.cmd.append('-plumed')
164 self.cmd.append(self.stage_io_dict["in"]["input_plumed_path"])
166 if self.stage_io_dict["in"].get("input_cpt_path"):
167 self.cmd.append('-cpi')
168 self.cmd.append(self.stage_io_dict["in"]["input_cpt_path"])
169 if self.stage_io_dict["out"].get("output_xtc_path"):
170 self.cmd.append('-x')
171 self.cmd.append(self.stage_io_dict["out"]["output_xtc_path"])
172 else:
173 self.tmp_files.append('traj_comp.xtc')
174 if self.stage_io_dict["out"].get("output_cpt_path"):
175 self.cmd.append('-cpo')
176 self.cmd.append(self.stage_io_dict["out"]["output_cpt_path"])
177 if self.checkpoint_time:
178 self.cmd.append('-cpt')
179 self.cmd.append(str(self.checkpoint_time))
180 if self.stage_io_dict["out"].get("output_dhdl_path"):
181 self.cmd.append('-dhdl')
182 self.cmd.append(self.stage_io_dict["out"]["output_dhdl_path"])
184 # general mpi properties
185 if self.mpi_bin:
186 mpi_cmd = [self.mpi_bin]
187 if self.mpi_np:
188 mpi_cmd.append('-n')
189 mpi_cmd.append(str(self.mpi_np))
190 if self.mpi_flags:
191 mpi_cmd.extend(self.mpi_flags)
192 self.cmd = mpi_cmd + self.cmd
194 # gromacs cpu mpi/openmp properties
195 if self.num_threads:
196 fu.log(
197 f'User added number of gmx threads: {self.num_threads}', self.out_log)
198 self.cmd.append('-nt')
199 self.cmd.append(self.num_threads)
200 if self.num_threads_mpi:
201 fu.log(
202 f'User added number of gmx mpi threads: {self.num_threads_mpi}', self.out_log)
203 self.cmd.append('-ntmpi')
204 self.cmd.append(self.num_threads_mpi)
205 if self.num_threads_omp:
206 fu.log(
207 f'User added number of gmx omp threads: {self.num_threads_omp}', self.out_log)
208 self.cmd.append('-ntomp')
209 self.cmd.append(self.num_threads_omp)
210 if self.num_threads_omp_pme:
211 fu.log(
212 f'User added number of gmx omp_pme threads: {self.num_threads_omp_pme}', self.out_log)
213 self.cmd.append('-ntomp_pme')
214 self.cmd.append(self.num_threads_omp_pme)
215 # GMX gpu properties
216 if self.use_gpu:
217 fu.log('Adding GPU specific settings adds: -nb gpu -pme gpu', self.out_log)
218 self.cmd += ["-nb", "gpu", "-pme", "gpu"]
219 if self.gpu_id:
220 fu.log(
221 f'list of unique GPU device IDs available to use: {self.gpu_id}', self.out_log)
222 self.cmd.append('-gpu_id')
223 self.cmd.append(self.gpu_id)
224 if self.gpu_tasks:
225 fu.log(
226 f'list of GPU device IDs, mapping each PP task on each node to a device: {self.gpu_tasks}', self.out_log)
227 self.cmd.append('-gputasks')
228 self.cmd.append(self.gpu_tasks)
230 if self.noappend:
231 self.cmd.append('-noappend')
233 if self.gmx_lib:
234 self.env_vars_dict['GMXLIB'] = self.gmx_lib
236 # Run Biobb block
237 self.run_biobb()
239 # Copy files to host
240 self.copy_to_host()
242 # Remove temporal files
243 self.remove_tmp_files()
245 self.check_arguments(output_files_created=True, raise_exception=False)
246 return self.return_code
248 def stage_files(self):
249 """
250 Stage the input/output files in a temporal unique directory aka sandbox.
252 Overwrite the parent class method to handle PLUMED input files.
253 """
255 # If PLUMED is requested, change the working directory to the sandbox
256 if self.io_dict["in"].get("input_plumed_path"):
257 fu.log("PLUMED detected: Enabling chdir_sandbox to ensure relative paths work.", self.out_log)
258 self.chdir_sandbox = True
260 super().stage_files()
262 # If plumed folder is provided, flatten its contents into the sandbox
263 if self.stage_io_dict["in"].get("input_plumed_folder"):
264 plumed_folder = self.stage_io_dict["in"]["input_plumed_folder"]
265 for item in os.listdir(plumed_folder):
266 s = os.path.join(plumed_folder, item)
267 d = os.path.join(self.stage_io_dict["unique_dir"], item)
268 if os.path.isdir(s):
269 shutil.copytree(s, d, dirs_exist_ok=True)
270 else:
271 shutil.copy2(s, d)
273 def copy_to_host(self):
274 """
275 Updates the path to the original output files in the sandbox,
276 to catch changes due to noappend restart.
278 GROMACS mdrun will change the output file names from md.gro to md.part0001.gro
279 if the noappend flag is used.
280 """
281 import pathlib
283 def capture_part_pattern(filename):
284 """
285 Captures the 'part' pattern followed by digits from a string.
286 """
287 import re
288 pattern = r'part\d+'
290 match = re.search(pattern, filename)
291 if match:
292 return match.group(0)
293 else:
294 return None
296 if self.noappend:
297 # List files in the staging directory
298 staging_path = self.stage_io_dict["unique_dir"]
299 files_in_staging = list(pathlib.Path(staging_path).glob('*'))
301 # Find the part000x pattern in the output files
302 for file in files_in_staging:
303 part_pattern = capture_part_pattern(file.name)
304 if part_pattern:
305 break
307 # Update expected output files
308 for file_ref, stage_file_path in self.stage_io_dict["out"].items():
309 if stage_file_path:
310 # Find the parent and the file name in the sandbox
311 parent_path = pathlib.Path(stage_file_path).parent
312 file_stem = pathlib.Path(stage_file_path).stem
313 file_suffix = pathlib.Path(stage_file_path).suffix
315 # Rename all output files except checkpoint files
316 if file_suffix != '.cpt':
317 # Create the new file name with the part pattern
318 if part_pattern:
319 new_file_name = f"{file_stem}.{part_pattern}{file_suffix}"
320 new_file_path = parent_path / new_file_name
321 # Update the stage_io_dict with the new file path
322 self.stage_io_dict["out"][file_ref] = str(
323 new_file_path)
325 super().copy_to_host()
327 # Bulk Copy PLUMED outputs
328 if self.io_dict["out"].get("output_plumed_folder"):
329 dest_folder = self.io_dict["out"]["output_plumed_folder"]
330 os.makedirs(dest_folder, exist_ok=True)
332 unique_dir = self.stage_io_dict["unique_dir"]
333 # We ignore files that were inputs
334 input_filenames = [os.path.basename(f) for f in self.io_dict["in"].values() if f]
335 # We ignore standard GMX outputs already copied
336 gmx_output_filenames = [os.path.basename(f) for f in self.stage_io_dict["out"].values() if f and isinstance(f, str)]
338 fu.log(f"Searching for PLUMED outputs in {unique_dir}...", self.out_log)
339 for item in os.listdir(unique_dir):
340 if item not in input_filenames and item not in gmx_output_filenames:
341 if os.path.isdir(os.path.join(unique_dir, item)):
342 # Skip directories
343 continue
344 # NOTE: Here we could list specific PLUMED output patterns or skip files contained in the input_plumed_folder
345 src = os.path.join(unique_dir, item)
346 dst = os.path.join(dest_folder, item)
347 fu.log(f"Copying PLUMED output: {item} --> {dest_folder}", self.out_log)
348 shutil.copy2(src, dst)
351def mdrun_plumed(input_tpr_path: str, output_gro_path: str, output_edr_path: str,
352 output_log_path: str, output_trr_path: Optional[str] = None, input_cpt_path: Optional[str] = None,
353 output_xtc_path: Optional[str] = None, output_cpt_path: Optional[str] = None,
354 output_dhdl_path: Optional[str] = None, input_plumed_path: Optional[str] = None,
355 input_plumed_folder: Optional[str] = None, output_plumed_folder: Optional[str] = None,
356 properties: Optional[dict] = None, **kwargs) -> int:
357 """Create :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` class and
358 execute the :meth:`launch() <gromacs.mdrun_plumed.MdrunPlumed.launch>` method."""
359 return MdrunPlumed(**dict(locals())).launch()
362mdrun_plumed.__doc__ = MdrunPlumed.__doc__
363main = MdrunPlumed.get_main(mdrun_plumed, "Wrapper for the GROMACS mdrun with PLUMED module.")
366if __name__ == '__main__':
367 main()