icolos.core.flow_control package

Submodules

icolos.core.flow_control.flow_control module

class icolos.core.flow_control.flow_control.BaseStepConfig(*, step_id: str = None, 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={}))

Bases: BaseModel

Minimal template class for the base config, without the unnecessary stuff that StepBase requires

as_dict()

data: StepData

execution: StepExecutionParameters

input: StepInputParameters

settings: StepSettingsParameters

step_id: str

type: str

work_dir: str

writeout: List[StepWriteoutParameters]

class icolos.core.flow_control.flow_control.FlowControlBase(*, base_config: List[BaseStepConfig] = None, initialized_steps: List[StepBase] = None)

Bases: BaseModel

base_config: List[BaseStepConfig]

initialized_steps: List[StepBase]

icolos.core.flow_control.iterator module

class icolos.core.flow_control.iterator.IterParallelizer(*, parallelize: bool = False, jobs: int = 1, dependent_steps: int = None)

Bases: BaseModel

dependent_steps: int

jobs: int

parallelize: bool

class icolos.core.flow_control.iterator.IterSettings(*, settings: Dict[str, IterSettingsParameters] = {}, iter_mode: IteratorEnum = IteratorEnum.N_ITERS, n_iters: int = None, remove_temporary_files: bool = True, single_directory: bool = False, parallelizer_settings: IterParallelizer = IterParallelizer(parallelize=False, jobs=1, dependent_steps=None))

Bases: BaseModel

iter_mode: IteratorEnum

n_iters: int

parallelizer_settings: IterParallelizer

remove_temporary_files: bool

settings: Dict[str, IterSettingsParameters]

single_directory: bool

class icolos.core.flow_control.iterator.IterSettingsParameters(*, flags: List = [], parameters: Dict = {}, additional: Dict = {}, work_dir: Union[List, str] = [])

Bases: BaseModel

additional: Dict

flags: List

parameters: Dict

work_dir: Union[List, str]

class icolos.core.flow_control.iterator.StepIterator(*, base_config: ~typing.List[~icolos.core.flow_control.flow_control.BaseStepConfig] = None, initialized_steps: ~typing.List[~icolos.core.workflow_steps.step.StepBase] = None, iter_settings: ~icolos.core.flow_control.iterator.IterSettings = IterSettings(settings={}, iter_mode=<IteratorEnum.N_ITERS: 'n_iters'>, n_iters=None, remove_temporary_files=True, single_directory=False, parallelizer_settings=IterParallelizer(parallelize=False, jobs=1, dependent_steps=None)), dispatcher: ~icolos.core.step_dispatch.dispatcher.StepDispatcher = None)

Bases: FlowControlBase, BaseModel

Implements iterator mechanism: wraps one or multiple steps and generates n copies of that set of steps according to iter_config Becomes master job when parallize=True, using Icolos JobControl to interface with external resources

dispatcher: StepDispatcher

iter_settings: IterSettings

Module contents