Coverage for biobb_cmip / cmip / cmip_run.py: 51%

1#!/usr/bin/env python3 

2 

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

4import os 

5import json 

6from typing import Optional 

7from typing import Any 

8import shutil 

9from pathlib import Path 

10from biobb_common.generic.biobb_object import BiobbObject 

11from biobb_common.tools import file_utils as fu 

12from biobb_common.tools.file_utils import launchlogger 

13from biobb_cmip.cmip.common import create_params_file 

14from biobb_cmip.cmip.common import params_preset 

15from biobb_cmip.cmip.common import get_grid 

16 

17 

18class CmipRun(BiobbObject): 

19 """ 

20 | biobb_cmip Titration 

21 | Wrapper class for the CMIP cmip module. 

22 | The CMIP cmip module. CMIP cmip module compute classical molecular interaction potentials. 

23 

24 Args: 

25 input_pdb_path (str): Path to the input PDB file. File type: input. `Sample file <https://raw.githubusercontent.com/bioexcel/biobb_cmip/master/biobb_cmip/test/data/cmip/1kim_h.pdb>`_. Accepted formats: pdb (edam:format_1476). 

26 input_probe_pdb_path (str): Path to the input probe file in PDB format. File type: input. `Sample file <https://raw.githubusercontent.com/bioexcel/biobb_cmip/master/biobb_cmip/test/data/cmip/RBD-hACE2.RBD.cmip.pdb>`_. Accepted formats: pdb (edam:format_1476). 

27 output_pdb_path (str) (Optional): Path to the output PDB file. File type: output. `Sample file <https://raw.githubusercontent.com/bioexcel/biobb_cmip/master/biobb_cmip/test/reference/cmip/1kim_neutral.pdb>`_. Accepted formats: pdb (edam:format_1476). 

28 output_grd_path (str) (Optional): Path to the output grid file in GRD format. File type: output. Accepted formats: grd (edam:format_2330). 

29 output_cube_path (str) (Optional): Path to the output grid file in cube format. File type: output. Accepted formats: cube (edam:format_2330). 

30 output_rst_path (str) (Optional): Path to the output restart file. File type: output. Accepted formats: txt (edam:format_2330). 

31 input_rst_path (str) (Optional): Path to the input restart file. File type: input. Accepted formats: txt (edam:format_2330). 

32 output_byat_path (str) (Optional): Path to the output atom by atom energy file. File type: output. Accepted formats: txt (edam:format_2330), out (edam:format_2330). 

33 output_log_path (str) (Optional): Path to the output CMIP log file LOG. File type: output. Accepted formats: log (edam:format_2330). 

34 input_vdw_params_path (str) (Optional): Path to the CMIP input Van der Waals force parameters, if not provided the CMIP conda installation one is used ("$CONDA_PREFIX/share/cmip/dat/vdwprm"). File type: input. Accepted formats: txt (edam:format_2330). 

35 input_params_path (str) (Optional): Path to the CMIP input parameters file. File type: input. Accepted formats: txt (edam:format_2330). 

36 output_json_box_path (str) (Optional): Path to the output CMIP box in JSON format. File type: output. Accepted formats: json (edam:format_3464). 

37 output_json_external_box_path (str) (Optional): Path to the output external CMIP box in JSON format. File type: output. Accepted formats: json (edam:format_3464). 

38 input_json_box_path (str) (Optional): Path to the input CMIP box in JSON format. File type: input. Accepted formats: json (edam:format_3464). 

39 input_json_external_box_path (str) (Optional): Path to the input CMIP box in JSON format. File type: input. Accepted formats: json (edam:format_3464). 

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

41 * **execution_type** (*str*) - ("mip_pos") Default options for the params file, each one creates a different params file. Values: check_only (Dry Run of CMIP), mip_pos (MIP O+ Mehler Solmajer dielectric), mip_neg (MIP O- Mehler Solmajer dielectric), mip_neu (MIP Oxygen Mehler Solmajer dielectric), solvation (Solvation & MEP), pb_interaction_energy (Docking Interaction energy calculation. PB electrostatics), docking (Docking Mehler Solmajer dielectric), docking_rst (Docking from restart file). 

42 * **params** (*dict*) - ({}) CMIP options specification. 

43 * **binary_path** (*str*) - ("cmip") Path to the CMIP cmip executable binary. 

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

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

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

47 * **container_path** (*str*) - (None) Path to the binary executable of your container. 

48 * **container_image** (*str*) - ("cmip/cmip:latest") Container Image identifier. 

49 * **container_volume_path** (*str*) - ("/data") Path to an internal directory in the container. 

50 * **container_working_dir** (*str*) - (None) Path to the internal CWD in the container. 

51 * **container_user_id** (*str*) - (None) User number id to be mapped inside the container. 

52 * **container_shell_path** (*str*) - ("/bin/bash") Path to the binary executable of the container shell. 

53 

54 

55 Examples: 

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

57 

58 from biobb_cmip.cmip.cmip import cmip 

59 prop = { 'binary_path': 'cmip' } 

60 cmip(input_pdb_path='/path/to/myStructure.pdb', 

61 output_pdb_path='/path/to/newStructure.pdb', 

62 output_log_path='/path/to/newStructureLog.log', 

63 properties=prop) 

64 

65 Info: 

66 * wrapped_software: 

67 * name: CMIP cmip 

68 * version: 2.7.0 

69 * license: Apache-2.0 

70 * ontology: 

71 * name: EDAM 

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

73 """ 

74 

75 def __init__(self, input_pdb_path: str, input_probe_pdb_path: Optional[str] = None, output_pdb_path: Optional[str] = None, 

76 output_grd_path: Optional[str] = None, output_cube_path: Optional[str] = None, output_rst_path: Optional[str] = None, 

77 input_rst_path: Optional[str] = None, output_byat_path: Optional[str] = None, output_log_path: Optional[str] = None, 

78 input_vdw_params_path: Optional[str] = None, input_params_path: Optional[str] = None, output_json_box_path: Optional[str] = None, 

79 output_json_external_box_path: Optional[str] = None, input_json_box_path: Optional[str] = None, 

80 input_json_external_box_path: Optional[str] = None, properties: Optional[dict] = None, **kwargs) -> None: 

81 

82 properties = properties or {} 

83 

84 # Call parent class constructor 

85 super().__init__(properties) 

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

87 

88 # Input/Output files 

89 self.io_dict = { 

90 "in": {"input_pdb_path": input_pdb_path, "input_probe_pdb_path": input_probe_pdb_path, 

91 "input_vdw_params_path": input_vdw_params_path, "input_params_path": input_params_path, 

92 "input_json_box_path": input_json_box_path, 

93 "input_json_external_box_path": input_json_external_box_path, 

94 "input_rst_path": input_rst_path}, 

95 "out": {"output_pdb_path": output_pdb_path, "output_grd_path": output_grd_path, 

96 "output_cube_path": output_cube_path, "output_rst_path": output_rst_path, 

97 "output_byat_path": output_byat_path, "output_log_path": output_log_path, 

98 "output_json_box_path": output_json_box_path, 

99 "output_json_external_box_path": output_json_external_box_path} 

100 } 

101 

102 # Properties specific for BB 

103 self.binary_path = properties.get('binary_path', 'cmip') 

104 self.execution_type = properties.get('execution_type', 'mip_pos') 

105 self.params = {k: str(v) for k, v in properties.get('params', dict()).items()} 

106 

107 if not self.io_dict['in'].get('input_vdw_params_path'): 

108 self.io_dict['in']['input_vdw_params_path'] = f"{os.environ.get('CONDA_PREFIX')}/share/cmip/dat/vdwprm" 

109 self.io_dict['in']['combined_params_path'] = properties.get('combined_params_path', 'params') 

110 

111 # Check the properties 

112 self.check_properties(properties) 

113 self.check_arguments() 

114 

115 @launchlogger 

116 def launch(self) -> int: 

117 """Execute the :class:`Cmip <cmip.cmip.Cmip>` object.""" 

118 

119 # Setup Biobb 

120 if self.check_restart(): 

121 return 0 

122 

123 # Check if output_pdb_path ends with ".pdb" and does not contain underscores 

124 if self.io_dict['out']['output_pdb_path']: 

125 if (not self.io_dict['out']['output_pdb_path'].endswith('.pdb')) or \ 

126 ("_" in str(Path(self.io_dict['out']['output_pdb_path']).name)): 

127 fu.log(f"ERROR: output_pdb_path ({self.io_dict['out']['output_pdb_path']}) " 

128 f"name must end in .pdb and not contain underscores", self.out_log, self.global_log) 

129 raise ValueError(f"ERROR: output_pdb_path ({self.io_dict['out']['output_pdb_path']})" 

130 f"name must end in .pdb and not contain underscores") 

131 

132 params_preset_dict: dict[str, Any] = params_preset(execution_type=self.execution_type) 

133 if self.io_dict['in']["input_json_external_box_path"]: 

134 params_preset_dict["readgrid0"] = 0 

135 origin, size, grid_params = get_grid(self.io_dict['in']["input_json_external_box_path"]) 

136 params_preset_dict['grid_int0'] = \ 

137 f"INTX0={grid_params['INT'][0]},INTY0={grid_params['INT'][1]},INTZ0={grid_params['INT'][2]}" 

138 params_preset_dict['grid_cen0'] = \ 

139 f"CENX0={grid_params['CEN'][0]},CENY0={grid_params['CEN'][1]},CENZ0={grid_params['CEN'][2]}" 

140 params_preset_dict['grid_dim0'] = \ 

141 f"DIMX0={grid_params['DIM'][0]},DIMY0={grid_params['DIM'][1]},DIMZ0={grid_params['DIM'][2]}" 

142 

143 if self.io_dict['in']["input_json_box_path"]: 

144 params_preset_dict["readgrid"] = 0 

145 origin, size, grid_params = get_grid(self.io_dict['in']["input_json_box_path"]) 

146 params_preset_dict['grid_int'] = \ 

147 f"INTX={grid_params['INT'][0]},INTY={grid_params['INT'][1]},INTZ={grid_params['INT'][2]}" 

148 params_preset_dict['grid_cen'] = \ 

149 f"CENX={grid_params['CEN'][0]},CENY={grid_params['CEN'][1]},CENZ={grid_params['CEN'][2]}" 

150 params_preset_dict['grid_dim'] = \ 

151 f"DIMX={grid_params['DIM'][0]},DIMY={grid_params['DIM'][1]},DIMZ={grid_params['DIM'][2]}" 

152 

153 if self.io_dict['out']['output_json_box_path'] or self.io_dict['out']['output_json_external_box_path']: 

154 params_preset_dict['WRITELOG'] = 1 

155 key_value_log_dir = fu.create_unique_dir() 

156 self.io_dict['out']['key_value_log_path'] = str(Path(key_value_log_dir).joinpath("key_value_cmip_log.log")) 

157 self.tmp_files.append(key_value_log_dir) 

158 

159 # Restart OUT 

160 if self.io_dict["out"].get("output_rst_path"): 

161 params_preset_dict['FULLRST'] = 1 # type: ignore 

162 params_preset_dict['OREST'] = 1 

163 

164 # Restart IN 

165 if self.io_dict['in']["input_rst_path"]: 

166 params_preset_dict['IREST'] = 2 

167 if not self.io_dict["out"].get("output_rst_path"): 

168 self.io_dict["out"]["output_rst_path"] = fu.create_unique_file_path() 

169 shutil.copy2(self.io_dict['in']["input_rst_path"], self.io_dict["out"]["output_rst_path"]) 

170 

171 else: 

172 params_preset_dict['IREST'] = 0 

173 

174 combined_params_dir = fu.create_unique_dir() 

175 self.io_dict['in']['combined_params_path'] = create_params_file( 

176 output_params_path=str(Path(combined_params_dir).joinpath(self.io_dict['in']['combined_params_path'])), 

177 input_params_path=self.io_dict['in'].get('input_params_path'), 

178 params_preset_dict=params_preset_dict, 

179 params_properties_dict=self.params) 

180 

181 self.stage_files() 

182 

183 self.cmd = [self.binary_path, 

184 '-i', self.stage_io_dict['in']['combined_params_path'], 

185 '-vdw', self.stage_io_dict['in']['input_vdw_params_path'], 

186 '-hs', self.stage_io_dict['in']['input_pdb_path']] 

187 

188 if self.stage_io_dict["in"].get("input_probe_pdb_path") and Path( 

189 self.io_dict["in"].get("input_probe_pdb_path", "")).exists(): 

190 self.cmd.append('-pr') 

191 self.cmd.append(self.stage_io_dict["in"].get("input_probe_pdb_path")) 

192 

193 if self.stage_io_dict["out"].get("output_pdb_path"): 

194 self.cmd.append('-outpdb') 

195 self.cmd.append(self.stage_io_dict['out']['output_pdb_path']) 

196 

197 if self.stage_io_dict["out"].get("output_grd_path"): 

198 self.cmd.append('-grdout') 

199 self.cmd.append(self.stage_io_dict["out"]["output_grd_path"]) 

200 

201 if self.stage_io_dict["out"].get("output_cube_path"): 

202 self.cmd.append('-cube') 

203 self.cmd.append(self.stage_io_dict["out"]["output_cube_path"]) 

204 

205 if self.stage_io_dict["out"].get("output_rst_path"): 

206 self.cmd.append('-rst') 

207 self.cmd.append(self.stage_io_dict["out"]["output_rst_path"]) 

208 

209 if self.stage_io_dict["out"].get("output_byat_path"): 

210 self.cmd.append('-byat') 

211 self.cmd.append(self.stage_io_dict["out"]["output_byat_path"]) 

212 

213 if self.stage_io_dict["out"].get("output_log_path"): 

214 self.cmd.append('-o') 

215 self.cmd.append(self.stage_io_dict["out"]["output_log_path"]) 

216 

217 if self.stage_io_dict['out'].get('output_json_box_path') or self.stage_io_dict['out'].get('output_json_external_box_path'): 

218 self.cmd.append('-l') 

219 self.cmd.append(self.stage_io_dict["out"]["key_value_log_path"]) 

220 

221 # Run Biobb block 

222 self.run_biobb() 

223 

224 # CMIP removes or adds a .pdb extension from pdb output name 

225 # manual copy_to_host or unstage 

226 if self.io_dict['out'].get('output_pdb_path'): 

227 output_pdb_path = str(Path(self.stage_io_dict["unique_dir"]).joinpath(Path(self.io_dict['out'].get('output_pdb_path', '')).name)) 

228 if Path(output_pdb_path[:-4]).exists(): 

229 shutil.move(output_pdb_path[:-4], self.io_dict['out'].get('output_pdb_path', '')) 

230 elif Path(output_pdb_path + ".pdb").exists(): 

231 shutil.move(output_pdb_path + ".pdb", self.io_dict['out'].get('output_pdb_path', '')) 

232 elif not Path(output_pdb_path).exists(): 

233 fu.log(f"WARNING: File not found output_pdb_path: {output_pdb_path}", self.out_log, self.global_log) 

234 

235 # Replace "ATOMTM" tag for "ATOM " 

236 

237 output_pdb_path = self.io_dict['out'].get('output_pdb_path', '') 

238 if output_pdb_path: 

239 if Path(output_pdb_path).exists(): 

240 with open(output_pdb_path) as pdb_file: 

241 list_pdb_lines = pdb_file.readlines() 

242 with open(output_pdb_path, 'w') as pdb_file: 

243 for line in list_pdb_lines: 

244 pdb_file.write(line.replace('ATOMTM', 'ATOM ')) 

245 else: 

246 fu.log(f"WARNING: File not found output_pdb_path: {output_pdb_path} Abs Path: {Path(output_pdb_path).resolve()}", self.out_log, self.global_log) 

247 

248 # Create json_box_path file from CMIP log file 

249 if self.io_dict['out'].get('output_json_box_path'): 

250 origin, size, grid_params = get_grid(self.stage_io_dict["out"]["output_log_path"]) 

251 grid_params['DIM'] = (int(grid_params['DIM'][0]), 

252 int(grid_params['DIM'][1]), 

253 int(grid_params['DIM'][2])) 

254 size_dict = {'x': round(grid_params['DIM'][0] * grid_params['INT'][0], 3), 

255 'y': round(grid_params['DIM'][1] * grid_params['INT'][1], 3), 

256 'z': round(grid_params['DIM'][2] * grid_params['INT'][2], 3)} 

257 origin_dict = {'x': round(grid_params['CEN'][0] - size_dict['x'] / 2, 3), 

258 'y': round(grid_params['CEN'][1] - size_dict['y'] / 2, 3), 

259 'z': round(grid_params['CEN'][0] - size_dict['z'] / 2, 3)} 

260 grid_dict = {'origin': origin_dict, 

261 'size': size_dict, 

262 'params': grid_params} 

263 with open(self.io_dict['out'].get('output_json_box_path', ''), 'w') as json_file: 

264 json_file.write(json.dumps(grid_dict, indent=4)) 

265 

266 # Create external_json_box_path file from CMIP log file 

267 if self.io_dict['out'].get('output_json_external_box_path'): 

268 origin, size, grid_params = get_grid(self.stage_io_dict["out"]["output_log_path"], True) 

269 grid_params['DIM'] = (int(grid_params['DIM'][0]), 

270 int(grid_params['DIM'][1]), 

271 int(grid_params['DIM'][2])) 

272 size_dict = {'x': round(grid_params['DIM'][0] * grid_params['INT'][0], 3), 

273 'y': round(grid_params['DIM'][1] * grid_params['INT'][1], 3), 

274 'z': round(grid_params['DIM'][2] * grid_params['INT'][2], 3)} 

275 origin_dict = {'x': round(grid_params['CEN'][0] - size_dict['x'] / 2, 3), 

276 'y': round(grid_params['CEN'][1] - size_dict['y'] / 2, 3), 

277 'z': round(grid_params['CEN'][0] - size_dict['z'] / 2, 3)} 

278 grid_dict = {'origin': origin_dict, 

279 'size': size_dict, 

280 'params': grid_params} 

281 with open(self.io_dict['out'].get('output_json_external_box_path', ''), 'w') as json_file: 

282 json_file.write(json.dumps(grid_dict, indent=4)) 

283 

284 # Copy files to host 

285 self.copy_to_host() 

286 

287 # remove temporary folder(s) 

288 self.tmp_files.append(combined_params_dir) 

289 self.remove_tmp_files() 

290 

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

292 

293 return self.return_code 

294 

295 

296def cmip_run(input_pdb_path: str, input_probe_pdb_path: Optional[str] = None, output_pdb_path: Optional[str] = None, 

297 output_grd_path: Optional[str] = None, output_cube_path: Optional[str] = None, output_rst_path: Optional[str] = None, 

298 output_byat_path: Optional[str] = None, output_log_path: Optional[str] = None, 

299 input_vdw_params_path: Optional[str] = None, input_params_path: Optional[str] = None, output_json_box_path: Optional[str] = None, 

300 output_json_external_box_path: Optional[str] = None, input_json_box_path: Optional[str] = None, 

301 input_json_external_box_path: Optional[str] = None, properties: Optional[dict] = None, **kwargs) -> int: 

302 """Create :class:`Cmip <cmip.cmip.Cmip>` class and 

303 execute the :meth:`launch() <cmip.cmip.Cmip.launch>` method.""" 

304 return CmipRun(**dict(locals())).launch() 

305 

306 

307cmip_run.__doc__ = CmipRun.__doc__ 

308main = CmipRun.get_main(cmip_run, "Wrapper of the CMIP cmip module.") 

309 

310if __name__ == '__main__': 

311 main()