17  Compose behaviour with decorators

The same behaviour repeatedly surrounds protocol functions: error handling, tip acquisition, and tip-volume control. This chapter wraps those concerns in decorators so the protocol body contains only the transfer itself.

NoteSetup

import functools, logging
from pathlib import Path

from pylabrobot.liquid_handling import LiquidHandler
from pylabrobot.liquid_handling.backends import LiquidHandlerChatterboxBackend
from pylabrobot.liquid_handling.errors import ChannelizedError
from pylabrobot.resources import (
    STARLetDeck, PLT_CAR_L5AC_A00, cor_96_wellplate_360uL_Fb,
    nest_1_troughplate_195000uL_Vb,
    TIP_CAR_288_C00, opentrons_96_filtertiprack_200ul,
    set_volume_tracking, set_tip_tracking,
)
from pylabrobot.resources.errors import NoTipError, TooLittleLiquidError

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] = plate = cor_96_wellplate_360uL_Fb(name="plate")
carrier[1] = trough = nest_1_troughplate_195000uL_Vb(name="buffer")

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

set_volume_tracking(True)
set_tip_tracking(True)
plate["A1"][0].set_volume(1000)
trough["A1"][0].set_volume(10000)

protocol_log = logging.getLogger("protocol")
protocol_log.setLevel(logging.INFO)
if not any(isinstance(h, logging.Handler) for h in protocol_log.handlers):
    run_log = logging.FileHandler("run.log")
    run_log.setFormatter(logging.Formatter(
        "%(asctime)s %(name)-24s %(levelname)-5s %(message)s"
    ))
    protocol_log.addHandler(run_log)

1

The logging chapter explains the handler split; here one file handler is enough, and the last recipe reads it back.

---

17.1 Compose behaviour with decorators

Decorators are useful when the same behaviour needs to surround many protocol functions. Three common laboratory examples are error handling, tip handling, and tip-volume control.

Two mechanics are worth stating before the patterns are written:

• Decorators apply bottom-up: @a over @b over def f is f = a(b(f)). An exception raised in the body travels outward through b first, so the outermost decorator gets the last word, and reordering the stack changes which handler acts first.
• Retrying re-runs everything in the retried unit. aspirate and dispense change deck state, so a step containing them is not idempotent; reads such as summary() and get_resource() are.

17.1.1 Centralize try/except behaviour

Try/ except decorators let you easily compose error-handling logic for specific exceptions, like sending a Slack notification to an operator if a tip pickup fails with a no tip detected exception. Suppose several liquid-handling steps should log failures consistently:

def handle_errors(fn):
    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        try:
            return await fn(*args, **kwargs)
        except Exception:
            protocol_log.exception(
                "%s failed", fn.__name__
            )
            raise
    return wrapper

Use it normally:

@handle_errors
async def transfer_sample(lh, source, dest):
    await lh.aspirate(source, vols=[50])
    await lh.dispense(dest, vols=[50])

await lh.pick_up_tips(rack["A6"])
await transfer_sample(lh, plate["A1"], plate["B1"])
await lh.return_tips()
print("transfer_sample ran cleanly")

More specific handlers can inspect errors that contain partial completion information. PLR’s ChannelizedError records errors by channel:

error = ChannelizedError(
    errors={2: NoTipError("no tip"), 5: TooLittleLiquidError("empty")}
)

failed_channels = sorted(error.errors)
print("failed channels:  ", failed_channels)
print("blind retry re-runs:", list(range(8)))
print("re-run only:       ", failed_channels)

failed channels:   [2, 5]
blind retry re-runs: [0, 1, 2, 3, 4, 5, 6, 7]
re-run only:        [2, 5]

This matters because blindly repeating the entire transfer can repeat channels that already succeeded: a retry of an eight-channel operation with failures on channels 2 and 5 would rerun the six successful channels too. The appropriate recovery depends on the operation, but the wrapper is a useful place to keep common exception policy out of the protocol body (chapter 10).

17.1.2 Wrap tip management

Tip acquisition and cleanup often surround otherwise simple liquid-handling functions:

def with_fresh_tip(fn):
    @functools.wraps(fn)
    async def wrapper(lh, tip, *args, **kwargs):
        await lh.pick_up_tips(tip)
        try:
            return await fn(lh, *args, **kwargs)
        finally:
            await lh.discard_tips()
    return wrapper

Then the liquid-transfer function only contains the transfer:

@with_fresh_tip
async def move_sample(lh, source, dest, volume):
    await lh.aspirate(source, vols=[volume])
    await lh.dispense(dest, vols=[volume])

await move_sample(
    lh,
    rack["A7"],
    plate["A1"],
    plate["B1"],
    50,
)
print("moved 50 uL on a fresh tip, tip discarded")

A different wrapper can express a different tip policy. For example, reuse within one group:

def reuse_tip(fn):
    @functools.wraps(fn)
    async def wrapper(lh, tip, transfers):
        await lh.pick_up_tips(tip)
        try:
            for transfer in transfers:
                await fn(lh, *transfer)
        finally:
            await lh.return_tips()
    return wrapper

@reuse_tip
async def move_sample(lh, source, dest, volume):
    await lh.aspirate(source, vols=[volume])
    await lh.dispense(dest, vols=[volume])

await move_sample(
    lh,
    rack["A8"],
    [
        (plate["A1"], plate["C1"], 20),
        (plate["B1"], plate["D1"], 20),
    ],
)
print("two transfers on one reused tip, tip returned")

The point is simply that the transfer logic (aspirate, dispense, mix) does not have to repeat the same pick-up / cleanup / return-or-discard logic everywhere.

17.1.3 Control tip volume when plating from a trough

A single aspirate limits how much can be dispensed to whatever fits in the tips. When plating a reagent from a trough into many wells, the tips can instead be tracked and topped back up from the trough whenever the tracked volume runs low, so one set of tips covers the whole plate:

def with_reagent_refill(source, capacity):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(lh, volume, *args, **kwargs):
            mounted = lh.get_mounted_tips()
            if not any(mounted):
                raise RuntimeError("no tips mounted")

            refill = [
                (channel, capacity - tip.tracker.get_used_volume())
                for channel, tip in enumerate(mounted)
                if tip is not None and tip.tracker.get_used_volume() < volume
            ]

            if refill:
                channels, top_ups = zip(*refill)

                protocol_log.info(
                    "re-aspirating reagent from %s on channels %s",
                    source.parent.name,
                    list(channels),
                )

                await lh.aspirate(
                    [source] * len(channels),
                    vols=list(top_ups),
                    use_channels=list(channels),
                )

            return await fn(lh, volume, *args, **kwargs)

        return wrapper
    return decorator

1

get_mounted_tips() reports the tips in channel order, with None for an empty channel. It relies on tip tracking, enabled in Setup.

2

A volume check is meaningless with no tips on the head, so fail loudly instead of passing silently. any() asks the question directly — “is anything mounted” — rather than building a filtered list and then testing whether it came out empty.

3

One comprehension answers both questions the refill needs — which channels own a tip, and which of those are low — and enumerate keeps the channel index attached, so the refill addresses the exact channels rather than assuming the contiguous block 0–7. Each tip’s tracker records the liquid in it: aspirate adds, dispense subtracts. The per-tip volumes need not be equal, so rather than a min() over all tips, only tips below what the next call needs are refilled — each back up to capacity. A tip carrying more than the others is left alone.

4

zip(*pairs) splits the pairs back into two parallel lists — the transpose from chapter 7, used here in the unpacking direction instead of walking refill twice with a comprehension apiece.

5

use_channels selects exactly the refilled channels. Repeating the single trough well once per channel is the single-resource idiom from chapter 4, and vols must match use_channels.

Then a plating function contains only the dispense:

trough_well = trough["A1"][0]
carrier[2] = dest = cor_96_wellplate_360uL_Fb(name="dest")   # a fresh plate

@with_reagent_refill(trough_well, capacity=200)
async def plate_column(lh, volume, dest_column):
    await lh.dispense(dest_column, vols=[volume] * 8)

await lh.pick_up_tips(rack["A1:H1"])
for column in range(12):
    await plate_column(lh, 60, dest.column(column))
remaining = [round(t.tracker.get_used_volume(), 1)
             for t in lh.get_mounted_tips()]
await lh.discard_tips()

200 µL tips dispensing 60 µL per column can plate three columns before the tracked volume runs low, so the trough is re-aspirated on columns 1, 4, 7, and 10 — the tips never run dry, and one tip set covers the whole plate. Each refill is a process-tier log record (the logging recipe), so the decision history survives:

for line in Path("run.log").read_text().splitlines():
    if "re-aspirating" in line:
        print(line)
print("leftover per tip:", remaining)

2026-08-27 22:14:23,324 protocol                 INFO  re-aspirating reagent from buffer on channels [0, 1, 2, 3, 4, 5, 6, 7]
2026-08-27 22:14:23,326 protocol                 INFO  re-aspirating reagent from buffer on channels [0, 1, 2, 3, 4, 5, 6, 7]
2026-08-27 22:14:23,328 protocol                 INFO  re-aspirating reagent from buffer on channels [0, 1, 2, 3, 4, 5, 6, 7]
2026-08-27 22:14:23,330 protocol                 INFO  re-aspirating reagent from buffer on channels [0, 1, 2, 3, 4, 5, 6, 7]
leftover per tip: [20.0, 20.0, 20.0, 20.0, 20.0, 20.0, 20.0, 20.0]

With real hardware the source trough must hold enough reagent for the sum of all refills; the wrapper only decides when to go back to the trough, it does not check the trough’s own tracked volume.

17.1.4 Recover by trying the next candidate

The wrappers so far log and re-raise, or clean up after a step. A different family of handlers recovers by substituting the argument: when an error means “this particular resource won’t work, but the next one might”, the wrapper advances to the next candidate and retries. Two failures that behave this way are an empty tip spot and a source that cannot supply the requested volume.

A pickup that fails because the spot is empty is a NoTipError. pick_up_tips raises it before anything is mounted — the failed pickup leaves the head unchanged — so retrying the next spot is safe, and no state needs undoing:

def try_next_tip(spots):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(lh, *args, **kwargs):
            for spot in spots:
                try:
                    return await fn(lh, [spot], *args, **kwargs)
                except NoTipError:
                    protocol_log.warning(
                        "no tip at %s; trying the next spot", spot.name
                    )
            raise RuntimeError(
                f"no usable tip left among {[s.name for s in spots]}"
            )
        return wrapper
    return decorator

1

The candidates are the spots the caller hands over — rack["A12:D12"] here, but any iterable of TipSpots works, so a partial rack or a single column can be the pool to search.

2

The candidate is handed over as a one-item list, because that is what pick_up_tips takes — one spot per channel. The wrapped function is responsible for the pickup; here that is with_fresh_tip.

3

NoTipError is raised by pick_up_tips when the spot has no tip; the tracker raises before the backend is involved, so nothing has been moved.

4

Only when every candidate is empty does the wrapper give up — and it says so explicitly.

Note what it does not catch: HasTipError means a channel already has a tip. Advancing to the next spot does not fix that — the same channel is still occupied. The recovery for an occupied channel is to drop or return the mounted tip first, then retry. Keeping the two failures separate is the point: try_next_tip handles “no tip here”, and the caller handles “the head already has tips” with cleanup.

rack.set_tip_state({"A12": False, "B12": False})   # two empty candidate spots

@try_next_tip(rack["A12:D12"])
@with_fresh_tip
async def transfer_sample(lh, source, dest, volume):
    await lh.aspirate(source, vols=[volume])
    await lh.dispense(dest, vols=[volume])

await transfer_sample(lh, plate["A1"], plate["B1"], 50)

The wrapper skips the two emptied spots and picks up from the first spot that has a tip. The decision is recorded at the process tier like every other wrapper decision:

for line in Path("run.log").read_text().splitlines():
    if "trying the next spot" in line:
        print(line)

2026-08-27 22:14:23,358 protocol                 WARNING no tip at rack_A12; trying the next spot
2026-08-27 22:14:23,358 protocol                 WARNING no tip at rack_B12; trying the next spot

The same shape substitutes a different source when aspiration fails. TooLittleLiquidError is raised by the volume tracker when a source cannot supply the requested volume — again before the backend call, so nothing has moved:

def try_next_source(sources):
    def decorator(fn):
        @functools.wraps(fn)
        async def wrapper(lh, *args, **kwargs):
            last_error = None
            for source in sources:
                try:
                    return await fn(lh, source, *args, **kwargs)
                except TooLittleLiquidError as error:
                    last_error = error
                    protocol_log.warning(
                        "%s cannot supply %s; trying the next source",
                        source.name, error,
                    )
            raise last_error or TooLittleLiquidError(
                f"all sources exhausted: {[s.name for s in sources]}"
            )
        return wrapper
    return decorator

1

The wrapped function receives the source as an argument, so the same aspirate body runs against whichever candidate is current.

2

TooLittleLiquidError is raised when the source cannot supply the requested volume; re-running against the next source is safe because nothing has been aspirated.

A nearly-dry spare trough then falls back to the full one:

carrier[3] = spare = nest_1_troughplate_195000uL_Vb(name="spare")
spare["A1"][0].set_volume(20)                        # nearly dry

@try_next_source([spare["A1"][0], trough["A1"][0]])
async def take_reagent(lh, source, volume):
    await lh.aspirate([source], vols=[volume])

await lh.pick_up_tips(rack["A11"])
await take_reagent(lh, 50)                           # spare can't supply 50 uL
await lh.discard_tips()

for line in Path("run.log").read_text().splitlines():
    if "cannot supply" in line:
        print(line)

2026-08-27 22:14:23,384 protocol                 WARNING spare_well_A1 cannot supply Not enough liquid in container: 50.0uL > 20uL.; trying the next source

The two wrappers differ in one respect worth stating: try_next_tip consumes the candidate through with_fresh_tip, while try_next_source passes the candidate into the wrapped function as its first argument. Both keep the recovery policy out of the protocol body, which is the whole point of this chapter.

---

17.2 What to remember

• Decorators apply bottom-up: the outermost decorator handles first, and reordering the stack changes which handler acts first.
• A retry re-runs everything in its unit — so a step containing aspirate/dispense is not safe to retry blindly. Reads are; mutations are not.
• Separate tip lifecycle (pick up, discard or return) from the scientific operation, and centralize exception policy in a wrapper instead of in each protocol body.
• Track the liquid in the tips and top up from a trough with use_channels targeting exactly the channels that need it.
• Some errors mean “this resource won’t work, try the next one”: NoTipError (empty tip spot) and TooLittleLiquidError (source can’t supply) both raise before anything is moved, so a wrapper can advance to the next candidate and retry safely. HasTipError (channel already occupied) does not work that way — clean up the mounted tip instead of advancing.