hpcflow.sdk.core.task.WorkflowTask#

class hpcflow.sdk.core.task.WorkflowTask(workflow, template, index, element_IDs)#

Bases: AppAware

Represents a Task that is bound to a Workflow.

Parameters:

workflow (Workflow) – The workflow that the task is bound to.
template (Task) – The task template that this binds.
index (int) – Where in the workflow’s list of tasks is this one.
element_IDs (list[int]) – The IDs of the elements of this task.

Methods

`add_elements`	Add elements to this task.
`ensure_input_sources`	Check valid input sources are specified for a new task to be added to the workflow in a given position.
`generate_new_elements`	Create information about new elements in this task.
`get`	Get a parameter known to this task by its path.
`get_all_element_iterations`	Get the iterations known by the task's elements.
`get_dependent_elements`	Get elements from downstream tasks that depend on this task.
`get_dependent_tasks`	Get tasks (insert ID or WorkflowTask objects) that depends on this task.
`get_dir_name`	Get the directory name for a particular iteration.
`get_element_dependencies`	Get elements from upstream tasks that this task depends on.
`get_task_dependencies`	Get tasks (insert ID or WorkflowTask objects) that this task depends on.
`initialise_EARs`	Try to initialise any uninitialised EARs of this task.
`new_empty_task`	Make a new instance without any elements set up yet.
`resolve_element_data_indices`	Find the index of the parameter group index list corresponding to each input data for all elements.

Attributes

`dir_name`	The name of the directory for the task's temporary files.
`downstream_tasks`	All workflow tasks that are downstream from this task.
`element_IDs`	The IDs of elements associated with this task.
`elements`	The elements associated with this task.
`index`	The index of this task within its workflow.
`inputs`	Inputs to this task.
`insert_ID`	The insertion ID of the template task.
`name`	The name of this task based on its template.
`num_actions`	The number of actions in this task.
`num_element_sets`	The number of element sets associated with this task.
`num_elements`	The number of elements associated with this task.
`outputs`	Outputs from this task.
`template`	The template for this task.
`unique_name`	The unique name for this task specifically.
`upstream_tasks`	All workflow tasks that are upstream from this task.
`workflow`	The workflow this task is bound to.

add_elements(*, base_element: Element | None = None, inputs: list[InputValue] | dict[str, Any] | None = None, input_files: list[InputFile] | None = None, sequences: list[ValueSequence] | None = None, resources: Resources = None, repeats: list[RepeatsDescriptor] | int | None = None, input_sources: dict[str, list[InputSource]] | None = None, nesting_order: dict[str, float] | None = None, element_sets: list[ElementSet] | None = None, sourceable_elem_iters: list[int] | None = None, propagate_to: list[ElementPropagation] | Mapping[str, ElementPropagation | Mapping[str, Any]] | None = None, return_indices: Literal[True]) → list[int]#

Add elements to this task.

Parameters:

sourceable_elem_iters (list of int, optional) – If specified, a list of global element iteration indices from which inputs may be sourced. If not specified, all workflow element iterations are considered sourceable.
propagate_to (dict[str, ElementPropagation]) – Propagate the new elements downstream to the specified tasks.
return_indices (bool) – If True, return the list of indices of the newly added elements. False by default.

property dir_name: str#: The name of the directory for the task’s temporary files.

property downstream_tasks: Iterator[WorkflowTask]#: All workflow tasks that are downstream from this task.

property element_IDs: list[int]#: The IDs of elements associated with this task.

property elements: Elements#: The elements associated with this task.

ensure_input_sources(element_set)#

Check valid input sources are specified for a new task to be added to the workflow in a given position. If none are specified, set them according to the default behaviour.

This method mutates element_set.input_sources.

Parameters:: element_set (ElementSet) –
Return type:: Mapping[str, Sequence[int]]

generate_new_elements(input_data_indices, output_data_indices, element_data_indices, sequence_indices, source_indices)#

Create information about new elements in this task.

Parameters:

input_data_indices (Mapping[str, Sequence[int | list[int]]]) –
output_data_indices (Mapping[str, Sequence[int]]) –
element_data_indices (Sequence[Mapping[str, int]]) –
sequence_indices (Mapping[str, Sequence[int]]) –
source_indices (Mapping[str, Sequence[int]]) –

Return type:

tuple[Sequence[DataIndex], Mapping[str, Sequence[int]], Mapping[str, Sequence[int]]]

get(path, *, raise_on_missing=False, default=None)#

Get a parameter known to this task by its path.

Parameters:

path (str) –
default (Any | None) –

Return type:

Parameters

get_all_element_iterations()#

Get the iterations known by the task’s elements.

Return type:: Mapping[int, ElementIteration]

get_dependent_elements(as_objects: Literal[False] = False) → set[int]#
get_dependent_elements(as_objects: Literal[True]) → list[Element]: Get elements from downstream tasks that depend on this task.

get_dependent_tasks(as_objects: Literal[False] = False) → set[int]#
get_dependent_tasks(as_objects: Literal[True]) → list[WorkflowTask]: Get tasks (insert ID or WorkflowTask objects) that depends on this task.

get_dir_name(loop_idx=None)#

Get the directory name for a particular iteration.

Parameters:: loop_idx (Mapping[str, int] | None) –
Return type:: str

get_element_dependencies(as_objects: Literal[False] = False) → set[int]#
get_element_dependencies(as_objects: Literal[True]) → list[Element]: Get elements from upstream tasks that this task depends on.

get_task_dependencies(as_objects: Literal[False] = False) → set[int]#

get_task_dependencies(as_objects: Literal[True]) → list[WorkflowTask]

Get tasks (insert ID or WorkflowTask objects) that this task depends on.

Dependencies may come from either elements from upstream tasks, or from locally defined inputs/sequences/defaults from upstream tasks.

property index: int#: The index of this task within its workflow.

initialise_EARs(iter_IDs=None)#

Try to initialise any uninitialised EARs of this task.

Parameters:: iter_IDs (list[int] | None) –
Return type:: Sequence[int]

property inputs: TaskInputParameters#: Inputs to this task.

property insert_ID: int#: The insertion ID of the template task.

property name: str#: The name of this task based on its template.

classmethod new_empty_task(workflow, template, index)#

Make a new instance without any elements set up yet.

Parameters:

workflow (Workflow) – The workflow that the task is bound to.
template (Task) – The task template that this binds.
index (int) – Where in the workflow’s list of tasks is this one.

Return type:

Self

property num_actions: int#: The number of actions in this task.

property num_element_sets: int#: The number of element sets associated with this task.

property num_elements: int#: The number of elements associated with this task.

property outputs: TaskOutputParameters#: Outputs from this task.

static resolve_element_data_indices(multiplicities)#

Find the index of the parameter group index list corresponding to each input data for all elements.

Parameters:

multiplicities (list of MultiplicityDescriptor) –

Each list item represents a sequence of values with keys:: multiplicity: int nesting_order: float path : str

Returns:

element_dat_idx – Each list item is a dict representing a single task element and whose keys are input data paths and whose values are indices that index the values of the dict returned by the task.make_persistent method.

Return type:

list of dict

Note

Non-integer nesting orders result in doing the dot product of that sequence with all the current sequences instead of just with the other sequences at the same nesting order (or as a cross product for other nesting orders entire).

property template: Task#: The template for this task.

property unique_name: str#: The unique name for this task specifically.

property upstream_tasks: Iterator[WorkflowTask]#: All workflow tasks that are upstream from this task.

property workflow: Workflow#: The workflow this task is bound to.