Tasks

Tasks are the fundamental building blocks of a LINQ workflow. A task represents a unit of work that can be executed on instruments, via custom code, through data integrations, or by operators. A workflow is fundamentally a directed graph of tasks.

Task Concepts

ActionTask — Executes an action on a workcell. This could be moving labware, triggering an instrument driver operation, or requiring manual operator confirmation.

CodeTask — Embeds custom logic into a workflow using Python functions or Lua scripts.

Conditional (If) — Enables branching logic in workflows, allowing different task sequences based on evaluated conditions.

ContinuousLoading — Supports dynamic loading of labware during execution, with each loaded item triggering a predefined sequence of tasks.

SftpDataConnector — Facilitates data transfer to/from external systems (SFTP, S3, LIMS, ELN, etc.).

Implicit Tasks — Transport tasks are automatically derived from labware inputs and outputs. Some driver actions (like load) are automatically compiled into required sequences.

Simple Example

from linq.task import ActionTask, LabwareOutput

seal_plate = ActionTask(
    id="seal_plate",
    description="Seal the microplate",
    instrument_type=sealer.type,
    action="seal",
    time_estimate=30,
    labware_sources=[previous_task.forward(labware=plate)],
    labware_outputs=[LabwareOutput(plate)],
)

Task Classes

ActionTask

An ActionTask represents the execution of an action on a workcell. This could be moving labware, triggering an instrument driver operation, or a manual operator step. It includes connections to labware sources and outputs.

class linq.task.ActionTask(*, id: str = '', description: str = '', dependencies: list[ActionTask | CodeTask | Conditional] = ..., action: str = '', time_estimate: int, instrument_type: str, labware_sources: list[LabwareSource | StoredLabware], labware_outputs: list[LabwareOutput | StoredLabware] = ..., instrument_mappings: list[str] | None = None, available_failure_actions: AvailableFailureActions = ..., skip: bool = False, mock_behaviour: MockBehaviour | None = None, inputs: Inputs = ..., static_arguments: dict[str, Any] | None = None, shared_arguments: dict[str, str] | None = None)

A Task executing an action.

action: str: ID of the action taken (see driver CLI for possible actions for each instrument.)

time_estimate: int: Estimated time to execute the task.

instrument_type: str: Type of the instrument this task runs on.

labware_sources: list[LabwareSource | StoredLabware]: Defines the source of each labware used in this task.

labware_outputs: list[LabwareOutput | StoredLabware]: Outputs of this task. If left empty, defaults to a labware output for each labware source. If no labware sources are defined, this will be empty.

instrument_mappings: list[str] | None: Restrict task to run on an instrument from one of these mappings.

available_failure_actions: AvailableFailureActions: Available actions to be taken in case of failure.

skip: bool: Set to true to skip this task.

mock_behaviour: MockBehaviour | None: Overrides for the behaviour of the task on mock drivers

inputs: Inputs: Inputs to be passed to the task. The key is the name of the input. Can be a TaskOutputReference to the output of another task, a ReferenceData to some stored data, or a literal value.

static_arguments: dict[str, Any] | None: Deprecated, use typed ‘inputs’ field instead

shared_arguments: dict[str, str] | None: Deprecated, use typed ‘inputs’ field instead

Create a labware source from an output of a previous task.

This can be used to forward previously used labware to another task without having to reference the source task multiple times.

out(output_name: str | None = None) → TaskOutputReference: Reference to the output of this task. If the task has only one output, output_name can be left as None.

CodeTask

A CodeTask allows embedding custom logic into your workflow, either through pre-defined Python functions or Lua scripts.

class linq.task.CodeTask(*, id: str = '', description: str = '', dependencies: list[Union[ForwardRef('ActionTask'), ForwardRef('CodeTask'), ForwardRef('Conditional')]] = ..., function: linq.task.LuaFunction | linq.task.PythonReferenceFunction, inputs: linq.task.Inputs = ...)

function: LuaFunction | PythonReferenceFunction: The function to be executed.

inputs: Inputs: The inputs to the function. The key is the name of the input. Can be a TaskOutputReference to the output of another task, a ReferenceData to some stored data, or a literal value.

out(output_name: str | None = None) → TaskOutputReference: Reference to the output of this task. If the task has only one output, output_name can be left as None.

Conditional (If)

class linq.task.Conditional(*, id: str = ..., is_expected: bool, comparison: linq.task.Comparison | Literal['Else'], description: str = 'Conditional', next: 'Conditional | None' = None, _prev: 'Conditional | None' = None)

Then(*dependents: ActionTask | CodeTask | Conditional) → Self: Register tasks to run when this condition is met.

class linq.task.If(*, id: str = ..., is_expected: bool = True, comparison: linq.task.Comparison | Literal['Else'], description: str = 'Conditional', next: 'Conditional | None' = None, _prev: 'Conditional | None' = None)

ContinuousLoading

Data Connector

These classes support data transfer and integration within workflows, such as interacting with SFTP endpoints.

Labware and Output Configuration

Some parameter types used across tasks are documented here:

LabwareSource – References labware used by a task.
StoredLabware – Declares static labware placement for outputs.
LabwareOutput – Specifies where labware ends up after a task.
OutputSlotFillingRule – Defines how labware fills output slots.
SlotFillingRule – Specifies slot filling behavior.

Specifies the labware required for a task.

labware: Labware: Labware being passed into a task.

destination_slot: int | SlotFillingRule: The instrument slot where the labware will be placed, or the rule for filling slots with labware (batching)

source_task: ActionTask | ConditionalSourceTask | None: ID of the task the labware is being transported from. None for tasks retrieving labware from their initial instrument.

dependencies: list[ActionTask | CodeTask | Conditional]: Additional tasks that have to be completed before the labware for the task is retrieved.

units_consumed: int | None: The number of units of the labware that are consumed in the task. Tied to the total_units of the labware. Leave blank for no units to be consumed.

A CodeTask allows embedding custom logic into your workflow, either through pre-defined Python functions or Lua scripts.

Data Connector Task Classes

These classes support data transfer and integration within workflows, such as interacting with SFTP endpoints.

class linq.task.OutputSlotFillingRule(behavior: Literal['match_input'] = 'match_input')

behavior: Literal['match_input']: The output slot filling behaviour.

class linq.task.SlotFillingRule(range: linq.labware.SlotRange | list[linq.labware.SlotRange], min_count: int | None = None, max_count: int | None = None, distribution: Literal['unique', 'non_unique'] = 'unique')

range: SlotRange | list[SlotRange]: The range of slots that the labware can be placed in.

min_count: int | None: The minimum number of labware items that must be placed in the slots.

max_count: int | None: The maximum number of labware items that can be placed in the slots.

distribution: Literal['unique', 'non_unique']: How the labware will be distributed across the slots. unique ensure each labware is placed in a unique slot. non_unique allows multiple labware to be placed in the same slot over many instances of the task

Task References and Conditionals

class linq.task.TaskOutputReference(*, task: ForwardRef('ActionTask') | ForwardRef('CodeTask') | ForwardRef('Conditional'), output_name: str | None = None)

task: ActionTask | CodeTask | Conditional: The task that produces the data

output_name: str | None: The name of the output from the task. Can leave as None if the task has only one output.

class linq.task.Comparison(*, lhs: 'TaskOutputReference | ReferenceData ', rhs: 'TaskOutputReference | ReferenceData | ParameterReference | workflow_api.JsonSerializable', operator: Literal['==', '!=', '<', '<=', '>', '>=', 'in', 'not in'])

class linq.task.ConditionalSourceTask(tasks: list[ActionTask | CodeTask | Conditional]): Specifies the potential source tasks that labware can be retrieved from.

Failure Actions

Both ActionTask and CodeTask support the optional available_failure_actions parameter. For configuration details and runtime semantics, see Error Handling.

class linq.task.AvailableFailureActions(*, complete_manually: bool = True, retry: bool = True)

Available actions that can be taken by an operator if a task fails.

complete_manually: bool: Allow manually completing the task.

retry: bool: Allow retrying.

Other Task Classes

class linq.task.Inputs(**kwargs: TaskOutputReference | ReferenceData | ParameterReference | JsonSerializable)

class linq.task.ReferenceData(*, key: str)

class linq.task.MockBehaviour(*, execution_time: float | None = None, success: bool = True, data_outputs: dict[str, JsonSerializable] | None = None)

execution_time: float | None: Execution time in seconds. Overrides the default execution time of the task. If set to 0, task will not start

success: bool: Whether the task should succeed or fail

data_outputs: dict[str, JsonSerializable] | None: key,value pairs of data to be stored in the data store after the task is completed