9  Worklists and data formats

Worklists are lists of source and destination pairs for liquid containers on a robot deck. They are one of the most common ways to specify how liquids are transfered in a script.

import csv, itertools, pathlib
from operator import itemgetter

from pylabrobot.liquid_handling import LiquidHandler
from pylabrobot.liquid_handling.backends import LiquidHandlerChatterboxBackend
from pylabrobot.resources import (
    STARLetDeck, PLT_CAR_L5AC_A00, cor_96_wellplate_360uL_Fb,
    opentrons_96_filtertiprack_200ul, TIP_CAR_288_C00,
)

lh = LiquidHandler(backend=LiquidHandlerChatterboxBackend(), deck=STARLetDeck())
await lh.setup()

carrier = PLT_CAR_L5AC_A00(name="carrier")
lh.deck.assign_child_resource(carrier, rails=10)
carrier[0] = source = cor_96_wellplate_360uL_Fb(name="source")
carrier[1] = target = cor_96_wellplate_360uL_Fb(name="target")

tips = TIP_CAR_288_C00(name="tip_carrier")
lh.deck.assign_child_resource(tips, rails=2)
tips[0] = tip_rack = opentrons_96_filtertiprack_200ul(name="tips")

Paths below are relative to the working directory, with this chapter’s inputs under data/. A real run puts inputs and outputs in one timestamped run directory — chapter 13 makes that the rule.

Python you need here

csv, pathlib, and itertools are the imports of this chapter. A worklist is plain data: the csv module reads and writes it, pathlib.Path handles paths without string surgery, and itertools.groupby (with operator.itemgetter as its key) collapses repeated rows into one robot move.

9.1 Drive a run from a CSV

Take a spreadsheet of transfers — one row per transfer — and execute it.

The worklist ships with the cookbook at data/worklist.csv; in real life it comes from a spreadsheet, a scheduler, or a LIMS export. Its entire contents:

target,volume
B1,25
B2,25
B3,25
B4,25

A header, then one transfer per row — four rows, so four transfers, all drawing from one source well (more on that below). Reading it and driving the robot from it:

worklist = pathlib.Path("data/worklist.csv")

with open(worklist, newline="") as f:
    rows = list(csv.DictReader(f))
1
The file lives next door to this chapter, under the cookbook’s shared data/ directory.
2
csv.DictReader gives each row as a dict keyed by the header — {"target": "B1", "volume": "25"}. Values are strings, so they are cast where they are used.

All four transfers draw from one source well, source["A1"], which changes how the run should be driven. One tip per transfer works, but wastes the head: with tips on channels 0–3 you can pick up once, aspirate the four volumes in series off the single source well, and dispense them in parallel, one channel per target:

n = len(rows)
assert n <= 8
tip_spots = tip_rack.row("A")[:n]
wells = [target[row["target"]][0] for row in rows]
vols = [float(row["volume"]) for row in rows]

await lh.pick_up_tips(tip_spots)
for i, vol in enumerate(vols):
    await lh.aspirate(source["A1"], vols=[vol], use_channels=[i])
await lh.dispense(wells, vols=vols)
await lh.drop_tips(tip_spots)
1
Hard cap: one tip per channel on an 8-channel head. Longer worklists get chunked — see #batching below.
2
row("A") is the rack’s top row as a list of spots (chapter 5); the slice takes the first n. Building the identifiers instead — [tip_rack[f"A{i + 1}"][0] for i in range(n)] — says the same thing in more characters and stops being true at A13.
3
One pickup for the whole batch, not one per transfer.
4
Four aspirations out of A1, in series — four channels cannot occupy one well at once. aspirate takes a list of wells (source["A1"] is the one-element list holding A1) and use_channels=[i] pins each draw to the tip picked up on channel i.
5
The four dispenses happen in parallel, one channel per target well.

pandas is not required. The stdlib csv module reads a worklist with no third-party dependency, and either way the seam produces the same thing: a list of (target, volume) pairs.

See also: #pairing gives the row→arguments mapping above a single name; #batching reorders rows for multi-channel heads.

9.2 One source per row

Real worklists rarely share one source. Here every row is an independent source→target pair, drawn from a second checked-in file — data/transfers.csv — whose rows are deliberately in no useful order. Its entire contents:

source,target,volume
C7,B1,30
A2,D4,30
B3,A5,30
E8,C2,30

Each row is one independent transfer: aspirate from the source well, dispense into the target well. Nothing about that order — C7’s column, then A2’s, then B3’s, then E8’s — is chosen for the robot’s benefit; it is just the order someone typed into a spreadsheet.

Reading it in is the same two lines as any CSV:

with open(pathlib.Path("data/transfers.csv"), newline="") as f:
    pair_rows = list(csv.DictReader(f))

The naive driver treats each row as its own mini-protocol: pick up a tip, transfer, drop the tip — four times over.

for row in pair_rows:
    await lh.pick_up_tips(tip_rack["A1"])
    await lh.transfer(
        source[row["source"]][0],
        targets=[target[row["target"]][0]],
        target_vols=[float(row["volume"])],
    )
    await lh.drop_tips(tip_rack["A1"])

With distinct sources there is no shared-well serialization to exploit, but the pickups can still be batched: pick up one tip per channel up front and run the whole worklist through one aspirate/dispense pair.

srcs = [source[row["source"]][0] for row in pair_rows]
dsts = [target[row["target"]][0] for row in pair_rows]
vols = [float(row["volume"]) for row in pair_rows]
tips = tip_rack.row("A")[:len(pair_rows)]

await lh.pick_up_tips(tips)
await lh.aspirate(srcs, vols=vols)
await lh.dispense(dsts, vols=vols)
await lh.drop_tips(tips)
1
Same cap as before: at most eight tips, one per channel.
2
One aspiration per row, in file order — the channels cannot share a move because every source is a different well.
3
One dispense per row. Notice the order the head visits wells in: C7, A2, B3, E8 is exactly the file’s order, and it zig-zags across the deck because we never asked it to do better.

That last point matters. The rows arrive in whatever order the spreadsheet was in; run as-is, the head traverses the deck per row. Sort the pairs spatially first — sources by column, targets riding along with their source — and the same aspirate/dispense pair sweeps instead of zig-zags. That is precisely what sort_by_xy_and_chunk_by_x (#batching) automates.

9.3 Aspirate once per source, not once per row

Reagent-addition worklists repeat their sources: a dozen rows, two or three reagents. Driven row by row, each one costs its own trip to the source well. Group the rows by source and each reagent is picked up once.

The file — data/reagents.csv — is six rows drawing on two sources, interleaved the way a spreadsheet accumulates them:

source,target,volume
A1,B1,20
A1,B2,20
A2,C1,15
A1,B3,20
A2,C2,15
A1,B4,20
with open(pathlib.Path("data/reagents.csv"), newline="") as f:
    reagent_rows = list(csv.DictReader(f))

by_source = itemgetter("source")

await lh.pick_up_tips(tip_rack["A1"])
for src, group in itertools.groupby(sorted(reagent_rows, key=by_source), key=by_source):
    group = list(group)
    total = sum(float(row["volume"]) for row in group)
    assert total <= 200

    await lh.aspirate(source[src], vols=[total], use_channels=[0])
    for row in group:
        await lh.dispense(target[row["target"]], vols=[float(row["volume"])], use_channels=[0])
await lh.drop_tips(tip_rack["A1"])
1
operator.itemgetter("source") is lambda row: row["source"] without the lambda — one key function, used twice, for the sort and for the grouping. Writing it once is what keeps the two in step: groupby batches only consecutive equal keys, so an unsorted groupby on this file yields three groups (A1, A2, A1) and quietly undoes the saving.
2
groupby hands out a shared iterator that is consumed when the next group is pulled. list() it before touching the rows twice — here, once for total and once for the dispenses.
3
The cap is the tip, not the worklist: 200 µL filter tips hold 200 µL. Longer runs of the same source split into tip-sized chunks.
4
Six rows, two aspirations. The row-by-row driver in #per-row-sources would make six, plus six returns to the source well.

Gotcha: the last dispense out of a shared aspiration is the least accurate.

One draw serving four dispenses means the fourth is pushed out of a nearly empty tip. Standard practice is to aspirate a little extra and discard the remainder, which is why this pattern belongs to reagent addition — where a percent matters little — and not to a serial dilution, where it matters a lot.

See also: #batching groups by position to shorten the head’s travel; this recipe groups by source to cut the number of aspirations. They compose — sort within a group.

9.4 Turn rows into call arguments

aspirate and dispense want parallel lists — wells here, volumes there. A worklist is the transpose of that: one row per transfer, all three fields together. Name the row→arguments mapping once, then turn the rows a quarter turn:

def as_call(row):
    """One worklist row -> (source well, target well, volume)."""
    return source[row["source"]][0], target[row["target"]][0], float(row["volume"])

srcs, dsts, vols = zip(*map(as_call, pair_rows))
1
The mapping from a row’s strings to the objects a call takes, stated once and given a name. The three separate comprehensions in #per-row-sources spread this same mapping across three lines and repeat the row[...] scaffolding on each; when a column is added or a cast changes, this is one edit instead of three that must stay in step.
2
zip(*rows) transposes: rows in, columns out. resources is typed Sequence, so the tuples zip returns go straight into a call — vols is annotated List and works the same, list(vols) if a type checker objects.

Both spellings walk the rows; the win is not speed on four rows, it is that the row shape is written down in one place.

The reverse direction has a failure mode. Whenever wells and volumes come from different places — two files, or one column that was filtered — zip silently stops at the shorter one and you get fewer operations than the worklist asked for, with no error:

rows = [{"target": "B1", "volume": "30"}, {"target": "B2", "volume": ""},
        {"target": "B3", "volume": "30"}]

wells = [target[row["target"]][0] for row in rows]
vols = [float(row["volume"]) for row in rows if row["volume"]]

print("plain zip:", len(list(zip(wells, vols))), "transfers from", len(rows), "rows")
try:
    list(zip(wells, vols, strict=True))
except ValueError as e:
    print("strict=True:", e)
2
A blank volume — the everyday form of a bad worklist row.
3
The filter drops the blank row from vols but not from wells. The lists are now ragged.
4
strict=True raises instead of truncating. Two transfers instead of three is a plate that looks finished and is not; the exception arrives before a channel moves, which is the whole point of catching it here rather than in chapter 10. 
plain zip: 2 transfers from 3 rows
strict=True: zip() argument 2 is shorter than argument 1

Where the index still matters. enumerate earns its keep in exactly one place in this chapter: #csv-run, where i becomes use_channels=[i] and pins a serialized draw to its own tip. When a call takes whole lists there is no index to carry, and the loop disappears with it. Past eight rows, the wrap-around is i % 8 — the periodicity arithmetic from chapter 5, which itertools.cycle and itertools.islice (#streaming) already hide.

See also: zip(*...) also inverts — list(zip(wells, vols)) rebuilds rows for writing a result CSV back out.

9.5 Compute volumes from a plate map

The volumes a run needs are rarely in the worklist. They are computed from something else — a quantification, a normalization target, a dilution factor — held in a second file keyed by well. data/plate_map.csv is that file, one row per sample:

well,sample_id,conc
B1,S001,45.0
D4,S002,20.0
A5,S003,80.0
C2,S004,31.5

The job: for every row of transfers.csv, look up the target well’s concentration and work out how much buffer dilutes 20 µL of it to 20 ng/µL. Written as the loop it looks like:

with open(pathlib.Path("data/plate_map.csv"), newline="") as f:
    plate_map = list(csv.DictReader(f))

TARGET, V_SAMPLE = 20.0, 20.0

vols = []
for row in pair_rows:
    for entry in plate_map:
        if entry["well"] == row["target"]:
            conc = float(entry["conc"])
            break
    vols.append(round(V_SAMPLE * (conc / TARGET - 1), 1))
1
The lookup is re-derived inside the driver loop, so the two jobs — find the concentration and apply the rule — are tangled together in one block.
2
append in a loop: three lines of bookkeeping around one line of arithmetic, and conc leaks out of the inner loop, so a target missing from the map silently reuses the previous row’s concentration instead of failing. That is a wrong volume in a well, and nothing says so.

That loop does collapse into a single comprehension — next() takes the first match, which is all the break was doing:

vols = [round(V_SAMPLE * (float(next(e["conc"] for e in plate_map
                                     if e["well"] == row["target"])) / TARGET - 1), 1)
        for row in pair_rows]
print(vols)
[25.0, 0.0, 60.0, 11.5]

It is worth writing out precisely because it is the wrong answer. It is four lines shorter, and it even fixes the leak — next raises StopIteration on a well the map does not have, instead of reusing the last row’s concentration. But the volume rule, the only part a reader came for, is now buried four parentheses deep behind a lookup that is not the point.

Shorter is not the same as clearer, and neither is what makes a run faster. All three versions on this page issue exactly the same moves; the robot cannot tell them apart, and no worklist a bench produces is large enough for the arithmetic to cost anything measurable. Pipetting steps are the budget in this chapter — #group-sources spends fewer of them, and so does the filter below. What the comprehension buys is a reader who can see the rule, and a failure that is loud. Separate the lookup from the rule with a dict comprehension, and the list comprehension is left saying only the rule:

conc = {row["well"]: float(row["conc"]) for row in plate_map}

vols = [round(V_SAMPLE * (conc[row["target"]] / TARGET - 1), 1)
        for row in pair_rows]

print(dict(zip((row["target"] for row in pair_rows), vols)))
1
The plate map, turned into the thing it is used as: a table from well to concentration, named conc and built once. A dict comprehension is the one-line spelling of that; the reader sees what the map is before seeing what is done with it.
2
The rule reads as the rule: this much buffer, for every row. conc[...] raises KeyError on a target the map does not know — the loud failure the break version turned into a silent one. 
{'B1': 25.0, 'D4': 0.0, 'A5': 60.0, 'C2': 11.5}

A comprehension filters in the same breath — and this is where the run actually gets shorter. Sample D4 is already at target, so its buffer volume is zero. Left in, it costs a tip, an aspiration, and a dispense to add nothing to a well:

plan = [(target[row["target"]][0], vol)
        for row, vol in zip(pair_rows, vols, strict=True) if vol > 0]

wells, buffer_vols = zip(*plan)

spots = tip_rack.row("B")[:len(wells)]
await lh.pick_up_tips(spots)
for i, vol in enumerate(buffer_vols):
    await lh.aspirate(source["A1"], vols=[vol], use_channels=[i])
await lh.dispense(wells, vols=list(buffer_vols))
await lh.drop_tips(spots)
3
The if clause drops the zero-volume row before it becomes a channel: three tips instead of four, three draws instead of four. That is the saving worth counting — one fewer physical operation, decided in data before anything moves. Filtering after the fact — building all four operations and then removing one — means keeping two lists in step by index, which is the bug strict=True exists to catch.
4
The same transpose as #pairing: pairs in, parallel lists out.
5
Buffer comes from one shared well, so the draws serialize and the index becomes use_channels — the #csv-run pattern, and the one place an index still earns its keep. Three channels cannot share a well: [source["A1"][0]] * 3 in a single call raises ValueError: Resource is too small to space channels.

Where a comprehension is the wrong tool. When the body does I/O or moves a channel — await, a db.execute, a log.info — it is a loop, and writing it as a comprehension for its own sake buys nothing and hides the effect. Comprehensions are for turning data into other data; the awaits above are deliberately still a loop.

See also: the same dict-as-index trick joins a run’s results back onto its worklist for chapter 13’s run record.

9.6 Group rows into channel-parallel batches

You have an 8-channel head. Turn a one-row-per-well worklist into batches your channels can do in parallel.

This is the tool the unsorted per-row worklist above was pointing at: its rows zig-zag because the file order is arbitrary. Sort spatially and the head sweeps column by column.

from pylabrobot.resources.utils import sort_by_xy_and_chunk_by_x

targets = [target[row["target"]][0] for row in reagent_rows]

batches = sort_by_xy_and_chunk_by_x(targets, max_chunk_size=8)
for batch in batches:
    print([w.name.split("well_")[1] for w in batch])
1
The reagent worklist’s targets, in file order: B1, B2, C1, B3, C2, B4 — columns 1, 2, 1, 3, 2, 4, which is the zig-zag this recipe exists to fix. target[...] returns a list, always — [0] unwraps the single well. Feed the raw lists to the chunker and it sorts lists instead of wells, which fails on the first coordinate lookup.
2
sort_by_xy_and_chunk_by_x(resources, max_chunk_size) sorts by x, then groups resources with the same x into chunks no larger than max_chunk_size. New in 0.2.2, purpose-built for this exact problem.
3
Each chunk shares a column — an 8-channel head fills it in one pass. 
['B3']
['B4']
['B1', 'C1']
['B2', 'C2']

Gotcha: chunk size is about channels, not wells.

max_chunk_size=8 is the number of channels on the head. It does not cap the total worklist — long lists split into many batches of eight. Pick it from your hardware (lh.backend.get_channel_spacings in chapter 4), not from the file length.

See also: sort_by_xy_and_chunk_by_x is the sibling of the query helpers from chapter 2, all in resources/utils.py.

9.7 Pick a format

Match the file format to the kind of data.

Job Format Why
Worklists (what to do) CSV one row per transfer; anyone can open, edit, diff, and export it
State (what is where) JSON nested structure; it is PLR’s own serialization format (chapter 11)
Audit (what happened) Log lines append-only, timestamped, greppable; a CSV you re-open is not a log

CSV — the worklist:

target,volume
B1,25
B2,25

JSON — the state (a slice of PLR’s serialize_state):

{"name": "target_well_A1", "volume": 25.0, "children": []}

Log lines — the audit (a slice of PLR’s _log_command output):

2026-08-12 14:03:11 DEBUG aspirate(resources=[<Well A1>], vols=[25.0], ...)

PLR gives you the audit for free.

Every operation is already logged through logging (chapter 13). A run log is three lines of basicConfig, and is greppable and diffable.

See also: sort_by_xy_and_chunk_by_x for the same data in a different shape; the format rule is restated at system level in chapter 13.

9.8 What to remember

  • PLR ships no worklist support. The seam is plain csv + pathlib + lh.transfer.
  • csv.DictReader gives rows as dicts; values are strings, cast before arithmetic.
  • zip(*rows) transposes a worklist into the parallel lists a call takes; write the row→arguments mapping once as a named function. strict=True raises on a ragged worklist before a channel moves.
  • groupby on the source well turns one aspiration per row into one per reagent. Sort by the key first, or the grouping silently does nothing.
  • A dict comprehension is how two files get joined: {well: value} built once and named, so the driver states the rule instead of re-deriving the lookup inside it.
  • A list comprehension states a volume rule as one expression, and its if clause drops the rows that should never become channels. When the body awaits, logs, or writes, keep it a loop.
  • A one-liner is not automatically a win. The nested lookup collapses into one comprehension and becomes harder to read, not better. Shorter is a legibility argument; faster means fewer pipetting steps, which is won by dropping operations — the zero-volume filter, groupby on the source — not by rearranging Python.
  • enumerate is for the case where the index is an argument (use_channels=[i]); when the call takes whole lists, the loop goes away.
  • sort_by_xy_and_chunk_by_x(targets, max_chunk_size) turns wells into channel-parallel batches.
  • CSV for worklists, JSON for state, log lines for audit. Three jobs, three formats.