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

1#!/usr/bin/env python3 

2 

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

4import argparse 

5import shutil 

6from pathlib import Path, PurePath 

7from biobb_common.generic.biobb_object import BiobbObject 

8from biobb_common.configuration import settings 

9from biobb_common.tools import file_utils as fu 

10from biobb_common.tools.file_utils import launchlogger 

11from biobb_amber.process.common import check_input_path, check_output_path 

12 

13 

14class ProcessMDOut(BiobbObject): 

15 """ 

16 | biobb_amber.process.process_mdout ProcessMDOut 

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

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

19 

20 Args: 

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

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

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

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

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

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

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

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

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

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

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

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

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

34 

35 Examples: 

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

37 

38 from biobb_amber.process.process_mdout import process_mdout 

39 prop = { 

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

41 } 

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

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

44 properties=prop) 

45 

46 Info: 

47 * wrapped_software: 

48 * name: AmberTools process_mdout 

49 * version: >20.9 

50 * license: LGPL 2.1 

51 * ontology: 

52 * name: EDAM 

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

54 

55 """ 

56 

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

58 

59 properties = properties or {} 

60 

61 # Call parent class constructor 

62 super().__init__(properties) 

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

64 

65 # Input/Output files 

66 self.io_dict = { 

67 'in': {'input_log_path': input_log_path}, 

68 'out': {'output_dat_path': output_dat_path} 

69 } 

70 

71 # Properties specific for BB 

72 self.properties = properties 

73 self.terms = properties.get('terms', ["ETOT"]) 

74 self.binary_path = properties.get('binary_path', 'process_mdout.perl') 

75 

76 # Check the properties 

77 self.check_properties(properties) 

78 self.check_arguments() 

79 

80 def check_data_params(self, out_log, err_log): 

81 """ Checks input/output paths correctness """ 

82 

83 # Check input(s) 

84 self.io_dict["in"]["input_log_path"] = check_input_path(self.io_dict["in"]["input_log_path"], "input_log_path", False, out_log, self.__class__.__name__) 

85 

86 # Check output(s) 

87 self.io_dict["out"]["output_dat_path"] = check_output_path(self.io_dict["out"]["output_dat_path"], "output_dat_path", False, out_log, self.__class__.__name__) 

88 

89 @launchlogger 

90 def launch(self): 

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

92 

93 # check input/output paths and parameters 

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

95 

96 # Setup Biobb 

97 if self.check_restart(): 

98 return 0 

99 self.stage_files() 

100 

101 if not self.container_path: 

102 self.tmp_folder = fu.create_unique_dir() 

103 fu.log('Creating %s temporary folder' % self.tmp_folder, self.out_log) 

104 self.cmd = ['cd', self.tmp_folder, ';', 

105 self.binary_path, 

106 str(Path(self.stage_io_dict['in']['input_log_path']).resolve()) 

107 ] 

108 else: 

109 self.tmp_folder = None 

110 self.cmd = [self.binary_path, 

111 self.stage_io_dict['in']['input_log_path'] 

112 ] 

113 

114 # Run Biobb block 

115 self.run_biobb() 

116 

117 # Copy files to host 

118 self.copy_to_host() 

119 

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

121 if self.container_path: 

122 shutil.copy(PurePath(self.stage_io_dict['unique_dir']).joinpath('summary.'+self.terms[0]), self.io_dict['out']['output_dat_path']) 

123 else: 

124 shutil.copy(PurePath(self.tmp_folder).joinpath('summary.'+self.terms[0]), self.io_dict['out']['output_dat_path']) 

125 else: 

126 

127 if self.container_path: 

128 tmp = self.stage_io_dict['unique_dir'] 

129 else: 

130 tmp = self.tmp_folder 

131 

132 ene_dict = {} 

133 for term in self.terms: 

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

135 for line in fp: 

136 x = line.split() 

137 if x: 

138 if (len(x) > 1): 

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

140 else: 

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

142 

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

144 fp_out.write("# TIME ") 

145 for term in self.terms: 

146 fp_out.write(term + " ") 

147 fp_out.write("\n") 

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

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

150 for term in self.terms: 

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

152 fp_out.write("\n") 

153 

154 # remove temporary folder(s) 

155 self.tmp_files.extend([ 

156 self.stage_io_dict.get("unique_dir"), 

157 list(Path().glob('summary*')), 

158 self.tmp_folder 

159 ]) 

160 self.remove_tmp_files() 

161 

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

163 

164 return self.return_code 

165 

166 

167def process_mdout(input_log_path: str, output_dat_path: str, 

168 properties: dict = None, **kwargs) -> int: 

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

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

171 

172 return ProcessMDOut(input_log_path=input_log_path, 

173 output_dat_path=output_dat_path, 

174 properties=properties).launch() 

175 

176 

177def main(): 

178 parser = argparse.ArgumentParser(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.', formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999)) 

179 parser.add_argument('--config', required=False, help='Configuration file') 

180 

181 # Specific args 

182 required_args = parser.add_argument_group('required arguments') 

183 required_args.add_argument('--input_log_path', required=True, help='AMBER (sander) MD output (log) file. Accepted formats: log, out, txt, o.') 

184 required_args.add_argument('--output_dat_path', required=True, help='Dat output file containing data from the specified terms along the MD process. File type: output. Accepted formats: dat, txt, csv.') 

185 

186 args = parser.parse_args() 

187 config = args.config if args.config else None 

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

189 

190 # Specific call 

191 process_mdout(input_log_path=args.input_log_path, 

192 output_dat_path=args.output_dat_path, 

193 properties=properties) 

194 

195 

196if __name__ == '__main__': 

197 main()