From 3fc50da765005a49dd1b46576043ef643f266b45 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:43:24 +0100 Subject: [PATCH 1/3] `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 6401df24261..6096b4b7a0c 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.""" @@ -641,14 +663,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 4c2505487da..2324ab0dfe5 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py @@ -382,3 +382,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 1ab4b52848e1d5a202985c651a77bc8d34dbfdb4 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 12:59:53 +0100 Subject: [PATCH 2/3] `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 6096b4b7a0c..6cee2acb001 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, @@ -314,6 +315,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" @@ -363,10 +393,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, @@ -394,10 +427,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 @@ -508,10 +544,14 @@ def _deck_to_robot_frame(self, location: Coordinate) -> Coordinate: return location - cast(OTDeck, self.deck).slot_locations[0] 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) @@ -583,10 +623,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) @@ -763,14 +807,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 4244008a6a0341d62fbbec69e71a327492860f44 Mon Sep 17 00:00:00 2001 From: Camillo Moschner Date: Mon, 29 Jun 2026 13:10:41 +0100 Subject: [PATCH 3/3] 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 | 65 ++----------------- 2 files changed, 4 insertions(+), 68 deletions(-) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py index 2324ab0dfe5..ab4d4158317 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..50e8d996f5b 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -7,13 +7,10 @@ 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 +from pylabrobot.resources.celltreat import celltreat_96_wellplate_350uL_Fb from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul @@ -37,7 +34,7 @@ async def asyncSetUp(self): 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.plate = celltreat_96_wellplate_350uL_Fb(name="plate") self.deck.assign_child_at_slot(self.plate, slot=2) async def asyncTearDown(self): @@ -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.""" @@ -148,17 +95,13 @@ async def asyncSetUp(self): 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.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."""