Coverage for biobb_cp2k/cp2k/cp2k_run.py: 73%

1#!/usr/bin/env python3 

2 

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

4import argparse 

5from typing import Optional 

6import shutil 

7import os 

8from pathlib import Path, PurePath 

9from biobb_common.generic.biobb_object import BiobbObject 

10from biobb_common.configuration import settings 

11from biobb_common.tools import file_utils as fu 

12from biobb_common.tools.file_utils import launchlogger 

13from biobb_cp2k.cp2k.common import check_input_path, check_output_path 

14 

15 

16class Cp2kRun(BiobbObject): 

17 """ 

18 | biobb_cp2k Cp2kRun 

19 | Wrapper of the `CP2K QM tool <https://www.cp2k.org/>`_ module. 

20 | Runs atomistic simulations of solid state, liquid, molecular, periodic, material, crystal, and biological systems using CP2K QM tool. 

21 

22 Args: 

23 input_inp_path (str): Input configuration file (CP2K run options). File type: input. `Sample file <https://github.com/bioexcel/biobb_cp2k/raw/master/biobb_cp2k/test/data/cp2k/cp2k_energy.inp>`_. Accepted formats: inp (edam:format_2330), in (edam:format_2330), txt (edam:format_2330), wfn (edam:format_2333). 

24 output_log_path (str): Output log file. File type: output. `Sample file <https://github.com/bioexcel/biobb_cp2k/raw/master/biobb_cp2k/test/reference/cp2k/cp2k_run_out.log>`_. Accepted formats: log (edam:format_2330), out (edam:format_2330), txt (edam:format_2330), o (edam:format_2330). 

25 output_outzip_path (str): Output files. File type: output. `Sample file <https://github.com/bioexcel/biobb_cp2k/raw/master/biobb_cp2k/test/reference/cp2k/cp2k_run_out.zip>`_. Accepted formats: zip (edam:format_3987), gzip (edam:format_3987), gz (edam:format_3987). 

26 output_rst_path (str): Output restart file. File type: output. `Sample file <https://github.com/bioexcel/biobb_cp2k/raw/master/biobb_cp2k/test/reference/cp2k/cp2k_run_out.wfn>`_. Accepted formats: wfn (edam:format_2333). 

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

28 * **binary_path** (*str*) - ("cp2k.sopt") CP2K binary path to be used. 

29 * **param_path** (*str*) - (None) Path to the CP2K parameter data files (BASIS_SET, POTENTIALS, etc.). If not provided, the parameter data files included in the package will be used. 

30 * **mpi_bin** (*str*) - (None) Path to the MPI runner. Usually "mpirun" or "srun". 

31 * **mpi_np** (*int*) - (0) [0~1000|1] Number of MPI processes. Usually an integer bigger than 1. 

32 * **mpi_flags** (*str*) - (None) Path to the MPI hostlist file. 

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

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

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

36 

37 Examples: 

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

39 

40 from biobb_cp2k.cp2k.cp2k_run import cp2k_run 

41 prop = { 

42 'binary_path': 'cp2k.sopt' 

43 } 

44 cp2k_run(input_inp_path='/path/to/cp2k_input.inp', 

45 output_log_path='/path/to/cp2k_log.log', 

46 output_outzip_path='/path/to/cp2k_out.zip', 

47 output_rst_path='/path/to/cp2k_rst.wfn', 

48 properties=prop) 

49 

50 Info: 

51 * wrapped_software: 

52 * name: CP2K 

53 * version: >=7.1.0 

54 * license: GPL-2.0 

55 * multinode: mpi 

56 * ontology: 

57 * name: EDAM 

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

59 

60 """ 

61 

62 def __init__(self, input_inp_path: str, output_log_path: str, 

63 output_outzip_path: str, output_rst_path: str, 

64 properties: Optional[dict] = None, **kwargs) -> None: 

65 

66 properties = properties or {} 

67 

68 # Call parent class constructor 

69 super().__init__(properties) 

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

71 

72 # Input/Output files 

73 self.io_dict = { 

74 'in': {'input_inp_path': input_inp_path}, 

75 'out': {'output_log_path': output_log_path, 

76 'output_outzip_path': output_outzip_path, 

77 'output_rst_path': output_rst_path} 

78 } 

79 

80 # Properties specific for BB 

81 self.properties = properties 

82 self.binary_path = properties.get('binary_path', 'cp2k.sopt') 

83 self.param_path = properties.get('param_path', None) 

84 

85 # Properties for MPI 

86 self.mpi_bin = properties.get('mpi_bin') 

87 self.mpi_np = properties.get('mpi_np') 

88 self.mpi_flags = properties.get('mpi_flags') 

89 

90 # Check the properties 

91 self.check_properties(properties) 

92 self.check_arguments() 

93 

94 def check_data_params(self, out_log, out_err): 

95 """ Checks input/output paths correctness """ 

96 

97 # Check input(s) 

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

99 

100 # Check output(s) 

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

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

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

104 

105 @launchlogger 

106 def launch(self): 

107 """Launches the execution of the Cp2kRun module.""" 

108 

109 # check input/output paths and parameters 

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

111 

112 # Setup Biobb 

113 if self.check_restart(): 

114 return 0 

115 self.stage_files() 

116 

117 # Creating temporary folder 

118 self.tmp_folder = fu.create_unique_dir() 

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

120 

121 shutil.copy2(self.io_dict["in"]["input_inp_path"], self.tmp_folder) 

122 

123 # set path to the CP2K parameter data files 

124 if not self.param_path: 

125 # os.environ["CP2K_DATA_DIR"] = str(PurePath(myself.__file__).parent.joinpath('cp2k_data')) 

126 os.environ["CP2K_DATA_DIR"] = str(Path(os.getenv("CONDA_PREFIX", "")).joinpath('cp2k_aux').joinpath('cp2k_data')) 

127 else: 

128 if not Path(PurePath(self.param_path)).exists(): 

129 fu.log(self.__class__.__name__ + ': Unexisting %s folder, exiting' % self.param_path, self.out_log) 

130 raise SystemExit(self.__class__.__name__ + ': Unexisting %s folder' % self.param_path) 

131 os.environ["CP2K_DATA_DIR"] = self.param_path 

132 

133 # Command line 

134 # cp2k.sopt -i benzene_dimer.inp -o mp2_test.out 

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

136 self.binary_path, 

137 '-i', PurePath(self.io_dict["in"]["input_inp_path"]).name, 

138 '-o', PurePath(self.io_dict["out"]["output_log_path"]).name] 

139 

140 # general mpi properties 

141 if self.mpi_bin: 

142 mpi_cmd = [self.mpi_bin] 

143 if self.mpi_np: 

144 mpi_cmd.append('-n') 

145 mpi_cmd.append(str(self.mpi_np)) 

146 if self.mpi_flags: 

147 mpi_cmd.extend(self.mpi_flags) 

148 self.cmd = mpi_cmd + self.cmd 

149 

150 # Run Biobb block 

151 self.run_biobb() 

152 

153 # Gather output files in a single zip file 

154 self.output = PurePath(self.tmp_folder).joinpath("cp2k_out.zip") 

155 out_files = [] 

156 restart = '' 

157 for root, dirs, files in os.walk(self.tmp_folder): 

158 for file in files: 

159 # fu.log('FILE %s ' % file, self.out_log) 

160 # if file.endswith('.out'): 

161 # out_files.append(file) 

162 # elif file.endswith('.wfn'): 

163 # restart = file 

164 

165 if file.endswith('.wfn'): 

166 restart = self.tmp_folder + '/' + file 

167 else: 

168 out_files.append(self.tmp_folder + '/' + file) 

169 

170 fu.zip_list(str(self.output), out_files, self.out_log) 

171 

172 # Copy outputs from temporary folder to output path 

173 shutil.copy2(self.output, PurePath(self.io_dict["out"]["output_outzip_path"])) 

174 shutil.copy2(PurePath(self.tmp_folder).joinpath(PurePath(self.io_dict["out"]["output_log_path"]).name), PurePath(self.io_dict["out"]["output_log_path"])) 

175 if restart: 

176 shutil.copy2(restart, PurePath(self.io_dict["out"]["output_rst_path"])) 

177 

178 # Copy files to host 

179 self.copy_to_host() 

180 

181 # remove temporary folder(s) 

182 self.tmp_files.extend([ 

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

184 self.tmp_folder 

185 ]) 

186 self.remove_tmp_files() 

187 

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

189 

190 return self.return_code 

191 

192 

193def cp2k_run(input_inp_path: str, 

194 output_log_path: str, output_outzip_path: str, output_rst_path: str, 

195 properties: Optional[dict] = None, **kwargs) -> int: 

196 """Create :class:`Cp2kRun <cp2k.cp2k_run.Cp2kRun>`cp2k.cp2k_run.Cp2kRun class and 

197 execute :meth:`launch() <cp2k.cp2k_run.Cp2kRun.launch>` method""" 

198 

199 return Cp2kRun(input_inp_path=input_inp_path, 

200 output_log_path=output_log_path, 

201 output_outzip_path=output_outzip_path, 

202 output_rst_path=output_rst_path, 

203 properties=properties).launch() 

204 

205 cp2k_run.__doc__ = Cp2kRun.__doc__ 

206 

207 

208def main(): 

209 parser = argparse.ArgumentParser(description='Running atomistic simulations of solid state, liquid, molecular, periodic, material, crystal, and biological systems using CP2K QM tool.', formatter_class=lambda prog: argparse.RawTextHelpFormatter(prog, width=99999)) 

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

211 

212 # Specific args 

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

214 required_args.add_argument('--input_inp_path', required=True, help='Input configuration file (QM run options). Accepted formats: inp, in, txt.') 

215 required_args.add_argument('--output_log_path', required=True, help='Output log file. Accepted formats: log, out, txt.') 

216 required_args.add_argument('--output_outzip_path', required=True, help='Output trajectory file. Accepted formats: zip, gz, gzip.') 

217 required_args.add_argument('--output_rst_path', required=True, help='Output restart file. Accepted formats: wfn.') 

218 

219 args = parser.parse_args() 

220 # config = args.config if args.config else None 

221 args.config = args.config or "{}" 

222 # properties = settings.ConfReader(config=config).get_prop_dic() 

223 properties = settings.ConfReader(config=args.config).get_prop_dic() 

224 

225 # Specific call 

226 cp2k_run(input_inp_path=args.input_inp_path, 

227 output_log_path=args.output_log_path, 

228 output_outzip_path=args.output_outzip_path, 

229 output_rst_path=args.output_rst_path, 

230 properties=properties) 

231 

232 

233if __name__ == '__main__': 

234 main()