OpentronsOT2Backend: head8 motion-safety guards - reachability and collision

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_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
@@ -127,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
 
@@ -138,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
 
@@ -188,6 +182,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)."""
@@ -239,6 +286,19 @@ def test_get_pickup_pipette_raises_when_tip_already_mounted(self):
     with self.assertRaises(NoChannelError):
       self.backend._get_pickup_pipette(ops)
 
+  # -- _deck_to_robot_frame --
+
+  def test_deck_to_robot_frame_maps_slot1_corner_to_robot_origin(self):
+    """The deck->robot transform subtracts slot 1's corner, so a deck-frame point at slot 1's
+    corner becomes the robot origin and a point offset from it keeps that offset."""
+    self.backend.set_deck(self.deck)
+    corner = self.deck.slot_locations[0]
+    self.assertEqual(self.backend._deck_to_robot_frame(corner), Coordinate(0, 0, 0))
+    self.assertEqual(
+      self.backend._deck_to_robot_frame(corner + Coordinate(10, 20, 3)),
+      Coordinate(10, 20, 3),
+    )
+
   # -- _get_drop_pipette --
 
   def test_get_drop_pipette_selects_right_for_20ul(self):
@@ -315,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.py

@@ -0,0 +1,184 @@
+"""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.
+"""
+
+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,
+)
+from pylabrobot.resources.opentrons import OT2RobotGeometry
+
+logger = logging.getLogger(__name__)
+
+
+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))
+    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:
+      print(rendered)
+
+  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,
+    allow_undeclared_tip_pickup: bool = False,
+    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
+    self.allow_undeclared_tip_pickup = allow_undeclared_tip_pickup
+    self.geometry = OT2RobotGeometry()
+
+    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,
+    }