From c6d356b25ed10b9c5a07cbafa18afc2dc5b52a24 Mon Sep 17 00:00:00 2001 From: miikee Date: Thu, 20 Aug 2026 10:28:16 -0400 Subject: [PATCH] OT-2 (legacy): fix _current_channel_position, and add move_channel_to _current_channel_position called ot_api.lh.save_position, which ot_api does not define, so every caller raised: move_channel_x, move_channel_y and move_channel_z could not work at all. It now enqueues savePosition itself and polls the command the way ot_api's own command wrapper does. The poll awaits rather than sleeping, so a robot that never answers does not hold the event loop for the whole 30s budget. With the read working, move_channel_to moves a channel to an absolute position and holds whichever axes are left out. Chaining the per-axis calls was the only way to reach an arbitrary point before, and each of those descends separately, so a three-axis move could clip labware between the steps. This lifts to the traversal height and travels once. get_channel_position exposes the read. Not verified on an OT-2. The savePosition command and its response shape are the same ones a Flex uses, where they are hardware verified. --- .../backends/opentrons_backend.py | 66 +++++++++++++++++-- .../backends/opentrons_backend_tests.py | 52 +++++++++++++++ 2 files changed, 113 insertions(+), 5 deletions(-) diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py index bff19ae91c4..950804d7908 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py +++ b/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 diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py index 7f753803de8..007e50a19a9 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py +++ b/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):