Coverage for mddb_workflow / utils / subsets.py: 74%
34 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 18:45 +0000
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 18:45 +0000
1from inspect import getfullargspec
3from mddb_workflow.utils.auxiliar import InputError
4from mddb_workflow.utils.formats import get_format_set_suitable_function
5from mddb_workflow.utils.pyt_spells import get_frames_count
6from mddb_workflow.utils.gmx_spells import get_trajectory_subset as gmx_get_trajectory_subset
7from mddb_workflow.utils.mdt_spells import get_trajectory_subset as mdt_get_trajectory_subset
8from mddb_workflow.utils.type_hints import *
10# Set the functions to perform single frame gettering
11subset_functions = [ gmx_get_trajectory_subset, mdt_get_trajectory_subset ]
13# Get specific frames from a trajectory
14def get_trajectory_subset (
15 input_structure_file : 'File',
16 input_trajectory_file : 'File',
17 output_trajectory_file : 'File',
18 start : int = 0,
19 end : int = None,
20 step : int = 1,
21 frames : list[int] = [],
22 skip : list[int] = [],
23):
24 # If there is no end set then the end as the last frame of the simulation
25 if end == None:
26 end = get_frames_count(input_structure_file, input_trajectory_file)
28 # How step and skip are combined is not intuitive and it could be missleading
29 # For this reason both arguments are not allowed together
30 if step and step != 1 and skip and len(skip) > 0:
31 raise InputError("Arguments 'step' and 'skip' are not allowed together. Please do it in 2 separated calls.")
33 # End must be grater than start
34 if end != None and end < start:
35 raise InputError('End frame must be posterior to start frame')
37 # We need an output trajectory filename
38 if not output_trajectory_file:
39 raise InputError('Missing output trajectory filename')
41 # In case the frames argument is passed
42 if frames and len(frames) > 0:
43 print('Specific frames were passed. Other arguments will be ignored (start, end, step and skip)')
44 if frames != sorted(frames):
45 print('WARNING: Note that the reduced trajectories will keep frames in their original order, not the input order')
47 # Get the input formats
48 input_structure_format = input_structure_file.format
49 input_trajectory_format = input_trajectory_file.format
50 output_trajectory_format = output_trajectory_file.format
51 format_set = {
52 'inputs': {
53 'input_structure_filename': None if input_structure_format == None else { input_structure_format },
54 'input_trajectory_filename': { input_trajectory_format }
55 },
56 'outputs': {
57 'output_trajectory_filename': { output_trajectory_format }
58 }
59 }
60 # Get a suitable function to do the job
61 suitables = next(get_format_set_suitable_function(
62 available_functions=subset_functions,
63 available_request_format_sets=[format_set],
64 ), None)
65 if not suitables:
66 raise InputError('There is no subset function which supports the requested formats')
67 suitable_function, formats = suitables
68 # Call the subset function
69 # Check if the subset function requires the input structure argument before
70 suitable_function_keywords = getfullargspec(suitable_function)[0]
71 required_structure = 'input_structure_filename' in suitable_function_keywords
72 if required_structure:
73 suitable_function(input_structure_file.path, input_trajectory_file.path, output_trajectory_file.path, start, end, step, frames, skip)
74 else:
75 suitable_function(input_trajectory_file.path, output_trajectory_file.path, start, end, step, frames, skip)