20  Define a custom liquid handler

NoteSetup

from typing import List

from pylabrobot.liquid_handling import LiquidHandler
from pylabrobot.liquid_handling.backends import LiquidHandlerBackend
from pylabrobot.liquid_handling.standard import (
    Pickup, Drop, SingleChannelAspiration, SingleChannelDispense,
)
from pylabrobot.resources import (
    STARLetDeck, PLT_CAR_L5AC_A00, cor_96_wellplate_360uL_Fb,
    TIP_CAR_288_C00, opentrons_96_filtertiprack_200ul,
    Tip, Coordinate, set_volume_tracking,
)

---

20.1 Step 1: Read the smallest complete backend

Start from a backend that already works.

pylabrobot/liquid_handling/backends/chatterbox.py is 242 lines and is a complete, correct implementation of the interface: every abstract member present, every operation handled. It prints instead of moving, which is the only thing separating it from a driver.

Two things to take from it: the shape of each method, async def aspirate(self, ops, use_channels), and that a backend receives lists of operation objects, not the arguments the caller passed to lh.aspirate.

---

20.2 Step 2: See what the interface demands

List what must be implemented, and what happens if it is not.

print(len(LiquidHandlerBackend.__abstractmethods__), "abstract members:")
for m in sorted(LiquidHandlerBackend.__abstractmethods__):
    print("  ", m)

14 abstract members:
   aspirate
   aspirate96
   can_pick_up_tip
   dispense
   dispense96
   drop_resource
   drop_tips
   drop_tips96
   move_picked_up_resource
   num_channels
   pick_up_resource
   pick_up_tips
   pick_up_tips96
   stop

They tier:

Tier	Members	This build
Core	num_channels, can_pick_up_tip, pick_up_tips, drop_tips, aspirate, dispense	implement — 6
Lifecycle	stop	implement — 1
96-head	pick_up_tips96, drop_tips96, aspirate96, dispense96	refuse — 4
Gripper	pick_up_resource, move_picked_up_resource, drop_resource	refuse — 3

class Incomplete(LiquidHandlerBackend):
    pass

try:
    Incomplete()
except TypeError as e:
    print("TypeError:", str(e)[:170])

1

Python refuses to instantiate a class with unimplemented abstract methods.

TypeError: Can't instantiate abstract class Incomplete without an implementation for abstract methods 'aspirate', 'aspirate96', 'can_pick_up_tip', 'dispense', 'dispense96', 'drop_re

Notesetup is not abstract, stop is

stop is on the list; setup is not. The base class provides a setup that asserts a deck has been set, so an override should call await super().setup() before touching hardware. There is no default stop.

Members with defaults you may override.

Not abstract, and already implemented: move_channel_x/y/z, prepare_for_manual_channel_operation, and request_tip_presence all raise NotImplementedError (chapter 12), and get_channel_spacings returns a generic occupancy diameter for every channel. Override the ones your machine can do.

---

20.3 Step 3: Implement the core six

Implement the six core members.

class FakeGantry:
    def __init__(self):
        self.sent: List[str] = []

    async def send(self, command: str) -> str:
        self.sent.append(command)
        return "ok"

1

Stands in for the transport, so this chapter runs. Step 6 replaces it.

class GantryBackend(LiquidHandlerBackend):
    def __init__(self, device: FakeGantry):
        super().__init__()
        self.device = device

    @property
    def num_channels(self) -> int:
        return 1

    def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
        return tip.maximal_volume <= 200

    async def _goto(self, resource, offset: Coordinate):
        loc = resource.get_absolute_location() + offset
        await self.device.send(f"MOVE X{loc.x:.2f} Y{loc.y:.2f} Z{loc.z:.2f}")

    async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]):
        for op in ops:
            await self._goto(op.resource, op.offset)
            await self.device.send("TIP ON")

    async def drop_tips(self, ops: List[Drop], use_channels: List[int]):
        for op in ops:
            await self._goto(op.resource, op.offset)
            await self.device.send("TIP OFF")

    async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int]):
        for op in ops:
            await self._goto(op.resource, op.offset)
            rate = op.flow_rate if op.flow_rate is not None else 100.0
            await self.device.send(f"ASP {op.volume:.1f} RATE {rate:.1f}")

    async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int]):
        for op in ops:
            await self._goto(op.resource, op.offset)
            rate = op.flow_rate if op.flow_rate is not None else 100.0
            await self.device.send(f"DIS {op.volume:.1f} RATE {rate:.1f}")

    async def setup(self):
        await super().setup()
        await self.device.send("HOME")

    async def stop(self):
        await self.device.send("PARK")

    # --- capabilities this machine does not have ---------------------
    async def pick_up_tips96(self, pickup):
        raise NotImplementedError("single-channel gantry: no 96-head")

    async def drop_tips96(self, drop):
        raise NotImplementedError("single-channel gantry: no 96-head")

    async def aspirate96(self, aspiration):
        raise NotImplementedError("single-channel gantry: no 96-head")

    async def dispense96(self, dispense):
        raise NotImplementedError("single-channel gantry: no 96-head")

    async def pick_up_resource(self, pickup):
        raise NotImplementedError("single-channel gantry: no gripper")

    async def move_picked_up_resource(self, move):
        raise NotImplementedError("single-channel gantry: no gripper")

    async def drop_resource(self, drop):
        raise NotImplementedError("single-channel gantry: no gripper")

print("defined:", GantryBackend.__name__)

1

super().__init__() sets up the base state, including _head96_installed = False.

2

A property, not a method. The frontend reads it to size lh.head.

3

Called before a pick-up. Returning False is how a channel declines a tip it cannot fit.

4

Not part of the interface — a helper. get_absolute_location() plus the op’s offset is the coordinate (chapter 3).

5

ops is a list with one entry per channel in use_channels. A one-channel machine gets one.

6

None means “backend default” (chapter 4) — the backend is what supplies that default.

7

super().setup() asserts a deck is set before the device is touched.

defined: GantryBackend

Each operation carries what the backend needs:

Op	Fields
Pickup / Drop	resource, offset, tip
SingleChannelAspiration / SingleChannelDispense	resource, offset, tip, volume, flow_rate, liquid_height, blow_out_air_volume, mix

---

20.4 Step 4: Run a protocol through it

Drive the backend with an ordinary LiquidHandler and read the device traffic.

device = FakeGantry()
lh = LiquidHandler(backend=GantryBackend(device), deck=STARLetDeck())
await lh.setup()
print("channels:", len(lh.head))

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")
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)
plate["A1"][0].set_volume(200)

await lh.pick_up_tips(rack["A1"])
await lh.aspirate(plate["A1"], vols=[50], flow_rates=[75.0])
await lh.dispense(plate["B1"], vols=[50])
await lh.drop_tips(rack["A1"])
await lh.stop()

print("\ndevice received:")
for command in device.sent:
    print("  ", command)

1

Sized from num_channels.

2

An explicit flow rate reaches the device as RATE 75.0.

3

No flow rate given, so the backend’s own default appears instead.

channels: 1

2026-08-27 22:14:36,745 - pylabrobot - WARNING - Resource 'rack' is very high on the deck: 279.19 mm. Be careful when traversing the deck.

device received:
   HOME
   MOVE X138.73 Y171.69 Z220.09
   TIP ON
   MOVE X317.37 Y142.27 Z186.15
   ASP 50.0 RATE 75.0
   MOVE X317.37 Y133.27 Z186.15
   DIS 50.0 RATE 100.0
   MOVE X138.73 Y171.69 Z220.09
   TIP OFF
   PARK

---

20.5 Step 5: Declare what you cannot do

What a caller sees when they ask for a capability the machine lacks.

lh2 = LiquidHandler(backend=GantryBackend(FakeGantry()), deck=STARLetDeck())
await lh2.setup()
tc = TIP_CAR_288_C00(name="tc2")
lh2.deck.assign_child_resource(tc, rails=2)
tc[0] = rack2 = opentrons_96_filtertiprack_200ul(name="rack2")

try:
    await lh2.pick_up_tips96(rack2)
except (NotImplementedError, KeyError) as e:
    print(f"{type(e).__name__}: {e}")

1

The backend raises NotImplementedError — but that is not what surfaces.

2026-08-27 22:14:36,799 - pylabrobot - WARNING - Resource 'rack2' is very high on the deck: 279.19 mm. Be careful when traversing the deck.

KeyError: 0

WarningGotcha: the 96-head methods are not reached

LiquidHandler.__init__ builds head96 only when the backend reports head96_installed, which defaults to False. pick_up_tips96 indexes that empty dict before calling the backend, so the caller gets a bare KeyError: 0 rather than the NotImplementedError the backend carefully raises.

A backend that does have a 96-head sets self._head96_installed = True in __init__, which is what makes the frontend build the head and route the call through.

See also: NotImplementedError with an empty message is what the generic chatterbox raises for the jog methods (chapter 12) — a message costs nothing and is worth supplying.

---

20.6 Step 6: Talk to a real device

Replace FakeGantry with a transport.

pylabrobot.io provides serial, usb, hid, ftdi, and socket:

from pylabrobot.io.serial import Serial

class GantryBackend(LiquidHandlerBackend):
    def __init__(self, port: str):
        super().__init__()
        self.io = Serial(
            human_readable_device_name="gantry",
            port=port,
            baudrate=115200,
            timeout=1,
        )

    async def setup(self):
        await super().setup()
        await self.io.setup()
        await self.io.write(b"HOME\n")

    async def stop(self):
        await self.io.write(b"PARK\n")
        await self.io.stop()

1

Serial wraps pyserial and takes the same parameters — port, baudrate, bytesize, parity, stopbits, timeout, rtscts, dsrdtr.

2

The transport has its own setup/stop, nested inside the backend’s.

---

20.7 Step 7: Coordinate frames and homing

Convert PLR’s coordinates to the machine’s.

PLR gives the backend deck coordinates in millimetres: resource.get_absolute_location() + op.offset, with the deck origin at its own front-left-bottom (chapter 3).

Three things stand between that and a motor command, and none is free:

1. Origin. The machine’s zero and the deck’s origin are different points. The offset between them is a constant you measure once, per instrument.
2. Axis direction and units. A machine whose y increases toward the back, or which counts in steps rather than millimetres, needs the conversion applied in the backend.
3. Homing. get_absolute_location() is meaningful only if the machine agrees where zero is, which means homing at setup() — the HOME command in step 3 — and after any event that can lose position.

DECK_ORIGIN_IN_MACHINE = Coordinate(x=12.5, y=8.0, z=0.0)
STEPS_PER_MM = 80

async def _goto(self, resource, offset: Coordinate):
    loc = resource.get_absolute_location() + offset + DECK_ORIGIN_IN_MACHINE
    await self.io.write(
        f"MOVE X{int(loc.x * STEPS_PER_MM)} Y{int(loc.y * STEPS_PER_MM)}\n".encode()
    )

1

Measured once for the instrument, not derived from anything PLR knows.

---

20.8 Step 8: Define the deck

A backend and a deck go together: the backend converts deck coordinates into motor commands, and the deck decides what those coordinates are. STARLetDeck positions by rails=; OTDeck positions by assign_child_at_slot(slot=) (chapter 3). A machine with its own layout needs its own Deck subclass.

This one is modelled on OTDeck: a grid of slots, each backed by a ResourceHolder.

from typing import List, Optional
from pylabrobot.resources import Deck, Resource, ResourceHolder, Coordinate

SLOT_SIZE_X, SLOT_SIZE_Y = 128.0, 86.0
SLOT_PITCH_X, SLOT_PITCH_Y = 132.5, 90.5


class GantryDeck(Deck):
    NUM_X, NUM_Y = 2, 3

    def __init__(self, name: str = "gantry_deck"):
        super().__init__(
            name=name,
            size_x=SLOT_PITCH_X * self.NUM_X,
            size_y=SLOT_PITCH_Y * self.NUM_Y,
            size_z=0,
        )
        self._slots: List[ResourceHolder] = []
        for i in range(self.NUM_X * self.NUM_Y):
            row, col = divmod(i, self.NUM_X)
            holder = ResourceHolder(
                name=f"{name}_slot_{i + 1}",
                size_x=SLOT_SIZE_X, size_y=SLOT_SIZE_Y, size_z=0,
            )
            self._slots.append(holder)
            super().assign_child_resource(
                holder,
                location=Coordinate(x=col * SLOT_PITCH_X, y=row * SLOT_PITCH_Y, z=0),
            )

    def assign_child_at_slot(self, resource: Resource, slot: int):
        if slot not in range(1, len(self._slots) + 1):
            raise ValueError(f"slot must be between 1 and {len(self._slots)}")
        holder = self._slots[slot - 1]
        if holder.resource is not None:
            raise ValueError(f"slot {slot} is already occupied")
        holder.assign_child_resource(resource)

    def get_slot(self, resource: Resource) -> Optional[int]:
        for i, holder in enumerate(self._slots):
            if holder.resource is resource:
                return i + 1
        return None

    @property
    def slots(self) -> List[Optional[Resource]]:
        return [h.resource for h in self._slots]


deck = GantryDeck()
print("deck:", deck.get_size_x(), "x", deck.get_size_y(), "mm |", len(deck.slots), "slots")

1

The labware footprint a slot accepts — an SBS plate is 127.76 × 85.48 mm.

2

Centre-to-centre slot spacing, which is larger than the footprint by the gap between slots.

3

One ResourceHolder per slot, created in __init__. The holders are the deck’s geometry, so there is a single source of truth and the layout serializes with the tree.

4

super().assign_child_resource places the holder. Calling the subclass’s own method here would recurse.

5

The positioning method. rails= and slot= are both this: a machine-specific name for a position, resolved to a Coordinate.

deck: 265.0 x 271.5 mm | 6 slots

Slots resolve to absolute positions, and labware lands where the slot says:

from pylabrobot.resources import cor_96_wellplate_360uL_Fb, opentrons_96_filtertiprack_200ul

plate = cor_96_wellplate_360uL_Fb(name="deck_plate")
rack = opentrons_96_filtertiprack_200ul(name="deck_rack")

deck.assign_child_at_slot(plate, slot=1)
deck.assign_child_at_slot(rack, slot=4)

print("plate:", plate.get_absolute_location())
print("rack: ", rack.get_absolute_location())
print("plate is in slot", deck.get_slot(plate))
print("occupied slots:", [i + 1 for i, r in enumerate(deck.slots) if r is not None])
print("well A1:", plate.get_item("A1").get_absolute_location())

1

Well coordinates compose through the slot holder, so the backend’s get_absolute_location() works unchanged (chapter 3).

plate: Coordinate(000.000, 000.000, 000.000)
rack:  Coordinate(132.500, 090.500, 000.000)
plate is in slot 1
occupied slots: [1, 4]
well A1: Coordinate(010.870, 070.770, 003.030)

Both failure modes raise from assign_child_at_slot:

for slot in (1, 9):
    try:
        deck.assign_child_at_slot(cor_96_wellplate_360uL_Fb(name=f"extra_{slot}"), slot=slot)
    except ValueError as e:
        print(f"slot {slot}: {e}")

slot 1: slot 1 is already occupied
slot 9: slot must be between 1 and 6

The deck goes to LiquidHandler in place of STARLetDeck, and the backend from step 3 needs no change:

lh_gantry = LiquidHandler(backend=GantryBackend(FakeGantry()), deck=GantryDeck())
await lh_gantry.setup()

sample = cor_96_wellplate_360uL_Fb(name="sample")
lh_gantry.deck.assign_child_at_slot(sample, slot=2)
print("on the gantry deck:", sample.get_absolute_location())
print("serializes as:", lh_gantry.deck.serialize()["type"])

1

The class name is what find_subclass resolves on load, so a saved layout needs GantryDeck imported before load_from_json_file (chapter 11).

on the gantry deck: Coordinate(132.500, 000.000, 000.000)
serializes as: GantryDeck

See also: resources/hamilton/hamilton_decks.py and resources/opentrons/deck.py are the two shipped implementations to compare — rails against slots.

---

20.9 What the interface guarantees

• The ABC is the capability declaration. All 14 members must exist before Python will instantiate the class, so a backend cannot accidentally omit one. What it cannot enforce is correctness: an aspirate that pipettes the wrong volume satisfies the interface.
• The frontend owns validation, the backend owns motion. Tracking, resource resolution, and argument checking happen before your code runs — which is why a protocol written against the chatterbox works against a new backend unchanged.
• Capability advertisement is separate from implementation. The ABC decides what you must write; flags like head96_installed decide what the frontend will call.

20.10 What to remember

• 14 abstract members: 6 core, 1 lifecycle (stop), 4 for the 96-head, 3 for the gripper. setup is not abstract; override it and call await super().setup().
• Backends receive lists of frozen operation dataclasses, one per channel — not the caller’s arguments.
• num_channels is a property. can_pick_up_tip returning False declines a tip.
• flow_rate=None means the backend supplies the default.
• Unsupported 96-head calls surface as KeyError: 0 from the frontend, not your NotImplementedError, unless the backend sets _head96_installed = True.
• Build on pylabrobot.io rather than raw pyserial: it is what makes capture and replay work.
• Homing at setup() is what makes absolute coordinates mean anything.
• A machine with its own layout needs a Deck subclass: hold one ResourceHolder per position and give it a positioning method, the way rails= and slot= work.