Time and Sequence

Overview

This section covers the ordering and timing of tasks within a workflow. It explains methods for controlling the sequence and constraining task execution to control exactly when tasks are completed.

Time Constraints

Time constraints define how two tasks are related in terms of their start and end times. They control task sequencing and timing to maintain order in workflows. These constraints are used by the planner when planning occurs. See planning and validation.

Available Constraint Types

START_TO_END: The first task starts before the second finishes.
START_TO_START: The second task starts after the first starts.
END_TO_END: The second task finishes after the first.
END_TO_START: The second task starts after the first finishes.
CONSISTENCY: Ensures timing consistency between two tasks.

plate_incubation = ActionTask(...)
dispense_reagent = ActionTask(...)

# Defining the time constraint for incubation
plate_incubation = TimeConstraint(
    id="incubate-to-dispense-constraint",
    constraint_type=TimeConstraintType(value="end_to_start"),
    start_task=plate_incubation,
    end_task=dispense_reagent,
    min_duration=600,
    max_duration=900
)

Time Constraints are added to a Workflow object. See Workflow.

In the event a time constraint is broken (for example due to an error or a large difference between estimated and real time) an error will be raised with mitigation options. See Error Handling.

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.

Max Idle Time

Max idle time sets the maximum duration a labware type can be left unattended during a workflow. This is important for labware containing sensitive materials like cells or samples that require timely handling to maintain their integrity.

To define max idle time, use the max_idle_time parameter in your labware type definition:

assay_plate = LabwareType(
    id="nest_flat_bottom_96_well_plate",
    max_idle_time=30  # Max idle time in seconds
)

Instrument Blocks

Instrument blocks prevent any task from executing on a specific instrument from a specified task until another specified task. This is useful when an instrument’s state is temporarily modified (e.g. holding a cap) meaning it is not strictly being used for a task, but is otherwise occupied.

A typical use case is an uncapping task which requires an instrument to hold a cap until the next task is completed. The specific uncapper can be blocked from use until a task completes.

Instrument blocks are added to a Workflow object. See Workflow.

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

Dependencies

Task dependencies ensure that certain tasks are completed before others can begin, maintaining proper task order. The first task of a chain of tasks will not have any dependencies. Any kind of task can have a dependency or be the target of a dependency.

Labware input and output creates labware dependencies automatically. In other words, if a given task has specified a previous task as containing labware that it needs as input, a dependency does not need to be specified. See Labware Flow.

Additional dependencies can be manually configured as follows:

dispense_reagent = ActionTask(...)

load_plate = ActionTask(
    id="load_plate",
    # ...
    dependencies=[dispense_reagent],
)

You can have multiple dependencies for the same task. Note that any type of task, e.g. a CodeTask or an ActionTask, can be specified in the dependency array or made dependent on any other task.

dependencies=[dispense_reagent, move_data],

Time estimates

Time estimates approximate how long each task will take, used by the planner to optimise task scheduling and predict workflow duration.

If the actual time is different from the estimated time for a task when executing on hardware, a replan may be triggered. See Planning and Validation.

dispense_reagent = ActionTask(...)

load_plate = ActionTask(
    id="load_plate",
    # ...
    time_estimate=20,
)

Transport Matrix and Robots

In workflows, transport actions are implicit, meaning they happen automatically between tasks (see Tasks and Transport. The transport matrix allows you to configure estimated timings for movements between workbenches and robots. These timings are used during planning to calculate transport tasks when the workflow is executed.

The default_transport_time serves as the fallback time for any unspecified transport paths. However, you can customize the timing between specific robots by defining transport paths. This lets you control how long it takes to move between robots while accounting for transport layer movements, ensuring more accurate planning.

workcell = Workcell(

    # ...

    transport_matrix=TransportMatrix(
            default_transport_time=20,
            paths=[
                TransportPath(source="robot_1", destination="robot_1", time=55),
                TransportPath(source="robot_2", destination="robot_2", time=20),
                TransportPath(source="robot_1", destination="robot_2", time=90),
                TransportPath(source="robot_2", destination="robot_1", time=100),
            ],
        ),
    )

class linq.workcell.TransportMatrix(*, default_transport_time: int, paths: list[TransportPath])

The transport matrix contains all estimates for transport times between benches.

default_transport_time: int: Default transport time if no specific estimate for a path is given.

paths: list[TransportPath]: List of paths describing the time to transport labware from one bench to another.