From 4472214482081ccb5b3a174865bb9f560a78b3cd Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:09:49 +0100 Subject: [PATCH 1/9] `OpentronsOT2Backend`: pin home/stop/modules/trash-drop wire calls Characterization tests ahead of the transport-seam refactor (Phase 0). The existing suite already asserts the pickup/drop/aspirate/dispense wire calls; this adds the four remaining ot_api call sites so the seam refactor cannot change them silently: - home() issues one ot_api.health.home() - list_connected_modules() returns ot_api.modules.list_connected_modules() verbatim - stop() cancels the active run via the requestor and clears mounted pipettes - a discard to the deck trash at api_version >= 7.1.0 routes through the addressable area (move_to_addressable_area_for_drop_tip + drop_tip_in_place), not drop_tip No production code changes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend_tests.py | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py index 05ea8e2845f..b892986837f 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py @@ -7,6 +7,7 @@ from pylabrobot.liquid_handling import LiquidHandler from pylabrobot.liquid_handling.backends.opentrons_backend import ( + _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, OpentronsOT2Backend, ) from pylabrobot.liquid_handling.errors import NoChannelError @@ -188,6 +189,59 @@ def assert_parameters( with no_volume_tracking(): await self.lh.dispense(self.plate["A1"], vols=[10]) + # -- characterization of the remaining ot_api call sites (Phase 0 safety net) -- + + @patch("ot_api.health.home") + async def test_home_calls_health_home(self, mock_home): + """home() issues exactly one ot_api.health.home() call.""" + await self.backend.home() + mock_home.assert_called_once_with() + + @patch("ot_api.modules.list_connected_modules") + async def test_list_connected_modules_passthrough(self, mock_modules): + """list_connected_modules() returns ot_api.modules.list_connected_modules() verbatim.""" + mock_modules.return_value = [{"id": "tempdeck"}] + result = await self.backend.list_connected_modules() + mock_modules.assert_called_once_with() + self.assertEqual(result, [{"id": "tempdeck"}]) + + @patch("ot_api.run_id", "run-id", create=True) + @patch("ot_api.requestor.post") + async def test_stop_cancels_active_run_and_clears_pipettes(self, mock_post): + """stop() cancels the active run through the requestor and clears mounted pipettes.""" + await self.backend.stop() + mock_post.assert_called_once_with("/runs/run-id/cancel") + self.assertIsNone(self.backend.left_pipette) + self.assertIsNone(self.backend.right_pipette) + + @patch("ot_api.lh.drop_tip_in_place") + @patch("ot_api.lh.move_to_addressable_area_for_drop_tip") + @patch("ot_api.lh.drop_tip") + @patch("ot_api.lh.pick_up_tip") + @patch("ot_api.labware.define") + @patch("ot_api.labware.add") + async def test_tip_drop_to_trash_uses_addressable_area( + self, + mock_add, + mock_define, + mock_pick_up_tip, + mock_drop_tip, + mock_to_trash, + mock_drop_in_place, + ): + """At api_version >= 7.1.0 a discard to the deck trash routes via the addressable + area (move_to_addressable_area_for_drop_tip + drop_tip_in_place), not drop_tip.""" + mock_define.side_effect = _mock_define + mock_add.side_effect = _mock_add + self.backend.ot_api_version = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION + + await self.lh.pick_up_tips(self.tip_rack["A1"]) + await self.lh.discard_tips() + + mock_to_trash.assert_called_once() + mock_drop_in_place.assert_called_once() + mock_drop_tip.assert_not_called() + def _make_backend_with_pipettes(left_name="p300_single_gen2", right_name="p20_single_gen2"): """Create a backend with pipette state set directly (no ot_api needed).""" From e87e78cb96fc05be18181868484f412bf228faf5 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:18:39 +0100 Subject: [PATCH 2/9] `OpentronsOT2Backend`: route all ot_api I/O through a self._ot handle Introduces a single transport seam (Phase 1): every ot_api call - including the run-cancel requestor calls in stop() - now goes through self._ot, set to the ot_api module in __init__. Behaviour is unchanged (self._ot is ot_api), but a subclass can now dry-run the backend by swapping the handle for a recording stand-in, the way STARChatterboxBackend overrides STAR's transport. The dedicated `import ot_api.requestor as _req` is dropped; ot_api.requestor is reachable through the handle (self._ot.requestor). Guarded by the Phase 0 characterization tests, which patch ot_api.* and stay green because self._ot is the same module object. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend.py | 59 ++++++++++--------- 1 file changed, 30 insertions(+), 29 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index f5e30322a9e..dc722e46d48 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -31,9 +31,6 @@ try: import ot_api - # for run cancellation - import ot_api.requestor as _req - USE_OT = True except ImportError as e: USE_OT = False @@ -77,8 +74,12 @@ def __init__(self, host: str, port: int = 31950): self.host = host self.port = port - ot_api.set_host(host) - ot_api.set_port(port) + # All hardware I/O goes through this handle so a subclass (e.g. the chatterbox) + # can dry-run the backend by swapping it for a recording stand-in. + self._ot = ot_api + + self._ot.set_host(host) + self._ot.set_port(port) self.ot_api_version: Optional[str] = None self.left_pipette: Optional[Dict[str, str]] = None @@ -97,16 +98,16 @@ def serialize(self) -> dict: async def setup(self, skip_home: bool = False): # create run - run_id = ot_api.runs.create() - ot_api.set_run(run_id) + run_id = self._ot.runs.create() + self._ot.set_run(run_id) # get pipettes, then assign them - self.left_pipette, self.right_pipette = ot_api.lh.add_mounted_pipettes() + self.left_pipette, self.right_pipette = self._ot.lh.add_mounted_pipettes() self.left_pipette_has_tip = self.right_pipette_has_tip = False # get api version - health = ot_api.health.get() + health = self._ot.health.get() self.ot_api_version = health["api_version"] if not skip_home: @@ -124,16 +125,16 @@ async def stop(self): self.right_pipette = None # cancel the HTTP-API run if it exists (helpful to make device available again in official Opentrons app) - run_id = getattr(ot_api, "run_id", None) + run_id = getattr(self._ot, "run_id", None) if run_id: try: - _req.post(f"/runs/{run_id}/cancel") + self._ot.requestor.post(f"/runs/{run_id}/cancel") except Exception: try: - _req.post(f"/runs/{run_id}/actions/cancel") + self._ot.requestor.post(f"/runs/{run_id}/actions/cancel") except Exception: try: - _req.delete(f"/runs/{run_id}") + self._ot.requestor.delete(f"/runs/{run_id}") except Exception: pass @@ -231,7 +232,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): ], } - data = ot_api.labware.define(lw) + data = self._ot.labware.define(lw) namespace, definition, version = data["data"]["definitionUri"].split("/") # assign labware to robot @@ -242,7 +243,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): slot = deck.get_slot(tip_rack) assert slot is not None, "tip rack must be on deck" - ot_api.labware.add( + self._ot.labware.add( load_name=definition, namespace=namespace, ot_location=slot, @@ -321,7 +322,7 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): offset_z += op.tip.total_tip_length - ot_api.lh.pick_up_tip( + self._ot.lh.pick_up_tip( labware_id=self.get_ot_name(tip_rack.name), well_name=self.get_ot_name(op.resource.name), pipette_id=pipette_id, @@ -361,15 +362,15 @@ async def drop_tips(self, ops: List[Drop], use_channels: List[int]): offset_z += 10 if use_fixed_trash: - ot_api.lh.move_to_addressable_area_for_drop_tip( + self._ot.lh.move_to_addressable_area_for_drop_tip( pipette_id=pipette_id, offset_x=offset_x, offset_y=offset_y, offset_z=offset_z, ) - ot_api.lh.drop_tip_in_place(pipette_id=pipette_id) + self._ot.lh.drop_tip_in_place(pipette_id=pipette_id) else: - ot_api.lh.drop_tip( + self._ot.lh.drop_tip( labware_id, well_name=self.get_ot_name(op.resource.name), pipette_id=pipette_id, @@ -461,18 +462,18 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[ if op.mix is not None: for _ in range(op.mix.repetitions): - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=volume, flow_rate=flow_rate, pipette_id=pipette_id, @@ -533,7 +534,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=volume, flow_rate=flow_rate, pipette_id=pipette_id, @@ -541,12 +542,12 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in if op.mix is not None: for _ in range(op.mix.repetitions): - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, @@ -563,7 +564,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in ) async def home(self): - ot_api.health.home() + self._ot.health.home() async def pick_up_tips96(self, pickup: PickupTipRack): raise NotImplementedError("The Opentrons backend does not support the 96 head.") @@ -590,7 +591,7 @@ async def drop_resource(self, drop: ResourceDrop): async def list_connected_modules(self) -> List[dict]: """List all connected temperature modules.""" - return cast(List[dict], ot_api.modules.list_connected_modules()) + return cast(List[dict], self._ot.modules.list_connected_modules()) def _pipette_id_for_channel(self, channel: int) -> str: pipettes = [] @@ -607,7 +608,7 @@ def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]: pipette_id = self._pipette_id_for_channel(channel) try: - res = ot_api.lh.save_position(pipette_id=pipette_id) + res = self._ot.lh.save_position(pipette_id=pipette_id) pos = res["data"]["result"]["position"] current = Coordinate(pos["x"], pos["y"], pos["z"]) except Exception as exc: # noqa: BLE001 @@ -676,7 +677,7 @@ async def move_pipette_head( if pipette_id is None: raise ValueError("No pipette id given or left/right pipette not available.") - ot_api.lh.move_arm( + self._ot.lh.move_arm( pipette_id=pipette_id, location_x=location.x, location_y=location.y, From 92fbaa85b479fda8c75754c035dd64210095108e Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:26:17 +0100 Subject: [PATCH 3/9] `OpentronsOT2ChatterboxBackend`: dry-run the OT-2 backend via the transport seam Phase 2. A chatterbox sibling to OpentronsOT2Backend that swaps the self._ot transport handle for a recorder, so the real backend logic - pipette selection, tip and volume tracking, the per-operation wire calls - runs unchanged with no hardware and no ot_api library. Mirrors STARChatterboxBackend (transport-only override) rather than OpentronsOT2Simulator (which reimplements the high-level methods single-channel). Issued calls are printed and collected in .commands. The recorder returns canned data for the reads setup() and the operations make back (mounted pipettes, api version, labware define, save_position, modules). Exported from backends/__init__. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../liquid_handling/backends/__init__.py | 1 + .../backends/opentrons_chatterbox.py | 173 ++++++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 pylabrobot/liquid_handling/backends/opentrons_chatterbox.py diff --git a/pylabrobot/liquid_handling/backends/__init__.py b/pylabrobot/liquid_handling/backends/__init__.py index 50c01b189ad..a039cc1ed4b 100644 --- a/pylabrobot/liquid_handling/backends/__init__.py +++ b/pylabrobot/liquid_handling/backends/__init__.py @@ -4,6 +4,7 @@ from .hamilton.STAR_backend import STAR, STARBackend from .hamilton.vantage_backend import Vantage, VantageBackend from .opentrons_backend import OpentronsOT2Backend +from .opentrons_chatterbox import OpentronsOT2ChatterboxBackend from .opentrons_simulator import OpentronsOT2Simulator from .serializing_backend import SerializingBackend from .tecan.EVO_backend import EVO, EVOBackend diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py new file mode 100644 index 00000000000..3894523dbfa --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py @@ -0,0 +1,173 @@ +"""A chatterbox backend for the Opentrons OT-2. + +Dry-runs the real OpentronsOT2Backend without hardware or the ``ot_api`` library +by swapping the backend's transport handle (``self._ot``) for a recorder that +logs every call and returns canned data for the few reads the backend makes back. + +This mirrors how ``STARChatterboxBackend`` dry-runs ``STARBackend``: only the +transport is replaced, so all the real high-level logic (pipette selection, tip +and volume bookkeeping, the per-operation wire calls) runs unchanged. Contrast +with ``OpentronsOT2Simulator``, which overrides the high-level methods themselves. +""" + +from typing import Dict, List, Optional, Tuple, cast + +from pylabrobot.liquid_handling.backends.backend import LiquidHandlerBackend +from pylabrobot.liquid_handling.backends.opentrons_backend import ( + _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + OpentronsOT2Backend, +) + + +class _RecordingNamespace: + """An ot_api sub-namespace (e.g. ``lh``, ``health``) that records every call. + + Unknown attributes resolve to a function that appends ``(name, args, kwargs)`` + to the shared recorder and returns the canned value registered in ``returns`` + (``None`` if none is registered). + """ + + def __init__(self, recorder: "_OTChatterboxModule", prefix: str, returns=None): + self._recorder = recorder + self._prefix = prefix + self._returns = returns or {} + + def __getattr__(self, name: str): + if name.startswith("_"): + raise AttributeError(name) + qualified = f"{self._prefix}.{name}" + returns = self._returns + recorder = self._recorder + + def _record(*args, **kwargs): + recorder.log(qualified, args, kwargs) + canned = returns.get(name) + return canned() if callable(canned) else canned + + return _record + + +class _OTChatterboxModule: + """Stand-in for the ``ot_api`` module that records calls instead of issuing them. + + Provides the sub-namespaces and reads the real backend touches: ``runs.create``, + ``lh.add_mounted_pipettes``, ``health.get``, ``labware.define``, + ``modules.list_connected_modules`` and ``lh.save_position`` return canned data; + everything else is recorded and returns ``None``. ``run_id`` stays ``None`` so + ``stop()`` skips the cancel request. + """ + + def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool = True): + self.calls: List[Tuple[str, tuple, dict]] = [] + self.run_id: Optional[str] = None + self._verbose = verbose + + self.runs = _RecordingNamespace(self, "runs", {"create": lambda: "chatterbox-run"}) + self.health = _RecordingNamespace(self, "health", {"get": lambda: {"api_version": api_version}}) + self.labware = _RecordingNamespace( + self, "labware", {"define": lambda: {"data": {"definitionUri": "pylabrobot/chatterbox/1"}}} + ) + self.modules = _RecordingNamespace(self, "modules", {"list_connected_modules": lambda: []}) + self.requestor = _RecordingNamespace(self, "requestor") + self.lh = _RecordingNamespace( + self, + "lh", + { + "add_mounted_pipettes": lambda: (left_pipette, right_pipette), + "save_position": lambda: {"data": {"result": {"position": {"x": 0, "y": 0, "z": 0}}}}, + }, + ) + + def log(self, qualified: str, args: tuple, kwargs: dict): + self.calls.append((qualified, args, kwargs)) + if self._verbose: + parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] + print(f"{qualified}({', '.join(parts)})") + + def __getattr__(self, name: str): + # top-level functions the backend calls directly: set_host, set_port, set_run + if name.startswith("_"): + raise AttributeError(name) + + def _record(*args, **kwargs): + self.log(name, args, kwargs) + return None + + return _record + + +class OpentronsOT2ChatterboxBackend(OpentronsOT2Backend): + """Chatterbox backend for the Opentrons OT-2. + + Runs the real OpentronsOT2Backend logic with its transport replaced by a + recorder - no hardware and no ``ot_api`` library required. Every issued call is + printed and collected in :attr:`commands`. + + Example: + >>> from pylabrobot.liquid_handling import LiquidHandler + >>> from pylabrobot.liquid_handling.backends import OpentronsOT2ChatterboxBackend + >>> from pylabrobot.resources.opentrons import OTDeck + >>> lh = LiquidHandler(backend=OpentronsOT2ChatterboxBackend(), deck=OTDeck()) + >>> await lh.setup() + """ + + def __init__( + self, + left_pipette_name: Optional[str] = "p300_single_gen2", + right_pipette_name: Optional[str] = "p20_single_gen2", + host: str = "chatterbox", + port: int = 31950, + api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + verbose: bool = True, + ): + """Initialize the chatterbox. + + Args: + left_pipette_name: pipette mounted on the left (``None`` for none). + right_pipette_name: pipette mounted on the right (``None`` for none). + api_version: reported Opentrons API version; defaults to the version at + which tip drops route through the addressable-area trash. + verbose: if True, print every recorded call. + """ + # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly. + LiquidHandlerBackend.__init__(self) + + pv = OpentronsOT2Backend.pipette_name2volume + if left_pipette_name is not None and left_pipette_name not in pv: + raise ValueError(f"Unknown left pipette: {left_pipette_name}") + if right_pipette_name is not None and right_pipette_name not in pv: + raise ValueError(f"Unknown right pipette: {right_pipette_name}") + + self._left_pipette_name = left_pipette_name + self._right_pipette_name = right_pipette_name + self.host = host + self.port = port + + left = ( + {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None + ) + right = ( + {"name": right_pipette_name, "pipetteId": "chatterbox-right"} if right_pipette_name else None + ) + self._ot = _OTChatterboxModule(left, right, api_version, verbose=verbose) + + self.ot_api_version: Optional[str] = None + self.left_pipette: Optional[Dict[str, str]] = None + self.right_pipette: Optional[Dict[str, str]] = None + self.traversal_height = 120 + self._tip_racks: Dict[str, int] = {} + self._plr_name_to_load_name: Dict[str, str] = {} + + @property + def commands(self) -> List[Tuple[str, tuple, dict]]: + """Recorded ``(qualified_name, args, kwargs)`` for every call issued so far.""" + return cast(List[Tuple[str, tuple, dict]], self._ot.calls) + + def serialize(self) -> dict: + return { + **LiquidHandlerBackend.serialize(self), + "left_pipette_name": self._left_pipette_name, + "right_pipette_name": self._right_pipette_name, + "host": self.host, + "port": self.port, + } From 4b5ad766099650646eafbf3702dd0bb9259dd2d2 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:34:06 +0100 Subject: [PATCH 4/9] `OpentronsOT2ChatterboxBackend`: tests + differential audit vs the simulator Phase 3. Direct tests run the chatterbox with no ot_api installed: setup resolves both mounts, a pickup -> aspirate -> dispense -> trash-discard records exactly one wire call per operation, an unknown pipette name raises, and serialize() captures the mounts. A differential audit runs the same single-channel protocol through the chatterbox and the reference OpentronsOT2Simulator and asserts identical tracked outcomes (tip mounted, source/destination volumes). The audit is scoped to the single-channel overlap, since the simulator cannot represent a multi-channel head. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_chatterbox_tests.py | 135 ++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py new file mode 100644 index 00000000000..d9ee5f370de --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -0,0 +1,135 @@ +"""Tests for OpentronsOT2ChatterboxBackend. + +Deliberately does NOT importorskip("ot_api"): running the real backend logic with +no hardware and no ot_api library is the whole point of the chatterbox. +""" + +import unittest + +from pylabrobot.liquid_handling import LiquidHandler +from pylabrobot.liquid_handling.backends import ( + OpentronsOT2ChatterboxBackend, + OpentronsOT2Simulator, +) +from pylabrobot.resources import set_tip_tracking, set_volume_tracking +from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb +from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul + + +def _names(backend: OpentronsOT2ChatterboxBackend): + return [call[0] for call in backend.commands] + + +class OpentronsChatterboxTests(unittest.IsolatedAsyncioTestCase): + """Direct tests: the chatterbox runs the real backend with no ot_api.""" + + async def asyncSetUp(self): + set_tip_tracking(True) + set_volume_tracking(True) + self.backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_single_gen2", + right_pipette_name="p20_single_gen2", + verbose=False, + ) + self.deck = OTDeck() + self.lh = LiquidHandler(backend=self.backend, deck=self.deck) + await self.lh.setup() + self.tips = opentrons_96_filtertiprack_20ul(name="tips") + self.deck.assign_child_at_slot(self.tips, slot=1) + self.plate = CellTreat_96_wellplate_350ul_Fb(name="plate") + self.deck.assign_child_at_slot(self.plate, slot=2) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_setup_resolves_two_channels_without_ot_api(self): + """setup() runs through the recorder and resolves both mounted pipettes.""" + self.assertEqual(self.backend.num_channels, 2) + assert self.backend.left_pipette is not None and self.backend.right_pipette is not None + self.assertEqual(self.backend.left_pipette["name"], "p20_single_gen2") + + async def test_full_protocol_records_one_wire_call_per_operation(self): + """A pickup -> aspirate -> dispense -> trash-discard records exactly one + wire call each, via the real backend logic.""" + self.plate.get_well("A1").tracker.set_volume(15) + await self.lh.pick_up_tips(self.tips["A1"]) + await self.lh.aspirate(self.plate["A1"], vols=[10]) + await self.lh.dispense(self.plate["B1"], vols=[10]) + await self.lh.discard_tips() + + names = _names(self.backend) + self.assertEqual(names.count("lh.pick_up_tip"), 1) + self.assertEqual(names.count("lh.aspirate_in_place"), 1) + self.assertEqual(names.count("lh.dispense_in_place"), 1) + # api_version defaults to 7.1.0, so the discard routes through the trash addressable area + self.assertEqual(names.count("lh.move_to_addressable_area_for_drop_tip"), 1) + self.assertEqual(names.count("lh.drop_tip_in_place"), 1) + + def test_unknown_pipette_name_raises(self): + """An unrecognised pipette name is rejected at construction.""" + with self.assertRaises(ValueError): + OpentronsOT2ChatterboxBackend(left_pipette_name="not_a_pipette") + + def test_serialize_includes_pipettes(self): + """serialize() captures the mounted-pipette names (None for an empty mount).""" + backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_single_gen2", right_pipette_name=None, verbose=False + ) + serialized = backend.serialize() + self.assertEqual(serialized["left_pipette_name"], "p20_single_gen2") + self.assertIsNone(serialized["right_pipette_name"]) + + +class OpentronsChatterboxVsSimulatorTests(unittest.IsolatedAsyncioTestCase): + """Differential audit: the chatterbox must produce the same tracked outcome as + the reference OpentronsOT2Simulator on the single-channel overlap. (The Simulator + is single-channel by construction, so the multi-channel head is out of scope here.)""" + + async def asyncSetUp(self): + set_tip_tracking(True) + set_volume_tracking(True) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def _run_single_channel_protocol(self, backend): + deck = OTDeck() + lh = LiquidHandler(backend=backend, deck=deck) + await lh.setup() + tips = opentrons_96_filtertiprack_20ul(name="tips") + deck.assign_child_at_slot(tips, slot=1) + plate = CellTreat_96_wellplate_350ul_Fb(name="plate") + deck.assign_child_at_slot(plate, slot=2) + plate.get_well("A1").tracker.set_volume(15) + + await lh.pick_up_tips(tips["A1"]) + await lh.aspirate(plate["A1"], vols=[10]) + await lh.dispense(plate["B1"], vols=[10]) + + outcome = ( + lh.head[0].has_tip, + round(plate.get_well("A1").tracker.get_used_volume(), 3), + round(plate.get_well("B1").tracker.get_used_volume(), 3), + ) + await lh.stop() + return outcome + + async def test_chatterbox_matches_simulator_single_channel(self): + simulator_outcome = await self._run_single_channel_protocol( + OpentronsOT2Simulator( + left_pipette_name="p20_single_gen2", right_pipette_name="p20_single_gen2" + ) + ) + chatterbox_outcome = await self._run_single_channel_protocol( + OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_single_gen2", right_pipette_name="p20_single_gen2", verbose=False + ) + ) + self.assertEqual(simulator_outcome, (True, 5.0, 10.0)) + self.assertEqual(chatterbox_outcome, simulator_outcome) + + +if __name__ == "__main__": + unittest.main() From ee2ecf82837283a91704e293eae4c5a933664f5e Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 14:02:10 +0100 Subject: [PATCH 5/9] Log OT-2 HTTP calls at LOG_LEVEL_IO, like other backends' io The OT-2 talks HTTP through ot_api rather than a pylabrobot.io transport, so its wire traffic never reached the IO log other backends produce. Wrap the real ot_api handle in a transparent _IOLogger proxy that logs every call (submodules recursed, plain attributes passed through) at LOG_LEVEL_IO. The chatterbox recorder logs the same way, so a dry run captures the same wire trace as a real run. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend.py | 42 +++++++++++++++++-- .../backends/opentrons_chatterbox.py | 11 ++++- 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index dc722e46d48..ba944988229 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -1,7 +1,10 @@ +import inspect +import logging import uuid -from typing import Dict, List, Optional, Tuple, Union, cast +from typing import Any, Dict, List, Optional, Tuple, Union, cast from pylabrobot import utils +from pylabrobot.io import LOG_LEVEL_IO from pylabrobot.liquid_handling.backends.backend import ( LiquidHandlerBackend, ) @@ -41,6 +44,38 @@ # https://labautomation.io/t/connect-pylabrobot-to-ot2/2862/18 _OT_DECK_IS_ADDRESSABLE_AREA_VERSION = "7.1.0" +logger = logging.getLogger(__name__) + + +class _IOLogger: + """Transparent proxy over the ``ot_api`` module that logs every call at + ``LOG_LEVEL_IO``. + + The OT-2 talks HTTP through ``ot_api`` rather than a pylabrobot.io transport, so + this wrapper gives it the same wire-level logging every other backend gets from + its io object. Submodules (``lh``, ``health``, ...) are wrapped recursively; + plain attributes (e.g. ``run_id``) pass through untouched. + """ + + def __init__(self, target: Any, prefix: str = ""): + object.__setattr__(self, "_target", target) + object.__setattr__(self, "_prefix", prefix) + + def __getattr__(self, name: str) -> Any: + attr = getattr(self._target, name) + qualified = f"{self._prefix}.{name}" if self._prefix else name + if inspect.ismodule(attr): + return _IOLogger(attr, qualified) + if callable(attr): + + def _logged(*args, **kwargs): + parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] + logger.log(LOG_LEVEL_IO, "%s(%s)", qualified, ", ".join(parts)) + return attr(*args, **kwargs) + + return _logged + return attr + class OpentronsOT2Backend(LiquidHandlerBackend): """Backends for the Opentrons OT2 liquid handling robots.""" @@ -75,8 +110,9 @@ def __init__(self, host: str, port: int = 31950): self.port = port # All hardware I/O goes through this handle so a subclass (e.g. the chatterbox) - # can dry-run the backend by swapping it for a recording stand-in. - self._ot = ot_api + # can dry-run the backend by swapping it for a recording stand-in. The real handle + # wraps ot_api to log every HTTP call at LOG_LEVEL_IO, like other backends' io. + self._ot: Any = _IOLogger(ot_api) self._ot.set_host(host) self._ot.set_port(port) diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py index 3894523dbfa..c5b421f4652 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py @@ -10,14 +10,18 @@ with ``OpentronsOT2Simulator``, which overrides the high-level methods themselves. """ +import logging from typing import Dict, List, Optional, Tuple, cast +from pylabrobot.io import LOG_LEVEL_IO from pylabrobot.liquid_handling.backends.backend import LiquidHandlerBackend from pylabrobot.liquid_handling.backends.opentrons_backend import ( _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, OpentronsOT2Backend, ) +logger = logging.getLogger(__name__) + class _RecordingNamespace: """An ot_api sub-namespace (e.g. ``lh``, ``health``) that records every call. @@ -80,9 +84,12 @@ def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool def log(self, qualified: str, args: tuple, kwargs: dict): self.calls.append((qualified, args, kwargs)) + parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] + rendered = f"{qualified}({', '.join(parts)})" + # log at LOG_LEVEL_IO so a dry run captures the same wire trace as a real run + logger.log(LOG_LEVEL_IO, "%s", rendered) if self._verbose: - parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] - print(f"{qualified}({', '.join(parts)})") + print(rendered) def __getattr__(self, name: str): # top-level functions the backend calls directly: set_host, set_port, set_run From 93a44d34bafacb9b16259951b306c479444091d5 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:43:24 +0100 Subject: [PATCH 6/9] `OpentronsOT2Backend`: model a multi mount as 8 channels (head8 step 1) First slice of building the 8-channel head into the backend itself. Adds the channel model: _pipette_channel_count (8 for a multi, 1 for a single), _channel_map (per-mount channel blocks, left then right), num_channels derived from it, and _pipette_id_for_channel routed through it. A multi mount now reports 8 channels (0 = back / row A) instead of 1; single-channel setups are unchanged, so the characterization net and the chatterbox tests stay green. Per-column command issuing and the pickup geometry guard build on this next. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend.py | 35 ++++++++++++++----- .../backends/opentrons_backend_tests.py | 27 ++++++++++++++ 2 files changed, 54 insertions(+), 8 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index ba944988229..d863bc83128 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -149,9 +149,31 @@ async def setup(self, skip_home: bool = False): if not skip_home: await self.home() + @staticmethod + def _pipette_channel_count(pipette: Optional[Dict[str, str]]) -> int: + """Number of channels a mounted pipette presents: 8 for a multi, 1 for a single.""" + if pipette is None: + return 0 + return 8 if "multi" in pipette["name"] else 1 + + def _channel_map(self) -> List[Tuple[Dict[str, str], int]]: + """Per-mount channel blocks: channel index -> (pipette, nozzle index within it). + + The left mount's channels come first, then the right mount's. A p20-multi on + the left plus a p300-single on the right gives channels 0-7 (the multi's + nozzles, 0 = back / row A) and channel 8 (the single). + """ + channels: List[Tuple[Dict[str, str], int]] = [] + for pipette in (self.left_pipette, self.right_pipette): + if pipette is None: + continue + for nozzle in range(self._pipette_channel_count(pipette)): + channels.append((pipette, nozzle)) + return channels + @property def num_channels(self) -> int: - return len([p for p in [self.left_pipette, self.right_pipette] if p is not None]) + return len(self._channel_map()) async def stop(self): """Cancel any active OT run, then clear labware definitions.""" @@ -630,14 +652,11 @@ async def list_connected_modules(self) -> List[dict]: return cast(List[dict], self._ot.modules.list_connected_modules()) def _pipette_id_for_channel(self, channel: int) -> str: - pipettes = [] - if self.left_pipette is not None: - pipettes.append(self.left_pipette["pipetteId"]) - if self.right_pipette is not None: - pipettes.append(self.right_pipette["pipetteId"]) - if channel < 0 or channel >= len(pipettes): + channel_map = self._channel_map() + if channel < 0 or channel >= len(channel_map): raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.") - return pipettes[channel] + pipette, _nozzle = channel_map[channel] + return cast(str, pipette["pipetteId"]) def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]: """Return the pipette id and current coordinate for a given channel.""" diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py index b892986837f..17d121b020b 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py @@ -369,3 +369,30 @@ def test_set_tip_state_right(self): self.backend._set_tip_state("right-id", True) self.assertFalse(self.backend.left_pipette_has_tip) self.assertTrue(self.backend.right_pipette_has_tip) + + # -- channel model (head8 step 1) -- + + def test_channel_map_two_singles_is_one_channel_per_mount(self): + """Two single-channel pipettes give two channels, one per mount (unchanged).""" + backend = _make_backend_with_pipettes("p300_single_gen2", "p20_single_gen2") + self.assertEqual(backend.num_channels, 2) + self.assertEqual(backend._pipette_id_for_channel(0), "left-id") + self.assertEqual(backend._pipette_id_for_channel(1), "right-id") + + def test_channel_map_multi_mount_is_eight_channels(self): + """A multi on the left + a single on the right gives channels 0-7 (the multi's + nozzles) and channel 8 (the single).""" + backend = _make_backend_with_pipettes("p20_multi_gen2", "p300_single_gen2") + self.assertEqual(backend.num_channels, 9) + self.assertTrue(all(pip is backend.left_pipette for pip, _ in backend._channel_map()[:8])) + self.assertIs(backend._channel_map()[8][0], backend.right_pipette) + self.assertEqual(backend._pipette_id_for_channel(0), "left-id") + self.assertEqual(backend._pipette_id_for_channel(7), "left-id") + self.assertEqual(backend._pipette_id_for_channel(8), "right-id") + + def test_pipette_id_for_channel_out_of_range_raises(self): + """Channels beyond the mounted pipettes raise NoChannelError.""" + backend = _make_backend_with_pipettes("p20_single_gen2", None) + self.assertEqual(backend.num_channels, 1) + with self.assertRaises(NoChannelError): + backend._pipette_id_for_channel(1) From f305d4868ffd38d9ba08f4971513935fe270410b Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:59:53 +0100 Subject: [PATCH 7/9] `OpentronsOT2Backend`: one command per column for multi ops (head8 step 2) Routes pick_up_tips / aspirate / dispense / drop_tips through a new _resolve_pipette_and_primary(ops, use_channels): it maps the channels to the single mount they address (rejecting a mix of mounts) and picks the primary op (lowest nozzle). A multi column op now issues exactly one ot_api command at the primary well while PLR tracks all 8 channels. can_pick_up_tip is likewise routed through the channel map so the multi's channels 1-7 are accepted (previously hardcoded channel 0 = left, 1 = right). The single-channel _get_*_pipette helpers stay - OpentronsOT2Simulator still uses them. Verified in simulation through the chatterbox: one pick_up_tip / one aspirate_in_place / one dispense_in_place for a full 8-channel column, all eight channels tracked, and a cross-mount selection rejected. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend.py | 87 ++++++++++++++----- .../backends/opentrons_chatterbox_tests.py | 61 +++++++++++++ 2 files changed, 124 insertions(+), 24 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index d863bc83128..b4f1e42ef51 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -1,7 +1,7 @@ import inspect import logging import uuid -from typing import Any, Dict, List, Optional, Tuple, Union, cast +from typing import Any, Dict, List, Optional, Sequence, Tuple, Union, cast from pylabrobot import utils from pylabrobot.io import LOG_LEVEL_IO @@ -18,6 +18,7 @@ MultiHeadDispensePlate, Pickup, PickupTipRack, + PipettingOp, ResourceDrop, ResourceMove, ResourcePickup, @@ -312,6 +313,35 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): self._tip_racks[tip_rack.name] = slot + def _resolve_pipette_and_primary( + self, ops: Sequence[PipettingOp], use_channels: List[int] + ) -> Tuple[str, PipettingOp]: + """Map ``use_channels`` to the single pipette they address, plus the primary op. + + All channels in one operation must belong to the same pipette/mount (a multi + pipette is a ganged head - the OT-2 cannot operate two mounts in one command; + issue separate calls per mount). The primary op is the one on the lowest + nozzle index (nozzle 0 = back / row A). The single ``ot_api`` command targets + the primary op's well; the firmware fans the remaining nozzles out from there. + """ + channel_map = self._channel_map() + pipette_ids = set() + primary_op: Optional[PipettingOp] = None + primary_nozzle: Optional[int] = None + for op, channel in zip(ops, use_channels): + if not 0 <= channel < len(channel_map): + raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.") + pipette, nozzle = channel_map[channel] + pipette_ids.add(cast(str, pipette["pipetteId"])) + if primary_nozzle is None or nozzle < primary_nozzle: + primary_nozzle, primary_op = nozzle, op + if len(pipette_ids) != 1 or primary_op is None: + raise NoChannelError( + "All channels in one operation must address the same pipette (mount); " + "issue separate calls per mount." + ) + return pipette_ids.pop(), primary_op + def _get_pickup_pipette(self, ops: List[Pickup]) -> str: """Get the pipette for a tip pick-up, or raise.""" assert len(ops) == 1, "only one channel supported for now" @@ -361,10 +391,13 @@ def _set_tip_state(self, pipette_id: str, has_tip: bool): raise ValueError(f"Unknown or unconfigured pipette_id {pipette_id!r} in _set_tip_state.") async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): - """Pick up tips from the specified resource.""" + """Pick up tips from the specified resource. - pipette_id = self._get_pickup_pipette(ops) - op = ops[0] + A multi-channel pickup (one op per channel) issues a single ``ot_api`` command + targeting the primary op's well; the firmware engages the remaining nozzles. + """ + + pipette_id, op = self._resolve_pipette_and_primary(ops, use_channels) offset_x, offset_y, offset_z = ( op.offset.x, @@ -392,10 +425,13 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): self._set_tip_state(pipette_id, True) async def drop_tips(self, ops: List[Drop], use_channels: List[int]): - """Drop tips from the specified resource.""" + """Drop tips from the specified resource. - pipette_id = self._get_drop_pipette(ops) - op = ops[0] + A multi-channel drop issues one ``ot_api`` command at the primary op's well. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(Drop, primary) use_fixed_trash = ( cast(str, self.ot_api_version) >= _OT_DECK_IS_ADDRESSABLE_AREA_VERSION @@ -497,10 +533,14 @@ def _get_default_aspiration_flow_rate(self, pipette_name: str) -> float: }[pipette_name] async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int]): - """Aspirate liquid from the specified resource using pip.""" + """Aspirate liquid from the specified resource using pip. - pipette_id = self._get_liquid_pipette(ops) - op = ops[0] + A multi-channel aspirate issues one ``ot_api`` command at the primary op's + well; all nozzles draw the same volume. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(SingleChannelAspiration, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) @@ -572,10 +612,14 @@ def _get_default_dispense_flow_rate(self, pipette_name: str) -> float: }[pipette_name] async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int]): - """Dispense liquid from the specified resource using pip.""" + """Dispense liquid from the specified resource using pip. - pipette_id = self._get_liquid_pipette(ops) - op = ops[0] + A multi-channel dispense issues one ``ot_api`` command at the primary op's + well; all nozzles dispense the same volume. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(SingleChannelDispense, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) @@ -752,14 +796,9 @@ def supports_tip(channel_vol: float, tip_vol: float) -> bool: return tip_vol in {1000} raise ValueError(f"Unknown channel volume: {channel_vol}") - if channel_idx == 0: - if self.left_pipette is None: - return False - left_volume = OpentronsOT2Backend.pipette_name2volume[self.left_pipette["name"]] - return supports_tip(left_volume, tip.maximal_volume) - if channel_idx == 1: - if self.right_pipette is None: - return False - right_volume = OpentronsOT2Backend.pipette_name2volume[self.right_pipette["name"]] - return supports_tip(right_volume, tip.maximal_volume) - return False + channel_map = self._channel_map() + if channel_idx < 0 or channel_idx >= len(channel_map): + return False + pipette, _nozzle = channel_map[channel_idx] + channel_volume = OpentronsOT2Backend.pipette_name2volume[pipette["name"]] + return supports_tip(channel_volume, tip.maximal_volume) diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py index d9ee5f370de..f14a858b97c 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -11,6 +11,7 @@ OpentronsOT2ChatterboxBackend, OpentronsOT2Simulator, ) +from pylabrobot.liquid_handling.errors import NoChannelError from pylabrobot.resources import set_tip_tracking, set_volume_tracking from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul @@ -131,5 +132,65 @@ async def test_chatterbox_matches_simulator_single_channel(self): self.assertEqual(chatterbox_outcome, simulator_outcome) +class OpentronsChatterboxHead8Tests(unittest.IsolatedAsyncioTestCase): + """Multi-channel (head8) column operations, dry-run through the chatterbox.""" + + async def asyncSetUp(self): + set_tip_tracking(True) + set_volume_tracking(True) + self.backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_multi_gen2", + right_pipette_name="p300_single_gen2", + verbose=False, + ) + self.deck = OTDeck() + self.lh = LiquidHandler(backend=self.backend, deck=self.deck) + await self.lh.setup() + self.tips = opentrons_96_filtertiprack_20ul(name="tips") + self.deck.assign_child_at_slot(self.tips, slot=1) + self.plate = CellTreat_96_wellplate_350ul_Fb(name="plate") + self.deck.assign_child_at_slot(self.plate, slot=2) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_multi_mount_reports_eight_channels(self): + """The multi on the left contributes 8 channels, the single on the right 1.""" + self.assertEqual(self.backend.num_channels, 9) + + async def test_eight_channel_column_issues_one_command_each_and_tracks_all(self): + """A full-column pickup -> aspirate -> dispense on the 8 multi channels issues + exactly one ot_api command each, tracks all 8 channels, and fills the column.""" + rows = "ABCDEFGH" + channels = list(range(8)) + for r in rows: + self.plate.get_well(f"{r}1").tracker.set_volume(20) + + await self.lh.pick_up_tips([self.tips.get_item(f"{r}1") for r in rows], use_channels=channels) + await self.lh.aspirate( + [self.plate.get_item(f"{r}1") for r in rows], vols=[5.0] * 8, use_channels=channels + ) + await self.lh.dispense( + [self.plate.get_item(f"{r}2") for r in rows], vols=[5.0] * 8, use_channels=channels + ) + + names = [call[0] for call in self.backend.commands] + self.assertEqual(names.count("lh.pick_up_tip"), 1) + self.assertEqual(names.count("lh.aspirate_in_place"), 1) + self.assertEqual(names.count("lh.dispense_in_place"), 1) + self.assertTrue(all(self.lh.head[c].has_tip for c in channels)) + for r in rows: + self.assertEqual(self.plate.get_item(f"{r}2").tracker.get_used_volume(), 5.0) + + async def test_channels_spanning_two_mounts_is_rejected(self): + """Channels addressing both the multi (0) and the single (8) cannot be one + command - the OT-2 drives a single mount per call. The resolver only pairs ops + with channels, so plain sentinels stand in for the ops here.""" + ops = [object(), object()] + with self.assertRaises(NoChannelError): + self.backend._resolve_pipette_and_primary(ops, use_channels=[0, 8]) # type: ignore[arg-type] + + if __name__ == "__main__": unittest.main() From 93dfdb99e075c9de4ca4d68ed778995203f393cb Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 13:10:41 +0100 Subject: [PATCH 8/9] Audit OT-2 tests: drop redundant/weak cases and tautological asserts - Remove test_multi_mount_reports_eight_channels: num_channels==9 is already pinned by test_channel_map_multi_mount_is_eight_channels and exercised by the 8-channel column test. - Remove the chatterbox-vs-simulator differential: both backends share the PLR frontend that does the tracking, so they can only diverge by raising, which the full-protocol recording test already guards. - Strip tautological assertEqual(offset_x, offset_x) (x3 in each of test_tip_pick_up and test_tip_drop) and a duplicate well_name assert in test_tip_drop. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend_tests.py | 7 --- .../backends/opentrons_chatterbox_tests.py | 59 +------------------ 2 files changed, 1 insertion(+), 65 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py index 17d121b020b..ab45960d818 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py @@ -128,9 +128,6 @@ def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, off self.assertEqual(labware_id, self.backend.get_ot_name("tip_rack")) self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(offset_x, offset_x) - self.assertEqual(offset_y, offset_y) - self.assertEqual(offset_z, offset_z) mock_pick_up_tip.side_effect = assert_parameters @@ -139,12 +136,8 @@ def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, off @patch("ot_api.lh.drop_tip") async def test_tip_drop(self, mock_drop_tip): def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, offset_z): - self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(offset_x, offset_x) - self.assertEqual(offset_y, offset_y) - self.assertEqual(offset_z, offset_z) mock_drop_tip.side_effect = assert_parameters diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py index f14a858b97c..f8b302380a7 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -7,10 +7,7 @@ import unittest from pylabrobot.liquid_handling import LiquidHandler -from pylabrobot.liquid_handling.backends import ( - OpentronsOT2ChatterboxBackend, - OpentronsOT2Simulator, -) +from pylabrobot.liquid_handling.backends import OpentronsOT2ChatterboxBackend from pylabrobot.liquid_handling.errors import NoChannelError from pylabrobot.resources import set_tip_tracking, set_volume_tracking from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb @@ -82,56 +79,6 @@ def test_serialize_includes_pipettes(self): self.assertIsNone(serialized["right_pipette_name"]) -class OpentronsChatterboxVsSimulatorTests(unittest.IsolatedAsyncioTestCase): - """Differential audit: the chatterbox must produce the same tracked outcome as - the reference OpentronsOT2Simulator on the single-channel overlap. (The Simulator - is single-channel by construction, so the multi-channel head is out of scope here.)""" - - async def asyncSetUp(self): - set_tip_tracking(True) - set_volume_tracking(True) - - async def asyncTearDown(self): - set_tip_tracking(False) - set_volume_tracking(False) - - async def _run_single_channel_protocol(self, backend): - deck = OTDeck() - lh = LiquidHandler(backend=backend, deck=deck) - await lh.setup() - tips = opentrons_96_filtertiprack_20ul(name="tips") - deck.assign_child_at_slot(tips, slot=1) - plate = CellTreat_96_wellplate_350ul_Fb(name="plate") - deck.assign_child_at_slot(plate, slot=2) - plate.get_well("A1").tracker.set_volume(15) - - await lh.pick_up_tips(tips["A1"]) - await lh.aspirate(plate["A1"], vols=[10]) - await lh.dispense(plate["B1"], vols=[10]) - - outcome = ( - lh.head[0].has_tip, - round(plate.get_well("A1").tracker.get_used_volume(), 3), - round(plate.get_well("B1").tracker.get_used_volume(), 3), - ) - await lh.stop() - return outcome - - async def test_chatterbox_matches_simulator_single_channel(self): - simulator_outcome = await self._run_single_channel_protocol( - OpentronsOT2Simulator( - left_pipette_name="p20_single_gen2", right_pipette_name="p20_single_gen2" - ) - ) - chatterbox_outcome = await self._run_single_channel_protocol( - OpentronsOT2ChatterboxBackend( - left_pipette_name="p20_single_gen2", right_pipette_name="p20_single_gen2", verbose=False - ) - ) - self.assertEqual(simulator_outcome, (True, 5.0, 10.0)) - self.assertEqual(chatterbox_outcome, simulator_outcome) - - class OpentronsChatterboxHead8Tests(unittest.IsolatedAsyncioTestCase): """Multi-channel (head8) column operations, dry-run through the chatterbox.""" @@ -155,10 +102,6 @@ async def asyncTearDown(self): set_tip_tracking(False) set_volume_tracking(False) - async def test_multi_mount_reports_eight_channels(self): - """The multi on the left contributes 8 channels, the single on the right 1.""" - self.assertEqual(self.backend.num_channels, 9) - async def test_eight_channel_column_issues_one_command_each_and_tracks_all(self): """A full-column pickup -> aspirate -> dispense on the 8 multi channels issues exactly one ot_api command each, tracks all 8 channels, and fills the column.""" From 8d55661bdabb0c8312531a3b516f655db0b82172 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 13:54:11 +0100 Subject: [PATCH 9/9] `OpentronsOT2Backend`: head8 partial-pickup geometry guard (head8 step 3) Adds _check_head8_pickup: a multi pickup must use a channel-0-anchored contiguous block [0..k] (the API anchors at the back nozzle and only declares front-anchored layouts), and the head also grabs occupied tipspots below the selection within its 8-nozzle reach. pick_up_tips rejects such a pickup unless allow_undeclared_tip_pickup is set (per call, keyword-only, or on the backend), in which case _absorb_undeclared_tips accounts for the grabbed tips so PLR tracking stays consistent, with a warning. The keyword-only override keeps OpentronsOT2Simulator's **kwargs signature compatible. Verified through the chatterbox: full column OK, back-anchored [1..7] rejected, grab-extra A1:F1 rejected, and allow_undeclared absorbs all eight with a warning. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../backends/opentrons_backend.py | 133 +++++++++++++++++- .../backends/opentrons_chatterbox.py | 2 + .../backends/opentrons_chatterbox_tests.py | 29 ++++ 3 files changed, 161 insertions(+), 3 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index b4f1e42ef51..9be532b7e1f 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -1,6 +1,7 @@ import inspect import logging import uuid +import warnings from typing import Any, Dict, List, Optional, Sequence, Tuple, Union, cast from pylabrobot import utils @@ -28,9 +29,10 @@ from pylabrobot.resources import ( Coordinate, Tip, + does_tip_tracking, ) from pylabrobot.resources.opentrons import OTDeck -from pylabrobot.resources.tip_rack import TipRack +from pylabrobot.resources.tip_rack import TipRack, TipSpot try: import ot_api @@ -98,7 +100,7 @@ class OpentronsOT2Backend(LiquidHandlerBackend): "p1000_single_gen3": 1000, } - def __init__(self, host: str, port: int = 31950): + def __init__(self, host: str, port: int = 31950, allow_undeclared_tip_pickup: bool = False): super().__init__() if not USE_OT: @@ -110,6 +112,10 @@ def __init__(self, host: str, port: int = 31950): self.host = host self.port = port + # Default for whether a multi pickup may absorb undeclared tips grabbed by unused + # nozzles; overridable per call via pick_up_tips(..., allow_undeclared_tip_pickup=). + self.allow_undeclared_tip_pickup = allow_undeclared_tip_pickup + # All hardware I/O goes through this handle so a subclass (e.g. the chatterbox) # can dry-run the backend by swapping it for a recording stand-in. The real handle # wraps ot_api to log every HTTP call at LOG_LEVEL_IO, like other backends' io. @@ -390,15 +396,136 @@ def _set_tip_state(self, pipette_id: str, has_tip: bool): raise ValueError(f"Unknown or unconfigured pipette_id {pipette_id!r} in _set_tip_state.") - async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): + def _check_head8_pickup( + self, ops: List[Pickup], use_channels: List[int] + ) -> List[Tuple[int, TipSpot]]: + """Validate an 8-channel pickup against what the Opentrons API can declare. + + A default-layout multi anchors the pickup at the referenced well - the back / + channel-0 nozzle - and fills toward the front (row H). The API only approves + front-anchored nozzle layouts, so use_channels must be a channel-0-anchored + contiguous block ``[0, 1, ..., k]``; a back-anchored set such as ``[1..7]`` + (leaving row A) cannot be declared at all. + + Returns the ``(nozzle, tipspot)`` pairs the head would also grab BELOW the + declared selection (occupied tipspots within its 8-nozzle reach) - undeclared + tips for the caller to reject or absorb. An empty list means a clean pickup. + """ + head8_pitch = 9.0 + num_nozzles = 8 + + ordered = sorted(use_channels) + if ordered != list(range(len(ordered))): + raise ValueError( + f"OT-2 8-channel pickup must use a channel-0-anchored contiguous block of channels " + f"[0, 1, ..., k]; got use_channels={ordered}. The Opentrons API anchors a multi pickup " + f"at the back (channel-0) nozzle and only approves front-anchored nozzle layouts, so a " + f"back-anchored selection such as [1..7] (which would leave row A and overhang the back " + f"edge) cannot be declared. To pick fewer tips, keep channel 0 and use the topmost rows." + ) + + def loc(resource): + return resource.get_absolute_location("c", "c", "b") + + tip_spots = [op.resource for op in ops] + rack = tip_spots[0].parent + if not isinstance(rack, TipRack) or any(s.parent is not rack for s in tip_spots): + raise ValueError("OT-2 8-channel pickup must come from a single tip rack.") + + col_x = loc(tip_spots[0]).x + if any(abs(loc(s).x - col_x) > 0.5 for s in tip_spots): + raise ValueError( + "OT-2 8-channel pickup must be within a single column (all tipspots must share x)." + ) + + col_spots = [s for s in rack.get_all_items() if abs(loc(s).x - col_x) < 0.5] + top_y = max(loc(s).y for s in col_spots) + + def row_of(y: float) -> int: + return int(round((top_y - y) / head8_pitch)) + + spot_by_row = {row_of(loc(s).y): s for s in col_spots} + + # The declared tipspots must run consecutively down the column in channel order + # (channel 0 = topmost well, then one row per channel at 9 mm pitch). + offsets = {row_of(loc(op.resource).y) - ch for op, ch in zip(ops, use_channels)} + if len(offsets) != 1: + raise ValueError( + "OT-2 8-channel pickup tipspots must run consecutively down the column in channel order " + "(channel 0 = topmost well, then one row per channel at 9 mm pitch). The given tipspots " + "do not line up 1:1 with the channels." + ) + top_row = offsets.pop() + + # The head also grabs any occupied tipspot BELOW the selection, within its 8-nozzle reach + # (unused nozzles k+1..7). Those are undeclared tips. + used = set(use_channels) + extras: List[Tuple[int, TipSpot]] = [] + for nozzle in range(num_nozzles): + if nozzle in used: + continue + spot = spot_by_row.get(nozzle + top_row) + if spot is not None and spot.has_tip(): + extras.append((nozzle, spot)) + return extras + + def _absorb_undeclared_tips(self, extras: List[Tuple[int, TipSpot]]) -> None: + """Account for tips grabbed by unused nozzles so PLR tracking stays consistent. + + Mirrors the frontend's pickup bookkeeping (add tip to the channel, remove it from + the tipspot) for each nozzle the caller did not declare. + """ + if self._head is None: + return + for nozzle, spot in extras: + self._head[nozzle].add_tip(spot.get_tip(), origin=spot, commit=True) + if does_tip_tracking() and not spot.tracker.is_disabled: + spot.tracker.remove_tip(commit=True) + + async def pick_up_tips( + self, + ops: List[Pickup], + use_channels: List[int], + *, + allow_undeclared_tip_pickup: Optional[bool] = None, + ): """Pick up tips from the specified resource. A multi-channel pickup (one op per channel) issues a single ``ot_api`` command targeting the primary op's well; the firmware engages the remaining nozzles. + + ``allow_undeclared_tip_pickup`` overrides the backend default for this call only; + when ``None`` the backend's ``self.allow_undeclared_tip_pickup`` is used. When the + 8-nozzle head would also grab occupied tipspots below the selection, the pickup is + rejected unless this is True, in which case those tips are absorbed (with a warning). """ pipette_id, op = self._resolve_pipette_and_primary(ops, use_channels) + allow_undeclared = ( + self.allow_undeclared_tip_pickup + if allow_undeclared_tip_pickup is None + else allow_undeclared_tip_pickup + ) + if "multi" in self.get_pipette_name(pipette_id): + extras = self._check_head8_pickup(ops, use_channels) + if extras: + where = ", ".join(spot.name for _, spot in extras) + if not allow_undeclared: + raise ValueError( + f"OT-2 8-channel pickup would also grab undeclared tips at {where}. The head fills " + f"from channel 0 down toward row H, and these occupied tipspots sit below your " + f"selection within its 8-nozzle reach, so the hardware would pick them up too. " + f"Extend use_channels down to include them, clear those spots first, or pass " + f"allow_undeclared_tip_pickup=True (per call or on the backend) to absorb them." + ) + warnings.warn( + f"OT-2 8-channel pickup is also grabbing undeclared tips at {where} (below the " + f"selection) and absorbing them into tracking (allow_undeclared_tip_pickup=True).", + stacklevel=2, + ) + self._absorb_undeclared_tips(extras) + offset_x, offset_y, offset_z = ( op.offset.x, op.offset.y, diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py index c5b421f4652..a390c17fc30 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py @@ -125,6 +125,7 @@ def __init__( host: str = "chatterbox", port: int = 31950, api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + allow_undeclared_tip_pickup: bool = False, verbose: bool = True, ): """Initialize the chatterbox. @@ -149,6 +150,7 @@ def __init__( self._right_pipette_name = right_pipette_name self.host = host self.port = port + self.allow_undeclared_tip_pickup = allow_undeclared_tip_pickup left = ( {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py index f8b302380a7..1f10bbaa751 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -5,6 +5,7 @@ """ import unittest +import warnings from pylabrobot.liquid_handling import LiquidHandler from pylabrobot.liquid_handling.backends import OpentronsOT2ChatterboxBackend @@ -134,6 +135,34 @@ async def test_channels_spanning_two_mounts_is_rejected(self): with self.assertRaises(NoChannelError): self.backend._resolve_pipette_and_primary(ops, use_channels=[0, 8]) # type: ignore[arg-type] + async def test_back_anchored_pickup_is_rejected(self): + """use_channels must be a channel-0-anchored block; [1..7] (leaving row A) is rejected.""" + with self.assertRaises(ValueError): + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "BCDEFGH"], use_channels=list(range(1, 8)) + ) + + async def test_pickup_grabbing_undeclared_tips_below_is_rejected(self): + """Picking A1:F1 from a full column would also grab the occupied G1/H1 below it within + the head's 8-nozzle reach; rejected by default.""" + with self.assertRaises(ValueError): + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEF"], use_channels=list(range(6)) + ) + + async def test_allow_undeclared_absorbs_grabbed_tips(self): + """allow_undeclared_tip_pickup=True absorbs the extra grabbed tips into tracking (with a + warning) so all eight nozzles are accounted for.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEF"], + use_channels=list(range(6)), + allow_undeclared_tip_pickup=True, + ) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(8))) + self.assertTrue(any("undeclared tips" in str(w.message) for w in caught)) + if __name__ == "__main__": unittest.main()