OT-2 (legacy): fix _current_channel_position, and add move_channel_to

pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py

@@ -1,5 +1,7 @@
+import asyncio
 import inspect
 import logging
+import time
 import uuid
 from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
@@ -44,6 +46,8 @@
 # https://labautomation.io/t/connect-pylabrobot-to-ot2/2862/18
 _OT_DECK_IS_ADDRESSABLE_AREA_VERSION = "7.1.0"
 
+_SAVE_POSITION_TIMEOUT = 30.0
+
 logger = logging.getLogger(__name__)
 
 
@@ -650,12 +654,34 @@ def _pipette_id_for_channel(self, channel: int) -> str:
       raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.")
     return pipettes[channel]
 
-  def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
+  async def _save_position(self, pipette_id: str) -> Dict[str, Any]:
+    """Ask the robot where a pipette is, and wait for the answer.
+
+    ``ot_api`` wraps no ``savePosition``, so this enqueues the command and polls it
+    the way ``ot_api``'s own command wrapper does.
+    """
+
+    command_id = self._ot.runs.enqueue_command(
+      "savePosition", {"pipetteId": pipette_id}, intent="setup"
+    )
+    deadline = time.monotonic() + _SAVE_POSITION_TIMEOUT
+    while time.monotonic() < deadline:
+      result = self._ot.runs.get_command(command_id)
+      status = result["data"]["status"]
+      if status == "failed":
+        error = result["data"]["error"]
+        raise RuntimeError(f"savePosition failed with {error['errorType']}: {error['detail']}")
+      if status not in ("queued", "running"):
+        return result
+      await asyncio.sleep(0.05)
+    raise RuntimeError("savePosition timed out")
+
+  async def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
     """Return the pipette id and current coordinate for a given channel."""
 
     pipette_id = self._pipette_id_for_channel(channel)
     try:
-      res = self._ot.lh.save_position(pipette_id=pipette_id)
+      res = await self._save_position(pipette_id)
       pos = res["data"]["result"]["position"]
       current = Coordinate(pos["x"], pos["y"], pos["z"])
     except Exception as exc:
@@ -668,10 +694,40 @@ async def prepare_for_manual_channel_operation(self, channel: int):
 
     _ = self._pipette_id_for_channel(channel)
 
+  async def get_channel_position(self, channel: int) -> Coordinate:
+    """Where a channel is right now, in deck coordinates."""
+
+    _, current = await self._current_channel_position(channel)
+    return current
+
+  async def move_channel_to(
+    self,
+    channel: int,
+    x: Optional[float] = None,
+    y: Optional[float] = None,
+    z: Optional[float] = None,
+  ):
+    """Move a channel to an absolute position, holding the axes left out.
+
+    One coordinated move rather than the per-axis calls chained: the robot lifts to the traversal
+    height and travels once, where three separate moves each descend and can clip labware between
+    them.
+    """
+
+    pipette_id, current = await self._current_channel_position(channel)
+    target = Coordinate(
+      x=current.x if x is None else x,
+      y=current.y if y is None else y,
+      z=current.z if z is None else z,
+    )
+    await self.move_pipette_head(
+      location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
+    )
+
   async def move_channel_x(self, channel: int, x: float):
     """Move a channel to an absolute x coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=x, y=current.y, z=current.z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
@@ -680,7 +736,7 @@ async def move_channel_x(self, channel: int, x: float):
   async def move_channel_y(self, channel: int, y: float):
     """Move a channel to an absolute y coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=current.x, y=y, z=current.z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
@@ -689,7 +745,7 @@ async def move_channel_y(self, channel: int, y: float):
   async def move_channel_z(self, channel: int, z: float):
     """Move a channel to an absolute z coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=current.x, y=current.y, z=z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id

pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py

@@ -151,6 +151,58 @@ def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, off
     await self.test_tip_pick_up()
     await self.lh.drop_tips(self.tip_rack["A1"])
 
+  @staticmethod
+  def _at(x: float, y: float, z: float) -> dict:
+    return {
+      "data": {"status": "succeeded", "result": {"position": {"x": x, "y": y, "z": z}}},
+    }
+
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  async def test_get_channel_position_asks_the_robot_to_save_its_position(
+    self, mock_enqueue, mock_get_command
+  ):
+    mock_enqueue.return_value = "cmd-1"
+    mock_get_command.return_value = self._at(11.0, 22.0, 33.0)
+
+    position = await self.backend.get_channel_position(0)
+
+    self.assertEqual(position, Coordinate(11.0, 22.0, 33.0))
+    self.assertEqual(mock_enqueue.call_args.args[0], "savePosition")
+    self.assertEqual(mock_enqueue.call_args.args[1], {"pipetteId": "left-pipette-id"})
+
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  async def test_a_failed_save_position_names_the_robot_error(self, mock_enqueue, mock_get_command):
+    mock_enqueue.return_value = "cmd-1"
+    mock_get_command.return_value = {
+      "data": {
+        "status": "failed",
+        "error": {"errorType": "MustHomeError", "detail": "Must home first"},
+      }
+    }
+
+    with self.assertRaises(RuntimeError):
+      await self.backend.get_channel_position(0)
+
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  async def test_move_channel_to_travels_once_holding_the_axes_left_out(
+    self, mock_enqueue, mock_get_command, mock_move_arm
+  ):
+    mock_enqueue.return_value = "cmd-1"
+    mock_get_command.return_value = self._at(11.0, 22.0, 33.0)
+
+    await self.backend.move_channel_to(0, x=50.0, z=5.0)
+
+    mock_move_arm.assert_called_once()
+    kwargs = mock_move_arm.call_args.kwargs
+    self.assertEqual(kwargs["location_x"], 50.0)
+    self.assertEqual(kwargs["location_y"], 22.0)  # not named, so held
+    self.assertEqual(kwargs["location_z"], 5.0)
+    self.assertEqual(kwargs["minimum_z_height"], self.backend.traversal_height)
+
   @patch("ot_api.lh.aspirate_in_place")
   @patch("ot_api.lh.move_arm")
   async def test_aspirate(self, mock_move=None, mock_aspirate=None):