Workflow

The workflow module contains the main classes for creating a workflow. When creating a workflow, it’s assumed that a workcell has already been created.

Workflow classes

class linq.workflow.Workflow(*, workcell: Workcell, name: str, author: str, description: str, version: str, scheduler_version: str, labware_types: list[LabwareType], labware: list[Labware], tasks: Sequence[ActionTask | CodeTask | Conditional] = ..., time_constraints: list[TimeConstraint] = ..., instrument_blocks: list[InstrumentBlock] = ..., options: Options = ..., run_instructions: list[RunInstruction] = ..., published: bool = False, hooks: Sequence[TaskStateChangeHook | RunStateChangeHook | LabwareMovementHook | SafetyStateChangeHook | NewPlanHook] = ..., python_scripts_version: str | None = None, parameter_definitions: list[Annotated[BooleanParameterDefinition | IntegerParameterDefinition | FloatParameterDefinition | StringParameterDefinition, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] = ..., batches: int | ParameterReference = 1)

Describes a workflow. This is the main entrypoint into the SDK, most CLI actions require a workflow to work.

workcell: Workcell: The workcell this workflow will be run in.

name: str: Workflow name.

author: str: Workflow author.

description: str: Workflow description.

version: str: Workflow version. This is arbitrary for now.

scheduler_version: str: Version of the scheduler to use. Use get_latest_scheduler_version to get the latest supported version.

labware_types: list[LabwareType]: List of all types of labware used in the workflow.

labware: list[Labware]: List of all labware used in the workflow.

tasks: Sequence[ActionTask | CodeTask | Conditional]: The workflow tasks.

time_constraints: list[TimeConstraint]: List of time constraints the workflow has to fulfill.

instrument_blocks: list[InstrumentBlock]: List of instrument blocks the workflow has to fulfill.

options: Options: Planning and execution options.

run_instructions: list[RunInstruction]: Instructions for running the workflow.

published: bool: Whether the workflow is published on the LINQ platform or not

hooks: Sequence[TaskStateChangeHook | RunStateChangeHook | LabwareMovementHook | SafetyStateChangeHook | NewPlanHook]: Hooks to be triggered when a run state or task state changes.

python_scripts_version: str | None: Version of the Python scripts repository managed by the organization. Leaving as None will pull the latest version if available.

batches: int | ParameterReference: Number of batches of the workflow to run. Default value 1 means no batching.

add_task(task: TaskLike) → TaskLike: Add a task to the workflow.

class linq.workflow.Options(*, planner: PlannerOptions | None = None, executor: ExecutorOptions | None = None, is_explicit: bool = False)

Planning and execution options for a workflow.

planner: PlannerOptions | None: Planning-specific options, used to tweak what the planner optimises for.

executor: ExecutorOptions | None: Execution-specific options.

is_explicit: bool: Set this to true to prevent the API from adding any implicit logic, like opening doors, loading/unloading instruments, etc.

class linq.workflow.ExecutorOptions(*, behaviour_on_failure: Literal['abort', 'pause', 'continue'])

Options for the execution of a workflow.

behaviour_on_failure: Literal['abort', 'pause', 'continue']: Behaviour of the system when a task fails. Defaults to cancelling the workflow.

class linq.workflow.PlannerOptions(*, target_objective: float | None = None, max_computation_time: int | None = None, max_dead_time: int | None = None, round_robin_load_balancing: bool = False, preset: Literal['DRAFT', 'TEST', 'PROD_STANDARD', 'PROD_LARGE'] | None = None, optimisation_target: Literal['minimize_total_plan_duration', 'minimize_labware_idle_time', 'balance_total_plan_duration_and_labware_idle_time'] = 'minimize_total_plan_duration')

Planning-specific options, used to tweak how the planner behaves and what it optimises for.

target_objective: float | None: Target objective value.

max_computation_time: int | None: Max computation time allowed for the planner, in seconds. Starting point governed by option guaranteed_solution.

max_dead_time: int | None: Max dead time allowed between tasks with a transport between them

round_robin_load_balancing: bool: Where there are equidistant instruments (instruments that take the same time to transport to), maestro uses round robin to assign instruments to those tasks.

preset: Literal['DRAFT', 'TEST', 'PROD_STANDARD', 'PROD_LARGE'] | None: The plan CLI has a number of presets that can be used to quickly adjust the behaviour of the planner to suit different user needs. The available presets are: - DRAFT: To be used when it is not important if the plan is accurate or feasible. A plan generated with this preset does not guarantee that the workflow is actually possible to run in the real world - TEST: The planner will try to find a plan quickly, but it may not be optimal. - PROD_STANDARD: A preset with fairly standard settings. This is useful for making sure that the planner is using the recommended settings. - PROD_LARGE: A preset with settings that are more suitable for large workflows. This preset is useful when the workflow is large and the planner is taking a long time to find a plan.

class linq.workflow.TimeConstraint(*, id: str, constraint_type: TimeConstraintType, start_task: ActionTask, end_task: ActionTask, min_duration: int | None, max_duration: int | None)

A time constraint for a sequence of tasks.

id: str: Unique ID for this time constraint.

constraint_type: TimeConstraintType: Time constraint type.

start_task: ActionTask: First task in the sequence that the constraint applies to.

end_task: ActionTask: Last task in the sequence that the constraint applies to.

min_duration: int | None: Minimum duration allowed for the sequence.

max_duration: int | None: Maximum duration allowed for the sequence.

enum linq.workflow.TimeConstraintType(value)

Member Type:: str

Valid values are as follows:

START_TO_END = <TimeConstraintType.START_TO_END: 'start_to_end'>

START_TO_START = <TimeConstraintType.START_TO_START: 'start_to_start'>

END_TO_END = <TimeConstraintType.END_TO_END: 'end_to_end'>

END_TO_START = <TimeConstraintType.END_TO_START: 'end_to_start'>

CONSISTENCY = <TimeConstraintType.CONSISTENCY: 'consistency'>

class linq.workflow.InstrumentBlock(*, start_task: ActionTask, end_task: ActionTask): An instrument block instruction, keeping the instrument from being used between two tasks.