Coverage for biobb_amber/process/process_mdout.py: 78%

1#!/usr/bin/env python3 

2 

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

4 

5import argparse 

6import shutil 

7from pathlib import Path, PurePath 

8from typing import Optional 

9 

10from biobb_common.configuration import settings 

11from biobb_common.generic.biobb_object import BiobbObject 

12from biobb_common.tools import file_utils as fu 

13from biobb_common.tools.file_utils import launchlogger 

14 

15from biobb_amber.process.common import ( 

16 _from_string_to_list, 

17 check_input_path, 

18 check_output_path, 

19) 

20 

21 

22class ProcessMDOut(BiobbObject): 

23 """ 

24 | biobb_amber.process.process_mdout ProcessMDOut 

25 | Wrapper of the `AmberTools (AMBER MD Package) process_mdout tool <https://ambermd.org/AmberTools.php>`_ module. 

26 | 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. 

27 

28 Args: 

29 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). 

30 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). 

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

32 * **terms** (*list*) - (["ETOT"]) Statistics descriptors. Values: VOLUME, TSOLVENT, TSOLUTE, TEMP, PRES, ETOT, ESCF, EPTOT, EKTOT, EKCMT, DENSITY. 

33 * **binary_path** (*str*) - ("process_mdout.perl") Path to the process_mdout.perl executable binary. 

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

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

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

37 * **container_path** (*str*) - (None) Container path definition. 

38 * **container_image** (*str*) - ('afandiadib/ambertools:serial') Container image definition. 

39 * **container_volume_path** (*str*) - ('/tmp') Container volume path definition. 

40 * **container_working_dir** (*str*) - (None) Container working directory definition. 

41 * **container_user_id** (*str*) - (None) Container user_id definition. 

42 * **container_shell_path** (*str*) - ('/bin/bash') Path to default shell inside the container. 

43 

44 Examples: 

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

46 

47 from biobb_amber.process.process_mdout import process_mdout 

48 prop = { 

49 'terms' : ['TEMP','VOLUME','DENSITY'] 

50 } 

51 process_mdout(input_log_path='/path/to/ambermd.log', 

52 output_dat_path='/path/to/newFeature.dat', 

53 properties=prop) 

54 

55 Info: 

56 * wrapped_software: 

57 * name: AmberTools process_mdout 

58 * version: >20.9 

59 * license: LGPL 2.1 

60 * ontology: 

61 * name: EDAM 

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

63 

64 """ 

65 

66 def __init__(self, input_log_path: str, output_dat_path: str, properties, **kwargs): 

67 properties = properties or {} 

68 

69 # Call parent class constructor 

70 super().__init__(properties) 

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

72 

73 # Input/Output files 

74 self.io_dict = { 

75 "in": {"input_log_path": input_log_path}, 

76 "out": {"output_dat_path": output_dat_path}, 

77 } 

78 

79 # Properties specific for BB 

80 self.properties = properties 

81 self.terms = _from_string_to_list(properties.get("terms", ["ETOT"])) 

82 self.binary_path = properties.get("binary_path", "process_mdout.perl") 

83 

84 # Check the properties 

85 self.check_properties(properties) 

86 self.check_arguments() 

87 

88 def check_data_params(self, out_log, err_log): 

89 """Checks input/output paths correctness""" 

90 

91 # Check input(s) 

92 self.io_dict["in"]["input_log_path"] = check_input_path( 

93 self.io_dict["in"]["input_log_path"], 

94 "input_log_path", 

95 False, 

96 out_log, 

97 self.__class__.__name__, 

98 ) 

99 

100 # Check output(s) 

101 self.io_dict["out"]["output_dat_path"] = check_output_path( 

102 self.io_dict["out"]["output_dat_path"], 

103 "output_dat_path", 

104 False, 

105 out_log, 

106 self.__class__.__name__, 

107 ) 

108 

109 @launchlogger 

110 def launch(self): 

111 """Launches the execution of the ProcessMDOut module.""" 

112 

113 # check input/output paths and parameters 

114 self.check_data_params(self.out_log, self.err_log) 

115 

116 # Setup Biobb 

117 if self.check_restart(): 

118 return 0 

119 self.stage_files() 

120 

121 if not self.container_path: 

122 self.tmp_folder = fu.create_unique_dir() 

123 fu.log("Creating %s temporary folder" % self.tmp_folder, self.out_log) 

124 self.cmd = [ 

125 "cd", 

126 self.tmp_folder, 

127 ";", 

128 self.binary_path, 

129 str(Path(self.stage_io_dict["in"]["input_log_path"]).resolve()), 

130 ] 

131 else: 

132 self.tmp_folder = None 

133 self.cmd = [self.binary_path, self.stage_io_dict["in"]["input_log_path"]] 

134 

135 # Run Biobb block 

136 self.run_biobb() 

137 

138 # Copy files to host 

139 self.copy_to_host() 

140 

141 if len(self.terms) == 1: 

142 if self.container_path: 

143 shutil.copy( 

144 PurePath(self.stage_io_dict["unique_dir"]).joinpath( 

145 "summary." + self.terms[0] 

146 ), 

147 self.io_dict["out"]["output_dat_path"], 

148 ) 

149 else: 

150 shutil.copy( 

151 PurePath(str(self.tmp_folder)).joinpath("summary." + self.terms[0]), 

152 self.io_dict["out"]["output_dat_path"], 

153 ) 

154 else: 

155 if self.container_path: 

156 tmp = self.stage_io_dict["unique_dir"] 

157 else: 

158 tmp = self.tmp_folder 

159 

160 ene_dict = {} 

161 for term in self.terms: 

162 with open(str(tmp) + "/summary." + term) as fp: 

163 for line in fp: 

164 x = line.split() 

165 if x: 

166 if len(x) > 1: 

167 ene_dict.setdefault(float(x[0]), {})[term] = x[1] 

168 else: 

169 ene_dict.setdefault(float(x[0]), {})[term] = "-" 

170 

171 with open(self.io_dict["out"]["output_dat_path"], "w") as fp_out: 

172 fp_out.write("# TIME ") 

173 for term in self.terms: 

174 fp_out.write(term + " ") 

175 fp_out.write("\n") 

176 for key in sorted(ene_dict.keys()): 

177 fp_out.write(str(key) + " ") 

178 for term in self.terms: 

179 fp_out.write(ene_dict[key][term] + " ") 

180 fp_out.write("\n") 

181 

182 # remove temporary folder(s) 

183 self.tmp_files.extend([ 

184 # self.stage_io_dict.get("unique_dir", ""), 

185 str(self.tmp_folder) 

186 ] + list(Path().glob("summary*")) 

187 ) 

188 self.remove_tmp_files() 

189 

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

191 

192 return self.return_code 

193 

194 

195def process_mdout( 

196 input_log_path: str, 

197 output_dat_path: str, 

198 properties: Optional[dict] = None, 

199 **kwargs, 

200) -> int: 

201 """Create :class:`ProcessMDOut <process.process_mdout.ProcessMDOut>`process.process_mdout.ProcessMDOut class and 

202 execute :meth:`launch() <process.process_mdout.ProcessMDOut.launch>` method""" 

203 

204 return ProcessMDOut( 

205 input_log_path=input_log_path, 

206 output_dat_path=output_dat_path, 

207 properties=properties, 

208 ).launch() 

209 

210 process_mdout.__doc__ = ProcessMDOut.__doc__ 

211 

212 

213def main(): 

214 parser = argparse.ArgumentParser( 

215 description="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.", 

216 formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999), 

217 ) 

218 parser.add_argument("--config", required=False, help="Configuration file") 

219 

220 # Specific args 

221 required_args = parser.add_argument_group("required arguments") 

222 required_args.add_argument( 

223 "--input_log_path", 

224 required=True, 

225 help="AMBER (sander) MD output (log) file. Accepted formats: log, out, txt, o.", 

226 ) 

227 required_args.add_argument( 

228 "--output_dat_path", 

229 required=True, 

230 help="Dat output file containing data from the specified terms along the MD process. File type: output. Accepted formats: dat, txt, csv.", 

231 ) 

232 

233 args = parser.parse_args() 

234 config = args.config if args.config else None 

235 properties = settings.ConfReader(config=config).get_prop_dic() 

236 

237 # Specific call 

238 process_mdout( 

239 input_log_path=args.input_log_path, 

240 output_dat_path=args.output_dat_path, 

241 properties=properties, 

242 ) 

243 

244 

245if __name__ == "__main__": 

246 main()