10 Tip management

Setup

import logging

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

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

carrier = TIP_CAR_288_C00(name="tip_carrier")
lh.deck.assign_child_resource(carrier, rails=2)
carrier[0] = rack_a = opentrons_96_filtertiprack_200ul(name="rack_a")
carrier[1] = rack_b = opentrons_96_filtertiprack_200ul(name="rack_b")
carrier[2] = rack_c = opentrons_96_filtertiprack_200ul(name="rack_c")

plt_carrier = PLT_CAR_L5AC_A00(name="plate_carrier")
lh.deck.assign_child_resource(plt_carrier, rails=10)
plt_carrier[0] = plate = cor_96_wellplate_360uL_Fb(name="plate")

set_tip_tracking(True)

logging.getLogger("pylabrobot.resources").setLevel(logging.WARNING)

1: TIP_CAR_288_C00 holds three 96-tip racks — 288 tips, which is where the name comes from.
2: Tip tracking is off by default, like volume tracking (chapter 1). With it off, pick_up_tips never updates the rack model and every spot reports has_tip() == True forever. The recipes below need it on.
3: linear_tip_spot_generator logs one INFO line per spot to pylabrobot.resources (Saved tip idx to disk: 8). Raising the level keeps the streaming recipe’s output readable — in a real run those lines are the audit trail of which tip went where.
4: A plate to pipette into, so the streaming recipe below can run a real pick-up → aspirate → dispense → discard cycle rather than just naming spots.

10.1 Drop, return, or discard

Put a used tip somewhere. There are three calls, and they differ in destination.

for spot in ("A1", "A2", "A3"):
    print(f"{spot}: has_tip = {rack_a.get_item(spot).has_tip()}")

A1: has_tip = True
A2: has_tip = True
A3: has_tip = False

await lh.pick_up_tips(rack_a["A1"])
await lh.return_tips()

await lh.pick_up_tips(rack_a["A2"])
await lh.discard_tips()

await lh.pick_up_tips(rack_a["A3"])
await lh.drop_tips(rack_a["A2"])

1: return_tips() puts each tip back in the spot it came from. It takes no destination — PLR recorded the origin at pick-up.
2: discard_tips() sends tips to the deck’s trash. The rack spot stays empty.
3: drop_tips(spots) takes an explicit destination, which can be any tip spot or a Trash. A2 is free because the previous line discarded its tip.

Call	Destination	`allow_nonzero_volume` default
`drop_tips(tip_spots)`	whatever you pass	`False`
`return_tips()`	the spot each tip came from	`False`
`discard_tips()`	the deck’s trash	`True`

Gotcha: the destination spot must be empty.

With tip tracking on, dropping into a spot that already holds a tip raises from the spot’s tracker:

HasTipError: Tip spot already has a tip.

A rack starts full, so drop_tips(rack["B3"]) fails on a fresh rack — the destination has to be a spot something was taken from, or one marked empty via set_tip_state. This is the tip-spot HasTipError, the suppressible one of the two in chapter 10.

Gotcha: the allow_nonzero_volume default is not the same across the three

drop_tips and return_tips default to False and raise if the tip still holds liquid:

RuntimeError: Cannot drop tip with volume 50.0

discard_tips defaults to True and discards the tip with the liquid still in it. A tip returned to a rack still holding liquid is a contamination path; a discarded one is not.

See also: the same RuntimeError appears in chapter 10 when a dispense fails and leaves the tip loaded; *96 variants (return_tips96, discard_tips96) work the same way for the 96-head.

10.2 Scope tips to a block

Pick up tips and put them away without writing the put-away call.

async with lh.use_tips(rack_b["A1"], channels=[0]):
    mounted_inside = lh.get_mounted_tips()[0] is not None
mounted_after = lh.get_mounted_tips()[0] is not None

async with lh.use_tips(rack_b["A2"], channels=[0], discard=False):
    pass

1: async with, not with. use_tips is decorated @contextlib.asynccontextmanager — unlike use_channels (chapter 6), which is synchronous.

print("mounted inside the block:", mounted_inside)
print("mounted after the block: ", mounted_after)
print("A1 has_tip (discarded):  ", rack_b.get_item("A1").has_tip())
print("A2 has_tip (returned):   ", rack_b.get_item("A2").has_tip())

1: The default is discard=True, so the tip went to the trash and the spot is empty.
2: discard=False returns the tips to their origin spots instead, so the spot is full again.

mounted inside the block: True
mounted after the block:  False
A1 has_tip (discarded):   False
A2 has_tip (returned):    True

Python you need here

Two context manager protocols exist: __enter__/__exit__ for with, and __aenter__/__aexit__ for async with. Using the wrong keyword raises TypeError: ... does not support the asynchronous context manager protocol (or its synchronous counterpart).

10.3 Stream tips across several racks

Run a protocol that needs more tips than one rack holds, without tracking rack boundaries by hand.

spots = F.get_all_tip_spots([rack_a, rack_b, rack_c])
print("total spots:", len(spots))

gen = F.linear_tip_spot_generator(
    spots,
    cache_file_path="tip_index.json",
)

batch = await gen.get(8)
print("batch:", [s.name.split("_")[-1] for s in batch])
print("tips left:", gen.get_num_tips_left())

1: get_all_tip_spots flattens a list of racks into one ordered list of spots.
2: linear_tip_spot_generator walks that list in order. With cache_file_path it writes its position to disk after every spot.
3: get(n) returns the next n spots. The generator is async — get is awaited, and iterating it directly uses async for.

total spots: 288
batch: ['A1', 'B1', 'C1', 'D1', 'E1', 'F1', 'G1', 'H1']
tips left: 280

The generator crosses rack boundaries without being told they exist:

gen2 = F.linear_tip_spot_generator(spots)
gen2.set_index(94)
crossing = await gen2.get(4)
for spot in crossing:
    print(f"  {spot.parent.name}  {spot.name.split('_')[-1]}")

1: set_index jumps to a position — here, near the end of the first 96-spot rack.

  rack_a  G12
  rack_a  H12
  rack_b  A1
  rack_b  B1

Driving a protocol from the generator.

This is the whole point: the loop body never names a rack or a spot, and never counts. Each round asks for eight more spots, and the generator has already moved past them by the time the pick-up runs.

gen3 = F.linear_tip_spot_generator(spots)
gen3.set_index(192)

for _ in range(3):
    fresh = await gen3.get(8)
    await lh.pick_up_tips(fresh)
    await lh.aspirate(plate["A1:H1"], vols=[10.0] * 8)
    await lh.dispense(plate["A2:H2"], vols=[10.0] * 8)
    await lh.discard_tips()

1: Start of rack_c, the one rack the recipes above have not taken tips from — the earlier cells left gaps in rack_a that a fresh generator walking from 0 would trip over with NoTipError, since the generator tracks position, not occupancy. A real run starts at 0 or resumes from its cache file.
2: Three rounds of eight. Twenty-four tips, no bookkeeping.
3: discard_tips() rather than return_tips(): the generator hands out each spot exactly once, so returning a used tip to a spot the generator has already passed would leave it in the rack forever, unreachable and indistinguishable from a fresh one.

print("index now at:  ", len(spots) - gen3.get_num_tips_left(), "(started at 192)")
print("tips left:     ", gen3.get_num_tips_left())
print("mounted after the loop:", [t is not None for t in lh.get_mounted_tips()][:2])

index now at:   216 (started at 192)
tips left:      72
mounted after the loop: [False, False]

Consuming one spot at a time.

The generator is an async iterator, so async for walks it spot by spot — useful when the channel count varies round to round, or when one tip serves one sample.

gen4 = F.linear_tip_spot_generator(spots[:3])
async for spot in gen4:
    print(spot.parent.name, spot.name.split("_")[-1])
print("loop ended cleanly")

1: A three-spot slice, so the loop terminates in the output rather than after 288 lines.
2: The loop ends on StopAsyncIteration, which async for swallows — see the gotcha below.

rack_a A1
rack_a B1
rack_a C1
loop ended cleanly

Gotcha: exhaustion is StopAsyncIteration, not a PLR error.

With repeat=False (the default), asking for a spot past the end raises StopAsyncIteration — Python’s iterator protocol, not one of the portable errors from chapter 10. Inside async for it ends the loop silently; a bare await gen.__anext__() propagates it. get(n) asserts against get_num_tips_left() first, so it raises AssertionError instead.

repeat=True cycles back to the first spot, and makes get_num_tips_left() raise RuntimeError.

Resuming after a crash or a restart.

cache_file_path is what makes the generator survive the process. A new generator built on the same path reads the index back and continues from it — the run that died half-way does not start over on rack A, on top of tips it already used.

resumed = F.linear_tip_spot_generator(
    spots,
    cache_file_path="tip_index.json",
)
print("index restored to:", len(spots) - resumed.get_num_tips_left())

nxt = (await resumed.get(1))[0]
print("next spot:", nxt.parent.name, nxt.name.split("_")[-1])

1: Same spot list, same cache path, brand new object — this is what the second process does.
2: The first cell in this recipe consumed 8 spots through this cache file. A fresh generator with no cache would report 0 here.
3: A2 of rack_a, not A1. gen3 and gen4 above passed no cache_file_path, so they wrote nothing and did not disturb this count.

index restored to: 8
next spot: rack_a A2

Two generators, one cache file

Nothing stops you pointing two live generators at the same path — they will overwrite each other’s index and hand out the same tips twice. One generator per cache file per run.

reset() sets the index back to 0 and writes that to the cache file, which is how you start a fresh run against a rack you have physically reloaded. Note the asymmetry with set_index, which also persists on the next spot handed out — neither of them touches the rack model, so a reset generator will happily hand out spots the trackers still believe are empty.

The randomized variant.

F.randomized_tip_spot_generator(tip_spots, K, cache_file_path=...) picks spots at random while refusing any used in the last K samples. It has the same get(n) interface, the same disk cache, and the same reset(), and raises RuntimeError when every spot is in the recent window.

Traversing one rack, in a direction you choose.

A TipRack is an ItemizedResource, so traverse (chapter 5) works on it exactly as it does on a plate — it is not a plate-only method. Use it when the order matters and one rack is enough; use the generator when you need to cross racks or resume across processes.

for batch in rack_b.traverse(batch_size=8, start="top_left", direction="down"):
    print([s.name.split("_")[-1] for s in batch])
    break

snake = rack_b.traverse(8, "top_left", "snake_down")
next(snake)
print("snake, batch 2:", [s.name.split("_")[-1] for s in next(snake)])

1: Same signature as the plate version: batch_size first, start required.
2: traverse is a lazy generator, so one batch costs one batch.
3: snake_down runs A1→H1, then H2→A2 — the shortest gantry path when the head moves column to column. Nothing about tip racks changes it.

['A1', 'B1', 'C1', 'D1', 'E1', 'F1', 'G1', 'H1']
snake, batch 2: ['H2', 'G2', 'F2', 'E2', 'D2', 'C2', 'B2', 'A2']

Gotcha: traverse reads the layout, not the trackers.

It yields every spot in the grid whether or not the spot still holds a tip. Filtering by has_tip() is the next recipe. The generators in this recipe have the same blind spot — position is theirs, occupancy is the rack’s.

See also: the batching helpers in chapter 7 chunk worklist rows to channel count.

10.4 Set what the rack model believes

Set which spots the model believes hold tips, after loading a partly used rack or a manual change.

rack_c.empty()
print("after empty(), A1:", rack_c.get_item("A1").has_tip())

rack_c.set_tip_state({"A1": True, "B1": True, "C1": False})
print("A1:", rack_c.get_item("A1").has_tip(),
      "B1:", rack_c.get_item("B1").has_tip(),
      "C1:", rack_c.get_item("C1").has_tip())

rack_c.fill()
print("after fill(), C1:", rack_c.get_item("C1").has_tip())

1: empty() marks every spot as having no tip.
2: set_tip_state takes a dict of identifier → bool, or a flat list of 96 bools.
3: fill() marks every spot as holding a tip.

after empty(), A1: False
A1: True B1: True C1: False
after fill(), C1: True

Turning the rack’s trackers off entirely.

rack.disable_tip_trackers() and rack.enable_tip_trackers() switch tracking for one rack rather than globally. A rack with disabled trackers accepts pick-ups from empty spots without raising.

See also: no_tip_tracking() and the two distinct HasTipErrors in chapter 10.

10.5 Find the next free tips from the rack itself

Work out where the tips are when you have no saved index — a rack that was partly used, or a session that reconnected to a machine mid-protocol.

rack_b.set_tip_state({spot: False for spot in
                      ["A1", "B1", "C1", "D1", "E1", "F1", "G1", "H1", "A2", "B2"]})

1: A rack somebody already took ten tips from: column 1, plus the top of column 2. set_tip_state sets rather than mutates, so it does not care which of those spots the recipes above already emptied — TipSpot.empty() in a loop would raise NoTipError on the first one that is already free. In a real session this state arrives from load_state_from_file (chapter 11) or from probe_tip_inventory below.

free = [s for s in rack_b.get_all_items() if s.has_tip()]
print("spots with tips:", len(free))
print("next 8:", [s.name.split("_")[-1] for s in free[:8]])

print(rack_b.summary())

1: get_all_items() returns every TipSpot in fill order; has_tip() asks that spot’s tracker. This is the whole “find the next tip” algorithm — PLR has no get_next_tip() to call.
2: Slice off as many as you have channels and hand them straight to pick_up_tips.
3: summary() prints the grid: V for a spot holding a tip, - for an empty one. TipRack overrides _occupied_func to read the tracker, so this is the fastest way to eyeball where a rack stands. (plate.summary() uses the same machinery for wells — chapter 5.)

spots with tips: 86
next 8: ['C2', 'D2', 'E2', 'F2', 'G2', 'H2', 'A3', 'B3']
TipRack(name='rack_b', size_x=127.76, size_y=85.48, size_z=64.49, location=Coordinate(000.000, 000.000, 000.000))
    1  2  3  4  5  6  7  8  9  10 11 12
A:  -  -  V  V  V  V  V  V  V  V  V  V
B:  -  -  V  V  V  V  V  V  V  V  V  V
C:  -  V  V  V  V  V  V  V  V  V  V  V
D:  -  V  V  V  V  V  V  V  V  V  V  V
E:  -  V  V  V  V  V  V  V  V  V  V  V
F:  -  V  V  V  V  V  V  V  V  V  V  V
G:  -  V  V  V  V  V  V  V  V  V  V  V
H:  -  V  V  V  V  V  V  V  V  V  V  V
12x8 TipRack

await lh.pick_up_tips(free[:8])
await lh.discard_tips()

Gotcha: this reads the model, not the machine.

has_tip() is only as good as what the trackers were told. If tracking was off for part of the run (it is off by default — see this chapter’s setup), or a rack was unloaded by hand, the model is fiction and the filter faithfully returns fiction. probe_tip_inventory below is the version that asks the hardware — on a backend that can answer.

See also: linear_tip_spot_generator above is the alternative when you do have continuity between runs — it tracks position in a file rather than re-deriving it from occupancy, which is the more robust of the two across a crash mid-pick-up.

10.6 What a tip is

A Tip carries the dimensions the backend needs, and the frontend uses them to decide whether a channel can take it.

tip = rack_a.get_tip("A6")
print("nominal volume: ", tip.nominal_volume)
print("maximal volume: ", tip.maximal_volume)
print("total length:   ", tip.total_tip_length)
print("fitting depth:  ", tip.fitting_depth)
print("filter:         ", tip.has_filter)

1: get_tip(identifier) returns the Tip object; get_tips(...) and get_all_tips() take several.
2: maximal_volume is the physical capacity; nominal_volume is what the tip is sold as. The backend’s can_pick_up_tip compares against these (chapter 18).
3: fitting_depth is how far the tip goes onto the channel, which is why a tip is not just a volume.

nominal volume:  200
maximal volume:  200
total length:    59.3
fitting depth:   7.47
filter:          True

Vendor subclasses add what their firmware needs. HamiltonTip takes a TipSize (LOW_VOLUME, STANDARD_VOLUME, HIGH_VOLUME, CORE_384_HEAD_TIP, XL) and a TipPickupMethod (OUT_OF_RACK or OUT_OF_WASH_LIQUID), which is what a STAR sends in its tip-type command.

See also: a TipSpot builds its tip on demand through make_tip(), which is how a rack definition specifies what it holds (chapter 17).

10.7 Move tips between racks

Relocate tips without pipetting — consolidating by hand, or filling a rack from a nested one.

for spot in rack_c["A7:D7"]:
    spot.empty()

await lh.move_tips(
    source_tip_spots=rack_a["A7:D7"],
    dest_tip_spots=rack_c["A7:D7"],
)

1: The destination has to be free. A rack starts full, so dropping into an occupied spot raises the tip-spot HasTipError — TipSpot.empty() marks one free in the model.
2: Source and destination are lists of tip spots, paired positionally.

print("rack_a A7:", rack_a.get_item("A7").has_tip())
print("rack_c A7:", rack_c.get_item("A7").has_tip())

rack_a A7: False
rack_c A7: True

consolidate_tip_inventory below is the planned version of this: it works out which partial racks to merge and in what order.

Nested tip racks. NestedTipRack is a TipRack with a stacking_z_height, for racks that stack inside one another (Hamilton NTR). It behaves as a tip rack in every other respect — hamilton_96_tiprack_50uL_NTR is one.

10.8 Tell PLR what is on the head

The channel head is model state like anything else, and can be set directly.

await lh.pick_up_tips(rack_a["A8"])
mounted_tip = lh.get_mounted_tips()[0]
mounted_before = [t is not None for t in lh.get_mounted_tips()]

lh.clear_head_state()
mounted_after = [t is not None for t in lh.get_mounted_tips()]

lh.update_head_state({0: mounted_tip})
mounted_restored = [t is not None for t in lh.get_mounted_tips()]

await lh.drop_tips(rack_a["A8"])

1: The Tip now on channel 0. Reading it from the head rather than the rack matters: with tracking on, the spot no longer holds it, so rack_a.get_tip("A8") raises NoTipError.
2: clear_head_state() forgets every mounted tip.
3: update_head_state({channel: tip}) sets specific channels. Passing None as the tip marks a channel empty.
4: Restoring the model made the tip droppable again. Had the state stayed cleared, this would raise NoTipError, and the next pick-up would raise HasTipError: Channel has tip.

print("after pick_up:  ", mounted_before[:2])
print("after clear:    ", mounted_after[:2])
print("after update:   ", mounted_restored[:2])

after pick_up:   [True, False]
after clear:     [False, False]
after update:    [True, False]

This is the tip equivalent of the off-deck bookkeeping in chapter 9: use it when a tip was fitted or removed without a PLR call, or when reconnecting to a machine that already has tips on.

Nothing physical happens. clear_head_state() on a machine with tips still fitted leaves the model claiming empty channels, and the next pick_up_tips will try to fit a second tip.

10.9 Probe and consolidate tip inventory

Read which spots hold tips from the machine, and pack partly used racks into fewer racks.

for spot, has_tip in presence.items():
    print(f"{spot}: {has_tip}")

rack_a_A5: True
rack_a_B5: True
rack_a_C5: True
rack_a_D5: True

presence = await lh.probe_tip_inventory(rack_a["A5:D5"])

1: Returns Dict[str, bool] keyed by tip spot name. The default probing function is probe_tip_presence_via_pickup, which attempts a pick-up and reads which channels failed.

await lh.consolidate_tip_inventory([rack_a, rack_b, rack_c])

1: Moves tips between partly filled racks so that as few racks as possible hold tips, grouped by tip model. Uses the first eight channels unless use_channels says otherwise.

Gotcha: probing cannot find discrepancies on the chatterbox

The probe reports presence by attempting a pick-up, and a simulated pick-up always succeeds, so every spot the model already believes is full comes back True.

For a spot the model believes is empty, the tip tracker raises NoTipError before the backend is reached — and NoTipError is not a ChannelizedError, so it propagates instead of being recorded as “absent”:

NoTipError: Tip spot does not have a tip.

Since only the STAR backend raises ChannelizedError (chapter 10), a probe that reports real presence is a Hamilton operation. Against the chatterbox it can only confirm the model against itself.

See also: F.get_all_tip_spots([...]) builds the spot list to hand to probe_tip_inventory; lh.move_tips relocates tips between spots without the consolidation planning.

10.10 What to remember

Three put-away calls: drop_tips(spots) to a destination you name, return_tips() to the origin, discard_tips() to the trash.
allow_nonzero_volume defaults to False for drop and return, True for discard.
use_tips is async with; use_channels is with.
F.linear_tip_spot_generator(F.get_all_tip_spots([...]), cache_file_path=...) streams across racks and resumes across processes. Exhaustion is StopAsyncIteration.
The streaming loop body is fresh = await gen.get(8) → pick_up_tips(fresh) → work → discard_tips(). async for walks one spot at a time; reset() starts a run over and persists.
rack.traverse(batch_size, start, direction) works on tip racks too — a TipRack is an ItemizedResource. It reads the layout, never the trackers.
With no saved index, the next free tips are [s for s in rack.get_all_items() if s.has_tip()]; rack.summary() prints the same thing as a V/- grid.
rack.empty() / fill() / set_tip_state({...}) write the model directly.
Tip tracking is off by default; with it off the rack model never changes.
probe_tip_inventory reports real presence only on a backend that raises ChannelizedError.
A Tip carries maximal_volume, total_tip_length, fitting_depth, and has_filter; HamiltonTip adds TipSize and TipPickupMethod.
lh.move_tips relocates tips directly; clear_head_state and update_head_state set what the head is believed to hold, and move nothing.