Support OT-2 GEN2 eight-channel full-head operations

pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py

@@ -11,7 +11,7 @@
 """
 
 import logging
-from typing import Dict, List, Optional, Tuple, cast
+from typing import Dict, List, Literal, Optional, Tuple, cast
 
 from pylabrobot.io import LOG_LEVEL_IO
 from pylabrobot.legacy.liquid_handling.backends.backend import LiquidHandlerBackend
@@ -126,6 +126,7 @@ def __init__(
     port: int = 31950,
     api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION,
     verbose: bool = True,
+    fixed_head_mount: Optional[Literal["left", "right"]] = None,
   ):
     """Initialize the chatterbox.
 
@@ -135,6 +136,7 @@ def __init__(
       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.
+      fixed_head_mount: Mount whose eight nozzles operate together through the fixed-head route.
     """
     # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly.
     LiquidHandlerBackend.__init__(self)
@@ -149,6 +151,9 @@ def __init__(
     self._right_pipette_name = right_pipette_name
     self.host = host
     self.port = port
+    if fixed_head_mount not in {None, "left", "right"}:
+      raise ValueError("fixed_head_mount must be left, right, or None")
+    self.fixed_head_mount = fixed_head_mount
 
     left = (
       {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None
@@ -171,10 +176,13 @@ def commands(self) -> List[Tuple[str, tuple, dict]]:
     return cast(List[Tuple[str, tuple, dict]], self._ot.calls)
 
   def serialize(self) -> dict:
-    return {
+    data = {
       **LiquidHandlerBackend.serialize(self),
       "left_pipette_name": self._left_pipette_name,
       "right_pipette_name": self._right_pipette_name,
       "host": self.host,
       "port": self.port,
     }
+    if self.fixed_head_mount is not None:
+      data["fixed_head_mount"] = self.fixed_head_mount
+    return data

pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py

@@ -11,9 +11,22 @@
   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
+from pylabrobot.legacy.liquid_handling.standard import Mix
+from pylabrobot.resources import (
+  does_tip_tracking,
+  does_volume_tracking,
+  set_tip_tracking,
+  set_volume_tracking,
+)
+from pylabrobot.resources.celltreat import (
+  CellTreat_96_wellplate_350ul_Fb,
+  celltreat_96_wellplate_350uL_Fb,
+)
+from pylabrobot.resources.opentrons import (
+  OTDeck,
+  opentrons_96_filtertiprack_20ul,
+  opentrons_96_tiprack_300ul,
+)
 
 
 def _names(backend: OpentronsOT2ChatterboxBackend):
@@ -131,5 +144,89 @@ async def test_chatterbox_matches_simulator_single_channel(self):
     self.assertEqual(chatterbox_outcome, simulator_outcome)
 
 
+class OpentronsFixedHeadChatterboxTests(unittest.IsolatedAsyncioTestCase):
+  """Exercise full-head operation through the normal frontend and recorded transport."""
+
+  async def test_both_models_and_mounts_preserve_eight_well_state_and_command_sequence(self):
+    """Pickup, mixing, transfer, and discard use the selected mount once per primitive."""
+    previous_tip_tracking = does_tip_tracking()
+    previous_volume_tracking = does_volume_tracking()
+    set_tip_tracking(True)
+    set_volume_tracking(True)
+    try:
+      for model, volume, flow, rack_factory in (
+        ("p20_multi_gen2", 1, 7.6, opentrons_96_filtertiprack_20ul),
+        ("p20_multi_gen2", 20, 7.6, opentrons_96_filtertiprack_20ul),
+        ("p300_multi_gen2", 20, 94, opentrons_96_tiprack_300ul),
+        ("p300_multi_gen2", 300, 94, opentrons_96_tiprack_300ul),
+      ):
+        for mount in ("left", "right"):
+          with self.subTest(model=model, mount=mount, volume=volume):
+            backend = OpentronsOT2ChatterboxBackend(
+              left_pipette_name=model if mount == "left" else None,
+              right_pipette_name=model if mount == "right" else None,
+              fixed_head_mount=mount,
+              verbose=False,
+            )
+            deck = OTDeck()
+            lh = LiquidHandler(backend=backend, deck=deck)
+            await lh.setup(skip_home=True)
+            self.assertEqual(list(lh.head), list(range(8)))
+            tips = rack_factory(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)
+            for well in plate["A1:H1"]:
+              well.tracker.set_volume(volume + 10)
+            await lh.pick_up_tips(tips["A1:H1"])
+            await lh.aspirate(plate["A1:H1"], vols=[volume] * 8, mix=[Mix(volume, 2, flow)] * 8)
+            await lh.dispense(plate["A2:H2"], vols=[volume] * 8)
+            await lh.discard_tips()
+
+            primitives = [
+              (name, args, kwargs)
+              for name, args, kwargs in backend.commands
+              if name
+              in {
+                "lh.pick_up_tip",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.drop_tip_in_place",
+              }
+            ]
+            self.assertEqual(
+              [name for name, _, _ in primitives],
+              [
+                "lh.pick_up_tip",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.drop_tip_in_place",
+              ],
+            )
+            self.assertTrue(
+              all(kwargs["pipette_id"] == f"chatterbox-{mount}" for _, _, kwargs in primitives)
+            )
+            self.assertTrue(
+              all(
+                kwargs["volume"] == volume and kwargs["flow_rate"] == flow
+                for name, _, kwargs in primitives
+                if name in {"lh.aspirate_in_place", "lh.dispense_in_place"}
+              )
+            )
+            self.assertEqual([well.tracker.get_used_volume() for well in plate["A1:H1"]], [10] * 8)
+            self.assertEqual(
+              [well.tracker.get_used_volume() for well in plate["A2:H2"]], [volume] * 8
+            )
+            self.assertTrue(all(not tracker.has_tip for tracker in lh.head.values()))
+            self.assertEqual(backend.serialize()["fixed_head_mount"], mount)
+    finally:
+      set_tip_tracking(previous_tip_tracking)
+      set_volume_tracking(previous_volume_tracking)
+
+
 if __name__ == "__main__":
   unittest.main()

pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py

@@ -50,6 +50,7 @@ def __init__(
     """
     # Skip OpentronsOT2Backend.__init__ (requires ot_api); call grandparent directly.
     LiquidHandlerBackend.__init__(self)
+    self.fixed_head_mount = None
 
     pv = OpentronsOT2Backend.pipette_name2volume
     if left_pipette_name is not None and left_pipette_name not in pv:

pylabrobot/legacy/liquid_handling/liquid_handler.py

@@ -752,14 +752,6 @@ async def pick_up_tips(
       for tip_spot, offset, tip in zip(tip_spots, offsets, tips)
     ]
 
-    # queue operations on the trackers
-    for channel, op in zip(use_channels, pickups):
-      if self.head[channel].has_tip:
-        raise HasTipError("Channel has tip")
-      if does_tip_tracking() and not op.resource.tracker.is_disabled:
-        op.resource.tracker.remove_tip()
-      self.head[channel].add_tip(op.tip, origin=op.resource, commit=False)
-
     # fix the backend kwargs
     extras = self._check_args(
       self.backend.pick_up_tips,
@@ -770,6 +762,21 @@ async def pick_up_tips(
     for extra in extras:
       del backend_kwargs[extra]
 
+    # Queue the complete request before entering the hardware boundary.
+    try:
+      for channel, op in zip(use_channels, pickups):
+        if self.head[channel].has_tip:
+          raise HasTipError("Channel has tip")
+        if does_tip_tracking() and not op.resource.tracker.is_disabled:
+          op.resource.tracker.remove_tip()
+        self.head[channel].add_tip(op.tip, origin=op.resource, commit=False)
+    except BaseException:
+      for channel, op in zip(use_channels, pickups):
+        if does_tip_tracking() and not op.resource.tracker.is_disabled:
+          op.resource.tracker.rollback()
+        self.head[channel].rollback()
+      raise
+
     # actually pick up the tips
     error: Optional[BaseException] = None
     try:
@@ -898,16 +905,6 @@ async def drop_tips(
       for tip_spot, tip, offset in zip(tip_spots, tips, offsets)
     ]
 
-    # queue operations on the trackers
-    for channel, op in zip(use_channels, drops):
-      if (
-        does_tip_tracking()
-        and isinstance(op.resource, TipSpot)
-        and not op.resource.tracker.is_disabled
-      ):
-        op.resource.tracker.add_tip(op.tip, commit=False)
-      self.head[channel].remove_tip()
-
     # fix the backend kwargs
     extras = self._check_args(
       self.backend.drop_tips,
@@ -918,6 +915,26 @@ async def drop_tips(
     for extra in extras:
       del backend_kwargs[extra]
 
+    try:
+      for channel, op in zip(use_channels, drops):
+        if (
+          does_tip_tracking()
+          and isinstance(op.resource, TipSpot)
+          and not op.resource.tracker.is_disabled
+        ):
+          op.resource.tracker.add_tip(op.tip, commit=False)
+        self.head[channel].remove_tip()
+    except BaseException:
+      for channel, op in zip(use_channels, drops):
+        if (
+          does_tip_tracking()
+          and isinstance(op.resource, TipSpot)
+          and not op.resource.tracker.is_disabled
+        ):
+          op.resource.tracker.rollback()
+        self.head[channel].rollback()
+      raise
+
     # actually drop the tips
     error: Optional[BaseException] = None
     try:

pylabrobot/legacy/tip_tracker.py

@@ -199,15 +199,20 @@ def serialize(self) -> dict:
     }
 
   def load_state(self, state: dict) -> None:
-    """Load a saved tip tracker state."""
+    """Load committed and pending tips, including the committed tip's liquid state."""
 
     tip_data, pending_tip_data = state.get("tip"), state.get("pending_tip")
     tip = Tip.deserialize(tip_data) if tip_data is not None else None
     pending_tip = Tip.deserialize(pending_tip_data) if pending_tip_data is not None else None
-    self._put(pending_tip)
     same = (tip is None) == (pending_tip is None) and (
       tip is None or tip.serialize() == cast(Tip, pending_tip).serialize()
     )
+    tip_state = state.get("tip_state")
+    if tip is not None and tip_state is not None:
+      tip.tracker.load_state(tip_state)
+    if same:
+      pending_tip = tip
+    self._put(pending_tip)
     self._before = _NOTHING_PENDING if same else tip
 
   def get_tip_origin(self) -> Optional["TipSpot"]:

pylabrobot/legacy/tip_tracker_tests.py

@@ -8,6 +8,7 @@
 from pylabrobot.resources import Coordinate
 from pylabrobot.resources.hamilton import HamiltonTip, hamilton_tip_300uL
 from pylabrobot.resources.hamilton.tip_creators import TIP_DIAMETER, TipSize
+from pylabrobot.resources.opentrons import opentrons_96_tiprack_300ul
 from pylabrobot.resources.tip import Tip
 
 
@@ -76,6 +77,64 @@ def test_load_empty_state(self):
     with self.assertRaises(NoTipError):
       tracker.get_tip()
 
+  def test_load_committed_tip_spot_state_remains_committed(self):
+    """A restored occupied spot retains its committed tip and liquid volume."""
+    spot = opentrons_96_tiprack_300ul("tips").get_item("A1")
+    spot.tracker.get_tip().tracker.set_volume(17)
+    state = spot.serialize_state()
+
+    spot.load_state(state)
+
+    self.assertEqual(spot.serialize_state(), state)
+    spot.tracker.rollback()
+    self.assertEqual(spot.serialize_state(), state)
+
+  def test_load_pending_tip_spot_removal_preserves_rollback(self):
+    """Rollback restores a removed tip with its saved liquid volume."""
+    spot = opentrons_96_tiprack_300ul("tips").get_item("A1")
+    spot.tracker.get_tip().tracker.set_volume(17)
+    committed = spot.serialize_state()
+    spot.tracker.remove_tip(commit=False)
+    pending = spot.serialize_state()
+
+    spot.load_state(pending)
+
+    self.assertFalse(spot.tracker.has_tip)
+    self.assertEqual(spot.tracker.get_tip().name, committed["tip"]["name"])
+    spot.tracker.rollback()
+    self.assertEqual(spot.serialize_state(), committed)
+
+  def test_load_channel_tip_preserves_pending_liquid_state(self):
+    """Restoration preserves uncommitted liquid changes until explicit rollback."""
+    tracker = TipTracker("channel")
+    tracker.add_tip(self.tip)
+    self.tip.tracker.set_volume(4)
+    self.tip.tracker.add_liquid(3)
+    saved = tracker.serialize()
+
+    restored = TipTracker("restored")
+    restored.load_state(saved)
+
+    self.assertEqual(restored.serialize(), saved)
+    restored.get_tip().tracker.rollback()
+    self.assertEqual(restored.get_tip().tracker.get_used_volume(), 4)
+
+  def test_load_pending_pickup_preserves_tip_transaction(self):
+    """A pending pickup remains pending and can be rolled back after restoration."""
+    tracker = TipTracker("channel")
+    tracker.add_tip(self.tip, commit=False)
+    saved = tracker.serialize()
+
+    restored = TipTracker("restored")
+    restored.load_state(saved)
+
+    self.assertEqual(restored.serialize(), saved)
+    self.assertTrue(restored.has_tip)
+    with self.assertRaises(NoTipError):
+      restored.get_tip()
+    restored.rollback()
+    self.assertFalse(restored.has_tip)
+
   def test_a_pending_removal_keeps_the_tip_readable_until_it_commits(self):
     """A liquid handler removes a spot's tip before its backend asks the spot which tip it is."""
     tracker = TipTracker(thing="tester")