Coverage for biobb_amber / process / process_mdout.py: 89%
71 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-15 15:57 +0000
1#!/usr/bin/env python3
3"""Module containing the ProcessMDOut class and the command line interface."""
5import shutil
6from pathlib import Path, PurePath
7from typing import Optional
9from biobb_common.generic.biobb_object import BiobbObject
10from biobb_common.tools import file_utils as fu
11from biobb_common.tools.file_utils import launchlogger
13from biobb_amber.process.common import (
14 _from_string_to_list,
15 check_input_path,
16 check_output_path,
17)
20class ProcessMDOut(BiobbObject):
21 """
22 | biobb_amber.process.process_mdout ProcessMDOut
23 | Wrapper of the `AmberTools (AMBER MD Package) process_mdout tool <https://ambermd.org/AmberTools.php>`_ module.
24 | Parses the AMBER (sander) md output file (log) and dumps statistics that can then be plotted. Using the process_mdout.pl tool from the AmberTools MD package.
26 Args:
27 input_log_path (str): AMBER (sander) MD output (log) file. File type: input. `Sample file <https://github.com/bioexcel/biobb_amber/raw/master/biobb_amber/test/data/process/sander.heat.log>`_. Accepted formats: log (edam:format_2330), out (edam:format_2330), txt (edam:format_2330), o (edam:format_2330).
28 output_dat_path (str): Dat output file containing data from the specified terms along the minimization process. File type: output. `Sample file <https://github.com/bioexcel/biobb_amber/raw/master/biobb_amber/test/reference/process/sander.md.temp.dat>`_. Accepted formats: dat (edam:format_1637), txt (edam:format_2330), csv (edam:format_3752).
29 properties (dic - Python dictionary object containing the tool parameters, not input/output files):
30 * **terms** (*list*) - (["ETOT"]) Statistics descriptors. Values: VOLUME, TSOLVENT, TSOLUTE, TEMP, PRES, ETOT, ESCF, EPTOT, EKTOT, EKCMT, DENSITY.
31 * **binary_path** (*str*) - ("process_mdout.perl") Path to the process_mdout.perl executable binary.
32 * **remove_tmp** (*bool*) - (True) [WF property] Remove temporal files.
33 * **restart** (*bool*) - (False) [WF property] Do not execute if output files exist.
34 * **sandbox_path** (*str*) - ("./") [WF property] Parent path to the sandbox directory.
35 * **container_path** (*str*) - (None) Container path definition.
36 * **container_image** (*str*) - ('afandiadib/ambertools:serial') Container image definition.
37 * **container_volume_path** (*str*) - ('/tmp') Container volume path definition.
38 * **container_working_dir** (*str*) - (None) Container working directory definition.
39 * **container_user_id** (*str*) - (None) Container user_id definition.
40 * **container_shell_path** (*str*) - ('/bin/bash') Path to default shell inside the container.
42 Examples:
43 This is a use example of how to use the building block from Python::
45 from biobb_amber.process.process_mdout import process_mdout
46 prop = {
47 'terms' : ['TEMP','VOLUME','DENSITY']
48 }
49 process_mdout(input_log_path='/path/to/ambermd.log',
50 output_dat_path='/path/to/newFeature.dat',
51 properties=prop)
53 Info:
54 * wrapped_software:
55 * name: AmberTools process_mdout
56 * version: >20.9
57 * license: LGPL 2.1
58 * ontology:
59 * name: EDAM
60 * schema: http://edamontology.org/EDAM.owl
62 """
64 def __init__(self, input_log_path: str, output_dat_path: str, properties, **kwargs):
65 properties = properties or {}
67 # Call parent class constructor
68 super().__init__(properties)
69 self.locals_var_dict = locals().copy()
71 # Input/Output files
72 self.io_dict = {
73 "in": {"input_log_path": input_log_path},
74 "out": {"output_dat_path": output_dat_path},
75 }
77 # Properties specific for BB
78 self.properties = properties
79 self.terms = _from_string_to_list(properties.get("terms", ["ETOT"]))
80 self.binary_path = properties.get("binary_path", "process_mdout.perl")
82 # Check the properties
83 self.check_properties(properties)
84 self.check_arguments()
86 def check_data_params(self, out_log, err_log):
87 """Checks input/output paths correctness"""
89 # Check input(s)
90 self.io_dict["in"]["input_log_path"] = check_input_path(
91 self.io_dict["in"]["input_log_path"],
92 "input_log_path",
93 False,
94 out_log,
95 self.__class__.__name__,
96 )
98 # Check output(s)
99 self.io_dict["out"]["output_dat_path"] = check_output_path(
100 self.io_dict["out"]["output_dat_path"],
101 "output_dat_path",
102 False,
103 out_log,
104 self.__class__.__name__,
105 )
107 @launchlogger
108 def launch(self):
109 """Launches the execution of the ProcessMDOut module."""
111 # check input/output paths and parameters
112 self.check_data_params(self.out_log, self.err_log)
114 # Setup Biobb
115 if self.check_restart():
116 return 0
117 self.stage_files()
119 if not self.container_path:
120 self.tmp_folder = fu.create_unique_dir()
121 fu.log("Creating %s temporary folder" % self.tmp_folder, self.out_log)
122 self.cmd = [
123 "cd",
124 self.tmp_folder,
125 ";",
126 self.binary_path,
127 str(Path(self.stage_io_dict["in"]["input_log_path"]).resolve()),
128 ]
129 else:
130 self.tmp_folder = None
131 self.cmd = [self.binary_path, self.stage_io_dict["in"]["input_log_path"]]
133 # Run Biobb block
134 self.run_biobb()
136 # Copy files to host
137 self.copy_to_host()
139 if len(self.terms) == 1:
140 if self.container_path:
141 shutil.copy(
142 PurePath(self.stage_io_dict["unique_dir"]).joinpath(
143 "summary." + self.terms[0]
144 ),
145 self.io_dict["out"]["output_dat_path"],
146 )
147 else:
148 shutil.copy(
149 PurePath(str(self.tmp_folder)).joinpath("summary." + self.terms[0]),
150 self.io_dict["out"]["output_dat_path"],
151 )
152 else:
153 if self.container_path:
154 tmp = self.stage_io_dict["unique_dir"]
155 else:
156 tmp = self.tmp_folder
158 ene_dict = {}
159 for term in self.terms:
160 with open(str(tmp) + "/summary." + term) as fp:
161 for line in fp:
162 x = line.split()
163 if x:
164 if len(x) > 1:
165 ene_dict.setdefault(float(x[0]), {})[term] = x[1]
166 else:
167 ene_dict.setdefault(float(x[0]), {})[term] = "-"
169 with open(self.io_dict["out"]["output_dat_path"], "w") as fp_out:
170 fp_out.write("# TIME ")
171 for term in self.terms:
172 fp_out.write(term + " ")
173 fp_out.write("\n")
174 for key in sorted(ene_dict.keys()):
175 fp_out.write(str(key) + " ")
176 for term in self.terms:
177 fp_out.write(ene_dict[key][term] + " ")
178 fp_out.write("\n")
180 # remove temporary folder(s)
181 self.tmp_files.extend([
182 str(self.tmp_folder)
183 ] + list(Path().glob("summary*"))
184 )
185 self.remove_tmp_files()
187 self.check_arguments(output_files_created=True, raise_exception=False)
189 return self.return_code
192def process_mdout(
193 input_log_path: str,
194 output_dat_path: str,
195 properties: Optional[dict] = None,
196 **kwargs,
197) -> int:
198 """Create :class:`ProcessMDOut <process.process_mdout.ProcessMDOut>`process.process_mdout.ProcessMDOut class and
199 execute :meth:`launch() <process.process_mdout.ProcessMDOut.launch>` method"""
200 return ProcessMDOut(**dict(locals())).launch()
203process_mdout.__doc__ = ProcessMDOut.__doc__
205main = ProcessMDOut.get_main(process_mdout, "Parses the AMBER (sander) MD output file (log) and dumps statistics that can then be plotted. Using the process_mdout.pl tool from the AmberTools MD package.")
207if __name__ == "__main__":
208 main()