Example Workflows

In this page you can see a simple example of a workflow and that same workflow with additional more complex tasks and feature usage. The example workcell is at the bottom of the page.

Simple example

This simple workflow demonstrates an end to end workflow where a plate is picked up from a stacker, peeled, placed in an incubator, removed from the incubator, sealed and placed back in the stacker.

from examples.docs.example_workcell import incubator, peeler, sealer, stacker, workcell
from linq.task import ActionTask, Inputs, LabwareOutput, LabwareSource, StoredLabware
from linq.labware import LabwareLocation
from linq.utils import get_latest_scheduler_version
from linq.workflow import Labware, LabwareType, Workflow

## Labware
microplate_type = LabwareType(id="microplate_96")

microplate = Labware(
    id="plate",
    labware_type=microplate_type,
    starting_location=LabwareLocation(
        instrument=stacker,
        slot=1,
    ),
    batch=1,
)


peel_plate = ActionTask(
    id="peel_plate",
    description="peel the plate",
    action="peel",
    instrument_type=peeler.type,
    labware_sources=[LabwareSource(labware=microplate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=22,
)

load_plate_for_incubation = ActionTask(
    id="load_incubation",
    description="Load plate into Liconic STX",
    action="load",
    static_arguments={"slot": 1},
    labware_sources=[LabwareSource(labware=microplate, destination_slot=0, source_task=peel_plate)],
    labware_outputs=[
        LabwareOutput(
            labware=microplate,
            slot=1,
        )
    ],
    instrument_type=incubator.type,
    time_estimate=35,
)

unload_plate = ActionTask(
    id="unload_incubation",
    description="Unload plate from Liconic STX",
    action="load",
    static_arguments={"slot": 1},
    labware_sources=[StoredLabware(labware=microplate, slot=1)],
    labware_outputs=[
        LabwareOutput(
            labware=microplate,
            slot=0,
        )
    ],
    instrument_type=incubator.type,
    time_estimate=35,
    dependencies=[
        load_plate_for_incubation
    ],  # This one needs an explicit dependency because it doesnt use source task.
)

seal_plate = ActionTask(
    id="seal_plate",
    description="seal the plate",
    action="seal",
    instrument_type=sealer.type,
    labware_sources=[LabwareSource(labware=microplate, source_task=unload_plate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=27,
    inputs=Inputs(wait_for_temperature=False),
)

return_to_stacker = ActionTask(
    id="return_to_stacker",
    description="put plate back on stacker",
    action="do_nothing",
    instrument_type=stacker.type,
    labware_sources=[LabwareSource(labware=microplate, source_task=seal_plate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=1,
)

workflow = Workflow(
    workcell=workcell,
    name="simple_example_workflow",
    author="John Doe",
    version="v0.1",
    description="Simple workflow",
    scheduler_version=get_latest_scheduler_version(),
    labware_types=[microplate_type],
    labware=[microplate],
    tasks=[
        peel_plate,
        load_plate_for_incubation,
        unload_plate,
        seal_plate,
        return_to_stacker,
    ],
    parameter_definitions=[],
)

Complex example

This simple workflow demonstrates a number of features, including a liquid handling step, and HTTP task, an operator pause, and a hook.

from examples.docs.example_workcell import http, incubator, liquid_handler, operator, peeler, sealer, stacker, workcell
from linq.hooks import HookParameters, TaskStateChangeHook
from linq.task import ActionTask, Inputs, LabwareOutput, LabwareSource, StoredLabware
from linq.labware import LabwareLocation
from linq.utils import get_latest_scheduler_version
from linq.workflow import (
    Labware,
    LabwareType,
    TimeConstraint,
    TimeConstraintType,
    Workflow,
)

## Labware
microplate_type = LabwareType(id="microplate_96")
reservoir_type = LabwareType(id="microplate_96")

microplate = Labware(
    id="plate",
    labware_type=microplate_type,
    starting_location=LabwareLocation(
        instrument=stacker,
        slot=1,
    ),
    batch=1,
)

reservoir = Labware(
    id="reservoir",
    labware_type=reservoir_type,
    starting_location=LabwareLocation(
        instrument=stacker,
        slot=2,
    ),
    batch=1,
)


peel_plate = ActionTask(
    id="peel_plate",
    description="peel the plate",
    action="peel",
    instrument_type=peeler.type,
    labware_sources=[LabwareSource(labware=microplate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=22,
)

liquid_handler = ActionTask(
    id="liquid_handler",
    description="Liquid Handler",
    action="run",
    instrument_type=liquid_handler.type,
    inputs=Inputs(protocol_name="protocol.json"),
    labware_sources=[
        LabwareSource(labware=microplate, source_task=peel_plate, destination_slot=1),
        LabwareSource(labware=reservoir, destination_slot=2),
    ],
    labware_outputs=[
        LabwareOutput(labware=microplate, slot=1),
        LabwareOutput(labware=reservoir, slot=2),
    ],
    time_estimate=300,
)

load_plate_for_incubation = ActionTask(
    id="load_incubation",
    description="Load plate into Liconic STX",
    action="load",
    inputs=Inputs(slot=1),
    labware_sources=[LabwareSource(labware=microplate, destination_slot=0, source_task=liquid_handler)],
    labware_outputs=[
        LabwareOutput(
            labware=microplate,
            slot=1,
        )
    ],
    instrument_type=incubator.type,
    time_estimate=35,
)

unload_plate = ActionTask(
    id="unload_incubation",
    description="Unload plate from Liconic STX",
    action="load",
    inputs=Inputs(slot=1),
    labware_sources=[StoredLabware(labware=microplate, slot=1)],
    labware_outputs=[
        LabwareOutput(
            labware=microplate,
            slot=0,
        )
    ],
    instrument_type=incubator.type,
    time_estimate=35,
    dependencies=[
        load_plate_for_incubation
    ],  # This one needs an explicit dependency because it doesnt use source task.
)

http_post_barcode = ActionTask(
    id="http_task",
    description="send barcode in post request",
    action="post",
    instrument_type="http",
    inputs=Inputs(base_request={"url_path": "/url_path_for_http", "additional_headers": []}),
    shared_arguments={"body": "datastore_key"},
    dependencies=[unload_plate],
    labware_sources=[LabwareSource(labware=microplate, source_task=unload_plate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=5,
)

seal_plate = ActionTask(
    id="seal_plate",
    description="seal the plate",
    action="seal",
    instrument_type=sealer.type,
    labware_sources=[LabwareSource(labware=microplate, source_task=http_post_barcode)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=27,
    inputs=Inputs(wait_for_temperature=False),
)

operator_pause = ActionTask(
    id="operator_pause",
    description="Pause for operator to confirm plate can be returned to stacker",
    action="pause",
    instrument_type=operator.type,
    inputs=Inputs(action_required=True, message="Confirm workflow is ok to continue"),
    labware_sources=[LabwareSource(labware=microplate, source_task=seal_plate)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=120,
)


return_to_stacker = ActionTask(
    id="return_to_stacker",
    description="put plate back on stacker",
    action="do_nothing",
    instrument_type=stacker.type,
    labware_sources=[LabwareSource(labware=microplate, source_task=operator_pause)],
    labware_outputs=[LabwareOutput(labware=microplate)],
    time_estimate=1,
)

# Defining the time constraint for incubation
plate_incubation = TimeConstraint(
    id="plate_incubation",
    constraint_type=TimeConstraintType(value="end_to_start"),
    start_task=load_plate_for_incubation,
    end_task=unload_plate,
    min_duration=600,
    max_duration=900,
)

# Defining Hooks
webhook_configuration = HookParameters(url="https://webhook.site/0000")
operator_task_hook = TaskStateChangeHook(
    parameters=webhook_configuration,
    task_ids=["operator_pause"],
)

# Now we have everything we need to create our workflow
workflow = Workflow(
    workcell=workcell,
    name="quickstart 4",
    author="John Doe",
    version="v0.1",
    description="Http",
    scheduler_version=get_latest_scheduler_version(),
    labware_types=[microplate_type],
    labware=[microplate, reservoir],
    tasks=[
        peel_plate,
        liquid_handler,
        load_plate_for_incubation,
        unload_plate,
        http_post_barcode,
        seal_plate,
        operator_pause,
        return_to_stacker,
    ],
    time_constraints=[plate_incubation],
    hooks=[operator_task_hook],
    parameter_definitions=[],
)

Example Workcell

This is the workcell used for the example workflows.

from linq.workcell import Instrument, TransportMatrix, TransportPath
from linq.workflow import Workcell

LIQUID_HANDLER = "liquid_handler"
INCUBATOR_TYPE = "incubator"
SEALER_TYPE = "sealer"
PEELER_TYPE = "peeler"
STACKER_TYPE = "stacker"
TRANSPORT_TYPE = "transport"

stacker = Instrument(
    id="stacker_1",
    name="stacker_1",
    type=STACKER_TYPE,
    driver="automata_passive",
    bench="robot1",
    capacity=10,
)

sealer = Instrument(
    id="kbio_sealer",
    name="kbio_sealer",
    type=SEALER_TYPE,
    capacity=1,
    driver="kbio_ultraseal_epro",
    bench="robot2",
    config={
        "serial_port": "/dev/serial/by-id/usb-FTDI_US232R_FT7GEVEI-if00-port0",
        "temperature": 165,
        "set_time": 2.0,
    },
)

peeler = Instrument(
    id="xpeel_peeler",
    name="xpeel_peeler",
    type=PEELER_TYPE,
    driver="brooks_xpeel",
    bench="robot1",
    config={"serial_port": "/dev/serial/by-id/usb-FTDI_UT232R_FT748EJM-if00-port0"},
)

incubator = Instrument(
    id="liconic_stx",
    name="liconic_stx",
    type=INCUBATOR_TYPE,
    bench="robot2",
    driver="liconic_stx220_v1",
    capacity=110,
    config={"serial_port": "/dev/serial/by-id/usb-XXXX-port0"},
)

liquid_handler = Instrument(
    id="opentron_flex_pre",
    name="opentron_flex_pre",
    type=LIQUID_HANDLER,
    bench="robot1",
    driver="opentrons_flex_v1",
    capacity=15,
    config={"robot_ip": "10.10.10.10"},
)

transport_instrument = Instrument(
    id="transport",
    name="transport",
    type=TRANSPORT_TYPE,
    driver="automata_transport",
    bench="na",
)

operator = Instrument(
    id="operator",
    name="Operator Interactions",
    type="operator",
    driver="automata_operator_v1",
    bench="na",
)

http = Instrument(
    id="http",
    name="HTTP",
    type="http",
    driver="automata_http_v1",
    config={
        "auth_type": 1,
        # Use basic authentication.
        # auth_secret_key will be used to retrieve a username:password string from Secrets Manager
        # the driver will add the Authorization header to the request
        # The IAM role active on the hub must have access to retrieve this secret.
        "auth_secret_key": "/secrets/API_AUTH",
    },
    bench="na",
)

transport_matrix = TransportMatrix(
    default_transport_time=30,
    paths=[
        TransportPath(source="robot1", destination="robot1", time=30),
        TransportPath(source="robot1", destination="robot2", time=60),
        TransportPath(source="robot2", destination="robot1", time=60),
    ],
)


# Now we have everything we need to create our workcell
workcell = Workcell(
    instruments=[
        stacker,
        sealer,
        peeler,
        incubator,
        liquid_handler,
        transport_instrument,
        operator,
        http,
    ],
    transport_matrix=transport_matrix,
)