icolos.core.workflow_steps package#

Subpackages#

Submodules#

icolos.core.workflow_steps.step module#

class icolos.core.workflow_steps.step.StepBase(*, step_id: str, work_dir: str = None, type: str = None, data: ~icolos.core.step_utils.input_preparator.StepData = StepData(compounds=[], generic=<icolos.core.containers.generic.GenericContainer object>, gmx_state=Gromacs Topology object: System: [] | Molecules: [] | FF: amber03 | itp files: [] | posre files []), input: ~icolos.core.step_utils.input_preparator.StepInputParameters = StepInputParameters(compounds=[], generic=[], gmx_state={}, perturbation_map=None, merge=StepMerge(compounds=True, enumerations=False, merge_compounds_by=<StepBaseEnum.INPUT_MERGE_BY_NAME: 'name'>, merge_enumerations_by=<StepBaseEnum.INPUT_MERGE_BY_ID: 'id'>), work_dir=None), writeout: ~typing.List[~icolos.core.step_utils.step_writeout.StepWriteoutParameters] = [], execution: ~icolos.core.workflow_steps.step.StepExecutionParameters = StepExecutionParameters(prefix_execution=None, binary_location=None, pipe_input=None, parallelization=StepExecutionParallelizationParameters(jobs=1, max_length_sublists=None), failure_policy=StepFailurePolicyParameters(n_tries=1, retry_wait_seconds=10), check_backend_availability=False, resources=StepExecutionResourceParameters(partition=<ExecutionPlatformEnum.CORE: 'core'>, time='12:00:00', gres=None, tasks=None, mem=None, cores=None, modules=[], other_args={}, additional_lines=[]), platform=<ExecutionPlatformEnum.LOCAL: 'local'>), settings: ~icolos.core.workflow_steps.step.StepSettingsParameters = StepSettingsParameters(arguments=StepSettingsArgsParameters(flags=[], parameters={}), additional={}))[source]#

Bases: BaseModel

class Config[source]#

Bases: object

underscore_attrs_are_private = True#
clone_compounds() → List[Compound][source]#

Deepcopy each compound in self.data.compounds

Return List[Compound]:

cloned list of Compounds

data: StepData#
execute()[source]#
execution: StepExecutionParameters#
generate_input()[source]#

Parse the input block, fetch objects and read them into self.data

get_arguments(defaults: Optional[dict] = None) → List[str][source]#

Construct pmx-specific arguments from the step defaults, overridden by arguments specified in the config file

Parameters:

defaults (dict) – optional set of key-value pairs for default arguments, defaults to None

Return List[str]:

formatted list of strings

get_compound_by_name(name: str) → Optional[Compound][source]#
get_compound_stats() → Tuple[int, int, int][source]#
get_compounds() → List[Compound][source]#
get_generic() → GenericContainer[source]#
get_perturbation_map() → PerturbationMap[source]#

return the perturbation map attached to the workflow

Return PerturbationMap:

the perturbation map in its current state

get_step_id() → str[source]#

Get the step’s unique string identifier

Return str:

self.step_id, as specified in the step’s config

get_topol() → GromacsState[source]#

Get the gromacs state attached to the step

Return GromacsState:

Current state of the GromacsState object

get_workflow_object()[source]#
input: StepInputParameters#
process_write_out()[source]#

write out data specified in self.writeout. Run after step execution

set_step_id(step_id: str)[source]#

Set the step’s unique ID

Parameters:

step_id (str) – unique identifier for the step

set_workflow_object(workflow_object)[source]#
settings: StepSettingsParameters#
step_id: str#
type: str#
work_dir: str#
write_conformers(path: str)[source]#

Convenience function for frequent conformer coordinate write-out. Better to use the WriteOutHandler class.

write_generic_by_extension(path: str, ext: str, join=True)[source]#

Write all generic data objects with a given extension

Parameters:

  • path (str) – path to write object to, by default just a directory

  • ext (str) – extension

  • join (bool) – controls join behaviour, if True, joins existing filename to path, defaults to True

write_generic_by_name(path: str, name: str)[source]#

Write a generic file by name

Parameters:

  • path (str) – directory to write to

  • name (str) – name of file to be written out

writeout: List[StepWriteoutParameters]#
class icolos.core.workflow_steps.step.StepExecutionParameters(*, prefix_execution: str = None, binary_location: str = None, pipe_input: str = None, parallelization: ~icolos.core.workflow_steps.step.StepExecutionParameters.StepExecutionParallelizationParameters = StepExecutionParallelizationParameters(jobs=1, max_length_sublists=None), failure_policy: ~icolos.core.workflow_steps.step.StepFailurePolicyParameters = StepFailurePolicyParameters(n_tries=1, retry_wait_seconds=10), check_backend_availability: bool = False, resources: ~icolos.core.workflow_steps.step.StepExecutionResourceParameters = StepExecutionResourceParameters(partition=<ExecutionPlatformEnum.CORE: 'core'>, time='12:00:00', gres=None, tasks=None, mem=None, cores=None, modules=[], other_args={}, additional_lines=[]), platform: ~icolos.utils.enums.execution_enums.ExecutionPlatformEnum = ExecutionPlatformEnum.LOCAL)[source]#

Bases: BaseModel

class StepExecutionParallelizationParameters(*, jobs: int = 1, max_length_sublists: int = None)[source]#

Bases: BaseModel

jobs: int#
max_length_sublists: int#
binary_location: str#
check_backend_availability: bool#
failure_policy: StepFailurePolicyParameters#
parallelization: StepExecutionParallelizationParameters#
pipe_input: str#
platform: ExecutionPlatformEnum#
prefix_execution: str#
resources: StepExecutionResourceParameters#
class icolos.core.workflow_steps.step.StepExecutionResourceParameters(*, partition: str = ExecutionPlatformEnum.CORE, time: str = '12:00:00', gres: str = None, tasks: str = None, mem: str = None, cores: int = None, modules: List = [], other_args: dict = {}, additional_lines: List = [])[source]#

Bases: BaseModel

additional_lines: List#
cores: int#
gres: str#
mem: str#
modules: List#
other_args: dict#
partition: str#
tasks: str#
time: str#
class icolos.core.workflow_steps.step.StepFailurePolicyParameters(*, n_tries: int = 1, retry_wait_seconds: int = 10)[source]#

Bases: BaseModel

n_tries: int#
retry_wait_seconds: int#
class icolos.core.workflow_steps.step.StepSettingsArgsParameters(*, flags: List = [], parameters: Dict = {})[source]#

Bases: BaseModel

flags: List#
parameters: Dict#
class icolos.core.workflow_steps.step.StepSettingsParameters(*, arguments: StepSettingsArgsParameters = StepSettingsArgsParameters(flags=[], parameters={}), additional: Dict = {})[source]#

Bases: BaseModel

additional: Dict#
arguments: StepSettingsArgsParameters#

Module contents#