Set up lineage for liquid handling

The Liquid Handling SDK records lineage from your deck setup and the actions you call. This page describes what to set up in your workflow so the events carry names, volumes and barcodes, and how to check them before you deploy.

Prerequisites

• unitelabs-liquid-handling ≥ 0.42.0
• UniteLabs SDK ≥ 0.20.0

Run inside a workflow

The library records into the active run context. Code inside @workflow, @phase, and @step functions has one by default. The SDK sends the events and the workspace when each phase exits and when the workflow exits.

Outside a workflow, nothing is recorded. To see the events in a script or notebook, use Audit.capture (see Check the events locally). It keeps the events in memory and sends nothing to the Platform.

Give the liquid handler a name

Events carry the name of the liquid handler as actor. It is part of the ID of the channels, the deck, the loading tray, the park sites and the wastes. The actor is the name you register the liquid handler under in ctx.devices. Without a registration, it is the liquid handler's identifier.

To use a different name, register the liquid handler before its first operation, for example ctx.devices["star"] = lh. The name is fixed at the first event of the run, and a later registration under another name is logged and ignored. Two liquid handlers in one run need different names.

Use unique labware identifiers

Labware IDs come from the identifier you give each item. Give every plate, rack, and carrier a readable identifier that is unique within the run. Two items with the same identifier share one history. The library logs a warning when this happens.

Set barcode on labware that has one. The library adds it to the workspace and to the labware's load and move events, and the Platform's trace search finds it.

Put the starting liquid into the model

The library only knows the liquid your deck setup describes. Add samples and reagents to their wells before the run starts:

for well in sample_plate["A1":"H1"]:
    well.container.add_liquid(Liquid(name="Sample"), 100)
reagent_trough.container.add_liquid(PredefinedLiquids.WATER, 50_000)

Each of these is a seed_container event, and every transfer from that well carries the liquid name in mixture. Wells you don't fill still record every transfer, but their history starts with no volume.

Check the events locally

Audit.capture records events without a workflow. It keeps them in ctx.state["events"] and sends nothing to the Platform. Use it with a mock liquid handler to check what a protocol records before you deploy it:

import asyncio

from unitelabs.labware import PredefinedLiquids, Standard96Plate
from unitelabs.labware.hamilton import PLT_CAR_L5MD_A00, TIP_CAR_480_A00, HamiltonTip_300, HamiltonTipRack_300
from unitelabs.liquid_handling.testing import MicrolabSTARMock
from unitelabs.sdk import Audit


async def main():
    lh = MicrolabSTARMock()
    await lh.configure()
    await lh.initialize()

    tip_carrier = TIP_CAR_480_A00(identifier="tip_carrier")
    tips = HamiltonTipRack_300(filled_with=HamiltonTip_300, identifier="tips")
    tip_carrier[0] = tips
    plate_carrier = PLT_CAR_L5MD_A00(identifier="plate_carrier")
    source = Standard96Plate(identifier="source")
    target = Standard96Plate(identifier="target")
    source["A1"].container.add_liquid(PredefinedLiquids.WATER, 200)
    plate_carrier[0] = source
    plate_carrier[1] = target
    lh.deck.add(tip_carrier, track=1)
    lh.deck.add(plate_carrier, track=10)

    with Audit.capture() as ctx:
        ctx.devices["star"] = lh
        await lh.pipettes.pick_up_tips(channels=[0], spots=[tips["A1"]])
        await lh.pipettes.aspirate([source["A1"]], volume=[50], channels=[0])
        await lh.pipettes.dispense([target["B1"]], volume=[50], channels=[0])
        await lh.pipettes.drop_tips(channels=[0])

    for event in ctx.state["events"]:
        print(event["operation"], event["inputs"].get("entity_id"), "->", event["outputs"].get("entity_id"))

    aspirate = next(event for event in ctx.state["events"] if event["operation"] == "aspirate")
    print(aspirate["version"], aspirate["inputs"], "->", aspirate["outputs"])


asyncio.run(main())

load_labware None -> deck:star:1
load_labware None -> tip_carrier:0
load_labware None -> deck:star:10
load_labware None -> plate_carrier:0
load_labware None -> plate_carrier:1
seed_container None -> source:A1
pick_up_tip tips:A1 -> channel:star:pipettes:0
aspirate source:A1 -> channel:star:pipettes:0
dispense channel:star:pipettes:0 -> target:B1
discard_tip channel:star:pipettes:0 -> waste:star:WasteBlock
2.0.0 {'entity_id': 'source:A1', 'kind': 'well', 'parent': 'source', 'address': 'A1'} -> {'entity_id': 'channel:star:pipettes:0', 'kind': 'channel', 'actor': 'star', 'module': 'pipettes', 'index': 0}

The first operation records the deck: two carriers on their tracks, the rack and plates on them, and the water in source:A1.

Next steps

• Migrate the Liquid Handling SDK: remove your own liquid-handling Audit.emit calls so each action is recorded once
• Emit lineage events: record events for devices the library doesn't drive
• Lineage delivery: choose when the SDK sends events, or fail the run when it can't
• Lineage from liquid handling: see what the library records for each operation