OpentronsOT2Backend: support the 8-channel pipette

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,
@@ -149,9 +150,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."""
@@ -292,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"
@@ -341,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,
@@ -372,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
@@ -486,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)
@@ -561,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)
@@ -641,14 +707,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."""
@@ -744,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)

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
 
@@ -382,3 +375,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)

pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py

@@ -7,12 +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
 
 
@@ -36,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):
@@ -81,54 +79,60 @@ 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.)"""
+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 _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),
+  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 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"
-      )
+    await self.lh.dispense(
+      [self.plate.get_item(f"{r}2") for r in rows], vols=[5.0] * 8, use_channels=channels
     )
-    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)
+
+    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__":