matflow.WorkflowTask

class matflow.WorkflowTask(workflow, template, index, element_IDs)

Bases: WorkflowTask

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.
export_array_output	Export this task's selected array output to a single .npy file.
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_array_output	Read an array output from this task's producing action runs.
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.
list_array_outputs	Return array output directory names for this task's output parameters.
new_empty_task	Make a new instance without any elements set up yet, validate any task condition, and add any missing parameter dependence required by that condition.
resolve_element_data_indices	Find the index of the parameter group index list corresponding to each input data for all elements.
validate_condition	Validate a task condition string.

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=None, inputs=None, input_files=None, sequences=None, resources=None, repeats=None, input_sources=None, nesting_order=None, element_sets=None, sourceable_elem_iters=None, propagate_to=None, return_indices=False)

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.

• base_element (Element | None) –

• inputs (list[InputValue] | dict[str, Any] | None) –

• input_files (list[InputFile] | None) –

• sequences (list[ValueSequence] | None) –

• resources (Resources) –

• repeats (RepeatsDescriptorType) –

• input_sources (dict[str, list[InputSource]] | None) –

• nesting_order (dict[str, float] | None) –

• element_sets (list[ElementSet] | None) –

Return type:

list[int] | None

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]]

export_array_output(name, path, *, iteration_idx=None, element_idx=None, sort=True, overwrite=False)

Export this task’s selected array output to a single .npy file.

Filters and sorting follow get_array_output(). All selected arrays must have matching shapes. Write to exactly path and return that path. Existing files raise FileExistsError unless overwrite=True.

Parameters:

• name (str) –

• path (str | Path) –

• iteration_idx (int | Sequence[int] | None) –

• element_idx (int | Sequence[int] | None) –

• sort (bool) –

• overwrite (bool) –

Return type:

Path

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_array_output(name, *, iteration_idx=None, element_idx=None, sort=True)

Read an array output from this task’s producing action runs.

The optional outputs. prefix is accepted. Task- and schema-level disable_array_outputs exclusions are combined. Only runs of the root parameter’s first source action from TaskSchema.get_action_parameter_flow() are included. By default, results are sorted lexicographically by element index, iteration index, action index, then run index. With sort=False, order follows Workflow.get_EAR_IDs_of_tasks(). Archives are read only by the workflow reader; missing saved data therefore raises rather than being skipped.

iteration_idx=None selects each element’s final non-loop-skipped iteration; an integer or sequence selects those iteration indices explicitly. element_idx=None includes all elements; an integer or sequence selects those element indices. Sequences filter without reordering or duplicating runs.

Raises:

ValueError – If the output is disabled or no producing runs are found.

Parameters:

• name (str) –

• iteration_idx (int | Sequence[int] | None) –

• element_idx (int | Sequence[int] | None) –

• sort (bool) –

Return type:

ArrayOutput

get_dependent_elements(as_objects=False)

Get elements from downstream tasks that depend on this task.

Parameters:

as_objects (bool) –

Return type:

set[int] | list[Element]

get_dependent_tasks(as_objects=False)

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

Parameters:

as_objects (bool) –

Return type:

set[int] | list[hpcflow.sdk.core.task.WorkflowTask]

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=False)

Get elements from upstream tasks that this task depends on.

Parameters:

as_objects (bool) –

Return type:

set[int] | list[Element]

get_task_dependencies(as_objects=False)

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.

Parameters:

as_objects (bool) –

Return type:

set[int] | list[hpcflow.sdk.core.task.WorkflowTask]

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.

list_array_outputs()

Return array output directory names for this task’s output parameters.

Names are sorted and exclude the optional outputs. prefix. No NPZ files are inspected; an existing directory does not guarantee that this task has saved entries.

Return type:

list[str]

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, validate any task condition, and add any missing parameter dependence required by that condition.

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.

validate_condition()

Validate a task condition string.

Return type:

TaskCondition | None

property workflow: Workflow

The workflow this task is bound to.