Coverage for mddb_workflow/mwf.py: 79%

1#!/usr/bin/env python 

2 

3# This is the starter script 

4 

5# Import python libraries 

6from os import chdir, rename, remove, walk, mkdir, getcwd 

7from os.path import exists, isdir, isabs, relpath, normpath, split, basename 

8from shutil import rmtree 

9import sys 

10import io 

11import re 

12import numpy 

13from glob import glob 

14from inspect import getfullargspec 

15import time 

16 

17# Constants 

18# Importing constants first is important 

19from mddb_workflow.utils.constants import * 

20 

21# Import local utils 

22#from mddb_workflow.utils.httpsf import mount 

23from mddb_workflow.utils.auxiliar import InputError, MISSING_TOPOLOGY 

24from mddb_workflow.utils.auxiliar import warn, load_json, save_json, load_yaml, save_yaml 

25from mddb_workflow.utils.auxiliar import is_directory_empty, is_glob, parse_glob, safe_getattr 

26from mddb_workflow.utils.auxiliar import read_ndict, write_ndict, get_git_version 

27from mddb_workflow.utils.arg_cksum import get_cksum_id 

28from mddb_workflow.utils.register import Register 

29from mddb_workflow.utils.cache import Cache 

30from mddb_workflow.utils.structures import Structure 

31from mddb_workflow.utils.topologies import Topology 

32from mddb_workflow.utils.file import File 

33from mddb_workflow.utils.remote import Remote 

34from mddb_workflow.utils.pyt_spells import get_frames_count, get_average_structure 

35from mddb_workflow.utils.selections import Selection 

36from mddb_workflow.utils.mda_spells import get_mda_universe 

37from mddb_workflow.utils.type_hints import * 

38 

39# Import local tools 

40from mddb_workflow.tools.get_first_frame import get_first_frame 

41from mddb_workflow.tools.get_bonds import find_safe_bonds, get_bonds_reference_frame 

42from mddb_workflow.tools.process_interactions import process_interactions 

43from mddb_workflow.tools.generate_metadata import prepare_project_metadata, generate_md_metadata 

44from mddb_workflow.tools.generate_ligands_desc import generate_ligand_mapping 

45from mddb_workflow.tools.chains import prepare_chain_references 

46from mddb_workflow.tools.generate_pdb_references import prepare_pdb_references 

47from mddb_workflow.tools.residue_mapping import generate_residue_mapping 

48from mddb_workflow.tools.generate_map import generate_protein_mapping 

49from mddb_workflow.tools.generate_lipid_references import generate_lipid_references 

50from mddb_workflow.tools.generate_membrane_mapping import generate_membrane_mapping 

51from mddb_workflow.tools.generate_topology import generate_topology 

52from mddb_workflow.tools.get_charges import get_charges 

53from mddb_workflow.tools.get_inchi_keys import get_inchikeys 

54from mddb_workflow.tools.remove_trash import remove_trash 

55from mddb_workflow.tools.get_screenshot import get_screenshot 

56from mddb_workflow.tools.process_input_files import process_input_files 

57from mddb_workflow.tools.provenance import produce_provenance 

58 

59# Import local analyses 

60from mddb_workflow.analyses.rmsds import rmsds 

61from mddb_workflow.analyses.tmscores import tmscores 

62from mddb_workflow.analyses.rmsf import rmsf 

63from mddb_workflow.analyses.rgyr import rgyr 

64from mddb_workflow.analyses.pca import pca 

65from mddb_workflow.analyses.density import density 

66from mddb_workflow.analyses.thickness import thickness 

67from mddb_workflow.analyses.area_per_lipid import area_per_lipid 

68from mddb_workflow.analyses.lipid_order import lipid_order 

69from mddb_workflow.analyses.lipid_interactions import lipid_interactions 

70from mddb_workflow.analyses.channels import channels 

71#from mddb_workflow.analyses.pca_contacts import pca_contacts 

72from mddb_workflow.analyses.rmsd_per_residue import rmsd_per_residue 

73from mddb_workflow.analyses.rmsd_pairwise import rmsd_pairwise 

74from mddb_workflow.analyses.clusters import clusters_analysis 

75from mddb_workflow.analyses.distance_per_residue import distance_per_residue 

76from mddb_workflow.analyses.hydrogen_bonds import hydrogen_bonds 

77from mddb_workflow.analyses.sasa import sasa 

78from mddb_workflow.analyses.energies import energies 

79from mddb_workflow.analyses.dihedral_energies import compute_dihedral_energies 

80from mddb_workflow.analyses.pockets import pockets 

81from mddb_workflow.analyses.rmsd_check import check_trajectory_integrity 

82from mddb_workflow.analyses.helical_parameters import helical_parameters 

83from mddb_workflow.analyses.markov import markov 

84 

85# Make the system output stream to not be buffered 

86# Only when not in a Jupyter notebook or using pytest 

87# Check if we're in an interactive Python shell like Jupyter 

88if not hasattr(sys, 'ps1') and not 'pytest' in sys.modules: 

89 # This is useful to make prints work on time in Slurm 

90 # Otherwise, output logs are written after the script has fully run 

91 # Note that this fix affects all modules and built-ins 

92 unbuffered = io.TextIOWrapper(open(sys.stdout.fileno(), 'wb', 0), write_through=True) 

93 sys.stdout = unbuffered 

94 

95# Set a special exception for missing inputs 

96MISSING_INPUT_EXCEPTION = Exception('Missing input') 

97 

98# Set a special exception for missing task function arguments 

99# This is used for easy debug when a new functions is added wrongly 

100MISSING_ARGUMENT_EXCEPTION = Exception('Missing argument') 

101 

102# Set a special exception for when a value is missing 

103MISSING_VALUE_EXCEPTION = Exception('Missing value') 

104 

105# Name of the argument used by all functions to know where to write output 

106OUTPUT_FILEPATH_ARG = 'output_filepath' 

107OUTPUT_DIRECTORY_ARG = 'output_directory' 

108 

109# Set some variables which are filled at the end but are referred by previously defined functions 

110requestables = {} 

111inverted_requestables = {} 

112 

113 

114class Task: 

115 """ Descriptor class to handle a generic task. 

116 It implements lazy properties, caching and overwriting. 

117 

118 Since its properties are static, results are stored in the parent object 

119 (MD/Project), or otherwise all MDs would share the same task values. """ 

120 

121 def __init__ (self, 

122 flag : str, 

123 name : str, 

124 func : Callable, 

125 args : dict = {}, 

126 output_filename : Optional[str] = None, 

127 use_cache : bool = True, 

128 debug : bool = False, 

129 ): 

130 """ 

131 Initialize the Task object. 

132 

133 Args: 

134 flag (str): 

135 The task flag. 

136 This name is used by the include/exclude/overwrite arguments and to name the analysis output directory. 

137 

138 name (str): 

139 The task user-friendly name is to be used in the logs. 

140 

141 func (Callable): 

142 The task function. 

143 Function argument names must correspond with Project/MD property names. 

144 

145 args (dict, optional): 

146 The task function additional arguments. 

147 Project/MD properties are automatically sent to the function as arguments. 

148 However some analyses have additional arguments (e.g. frames limit, cutoffs, etc.) 

149 

150 output_filename (str, optional): 

151 The task output filename. 

152 Path will be set automatically relative to its project/MD. 

153 For those tasks which generate a directory with multiple outputs this is not necessary. 

154 However this may come in handy by tasks with a single file output. 

155 Specially when this output file is used later in this workflow. 

156 

157 use_cache (bool, optional): 

158 Set if the returned output is to be cached. 

159 Note that argument values are always cached, this is not optional. 

160 

161 debug (bool, optional): 

162 If the task is run in debug mode, producing more output logs. Defaults to False. 

163 """ 

164 # Save input arguments 

165 self.flag = flag 

166 self.name = name 

167 self.func = func 

168 self.args = args 

169 self.output_filename = output_filename 

170 self.use_cache = use_cache 

171 self.debug = debug 

172 # Set the key used to store and retireve data in the parent and cache 

173 self.parent_output_key = f'_{self.flag}_task_output' 

174 self.parent_output_filepath_key = f'{self.flag}_task_output_filepath' 

175 self.cache_output_key = f'{self.flag}_task_output' 

176 self.cache_arg_cksums = f'{self.flag}_task_arg_cksums' 

177 # Para el arg_cksum 

178 self.__name__ = self.flag 

179 

180 # Set internal functions to handle parent saved output 

181 # This output is not saved in the task itself, but in the parent, because the task is static 

182 def _get_parent_output (self, parent): 

183 return safe_getattr(parent, self.parent_output_key, MISSING_VALUE_EXCEPTION) 

184 def _set_parent_output (self, parent, new_output): 

185 return setattr(parent, self.parent_output_key, new_output) 

186 def _get_parent_output_filepath (self, parent): 

187 return safe_getattr(parent, self.parent_output_filepath_key, MISSING_VALUE_EXCEPTION) 

188 def _set_parent_output_filepath (self, parent, new_filepath): 

189 return setattr(parent, self.parent_output_filepath_key, new_filepath) 

190 # Get the task output, running the task if necessary 

191 def get_output (self, parent): 

192 # If we already have a value stored from this run then return it 

193 output = self._get_parent_output(parent) 

194 if output != MISSING_VALUE_EXCEPTION: return output 

195 # Otherwise run the task and return the output 

196 return self(parent) 

197 output = property(get_output, None, None, "Task output (read only)") 

198 # Asking for the output file or filepath implies running the Task, then returning the file/filepath 

199 def get_output_filepath (self, parent) -> str: 

200 # If we already have a filepath stored from this run then return it 

201 filepath = self._get_parent_output_filepath(parent) 

202 if filepath != MISSING_VALUE_EXCEPTION: return filepath 

203 # Otherwise run the task and return the filepath 

204 self(parent) 

205 filepath = self._get_parent_output_filepath(parent) 

206 if filepath != MISSING_VALUE_EXCEPTION: return filepath 

207 raise ValueError(f'Task {self.flag} has no output filepath after running') 

208 output_filepath = property(get_output_filepath, None, None, "Task output filepath (read only)") 

209 def get_output_file (self, parent) -> str: 

210 filepath = self.get_output_filepath(parent) 

211 return File(filepath) 

212 output_file = property(get_output_file, None, None, "Task output file (read only)") 

213 

214 # When the task is printed, show the flag 

215 def __repr__ (self): return f'<Task ({self.flag})>' 

216 

217 # When a task is called 

218 def __call__(self, parent: Union['Project', 'MD']): 

219 # First of all check if this task has been already done in this very run 

220 # If so then return the stored vale 

221 output = self._get_parent_output(parent) 

222 if output != MISSING_VALUE_EXCEPTION: return output 

223 # Process the task function arguments 

224 processed_args = {} 

225 # Get the task function expected arguments 

226 specification = getfullargspec(self.func) 

227 expected_arguments = specification.args 

228 n_default_arguments = len(specification.defaults) if specification.defaults else 0 

229 # Find out which arguments are optional since they have default values 

230 default_arguments = set(expected_arguments[::-1][:n_default_arguments]) 

231 # If one of the expected arguments is the output_filename then set it here 

232 output_filepath = None 

233 writes_output_file = OUTPUT_FILEPATH_ARG in expected_arguments 

234 if writes_output_file: 

235 # The task should have a defined output file 

236 if not self.output_filename: 

237 raise RuntimeError(f'Task {self.flag} must have an "output_filename"') 

238 # Set the output file path 

239 output_filepath = parent.pathify(self.output_filename) 

240 self._set_parent_output_filepath(parent, output_filepath) 

241 # Add it to the processed args 

242 processed_args[OUTPUT_FILEPATH_ARG] = output_filepath 

243 # Remove the expected argument from the list 

244 expected_arguments.remove(OUTPUT_FILEPATH_ARG) 

245 # If one of the expected arguments is the output_directory then set it here 

246 # We will set a new directory with the flag name of the task, in the correspoding path 

247 # Note that while the task is beeing done the output directory has a different name 

248 # Thus the directory is hidden and marked as incomplete 

249 # The final output directory is the one without the incomplete prefix 

250 writes_output_dir = OUTPUT_DIRECTORY_ARG in expected_arguments 

251 incomplete_output_directory = None 

252 final_output_directory = None 

253 if writes_output_dir: 

254 # Set the output directory path 

255 incomplete_output_directory = parent.pathify(INCOMPLETE_PREFIX + self.flag) 

256 final_output_directory = incomplete_output_directory.replace(INCOMPLETE_PREFIX, '') 

257 # Add it to the processed args 

258 processed_args[OUTPUT_DIRECTORY_ARG] = incomplete_output_directory 

259 # Remove the expected argument from the list 

260 expected_arguments.remove(OUTPUT_DIRECTORY_ARG) 

261 # Iterate the reamining expected arguments 

262 for arg in expected_arguments: 

263 # First find the argument among the parent properties 

264 arg_value = self.find_arg_value(arg, parent, default_arguments) 

265 if arg_value == MISSING_ARGUMENT_EXCEPTION: continue 

266 # Add the processed argument 

267 processed_args[arg] = arg_value 

268 # Check again if the task has output already 

269 # It may happend that some dependencies assign output on their own 

270 # e.g. charges, bonds 

271 # If so then return the stored vale 

272 output = self._get_parent_output(parent) 

273 if output != MISSING_VALUE_EXCEPTION: return output 

274 # Find if we have cached output 

275 if self.use_cache: 

276 output = parent.cache.retrieve(self.cache_output_key, MISSING_VALUE_EXCEPTION) 

277 self._set_parent_output(parent, output) 

278 # Check if this dependency is to be overwriten 

279 forced_overwrite = self.flag in parent.overwritables 

280 # Get the list of inputs which have changed compared to a previous run 

281 # WARNING: Always get changed inputs, since this function updates the cache 

282 # If had_cache is false then it means this is the first time the task is ever done 

283 changed_inputs, had_cache, cache_cksums = self.get_changed_inputs(parent, processed_args) 

284 any_input_changed = len(changed_inputs) > 0 

285 # Update the cache inputs 

286 parent.cache.update(self.cache_arg_cksums, cache_cksums) 

287 # We must overwrite outputs either if inputs changed or if it was forced by the user 

288 must_overwrite = forced_overwrite or any_input_changed 

289 # Check if output already exists 

290 # If the final directory already exists then it means the task was started in a previous run 

291 existing_incomplete_output = writes_output_dir and exists(incomplete_output_directory) 

292 # If the final directory already exists then it means the task was done in a previous run 

293 existing_final_output = writes_output_dir and exists(final_output_directory) 

294 # If the output file already exists then it also means the task was done in a previous run 

295 existing_output_file = writes_output_file and exists(output_filepath) 

296 # If we already have a cached output result 

297 existing_output_data = output != MISSING_VALUE_EXCEPTION 

298 # If we must overwrite then purge previous outputs 

299 if must_overwrite: 

300 if existing_incomplete_output: rmtree(incomplete_output_directory) 

301 if existing_final_output: rmtree(final_output_directory) 

302 if existing_output_file: remove(output_filepath) 

303 if existing_output_data: parent.cache.delete(self.cache_output_key) 

304 # If already existing output is not to be overwritten then check if it is already what we need 

305 else: 

306 # If output files/directories are expected then they must exist 

307 # If output data is expected then it must be cached 

308 satisfied_output = (not writes_output_dir or exists(final_output_directory)) \ 

309 and (not writes_output_file or exists(output_filepath)) \ 

310 and (output != MISSING_VALUE_EXCEPTION) 

311 # If we already have the expected output then we can skip the task at all 

312 if satisfied_output: 

313 print(f'{GREY_HEADER}-> Task {self.flag} ({self.name}) already completed{COLOR_END}') 

314 return output 

315 # If we are at this point then we are missing some output so we must proceed to run the task 

316 # Use the final output directory instead of the incomplete one if exists 

317 # Note that we must check if it exists again since it may have been deleted since the last check 

318 if writes_output_dir and exists(final_output_directory): 

319 processed_args[OUTPUT_DIRECTORY_ARG] = final_output_directory 

320 # Create the incomplete output directory, if necessary 

321 missing_incomplete_output = writes_output_dir \ 

322 and not exists(incomplete_output_directory) \ 

323 and not exists(final_output_directory) 

324 if missing_incomplete_output: mkdir(incomplete_output_directory) 

325 # Finally call the function 

326 print(f'{GREEN_HEADER}-> Running task {self.flag} ({self.name}){COLOR_END}') 

327 start_time = time.time() 

328 # If the task is to be run again because an inputs changed then let the user know 

329 if any_input_changed and had_cache and not forced_overwrite: 

330 changes = ''.join([ '\n - ' + inp for inp in changed_inputs ]) 

331 print(f'{GREEN_HEADER} The task is run again since the following inputs changed:{changes}{COLOR_END}') 

332 # Save a few internal values the task although the task is static 

333 # We save it right before calling the function in case the function uses this task as input 

334 self.changed_inputs = changed_inputs 

335 self.cache_cksums = cache_cksums 

336 # Run the actual task 

337 output = self.func(**processed_args) 

338 end_time = time.time() 

339 print(f' Task {self.flag} completed in {end_time - start_time:.2f} seconds{COLOR_END}') 

340 self._set_parent_output(parent, output) 

341 # Set the output to be saved in cache 

342 # Note that all must be JSON serializable values 

343 cache_output = output 

344 # Update cache output unless it is marked to not save it 

345 if self.use_cache: parent.cache.update(self.cache_output_key, cache_output) 

346 # Update the overwritables so this is not remade further in the same run 

347 parent.overwritables.discard(self.flag) 

348 # As a brief cleanup, if the output directory is empty at the end, then remove it 

349 # Otheriwse, change the incomplete directory name to its final name 

350 if writes_output_dir and exists(incomplete_output_directory): 

351 if is_directory_empty(incomplete_output_directory): rmtree(incomplete_output_directory) 

352 else: rename(incomplete_output_directory, final_output_directory) 

353 # Now return the function result 

354 return output 

355 

356 def find_arg_value (self, arg : str, parent : Union['Project', 'MD'], default_arguments : set): 

357 """ Find argument values, thus running any dependency if necessary. """ 

358 # Word 'task' is reserved for getting the task itself 

359 if arg == 'task': return self 

360 # Word 'self' is reserved for getting the caller Project/MD 

361 if arg == 'self': return parent 

362 # Check if the argument is an MD property 

363 arg_value = safe_getattr(parent, arg, MISSING_ARGUMENT_EXCEPTION) 

364 if arg_value != MISSING_ARGUMENT_EXCEPTION: return arg_value 

365 # If the parent is an MD then it may happen the property is from the Project 

366 if isinstance(parent, MD): 

367 arg_value = safe_getattr(parent.project, arg, MISSING_ARGUMENT_EXCEPTION) 

368 if arg_value != MISSING_ARGUMENT_EXCEPTION: return arg_value 

369 # If the property is missing then search among the additional arguments 

370 arg_value = self.args.get(arg, MISSING_ARGUMENT_EXCEPTION) 

371 if arg_value != MISSING_ARGUMENT_EXCEPTION: return arg_value 

372 # It may also happen that the argument has a default value 

373 # If this is the case then we can skip it 

374 if arg in default_arguments: return MISSING_ARGUMENT_EXCEPTION 

375 # NEVER FORGET: Function arguments must have the same name that the Project/MD property 

376 # If the argument is still missing then you programmed the function wrongly or... 

377 # You may have forgotten the additional argument in the task args 

378 raise RuntimeError(f'Function "{self.func.__name__}" from task "{self.flag}" expects argument "{arg}" but it is missing') 

379 

380 def get_changed_inputs (self, 

381 parent : Union['Project', 'MD'], 

382 processed_args : dict) -> tuple[ list[str], bool ]: 

383 """ Find out if input arguments changed regarding the last run. """ 

384 # Get cache argument references 

385 cache_cksums = parent.cache.retrieve(self.cache_arg_cksums, MISSING_VALUE_EXCEPTION) 

386 had_cache = False if cache_cksums == MISSING_VALUE_EXCEPTION else True 

387 if cache_cksums == MISSING_VALUE_EXCEPTION: cache_cksums = {} 

388 # Check argument by argument 

389 # Keep a list with arguments which have changed 

390 unmatched_arguments = [] 

391 for arg_name, arg_value in processed_args.items(): 

392 # Skip the output directory argument 

393 # Changes in this argument are not actual changes 

394 if arg_name == OUTPUT_DIRECTORY_ARG: continue 

395 # Get the cksum from the new argument value 

396 new_cksum = get_cksum_id(arg_value) 

397 # Retrieve the cksum from the old argument value 

398 old_cksum = cache_cksums.get(arg_name, None) 

399 if self.debug: print(f'Task "{self.name}" -> argument "{arg_name}"\n' + 

400 f' new value: {arg_value}\n' + 

401 f' new value cksum: {new_cksum}\n' + 

402 f' old value cksum: {old_cksum}\n' + 

403 f' match: {new_cksum == old_cksum}') 

404 # Compare new and old cksums 

405 if new_cksum != old_cksum: 

406 # If we found a missmatch then add it to the list 

407 unmatched_arguments.append(arg_name) 

408 # Update the references 

409 cache_cksums[arg_name] = new_cksum 

410 return unmatched_arguments, had_cache, cache_cksums 

411 

412 

413class MD: 

414 """ A Molecular Dynamics (MD) is the union of a structure and a trajectory. 

415 Having this data several analyses are possible. 

416 Note that an MD is always defined inside of a Project and thus it has additional topology and metadata. """ 

417 

418 def __init__ (self, 

419 project : 'Project', 

420 number : int, 

421 directory : str, 

422 input_structure_filepath : str, 

423 input_trajectory_filepaths : list[str], 

424 ): 

425 """ 

426 Initialize the MD object. 

427  

428 Args: 

429 project (Project): The parent project this MD belongs to. 

430 number (int): The number of the MD according to its accession. 

431 directory (str): The local directory where the MD takes place. 

432 input_structure_filepath (str): The input structure file path. 

433 input_trajectory_filepaths (list[str]): The input trajectory file paths. 

434 """ 

435 

436 # Save the inputs 

437 self.project = project 

438 if not project: 

439 raise Exception('Project is mandatory to instantiate a new MD') 

440 # Save the MD number and index 

441 self.number = number 

442 self.index = number - 1 

443 # Set the MD accession and request URL 

444 self.accession = None 

445 self.remote = None 

446 if self.project.database_url and self.project.accession: 

447 self.accession = f'{self.project.accession}.{self.number}' 

448 self.remote = Remote(self.project.database_url, self.accession) 

449 # Save the directory 

450 self.directory = normpath(directory) 

451 # Now set the director relative to the project 

452 self.directory = self.project.pathify(self.directory) 

453 if normpath(self.directory) == normpath(self.project.directory): 

454 raise InputError(f'MD {self.number} has the same directory as the project: {self.directory}') 

455 # Save the directory name alone apart 

456 self.directory_location, self.directory_name = split(self.directory) 

457 # If the directory does not exists then create it 

458 if not exists(self.directory): 

459 mkdir(self.directory) 

460 # Save the input structure filepath 

461 # They may be relative to the project directory (unique) or relative to the MD directory (one per MD) 

462 # If the path is absolute then it is considered unique 

463 # If the file does not exist and it is to be downloaded then it is downloaded for each MD 

464 # Priorize the MD directory over the project directory 

465 self.arg_input_structure_filepath = input_structure_filepath 

466 self._input_structure_filepath = None 

467 # Set the internal variable for the input structure file, to be assigned later 

468 self._input_structure_file = None 

469 # Save the input trajectory filepaths 

470 self.arg_input_trajectory_filepaths = input_trajectory_filepaths 

471 self._input_trajectory_filepaths = None 

472 # Set the internal variable for the input trajectory files, to be assigned later 

473 self._input_trajectory_files = None 

474 

475 # Processed structure and trajectory files 

476 self._structure_file = None 

477 self._trajectory_file = None 

478 

479 # Other values which may be found/calculated on demand 

480 self._md_inputs = None 

481 self._structure = None 

482 

483 # Tests 

484 self._trajectory_integrity = None 

485 

486 # Set a new MD specific register 

487 # In case the directory is the project directory itself, use the project register 

488 register_filepath = self.pathify(REGISTER_FILENAME) 

489 register_file = File(register_filepath) 

490 if register_file.path == self.project.register.file.path: 

491 self.register = self.project.register 

492 else: 

493 self.register = Register(register_file) 

494 # Save also warnings apart since they are to be used as an input for metadata tasks 

495 self.warnings = self.register.warnings 

496 

497 # Set a new MD specific cache 

498 # In case the directory is the project directory itself, use the project cache 

499 cache_filepath = self.pathify(CACHE_FILENAME) 

500 cache_file = File(cache_filepath) 

501 if cache_file.path == self.project.cache.file.path: 

502 self.cache = self.project.cache 

503 else: 

504 self.cache = Cache(cache_file) 

505 

506 # Set tasks whose output is to be overwritten 

507 self.overwritables = set() 

508 

509 # Get MD inputs just to fill the inputs' "mds" value 

510 # Some functions may fail otherwise when its value is missing 

511 self.get_md_inputs() 

512 

513 def __repr__ (self): 

514 return 'MD' 

515 

516 def pathify (self, filename_or_relative_path : str) -> str: 

517 """ Given a filename or relative path, add the MD directory path at the beginning. """ 

518 return normpath(self.directory + '/' + filename_or_relative_path) 

519 

520 # Input structure file ------------ 

521 

522 def get_input_structure_filepath (self) -> str: 

523 """ Set a function to get input structure file path. """ 

524 # Return the internal value if it is already assigned 

525 if self._input_structure_filepath != None: 

526 return self._input_structure_filepath 

527 # Set a function to find out if a path is relative to MD directories or to the project directory 

528 # To do so just check if the file exists in any of those 

529 # In case it exists in both or none then assume it is relative to MD directory 

530 # Parse glob notation in the process 

531 def relativize_and_parse_paths (input_path : str, may_not_exist : bool = False) -> Optional[str]: 

532 # Check if it is an absolute path 

533 if isabs(input_path): 

534 abs_glob_parse = parse_glob(input_path) 

535 # If we had multiple results then we complain 

536 if len(abs_glob_parse) > 1: 

537 raise InputError(f'Multiple structures found with "{input_path}": {", ".join(abs_glob_parse)}') 

538 # If we had no results then we complain 

539 if len(abs_glob_parse) == 0: 

540 if self.remote: 

541 warn('Spread syntax is not supported to download remote files') 

542 raise InputError(f'No structure found with "{input_path}"') 

543 abs_parsed_filepath = abs_glob_parse[0] 

544 return abs_parsed_filepath 

545 # Check the MD directory 

546 md_relative_filepath = self.pathify(input_path) 

547 md_glob_parse = parse_glob(md_relative_filepath) 

548 if len(md_glob_parse) > 1: 

549 raise InputError(f'Multiple structures found with "{input_path}": {", ".join(md_glob_parse)}') 

550 md_parsed_filepath = md_glob_parse[0] if len(md_glob_parse) == 1 else None 

551 if md_parsed_filepath and File(md_parsed_filepath).exists: 

552 return md_parsed_filepath 

553 # Check the project directory 

554 project_relative_filepath = self.project.pathify(input_path) 

555 project_glob_parse = parse_glob(project_relative_filepath) 

556 if len(project_glob_parse) > 1: 

557 raise InputError(f'Multiple structures found with "{input_path}": {", ".join(project_glob_parse)}') 

558 project_parsed_filepath = project_glob_parse[0] if len(project_glob_parse) == 1 else None 

559 if project_parsed_filepath and File(project_parsed_filepath).exists: 

560 return project_parsed_filepath 

561 # At this point we can conclude the input structure file does not exist 

562 # If we have no paths at all then it means a glob pattern was passed and it didn't match 

563 # Note that if a glob pattern existed then it would mean the file actually existed 

564 if len(md_glob_parse) == 0 and len(project_glob_parse) == 0: 

565 # Warn the user in case it was trying to use glob syntax to donwload remote files 

566 if self.remote: 

567 warn('Spread syntax is not supported to download remote files') 

568 raise InputError('No trajectory file was reached neither in the project directory or MD directories in path(s) ' + ', '.join(input_path)) 

569 # If the path does not exist anywhere then we asume it will be downloaded and set it relative to the MD 

570 # However make sure we have a remote 

571 # As an exception, if the 'may not exist' flag is passed then we return the result even if there is no remote 

572 if not may_not_exist and not self.remote: 

573 raise InputError(f'Cannot find a structure file by "{input_path}" anywhere') 

574 md_parsed_filepath = self.project.pathify(input_path) if f'{self.directory_name}/' in md_parsed_filepath else self.pathify(input_path) 

575 return md_parsed_filepath 

576 # If we have a value passed through command line 

577 if self.arg_input_structure_filepath: 

578 # Find out if it is relative to MD directories or to the project directory 

579 self._input_structure_filepath = relativize_and_parse_paths(self.arg_input_structure_filepath) 

580 # Save the parsed value in the inputs file 

581 self.project.update_inputs( 

582 f'mds.{self.index}.input_structure_filepath', 

583 self._input_structure_filepath) 

584 return self._input_structure_filepath 

585 # If we have a value passed through the inputs file has the value 

586 if self.project.is_inputs_file_available(): 

587 # Get the input value, whose key must exist 

588 inputs_value = self.get_input('input_structure_filepath') 

589 # If there is a valid input then use it 

590 if inputs_value: 

591 self._input_structure_filepath = relativize_and_parse_paths(inputs_value) 

592 return self._input_structure_filepath 

593 # If there is not input structure anywhere then use the input topology 

594 # We will extract the structure from it using a sample frame from the trajectory 

595 # Note that topology input filepath must exist and an input error will raise otherwise 

596 # However if we are using the standard topology file we can not extract the PDB from it (yet) 

597 if self.project.input_topology_file != MISSING_TOPOLOGY and \ 

598 self.project.input_topology_file.filename != STANDARD_TOPOLOGY_FILENAME: 

599 return self.project.input_topology_file.path 

600 # If we can not use the topology either then surrender 

601 raise InputError('There is not input structure at all') 

602 

603 def get_input_structure_file (self) -> str: 

604 """ Get the input pdb filename from the inputs. 

605 If the file is not found try to download it. """ 

606 # If the input structure file is already defined then return it 

607 if self._input_structure_file: 

608 return self._input_structure_file 

609 # Otherwise we must set it 

610 # First set the input structure filepath 

611 input_structure_filepath = self.get_input_structure_filepath() 

612 # Now set the input structure file 

613 self._input_structure_file = File(input_structure_filepath) 

614 # If the file already exists then return it 

615 if self._input_structure_file.exists: 

616 return self._input_structure_file 

617 # Try to download it 

618 # If we do not have the required parameters to download it then we surrender here 

619 if not self.remote: 

620 raise InputError(f'Missing input structure file "{self._input_structure_file.path}"') 

621 # Download the structure 

622 # If the structure filename is the standard structure filename then use the structure endpoint instead 

623 if self._input_structure_file.filename == STRUCTURE_FILENAME: 

624 self.remote.download_standard_structure(self._input_structure_file) 

625 # Otherwise download the input strucutre file by its filename 

626 else: 

627 self.remote.download_file(self._input_structure_file) 

628 return self._input_structure_file 

629 input_structure_file = property(get_input_structure_file, None, None, "Input structure filename (read only)") 

630 

631 # Input trajectory filename ------------ 

632 

633 def get_input_trajectory_filepaths (self) -> str: 

634 """ Get the input trajectory file paths. """ 

635 # Return the internal value if it is already assigned 

636 if self._input_trajectory_filepaths != None: 

637 return self._input_trajectory_filepaths 

638 # Set a function to check and fix input trajectory filepaths 

639 # Also relativize paths to the current MD directory and parse glob notation 

640 def relativize_and_parse_paths (input_paths : list[str]) -> list[str]: 

641 checked_paths = input_paths 

642 # Input trajectory filepaths may be both a list or a single string 

643 # However we must keep a list 

644 if type(checked_paths) == list: 

645 pass 

646 elif type(checked_paths) == str: 

647 checked_paths = [ checked_paths ] 

648 else: 

649 raise InputError('Input trajectory filepaths must be a list of strings or a string') 

650 # Make sure all or none of the trajectory paths are absolute 

651 abs_count = sum([ isabs(path) for path in checked_paths ]) 

652 if not (abs_count == 0 or abs_count == len(checked_paths)): 

653 raise InputError('All trajectory paths must be relative or absolute. Mixing is not supported') 

654 # Set a function to glob-parse and merge all paths 

655 def parse_all_glob (paths : list[str]) -> list[str]: 

656 parsed_paths = [] 

657 for path in paths: 

658 parsed_paths += parse_glob(path) 

659 return parsed_paths 

660 # In case trajectory paths are absolute 

661 if abs_count > 0: 

662 absolute_parsed_paths = parse_all_glob(checked_paths) 

663 # Check we successfully defined some trajectory file 

664 if len(absolute_parsed_paths) == 0: 

665 # Warn the user in case it was trying to use glob syntax to donwload remote files 

666 if self.remote: 

667 warn('Spread syntax is not supported to download remote files') 

668 raise InputError('No trajectory file was reached neither in the project directory or MD directories in path(s) ' + ', '.join(input_paths)) 

669 return absolute_parsed_paths 

670 # If trajectory paths are not absolute then check if they are relative to the MD directory 

671 # Get paths relative to the current MD directory 

672 md_relative_paths = [ self.pathify(path) for path in checked_paths ] 

673 # In case there are glob characters we must parse the paths 

674 md_parsed_paths = parse_all_glob(md_relative_paths) 

675 # Check we successfully defined some trajectory file 

676 if len(md_parsed_paths) > 0: 

677 # If so, check at least one of the files do actually exist 

678 if any([ File(path).exists for path in md_parsed_paths ]): 

679 return md_parsed_paths 

680 # If no trajectory files where found then asume they are relative to the project 

681 # Get paths relative to the project directory 

682 project_relative_paths = [ self.project.pathify(path) for path in checked_paths ] 

683 # In case there are glob characters we must parse the paths 

684 project_parsed_paths = parse_all_glob(project_relative_paths) 

685 # Check we successfully defined some trajectory file 

686 if len(project_parsed_paths) > 0: 

687 # If so, check at least one of the files do actually exist 

688 if any([ File(path).exists for path in project_parsed_paths ]): 

689 return project_parsed_paths 

690 # At this point we can conclude the input trajectory file does not exist 

691 # If we have no paths at all then it means a glob pattern was passed and it didn't match 

692 # Note that if a glob pattern existed then it would mean the file actually existed 

693 if len(md_parsed_paths) == 0 and len(project_parsed_paths) == 0: 

694 # Warn the user in case it was trying to use glob syntax to donwload remote files 

695 if self.remote: 

696 warn('Spread syntax is not supported to download remote files') 

697 raise InputError('No trajectory file was reached neither in the project directory or MD directories in path(s) ' + ', '.join(input_paths)) 

698 # If we have a path however it may be downloaded from the database if we have a remote 

699 if not self.remote: 

700 raise InputError(f'Cannot find anywhere a trajectory file with path(s) "{", ".join(input_paths)}"') 

701 # Note that if input path was not glob based it will be both as project relative and MD relative 

702 if len(md_parsed_paths) == 0: raise ValueError('This should never happen') 

703 # If file is to be downloaded then we must make sure the path is relative to the project 

704 project_relative_paths = [ 

705 self.project.pathify(path) if f'{self.directory_name}/' in path else self.pathify(path) for path in checked_paths 

706 ] 

707 return project_relative_paths 

708 # If we have a value passed through command line 

709 if self.arg_input_trajectory_filepaths: 

710 self._input_trajectory_filepaths = relativize_and_parse_paths(self.arg_input_trajectory_filepaths) 

711 # Save the parsed value in the inputs file 

712 self.project.update_inputs( 

713 f'mds.{self.index}.input_trajectory_filepaths', 

714 self._input_trajectory_filepaths) 

715 return self._input_trajectory_filepaths 

716 # Check if the inputs file has the value 

717 if self.project.is_inputs_file_available(): 

718 # Get the input value 

719 inputs_value = self.get_input('input_trajectory_filepaths') 

720 if inputs_value: 

721 self._input_trajectory_filepaths = relativize_and_parse_paths(inputs_value) 

722 return self._input_trajectory_filepaths 

723 # If there is no trajectory available then we surrender 

724 raise InputError('There is not input trajectory at all') 

725 

726 def get_input_trajectory_files (self) -> str: 

727 """ Get the input trajectory filename(s) from the inputs. 

728 If file(s) are not found try to download it. """ 

729 # If we already defined input trajectory files then return them 

730 if self._input_trajectory_files != None: 

731 return self._input_trajectory_files 

732 # Otherwise we must set the input trajectory files 

733 input_trajectory_filepaths = self.get_input_trajectory_filepaths() 

734 self._input_trajectory_files = [ File(path) for path in input_trajectory_filepaths ] 

735 # Find missing trajectory files 

736 missing_input_trajectory_files = [] 

737 for trajectory_file in self._input_trajectory_files: 

738 if not trajectory_file.exists: 

739 missing_input_trajectory_files.append(trajectory_file) 

740 # If all files already exists then we are done 

741 if len(missing_input_trajectory_files) == 0: 

742 return self._input_trajectory_files 

743 # Try to download the missing files 

744 # If we do not have the required parameters to download it then we surrender here 

745 if not self.remote: 

746 missing_filepaths = [ trajectory_file.path for trajectory_file in missing_input_trajectory_files ] 

747 raise InputError('Missing input trajectory files: ' + ', '.join(missing_filepaths)) 

748 # Download each trajectory file (ususally it will be just one) 

749 for trajectory_file in self._input_trajectory_files: 

750 # If this is the main trajectory (the usual one) then use the dedicated endpoint 

751 if trajectory_file.filename == TRAJECTORY_FILENAME: 

752 frame_selection = f'1:{self.project.sample_trajectory}:1' if self.project.sample_trajectory else None 

753 self.remote.download_trajectory(trajectory_file, frame_selection=frame_selection, format='xtc') 

754 # Otherwise, download it by its filename 

755 else: 

756 self.remote.download_file(trajectory_file) 

757 return self._input_trajectory_files 

758 input_trajectory_files = property(get_input_trajectory_files, None, None, "Input trajectory filenames (read only)") 

759 

760 def get_md_inputs (self) -> dict: 

761 """ Get MD specific inputs. """ 

762 # If we already have a value stored then return it 

763 if self._md_inputs: 

764 return self._md_inputs 

765 # Otherwise we must find its value 

766 # If we have MD inputs in the inputs file then use them 

767 if self.project.input_mds: 

768 # Iterate over the different MD inputs to find out each directory 

769 # We must find the MD inputs whcih belong to this specific MD according to this directory 

770 for md in self.project.input_mds: 

771 # Get the directory according to the inputs 

772 directory = md.get(MD_DIRECTORY, None) 

773 if directory: 

774 check_directory(directory) 

775 # If no directory is specified in the inputs then guess it from the MD name 

776 else: 

777 name = md.get('name', None) 

778 if not name: raise InputError('There is a MD with no name and no directory. Please define at least one of them.') 

779 directory = name_2_directory(name) 

780 # If the directory matches then this is our MD inputs 

781 if self.project.pathify(directory) == self.directory: 

782 self._md_inputs = md 

783 return self._md_inputs 

784 # If this MD directory has not associated inputs then it means it was passed through command line 

785 # We set a new MD inputs for it 

786 new_md_name = directory_2_name(self.directory) 

787 self._md_inputs = { 'name': new_md_name, 'mdir': self.directory } 

788 # Update the inputs file with the new MD inputs 

789 mds = self.project.inputs.get('mds', None) 

790 if mds == None: mds = [] 

791 new_mds_inputs = [ *mds, self._md_inputs ] 

792 self.project.update_inputs('mds', new_mds_inputs) 

793 return self._md_inputs 

794 

795 md_inputs = property(get_md_inputs, None, None, "MD specific inputs (read only)") 

796 

797 def get_input (self, name: str): 

798 """ Get a specific 'input' value from MD inputs. """ 

799 value = self.md_inputs.get(name, MISSING_INPUT_EXCEPTION) 

800 # If we had a value then return it 

801 if value != MISSING_INPUT_EXCEPTION: 

802 return value 

803 return self.project.get_input(name) 

804 

805 # --------------------------------- 

806 

807 def get_file (self, target_file : File) -> bool: 

808 """ Check if a file exists. If not, try to download it from the database. 

809 If the file is not found in the database it is fine, we do not even warn the user. 

810 Note that this function is used to get populations and transitions files, which are not common. """ 

811 # If it exists we are done 

812 if target_file.exists: 

813 return True 

814 # Try to download the missing file 

815 # If we do not have the required parameters to download it then we surrender here 

816 if not self.remote: 

817 return False 

818 # Check if the file is among the available remote files 

819 # If it is no then stop here 

820 if target_file.filename not in self.remote.available_files: 

821 return False 

822 # Download the file 

823 self.remote.download_file(target_file) 

824 return True 

825 

826 def print_tests_summary (self): 

827 """ Make a summary of tests and their status. """ 

828 print('Tests summary:') 

829 for test_name in AVAILABLE_CHECKINGS: 

830 test_result = self.register.tests.get(test_name, None) 

831 # Print things pretty 

832 test_nice_name = NICE_NAMES[test_name] 

833 test_nice_result = None