Conditionals

We are taking a plate from a stacker and randomly deciding with a dice roll and a coin file whether it goes to the bin, stays where is is or makes it through to a plate hotel.

from linq.task import (
    ActionTask,
    CodeTask,
    ConditionalSourceTask,
    If,
    LabwareSource,
    LuaFunction,
)

HEAD = 1
TAILS = 0

plate_type = labware.LabwareType(id="microplate_96")

plate = labware.Labware(
    id="plate_1",
    labware_type=plate_type,
    starting_location=LabwareLocation(
        instrument=stacker,
        slot=1
    ),
    batch=1,
)

workflow = Workflow(
    workcell=conditional_workcell,
    name="Conditional Workflow",
    author="Automata",
    description="Conditional Workflow",
    version="0.1",
    scheduler_version=get_latest_scheduler_version(),
    labware_types=[plate_type],
    labware=[plate],
)

Single conditional

workflow.If(roll_the_dice.out().equals_any([1, 6]), id="dice-roll-conditional").Then(
    # Move to hotel
    move_plate_dice_roll := ActionTask(
        id="move-plate-to-hotel-on-dice-roll",
        description="Dice is 1 or 6, the plate survives and moves to the hotel.",
        time_estimate=30,
        instrument_type=hotel.type,
        labware_sources=[LabwareSource(labware=plate)],
    ),
).ElseIf(roll_the_dice.out() == 2).Then(
    # Bin it
    ActionTask(
        id="bin-plate",
        description="Dice is 2, the plate does not make it.",
        time_estimate=30,
        instrument_type=bin1.type,
        labware_sources=[LabwareSource(labware=plate)],
    ),
).Else(
    # Leave it where it is
    leave_the_plate_on_stacker := ActionTask(
        id="leave-plate-on-stacker",
        description="Dice is not 1, 2 or 6, the plate stays where it is.",
        time_estimate=30,
        instrument_type=stacker.type,
        action="do_nothing",
        labware_sources=[LabwareSource(labware=plate)],
    )
)

Nested conditional

workflow.If(roll_the_dice.out().equals_any([1, 6]), id="dice-roll-conditional").Then(
    move_plate_dice_roll := ActionTask(
        id="move-plate-to-hotel-on-dice-roll",
        description="Dice is 1 or 6, the plate survives.",
        time_estimate=30,
        instrument_type=hotel.type,
        labware_sources=[LabwareSource(labware=plate)],
    ),
).Else(
    If(flip_the_coin.out() == HEADS, id="flip-coin-conditional")
    .Then(
        move_plate_coin_flip := ActionTask(
            id="move-plate-to-hotel-on-coin-flip",
            description="Dice is not 1 or 6 but Coin is heads, the plate survives.",
            time_estimate=30,
            instrument_type=hotel.type,
            labware_sources=[LabwareSource(labware=plate)],
        ),
    )
    .Else(
        ActionTask(
            id="bin-plate",
            description="Dice is not 1 or 6, Coin is tails, the plate does not make it.",
            time_estimate=30,
            instrument_type=bin1.type,
            labware_sources=[LabwareSource(labware=plate)],
        ),
    ),
    # NOTE: We need a do nothing task here to ensure that the labware source is defined for
    # the final task store-in-hotel which exists outside the If block
    do_nothing := ActionTask(
        id="do-nothing-after-move_plate_coin_flip",
        description="Do nothing task is required to give tasks outside the If block a labware source",
        time_estimate=1,
        instrument_type=hotel.type,
        action="do_nothing",
        labware_sources=[LabwareSource(labware=plate, source_task=move_plate_coin_flip)],
    ),
)

Using ConditionalSourceTask for Downstream Tasks

When multiple conditional branches produce labware that needs to be used by a downstream task, use ConditionalSourceTask to specify all possible labware sources. This is necessary because at planning time, the actual outcome of the conditional is not known, so you must provide all options that may evaluate to true.

# After the nested conditional above, we want a final task that uses the plate
# regardless of which branch was taken (hotel, bin, or stacker)
# ConditionalSourceTask allows us to have one final task that accepts labware from any branch

final_plate_processing = ActionTask(
    id="final-plate-processing",
    description="Final processing for plate, regardless of routing path",
    time_estimate=60,
    instrument_type=workstation.type,
    labware_sources=[
        LabwareSource(
            labware=plate,
            # Specify all possible source tasks from conditional branches
            # The planner will account for all possibilities during planning
            source_task=ConditionalSourceTask(
                tasks=[
                    move_plate_dice_roll,          # If branch: dice roll is 1 or 6
                    move_plate_coin_flip,          # Nested Else->Then: coin flip is heads
                    leave_the_plate_on_stacker,    # Nested Else->Else: all other cases
                ]
            ),
        )
    ],
)

# Add the final task to the workflow
workflow.add_task(final_plate_processing)

Key Points about ConditionalSourceTask

Multiple conditional branches: When a downstream task could receive labware from multiple conditional branches, use ConditionalSourceTask to reference all possible source tasks
Planning constraint: The planner accounts for all possibilities and generates a plan that is valid regardless of which branch executes at runtime
Runtime execution: At execution time, only the actual labware from the branch that executed will be used as input to the downstream task
Workflow simplification: Eliminates the need for duplicate downstream tasks for each conditional branch or complex intermediate routing logic