OpentronsOT2ChatterboxBackend: a dry-run OT-2 backend with I/O logging, via a transport seam (#1132)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: BioCam

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

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,
 )
@@ -31,9 +34,6 @@
 try:
   import ot_api
 
-  # for run cancellation
-  import ot_api.requestor as _req
-
   USE_OT = True
 except ImportError as e:
   USE_OT = False
@@ -44,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 = ""):
+    self._target = target
+    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."""
@@ -77,8 +109,13 @@ 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. 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)
 
     self.ot_api_version: Optional[str] = None
     self.left_pipette: Optional[Dict[str, str]] = None
@@ -97,16 +134,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 +161,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 +268,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
@@ -244,7 +281,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,
@@ -323,7 +360,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,
@@ -363,15 +400,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,
@@ -472,18 +509,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,
@@ -544,20 +581,20 @@ 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,
     )
 
     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,
@@ -574,7 +611,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.")
@@ -601,7 +638,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 = []
@@ -618,7 +655,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
@@ -687,7 +724,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,

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)."""