Coverage for biobb_gromacs/gromacs/mdrun_plumed.py: 50%
175 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-05-28 06:50 +0000
1#!/usr/bin/env python3
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.generic.biobb_object import BiobbObject
9from biobb_common.tools import file_utils as fu
10from biobb_common.tools.file_utils import launchlogger
11from biobb_gromacs.gromacs.common import get_gromacs_version
14class MdrunPlumed(BiobbObject):
15 """
16 | biobb_gromacs MdrunPlumed
17 | Wrapper of the `GROMACS mdrun <http://manual.gromacs.org/current/onlinehelp/gmx-mdrun.html>`_ module.
18 | 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.
20 Args:
21 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).
22 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).
23 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).
24 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).
25 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).
26 input_cpt_path (str) (Optional): Path to the input GROMACS checkpoint file CPT. File type: input. Accepted formats: cpt (edam:format_2333).
27 output_xtc_path (str) (Optional): Path to the GROMACS compressed trajectory file XTC. File type: output. Accepted formats: xtc (edam:format_3875).
28 output_cpt_path (str) (Optional): Path to the output GROMACS checkpoint file CPT. File type: output. Accepted formats: cpt (edam:format_2333).
29 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).
30 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).
31 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)
32 output_plumed_folder (dir) (Optional): Folder where PLUMED generated output files will be saved. File type: output. Accepted formats: directory (edam:format_1915)
33 properties (dict - Python dictionary object containing the tool parameters, not input/output files):
34 * **mpi_bin** (*str*) - (None) Path to the MPI runner. Usually "mpirun" or "srun".
35 * **mpi_np** (*int*) - (0) [0~1000|1] Number of MPI processes. Usually an integer bigger than 1.
36 * **mpi_flags** (*str*) - (None) Path to the MPI hostlist file.
37 * **checkpoint_time** (*int*) - (15) [0~1000|1] Checkpoint writing interval in minutes. Only enabled if an output_cpt_path is provided.
38 * **noappend** (*bool*) - (False) Include the noappend flag to open new output files and add the simulation part number to all output file names
39 * **num_threads** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of threads that are going to be used.
40 * **num_threads_mpi** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of GROMACS MPI threads that are going to be used.
41 * **num_threads_omp** (*int*) - (0) [0~1000|1] Let GROMACS guess. The number of GROMACS OPENMP threads that are going to be used.
42 * **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.
43 * **use_gpu** (*bool*) - (False) Use settings appropriate for GPU. Adds: -nb gpu -pme gpu
44 * **gpu_id** (*str*) - (None) list of unique GPU device IDs available to use.
45 * **gpu_tasks** (*str*) - (None) list of GPU device IDs, mapping each PP task on each node to a device.
46 * **gmx_lib** (*str*) - (None) Path set GROMACS GMXLIB environment variable.
47 * **binary_path** (*str*) - ("gmx") Path to the GROMACS executable binary.
48 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files.
49 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist.
50 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory.
51 * **container_path** (*str*) - (None) Path to the binary executable of your container.
52 * **container_image** (*str*) - (None) Container Image identifier.
53 * **container_volume_path** (*str*) - ("/data") Path to an internal directory in the container.
54 * **container_working_dir** (*str*) - (None) Path to the internal CWD in the container.
55 * **container_user_id** (*str*) - (None) User number id to be mapped inside the container.
56 * **container_shell_path** (*str*) - ("/bin/bash") Path to the binary executable of the container shell.
58 Examples:
59 This is a use example of how to use the building block from Python::
61 from biobb_gromacs.gromacs.mdrun_plumed import mdrun_plumed
62 prop = { 'num_threads': 0,
63 'binary_path': 'gmx' }
64 mdrun_plumed(input_tpr_path='/path/to/myPortableBinaryRunInputFile.tpr',
65 output_trr_path='/path/to/newTrajectory.trr',
66 output_gro_path='/path/to/newStructure.gro',
67 output_edr_path='/path/to/newEnergy.edr',
68 output_log_path='/path/to/newSimulationLog.log',
69 properties=prop)
71 Info:
72 * wrapped_software:
73 * name: GROMACS Mdrun with PLUMED
74 * version: 2025.2
75 * license: LGPL 2.1
76 * multinode: mpi
77 * ontology:
78 * name: EDAM
79 * schema: http://edamontology.org/EDAM.owl
80 """
82 def __init__(self, input_tpr_path: str, output_gro_path: str, output_edr_path: str,
83 output_log_path: str, output_trr_path: Optional[str] = None, input_cpt_path: Optional[str] = None,
84 output_xtc_path: Optional[str] = None, output_cpt_path: Optional[str] = None,
85 output_dhdl_path: Optional[str] = None, input_plumed_path: Optional[str] = None,
86 input_plumed_folder: Optional[str] = None, output_plumed_folder: Optional[str] = None,
87 properties: Optional[dict] = None, **kwargs) -> None:
88 properties = properties or {}
90 # Call parent class constructor
91 super().__init__(properties)
92 self.locals_var_dict = locals().copy()
94 # Input/Output files
95 self.io_dict = {
96 "in": {"input_tpr_path": input_tpr_path, "input_cpt_path": input_cpt_path,
97 "input_plumed_path": input_plumed_path, "input_plumed_folder": input_plumed_folder},
98 "out": {"output_trr_path": output_trr_path, "output_gro_path": output_gro_path,
99 "output_edr_path": output_edr_path, "output_log_path": output_log_path,
100 "output_xtc_path": output_xtc_path, "output_cpt_path": output_cpt_path,
101 "output_dhdl_path": output_dhdl_path, "output_plumed_folder": output_plumed_folder}
102 }
104 # Properties specific for BB
105 # general mpi properties
106 self.mpi_bin = properties.get('mpi_bin')
107 self.mpi_np = properties.get('mpi_np')
108 self.mpi_flags = properties.get('mpi_flags')
109 # gromacs cpu mpi/openmp properties
110 self.num_threads = str(properties.get('num_threads', ''))
111 self.num_threads_mpi = str(properties.get('num_threads_mpi', ''))
112 self.num_threads_omp = str(properties.get('num_threads_omp', ''))
113 self.num_threads_omp_pme = str(
114 properties.get('num_threads_omp_pme', ''))
115 # gromacs gpus
116 self.use_gpu = properties.get(
117 'use_gpu', False) # Adds: -nb gpu -pme gpu
118 self.gpu_id = str(properties.get('gpu_id', ''))
119 self.gpu_tasks = str(properties.get('gpu_tasks', ''))
120 # gromacs
121 self.checkpoint_time = properties.get('checkpoint_time')
122 self.noappend = properties.get('noappend', False)
124 # Properties common in all GROMACS BB
125 self.gmx_lib = properties.get('gmx_lib', None)
126 self.binary_path: str = properties.get('binary_path', 'gmx')
127 self.gmx_nobackup = properties.get('gmx_nobackup', True)
128 self.gmx_nocopyright = properties.get('gmx_nocopyright', True)
129 if self.gmx_nobackup:
130 self.binary_path += ' -nobackup'
131 if self.gmx_nocopyright:
132 self.binary_path += ' -nocopyright'
133 if (not self.mpi_bin) and (not self.container_path):
134 self.gmx_version = get_gromacs_version(self.binary_path)
136 # Check the properties
137 self.check_properties(properties)
138 self.check_arguments()
140 @launchlogger
141 def launch(self) -> int:
142 """Execute the :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` object."""
144 # Setup Biobb
145 if self.check_restart():
146 return 0
148 # Optional output files (if not added mrun will create them using a generic name)
149 if not self.stage_io_dict["out"].get("output_trr_path"):
150 self.stage_io_dict["out"]["output_trr_path"] = fu.create_name(
151 prefix=self.prefix, step=self.step, name='trajectory.trr')
152 self.tmp_files.append(self.stage_io_dict["out"]["output_trr_path"])
154 self.stage_files()
156 if self.container_path:
157 working_dir = self.container_volume_path if self.container_volume_path else "/data"
158 else:
159 working_dir = self.stage_io_dict.get('unique_dir', '')
161 self.cmd = [self.binary_path, 'mdrun',
162 '-o', PurePath(self.stage_io_dict["out"]["output_trr_path"]).name,
163 '-s', PurePath(self.stage_io_dict["in"]["input_tpr_path"]).name,
164 '-c', PurePath(self.stage_io_dict["out"]["output_gro_path"]).name,
165 '-e', PurePath(self.stage_io_dict["out"]["output_edr_path"]).name,
166 '-g', PurePath(self.stage_io_dict["out"]["output_log_path"]).name]
168 if self.stage_io_dict["in"].get("input_plumed_path"):
169 self.cmd.append('-plumed')
170 self.cmd.append(PurePath(self.stage_io_dict["in"]["input_plumed_path"]).name)
172 if self.stage_io_dict["in"].get("input_cpt_path"):
173 self.cmd.append('-cpi')
174 self.cmd.append(PurePath(self.stage_io_dict["in"]["input_cpt_path"]).name)
175 if self.stage_io_dict["out"].get("output_xtc_path"):
176 self.cmd.append('-x')
177 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_xtc_path"]).name)
178 else:
179 self.tmp_files.append('traj_comp.xtc')
180 if self.stage_io_dict["out"].get("output_cpt_path"):
181 self.cmd.append('-cpo')
182 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_cpt_path"]).name)
183 if self.checkpoint_time:
184 self.cmd.append('-cpt')
185 self.cmd.append(str(self.checkpoint_time))
186 if self.stage_io_dict["out"].get("output_dhdl_path"):
187 self.cmd.append('-dhdl')
188 self.cmd.append(PurePath(self.stage_io_dict["out"]["output_dhdl_path"]).name)
190 # general mpi properties
191 if self.mpi_bin:
192 mpi_cmd = [self.mpi_bin]
193 if self.mpi_np:
194 mpi_cmd.append('-n')
195 mpi_cmd.append(str(self.mpi_np))
196 if self.mpi_flags:
197 mpi_cmd.extend(self.mpi_flags)
198 self.cmd = mpi_cmd + self.cmd
200 self.cmd = ["cd", working_dir, ";"] + self.cmd
202 # gromacs cpu mpi/openmp properties
203 if self.num_threads:
204 fu.log(
205 f'User added number of gmx threads: {self.num_threads}', self.out_log)
206 self.cmd.append('-nt')
207 self.cmd.append(self.num_threads)
208 if self.num_threads_mpi:
209 fu.log(
210 f'User added number of gmx mpi threads: {self.num_threads_mpi}', self.out_log)
211 self.cmd.append('-ntmpi')
212 self.cmd.append(self.num_threads_mpi)
213 if self.num_threads_omp:
214 fu.log(
215 f'User added number of gmx omp threads: {self.num_threads_omp}', self.out_log)
216 self.cmd.append('-ntomp')
217 self.cmd.append(self.num_threads_omp)
218 if self.num_threads_omp_pme:
219 fu.log(
220 f'User added number of gmx omp_pme threads: {self.num_threads_omp_pme}', self.out_log)
221 self.cmd.append('-ntomp_pme')
222 self.cmd.append(self.num_threads_omp_pme)
223 # GMX gpu properties
224 if self.use_gpu:
225 fu.log('Adding GPU specific settings adds: -nb gpu -pme gpu', self.out_log)
226 self.cmd += ["-nb", "gpu", "-pme", "gpu"]
227 if self.gpu_id:
228 fu.log(
229 f'list of unique GPU device IDs available to use: {self.gpu_id}', self.out_log)
230 self.cmd.append('-gpu_id')
231 self.cmd.append(self.gpu_id)
232 if self.gpu_tasks:
233 fu.log(
234 f'list of GPU device IDs, mapping each PP task on each node to a device: {self.gpu_tasks}', self.out_log)
235 self.cmd.append('-gputasks')
236 self.cmd.append(self.gpu_tasks)
238 if self.noappend:
239 self.cmd.append('-noappend')
241 if self.gmx_lib:
242 self.env_vars_dict['GMXLIB'] = self.gmx_lib
244 # Run Biobb block
245 self.run_biobb()
247 # Copy files to host
248 self.copy_to_host()
250 # Remove temporal files
251 self.remove_tmp_files()
253 self.check_arguments(output_files_created=True, raise_exception=False)
254 return self.return_code
256 def stage_files(self):
257 """
258 Stage the input/output files in a temporal unique directory aka sandbox.
260 Overwrite the parent class method to handle PLUMED input files.
261 """
263 # If PLUMED is requested, change the working directory to the sandbox
264 if self.io_dict["in"].get("input_plumed_path"):
265 fu.log("PLUMED detected: Enabling chdir_sandbox to ensure relative paths work.", self.out_log)
266 self.chdir_sandbox = True
268 super().stage_files()
270 # If plumed folder is provided, flatten its contents into the sandbox
271 if self.stage_io_dict["in"].get("input_plumed_folder"):
272 plumed_folder = self.stage_io_dict["in"]["input_plumed_folder"]
273 for item in os.listdir(plumed_folder):
274 s = os.path.join(plumed_folder, item)
275 d = os.path.join(self.stage_io_dict["unique_dir"], item)
276 if os.path.isdir(s):
277 shutil.copytree(s, d, dirs_exist_ok=True)
278 else:
279 shutil.copy2(s, d)
281 def copy_to_host(self):
282 """
283 Updates the path to the original output files in the sandbox,
284 to catch changes due to noappend restart.
286 GROMACS mdrun will change the output file names from md.gro to md.part0001.gro
287 if the noappend flag is used.
288 """
289 import pathlib
291 def capture_part_pattern(filename):
292 """
293 Captures the 'part' pattern followed by digits from a string.
294 """
295 import re
296 pattern = r'part\d+'
298 match = re.search(pattern, filename)
299 if match:
300 return match.group(0)
301 else:
302 return None
304 if self.noappend:
305 # List files in the staging directory
306 staging_path = self.stage_io_dict["unique_dir"]
307 files_in_staging = list(pathlib.Path(staging_path).glob('*'))
309 # Find the part000x pattern in the output files
310 for file in files_in_staging:
311 part_pattern = capture_part_pattern(file.name)
312 if part_pattern:
313 break
315 # Update expected output files
316 for file_ref, stage_file_path in self.stage_io_dict["out"].items():
317 if stage_file_path:
318 # Find the parent and the file name in the sandbox
319 parent_path = pathlib.Path(stage_file_path).parent
320 file_stem = pathlib.Path(stage_file_path).stem
321 file_suffix = pathlib.Path(stage_file_path).suffix
323 # Rename all output files except checkpoint files
324 if file_suffix != '.cpt':
325 # Create the new file name with the part pattern
326 if part_pattern:
327 new_file_name = f"{file_stem}.{part_pattern}{file_suffix}"
328 new_file_path = parent_path / new_file_name
329 # Update the stage_io_dict with the new file path
330 self.stage_io_dict["out"][file_ref] = str(
331 new_file_path)
333 super().copy_to_host()
335 # Bulk Copy PLUMED outputs
336 if self.io_dict["out"].get("output_plumed_folder"):
337 dest_folder = self.io_dict["out"]["output_plumed_folder"]
338 os.makedirs(dest_folder, exist_ok=True)
340 unique_dir = self.stage_io_dict["unique_dir"]
341 # We ignore files that were inputs
342 input_filenames = [os.path.basename(f) for f in self.io_dict["in"].values() if f]
343 # We ignore standard GMX outputs already copied
344 gmx_output_filenames = [os.path.basename(f) for f in self.stage_io_dict["out"].values() if f and isinstance(f, str)]
346 fu.log(f"Searching for PLUMED outputs in {unique_dir}...", self.out_log)
347 for item in os.listdir(unique_dir):
348 if item not in input_filenames and item not in gmx_output_filenames:
349 if os.path.isdir(os.path.join(unique_dir, item)):
350 # Skip directories
351 continue
352 # NOTE: Here we could list specific PLUMED output patterns or skip files contained in the input_plumed_folder
353 src = os.path.join(unique_dir, item)
354 dst = os.path.join(dest_folder, item)
355 fu.log(f"Copying PLUMED output: {item} --> {dest_folder}", self.out_log)
356 shutil.copy2(src, dst)
359def mdrun_plumed(input_tpr_path: str, output_gro_path: str, output_edr_path: str,
360 output_log_path: str, output_trr_path: Optional[str] = None, input_cpt_path: Optional[str] = None,
361 output_xtc_path: Optional[str] = None, output_cpt_path: Optional[str] = None,
362 output_dhdl_path: Optional[str] = None, input_plumed_path: Optional[str] = None,
363 input_plumed_folder: Optional[str] = None, output_plumed_folder: Optional[str] = None,
364 properties: Optional[dict] = None, **kwargs) -> int:
365 """Create :class:`MdrunPlumed <gromacs.mdrun_plumed.MdrunPlumed>` class and
366 execute the :meth:`launch() <gromacs.mdrun_plumed.MdrunPlumed.launch>` method."""
367 return MdrunPlumed(**dict(locals())).launch()
370mdrun_plumed.__doc__ = MdrunPlumed.__doc__
371main = MdrunPlumed.get_main(mdrun_plumed, "Wrapper for the GROMACS mdrun with PLUMED module.")
374if __name__ == '__main__':
375 main()