From c6d356b25ed10b9c5a07cbafa18afc2dc5b52a24 Mon Sep 17 00:00:00 2001 From: miikee Date: Thu, 20 Aug 2026 10:28:16 -0400 Subject: [PATCH 1/2] 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): From a12122c2e31f3cbb005e5af1c8be2be861339d47 Mon Sep 17 00:00:00 2001 From: miikee Date: Thu, 20 Aug 2026 18:23:28 -0400 Subject: [PATCH 2/2] OT-2 (legacy): bound every request, take it off the event loop, and own the command queue ot_api reaches the robot through urlopen and passes it no timeout, so a request the robot never answers blocked the process for good. All 24 ot_api calls are issued from an async def and ran inline, so that block also froze every other device's coordination behind it. Every call now goes through one _request helper. A daemon thread keeps the loop free, asyncio.wait_for ends the wait at a real deadline, and a lock keeps commands one at a time the way the blocking calls used to for free. The thread is not the loop's shared executor, because asyncio.run() joins that executor on the way out: an unanswered request would move the hang from mid-command to shutdown and burn a pool slot every other backend shares. Three budgets on the constructor, named to match the Flex backend: request_timeout for one request/response, command_timeout for a command that waits on motion, status_poll_interval between status reads. The budget wraps the lock as well as the request, so a read documented at 7.5s cannot silently block for as long as whatever is ahead of it. Non-positive values are refused. Robot commands are enqueued and polled here rather than through ot_api's @command decorator, whose 30s ceiling no caller can raise and whose poll loop has no sleep in it. command_timeout is the real ceiling now, so a mix declared at ten minutes is not killed at thirty seconds inside ot_api. A command that times out is contained rather than abandoned: the run is stopped so what is still queued cannot execute, and further commands are refused until setup() starts a fresh run. Retrying an aspirate used to make the pipette aspirate twice from one well. stop() halts the run through the endpoint the robot-server actually has; the two it used before do not exist, so it spent two full request budgets on 404s. The give-up type is builtins.TimeoutError on every supported version. asyncio.wait_for raises asyncio.exceptions.TimeoutError below 3.11, which is not an OSError, so a downstream "except TimeoutError" missed every OT-2 timeout on 3.9 and 3.10 and caught it on 3.11+. _save_position and _current_channel_position are async now, and the simulator's override follows. The chatterbox's canned savePosition data returned None, so a dry run could not read a channel back at all; it now answers with the position the recorded moves sent the arm to. The tests fake the robot at its command queue, which is where the backend reaches it, so they pin the params a real robot would receive. --- .../backends/opentrons_backend.py | 462 +++++++++--- .../backends/opentrons_backend_tests.py | 692 +++++++++++++----- .../backends/opentrons_chatterbox.py | 67 +- .../backends/opentrons_chatterbox_tests.py | 51 +- .../backends/opentrons_simulator.py | 2 +- 5 files changed, 979 insertions(+), 295 deletions(-) diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py index 950804d7908..bf186e253db 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py @@ -1,9 +1,10 @@ import asyncio import inspect import logging +import threading import time import uuid -from typing import Any, Dict, List, Optional, Tuple, Union, cast +from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast from pylabrobot import utils from pylabrobot.io import LOG_LEVEL_IO @@ -46,10 +47,81 @@ # 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__) +# One request/response with the robot-server. A read of robot state, not a motion. +DEFAULT_REQUEST_TIMEOUT = 30.0 + +# One command, including the motion it performs. The OT-2's slowest single move is a +# full-stroke aspirate or dispense at a viscous-liquid flow rate. +DEFAULT_COMMAND_TIMEOUT = 120.0 + +# Delay between two reads of a running command's status. +DEFAULT_STATUS_POLL_INTERVAL = 0.05 + + +def _seconds_left(deadline: float) -> float: + return max(deadline - time.monotonic(), 0.0) + + +def _well_location(x: float, y: float, z: float, origin: str = "bottom") -> Dict[str, Any]: + """The ``wellLocation`` shape every well-addressed command takes.""" + return {"origin": origin, "offset": {"x": x, "y": y, "z": z}} + + +def _in_place( + volume: float, + flow_rate: float, + pipette_id: str, + push_out: Optional[bool] = None, +) -> Dict[str, Any]: + """Params for an ``aspirateInPlace``/``dispenseInPlace`` command.""" + params: Dict[str, Any] = {"flowRate": flow_rate, "volume": volume, "pipetteId": pipette_id} + if push_out is not None: + params["pushOut"] = push_out + return params + + +async def _call_off_loop(call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """Run a blocking call on a thread of its own and await its result. + + Not ``asyncio.to_thread``: that borrows the loop's shared executor, whose threads + ``asyncio.run()`` joins on the way out, so one request nobody ever answers would + hang the process at shutdown and burn a pool slot every other backend shares. A + daemon thread holds neither. + + A thread per call, not a queue of one, and that is deliberate. ``ot_api`` gives + ``urlopen`` no socket timeout, so a request the robot never answers leaks its + thread for the life of the process. Queueing behind it would mean the stop that + contains a timed-out move could never reach the robot either, which is the worse + failure: the robot would keep executing what we stopped waiting for. The cost is + that after a timeout a second call can be in flight against the same robot while + the first is still stuck. + """ + loop = asyncio.get_running_loop() + future: "asyncio.Future[Any]" = loop.create_future() + + def _settle(setter: Callable[[Any], None], value: Any) -> None: + if not future.done(): + setter(value) + + def _deliver(setter: Callable[[Any], None], value: Any) -> None: + try: + loop.call_soon_threadsafe(_settle, setter, value) + except RuntimeError: + pass # the loop is gone, so nobody is waiting for this answer any more + + def _run() -> None: + try: + result = call(*args, **kwargs) + except BaseException as exc: + _deliver(future.set_exception, exc) + else: + _deliver(future.set_result, result) + + threading.Thread(target=_run, name="opentrons-request", daemon=True).start() + return await future + class _IOLogger: """Transparent proxy over the ``ot_api`` module that logs every call at @@ -77,6 +149,9 @@ def _logged(*args, **kwargs): logger.log(LOG_LEVEL_IO, "%s(%s)", qualified, ", ".join(parts)) return attr(*args, **kwargs) + # Without this every wrapped call answers to "_logged", and anything that + # names the call it is reporting on (a timeout message) names the proxy. + _logged.__name__ = _logged.__qualname__ = qualified return _logged return attr @@ -101,9 +176,27 @@ class OpentronsOT2Backend(LiquidHandlerBackend): "p1000_single_gen3": 1000, } - def __init__(self, host: str, port: int = 31950): + def __init__( + self, + host: str, + port: int = 31950, + request_timeout: float = DEFAULT_REQUEST_TIMEOUT, + command_timeout: float = DEFAULT_COMMAND_TIMEOUT, + status_poll_interval: float = DEFAULT_STATUS_POLL_INTERVAL, + ): + """Args: + host: the robot's address. + port: the robot-server's port. + request_timeout: how long one request/response with the robot may take, in + seconds. Includes the wait for whatever request is already in flight. + command_timeout: how long a command that moves the robot may take, in seconds. + Covers the motion itself, not just the request that started it. + status_poll_interval: delay between two reads of a running command's status. + """ super().__init__() + self._init_wire_state(request_timeout, command_timeout, status_poll_interval) + if not USE_OT: raise RuntimeError( "Opentrons is not installed. Please run pip install pylabrobot[opentrons]." @@ -129,6 +222,34 @@ def __init__(self, host: str, port: int = 31950): self._tip_racks: Dict[str, int] = {} # tip_rack.name -> slot index self._plr_name_to_load_name: Dict[str, str] = {} + def _init_wire_state( + self, + request_timeout: float, + command_timeout: float, + status_poll_interval: float, + ) -> None: + """Check and record the three budgets, and the state the wire layer keeps. + + Shared with the chatterbox, which skips this ``__init__`` because it has no + ``ot_api`` to talk to and would otherwise drift from what the wire layer expects. + """ + for name, value in ( + ("request_timeout", request_timeout), + ("command_timeout", command_timeout), + ("status_poll_interval", status_poll_interval), + ): + if value <= 0: + raise ValueError(f"{name} must be greater than 0, got {value}") + self.request_timeout = request_timeout + self.command_timeout = command_timeout + self.status_poll_interval = status_poll_interval + # Built on first use, not here: on 3.9 a Lock binds to whatever loop is current + # when it is constructed, and a backend is routinely built before asyncio.run(). + self._request_lock: Optional[asyncio.Lock] = None + # Set by a command timeout: the robot is still holding a command we stopped + # waiting for, so its pose is no longer ours to describe. + self._robot_state_unknown = False + def serialize(self) -> dict: return { **super().serialize(), @@ -136,18 +257,175 @@ def serialize(self) -> dict: "port": self.port, } + async def _request( + self, + call: Callable[..., Any], + *args: Any, + timeout: Optional[float] = None, + **kwargs: Any, + ) -> Any: + """Issue one ``ot_api`` call off the event loop, and give up after ``timeout``. + + ``ot_api`` reaches the robot with ``urlopen`` and passes it no socket timeout, so + an unanswered request blocks its thread for good. Running it off the loop keeps + the rest of the process going, and the wait_for ends OUR wait: a thread cannot be + cancelled, so the abandoned one finishes on its own. + + The budget covers the wait for the lock as well as the request. A caller told a + read is bounded at seven seconds must not sit behind someone else's mix for ten + minutes first, so the clock starts before the queue, not after it. + """ + + budget = self.request_timeout if timeout is None else timeout + try: + return await asyncio.wait_for(self._locked_call(call, *args, **kwargs), timeout=budget) + except asyncio.TimeoutError as exc: + # Before 3.11 asyncio.TimeoutError is not builtins.TimeoutError, so re-raising + # is what gives this backend one give-up type on every supported version. + raise TimeoutError( + f"{getattr(call, '__name__', call)} did not answer within {budget:g}s" + ) from exc + + async def _locked_call(self, call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + """One ``ot_api`` call at a time: the robot runs one queue and ``ot_api`` keeps the + run id in a module global. + + Per call, not per command. The blocking calls this replaced held the event loop for + a whole command; here the lock is released between an enqueue and its polls, so + keeping two commands off one robot is the caller's job. + """ + if self._request_lock is None: + self._request_lock = asyncio.Lock() + async with self._request_lock: + return await _call_off_loop(call, *args, **kwargs) + + async def _command( + self, + command_type: str, + params: Dict[str, Any], + timeout: Optional[float] = None, + abandon_run_on_timeout: bool = True, + ) -> Dict[str, Any]: + """Enqueue one run command and wait for the robot to finish it. + + ``ot_api``'s own command wrappers are not used for anything that moves: their + decorator hard-codes a 30s ceiling no caller can raise, and polls with no delay + between reads. Enqueue-and-poll here honours ``command_timeout`` instead. + + ``timeout`` bounds the waiting, not the whole call: the enqueue and each status + read carry a request budget of their own, so one unanswered request can carry the + call past its deadline by up to ``request_timeout``. A read that does not answer + is retried until the deadline: one lost GET says nothing about the motion, and + ending a ten-minute mix at the eight-second mark would take the abort decision + away from whoever asked for ten minutes. + + Set ``abandon_run_on_timeout`` False for a command that moves nothing. Giving up + on one of those leaves no motion outstanding, so halting the robot and refusing + everything after it would cost more than the timeout did. + """ + + self._refuse_if_state_unknown() + budget = self.command_timeout if timeout is None else timeout + deadline = time.monotonic() + budget + try: + command_id = await self._request( + self._ot.runs.enqueue_command, + command_type, + params, + intent="setup", + timeout=min(self.request_timeout, _seconds_left(deadline)), + ) + while True: + # A read is a request, so it gets a request budget. Handing it whatever is + # left of the deadline gives it milliseconds it cannot answer in, and then a + # command the robot finished reads as a timeout. + try: + result: Dict[str, Any] = await self._request(self._ot.runs.get_command, command_id) + except TimeoutError as exc: + remaining = _seconds_left(deadline) + if remaining <= 0: + raise TimeoutError(f"{command_type} did not finish within {budget:g}s") from exc + logger.warning( + "status read for %s did not answer; %.0fs of its budget left", command_type, remaining + ) + await asyncio.sleep(min(self.status_poll_interval, remaining)) + continue + data = result["data"] + status = data["status"] + if status == "failed": + error = data["error"] + raise RuntimeError(f"{command_type} failed with {error['errorType']}: {error['detail']}") + if status not in ("queued", "running"): + return result + # The deadline is checked after the read, so the read that follows the last + # sleep still happens: that is the one that sees a command which finished + # while we were asleep. + remaining = _seconds_left(deadline) + if remaining <= 0: + raise TimeoutError(f"{command_type} did not finish within {budget:g}s") + await asyncio.sleep(min(self.status_poll_interval, remaining)) + except TimeoutError: + if abandon_run_on_timeout: + await self._abandon_run() + raise + + def _refuse_if_state_unknown(self) -> None: + if self._robot_state_unknown: + raise RuntimeError( + "A command timed out and the OT-2 was left holding it, so its pose is " + "unknown. Recover with setup(), which starts a fresh run the stale command " + "cannot execute in, and homes. Check the pipettes by eye first: the OT-2 has " + "no tip sensor, ending a run does not drop tips, and setup() records both " + "mounts as empty, so a tip left on will be pressed into the rack." + ) + + async def _abandon_run(self) -> None: + """Stop the run a timed-out command is still sitting in, and refuse the next one. + + Giving up on the wait does not take the command out of the robot's queue: it will + still execute, so a caller who retries makes the robot aspirate twice from one + well. The stop action is what prevents that, but the robot-server only schedules + it and answers 201 straight away, so nothing here can confirm the run halted. The + refusal is the part that holds: it stands until ``setup()`` builds a new run, + which the stale command cannot execute in whatever the old one did. + """ + self._robot_state_unknown = True + run_id = getattr(self._ot, "run_id", None) + if not run_id: + return + try: + await self._stop_run(run_id) + except Exception: + logger.warning("could not stop run %s after a command timed out", run_id, exc_info=True) + + async def _stop_run(self, run_id: str) -> None: + """Halt a run, so nothing still queued in it executes.""" + await self._request( + self._ot.requestor.post, + f"/runs/{run_id}/actions", + {"data": {"actionType": "stop"}}, + ) + async def setup(self, skip_home: bool = False): # create run - run_id = self._ot.runs.create() + run_id = await self._request(self._ot.runs.create) self._ot.set_run(run_id) - # get pipettes, then assign them - self.left_pipette, self.right_pipette = self._ot.lh.add_mounted_pipettes() + # Only now is the unknown-state refusal lifted. Creating the run is what orphans + # whatever an earlier timeout left queued, and it is also the step most likely to + # fail here: the robot-server answers 409 while it still holds the old run. + self._robot_state_unknown = False + + # get pipettes, then assign them. This reads /pipettes and then loads each one, + # so it needs the command budget rather than a single request's. + self.left_pipette, self.right_pipette = await self._request( + self._ot.lh.add_mounted_pipettes, timeout=self.command_timeout + ) self.left_pipette_has_tip = self.right_pipette_has_tip = False # get api version - health = self._ot.health.get() + health = await self._request(self._ot.health.get) self.ot_api_version = health["api_version"] if not skip_home: @@ -164,19 +442,18 @@ async def stop(self): self.left_pipette = None self.right_pipette = None - # cancel the HTTP-API run if it exists (helpful to make device available again in official Opentrons app) + # release the run so the official Opentrons app can drive the robot again. Halt + # first: deleting a run the robot is still working through leaves it working. run_id = getattr(self._ot, "run_id", None) if run_id: try: - self._ot.requestor.post(f"/runs/{run_id}/cancel") + await self._stop_run(run_id) except Exception: - try: - self._ot.requestor.post(f"/runs/{run_id}/actions/cancel") - except Exception: - try: - self._ot.requestor.delete(f"/runs/{run_id}") - except Exception: - pass + logger.warning("could not stop run %s", run_id, exc_info=True) + try: + await self._request(self._ot.requestor.delete, f"/runs/{run_id}") + except Exception: + logger.warning("could not delete run %s", run_id, exc_info=True) def get_ot_name(self, plr_resource_name: str) -> str: """Opentrons only allows names in ^[a-z0-9._]+$, but in PLR we are flexible. @@ -272,7 +549,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): ], } - data = self._ot.labware.define(lw) + data = await self._request(self._ot.labware.define, lw) namespace, definition, version = data["data"]["definitionUri"].split("/") # assign labware to robot @@ -285,13 +562,16 @@ 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" - self._ot.labware.add( - load_name=definition, - namespace=namespace, - ot_location=slot, - version=version, - labware_id=labware_uuid, - display_name=self.get_ot_name(tip_rack.name), + await self._command( + "loadLabware", + { + "location": {"slotName": str(slot)}, + "loadName": definition, + "namespace": namespace, + "version": version, + "labwareId": labware_uuid, + "displayName": self.get_ot_name(tip_rack.name), + }, ) self._tip_racks[tip_rack.name] = slot @@ -364,13 +644,14 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): offset_z += op.tip.total_tip_length - 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, - offset_x=offset_x, - offset_y=offset_y, - offset_z=offset_z, + await self._command( + "pickUpTip", + { + "labwareId": self.get_ot_name(tip_rack.name), + "wellName": self.get_ot_name(op.resource.name), + "wellLocation": _well_location(offset_x, offset_y, offset_z), + "pipetteId": pipette_id, + }, ) self._set_tip_state(pipette_id, True) @@ -404,21 +685,26 @@ async def drop_tips(self, ops: List[Drop], use_channels: List[int]): offset_z += 10 if use_fixed_trash: - 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, + await self._command( + "moveToAddressableAreaForDropTip", + { + "pipetteId": pipette_id, + "addressableAreaName": "fixedTrash", + "wellName": "A1", + "wellLocation": _well_location(offset_x, offset_y, offset_z, origin="default"), + "alternateDropLocation": False, + }, ) - self._ot.lh.drop_tip_in_place(pipette_id=pipette_id) + await self._command("dropTipInPlace", {"pipetteId": pipette_id}) else: - self._ot.lh.drop_tip( - labware_id, - well_name=self.get_ot_name(op.resource.name), - pipette_id=pipette_id, - offset_x=offset_x, - offset_y=offset_y, - offset_z=offset_z, + await self._command( + "dropTip", + { + "labwareId": labware_id, + "wellName": self.get_ot_name(op.resource.name), + "wellLocation": _well_location(offset_x, offset_y, offset_z), + "pipetteId": pipette_id, + }, ) self._set_tip_state(pipette_id, False) @@ -513,22 +799,14 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[ if op.mix is not None: for _ in range(op.mix.repetitions): - self._ot.lh.aspirate_in_place( - volume=op.mix.volume, - flow_rate=op.mix.flow_rate, - pipette_id=pipette_id, + await self._command( + "aspirateInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id) ) - self._ot.lh.dispense_in_place( - volume=op.mix.volume, - flow_rate=op.mix.flow_rate, - pipette_id=pipette_id, + await self._command( + "dispenseInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id, push_out=False) ) - self._ot.lh.aspirate_in_place( - volume=volume, - flow_rate=flow_rate, - pipette_id=pipette_id, - ) + await self._command("aspirateInPlace", _in_place(volume, flow_rate, pipette_id)) traversal_location = self._deck_to_robot_frame( op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset @@ -585,23 +863,15 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in pipette_id=pipette_id, ) - self._ot.lh.dispense_in_place( - volume=volume, - flow_rate=flow_rate, - pipette_id=pipette_id, - ) + await self._command("dispenseInPlace", _in_place(volume, flow_rate, pipette_id, push_out=False)) if op.mix is not None: for _ in range(op.mix.repetitions): - self._ot.lh.aspirate_in_place( - volume=op.mix.volume, - flow_rate=op.mix.flow_rate, - pipette_id=pipette_id, + await self._command( + "aspirateInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id) ) - self._ot.lh.dispense_in_place( - volume=op.mix.volume, - flow_rate=op.mix.flow_rate, - pipette_id=pipette_id, + await self._command( + "dispenseInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id, push_out=False) ) traversal_location = self._deck_to_robot_frame( @@ -615,7 +885,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in ) async def home(self): - self._ot.health.home() + await self._request(self._ot.health.home, timeout=self.command_timeout) async def pick_up_tips96(self, pickup: PickupTipRack): raise NotImplementedError("The Opentrons backend does not support the 96 head.") @@ -642,7 +912,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], self._ot.modules.list_connected_modules()) + return cast(List[dict], await self._request(self._ot.modules.list_connected_modules)) def _pipette_id_for_channel(self, channel: int) -> str: pipettes = [] @@ -657,24 +927,16 @@ def _pipette_id_for_channel(self, channel: int) -> str: 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. + A read rather than a move, so it gets the request budget rather than the command + one: nothing here waits on motion. """ - command_id = self._ot.runs.enqueue_command( - "savePosition", {"pipetteId": pipette_id}, intent="setup" + return await self._command( + "savePosition", + {"pipetteId": pipette_id}, + timeout=self.request_timeout, + abandon_run_on_timeout=False, ) - 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.""" @@ -685,7 +947,7 @@ async def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate pos = res["data"]["result"]["position"] current = Coordinate(pos["x"], pos["y"], pos["z"]) except Exception as exc: - raise RuntimeError("Failed to query current pipette position") from exc + raise RuntimeError(f"Failed to query current pipette position: {exc}") from exc return pipette_id, current @@ -695,7 +957,8 @@ 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.""" + """Where a channel is right now, in the OT-2 robot frame (this file's own name + for the frame ``_deck_to_robot_frame`` converts PLR deck coordinates into).""" _, current = await self._current_channel_position(channel) return current @@ -780,15 +1043,16 @@ async def move_pipette_head( if pipette_id is None: raise ValueError("No pipette id given or left/right pipette not available.") - self._ot.lh.move_arm( - pipette_id=pipette_id, - location_x=location.x, - location_y=location.y, - location_z=location.z, - minimum_z_height=minimum_z_height, - speed=speed, - force_direct=force_direct, - ) + params: Dict[str, Any] = { + "pipetteId": pipette_id, + "coordinates": {"x": location.x, "y": location.y, "z": location.z}, + "forceDirect": force_direct, + } + if minimum_z_height is not None: + params["minimumZHeight"] = minimum_z_height + if speed is not None: + params["speed"] = speed + await self._command("moveToCoordinates", params) def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool: def supports_tip(channel_vol: float, tip_vol: float) -> bool: diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py index 007e50a19a9..df65364e0f7 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py @@ -1,4 +1,8 @@ +import asyncio +import threading +import time import unittest +from typing import Any, Dict, List, Optional, Tuple from unittest.mock import patch import pytest @@ -21,46 +25,67 @@ from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul from pylabrobot.resources.well import Well +_PIPETTES = ( + {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"}, + {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"}, +) + def _mock_define(lw): return {"data": {"definitionUri": f'lw["namespace"]/{lw["metadata"]["displayName"]}/1'}} -def _mock_add(load_name, namespace, ot_location, version, labware_id, display_name): - return labware_id - - def _mock_health_get(): return { "api_version": "7.0.1", } +class _FakeRobot: + """Stands in for the robot-server's command queue. + + Records what the backend enqueues and answers every poll "succeeded", so a test + can read the exact wire params the robot would have been handed. + """ + + def __init__(self, position: Tuple[float, float, float] = (0.0, 0.0, 0.0)) -> None: + self.commands: List[Tuple[str, Dict[str, Any]]] = [] + self.position = position + self.error: Optional[Dict[str, str]] = None + + def enqueue_command(self, command_type, params, intent="setup", **kwargs) -> str: + self.commands.append((command_type, dict(params))) + return f"cmd-{len(self.commands)}" + + def get_command(self, command_id, **kwargs) -> Dict[str, Any]: + if self.error is not None: + return {"data": {"status": "failed", "error": self.error}} + x, y, z = self.position + return {"data": {"status": "succeeded", "result": {"position": {"x": x, "y": y, "z": z}}}} + + def command_types(self) -> List[str]: + return [command_type for command_type, _params in self.commands] + + def params(self, command_type: str) -> Dict[str, Any]: + return dict(self.commands)[command_type] + + class OpentronsBackendSetupTests(unittest.IsolatedAsyncioTestCase): """Tests for setup and stop""" @patch("ot_api.runs.create") @patch("ot_api.health.home") @patch("ot_api.lh.add_mounted_pipettes") - @patch("ot_api.labware.add") - @patch("ot_api.labware.define") @patch("ot_api.health.get") async def test_setup( self, mock_health_get, - mock_define, - mock_add, mock_add_mounted_pipettes, mock_home, mock_create, ): mock_create.return_value = "run-id" - mock_add_mounted_pipettes.return_value = ( - {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"}, - {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"}, - ) - mock_add.side_effect = _mock_add - mock_define.side_effect = _mock_define + mock_add_mounted_pipettes.return_value = _PIPETTES mock_health_get.side_effect = _mock_health_get self.backend = OpentronsOT2Backend(host="localhost", port=1338) @@ -78,33 +103,40 @@ def test_serialize(self): "OpentronsOT2Backend", ) + def test_a_budget_of_zero_or_less_is_refused(self): + """A zero poll interval spins the robot-server; a zero command budget never polls.""" + for kwargs in ( + {"request_timeout": 0.0}, + {"command_timeout": -1.0}, + {"status_poll_interval": 0.0}, + ): + with self.subTest(**kwargs): + with self.assertRaises(ValueError): + OpentronsOT2Backend(host="localhost", port=1338, **kwargs) -class OpentronsBackendCommandTests(unittest.IsolatedAsyncioTestCase): - """Tests Opentrons commands""" - @patch("ot_api.runs.create") - @patch("ot_api.health.home") - @patch("ot_api.lh.add_mounted_pipettes") - @patch("ot_api.labware.add") - @patch("ot_api.labware.define") - @patch("ot_api.health.get") - async def asyncSetUp( - self, - mock_health_get, - mock_define, - mock_add, - mock_add_mounted_pipettes, - mock_home, - mock_create, - ): - mock_add.side_effect = _mock_add - mock_define.side_effect = _mock_define - mock_add_mounted_pipettes.return_value = ( - {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"}, - {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"}, - ) - mock_create.return_value = "run-id" - mock_health_get.side_effect = _mock_health_get +class OpentronsBackendCommandTests(unittest.IsolatedAsyncioTestCase): + """Tests Opentrons commands. + + The robot is faked at the command queue (``runs.enqueue_command`` / + ``runs.get_command``), which is where the backend actually reaches it, so every + assertion here is about the params a real robot would receive. + """ + + async def asyncSetUp(self): + self.robot = _FakeRobot() + for target, kwargs in ( + ("ot_api.runs.create", {"return_value": "run-id"}), + ("ot_api.runs.enqueue_command", {"side_effect": self.robot.enqueue_command}), + ("ot_api.runs.get_command", {"side_effect": self.robot.get_command}), + ("ot_api.lh.add_mounted_pipettes", {"return_value": _PIPETTES}), + ("ot_api.health.get", {"side_effect": _mock_health_get}), + ("ot_api.health.home", {}), + ("ot_api.labware.define", {"side_effect": _mock_define}), + ): + patcher = patch(target, **kwargs) + patcher.start() + self.addCleanup(patcher.stop) self.backend = OpentronsOT2Backend(host="localhost", port=1338) self.deck = OTDeck() @@ -115,184 +147,156 @@ async def asyncSetUp( self.deck.assign_child_at_slot(self.tip_rack, slot=1) self.plate = celltreat_96_wellplate_350uL_Fb(name="plate") self.deck.assign_child_at_slot(self.plate, slot=11) + self.robot.commands.clear() - @patch("ot_api.lh.pick_up_tip") - @patch("ot_api.labware.define") - @patch("ot_api.labware.add") - async def test_tip_pick_up(self, mock_add=None, mock_define=None, mock_pick_up_tip=None): - assert mock_pick_up_tip is not None and mock_define is not None and mock_add is not None - mock_define.side_effect = _mock_define - mock_add.side_effect = _mock_add - - def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, offset_z): - 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) + async def test_tip_pick_up(self): + await self.lh.pick_up_tips(self.tip_rack["A1"]) - mock_pick_up_tip.side_effect = assert_parameters + params = self.robot.params("pickUpTip") + self.assertEqual(params["labwareId"], self.backend.get_ot_name("tip_rack")) + self.assertEqual( + params["wellName"], self.backend.get_ot_name(self.tip_rack.get_item("A1").name) + ) + self.assertEqual(params["pipetteId"], "left-pipette-id") + self.assertEqual(params["wellLocation"]["origin"], "bottom") + async def test_a_tip_rack_is_loaded_into_the_run_before_its_first_pickup(self): await self.lh.pick_up_tips(self.tip_rack["A1"]) - @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 + params = self.robot.params("loadLabware") + self.assertEqual(params["location"], {"slotName": "1"}) + self.assertEqual(params["labwareId"], self.backend.get_ot_name("tip_rack")) + self.assertLess( + self.robot.command_types().index("loadLabware"), + self.robot.command_types().index("pickUpTip"), + ) - await self.test_tip_pick_up() + async def test_tip_drop(self): + await self.lh.pick_up_tips(self.tip_rack["A1"]) 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}}}, - } + params = self.robot.params("dropTip") + self.assertEqual( + params["wellName"], self.backend.get_ot_name(self.tip_rack.get_item("A1").name) + ) + self.assertEqual(params["pipetteId"], "left-pipette-id") - @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) + async def test_get_channel_position_asks_the_robot_to_save_its_position(self): + self.robot.position = (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"}) + self.assertEqual(self.robot.params("savePosition"), {"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"}, - } - } + async def test_a_failed_save_position_names_the_robot_error(self): + self.robot.error = {"errorType": "MustHomeError", "detail": "Must home first"} - with self.assertRaises(RuntimeError): + with self.assertRaises(RuntimeError) as caught: 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) + self.assertIn("MustHomeError", str(caught.exception)) - await self.backend.move_channel_to(0, x=50.0, z=5.0) + async def test_move_channel_to_travels_once_holding_the_axes_left_out(self): + self.robot.position = (11.0, 22.0, 33.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): - assert mock_aspirate is not None and mock_move is not None - - def assert_parameters( - volume, - flow_rate, - pipette_id, - ): - self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(volume, 10) - self.assertEqual(flow_rate, 3.78) + await self.backend.move_channel_to(0, x=50.0, z=5.0) - mock_aspirate.side_effect = assert_parameters + self.assertEqual(self.robot.command_types().count("moveToCoordinates"), 1) + params = self.robot.params("moveToCoordinates") + self.assertEqual(params["coordinates"]["x"], 50.0) + self.assertEqual(params["coordinates"]["y"], 22.0) # not named, so held + self.assertEqual(params["coordinates"]["z"], 5.0) + self.assertEqual(params["minimumZHeight"], self.backend.traversal_height) - await self.test_tip_pick_up() + async def test_aspirate(self): + await self.lh.pick_up_tips(self.tip_rack["A1"]) self.plate.get_well("A1").tracker.set_volume(10) - await self.lh.aspirate(self.plate["A1"], vols=[10]) - @patch("ot_api.lh.dispense_in_place") - @patch("ot_api.lh.move_arm") - async def test_dispense(self, mock_move, mock_dispense): - def assert_parameters( - volume, - flow_rate, - pipette_id, - ): - self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(volume, 10) - self.assertEqual(flow_rate, 7.56) + await self.lh.aspirate(self.plate["A1"], vols=[10]) - mock_dispense.side_effect = assert_parameters + self.assertEqual( + self.robot.params("aspirateInPlace"), + {"flowRate": 3.78, "volume": 10, "pipetteId": "left-pipette-id"}, + ) - await self.test_aspirate() # aspirate first + async def test_dispense(self): + await self.test_aspirate() 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) -- + self.assertEqual( + self.robot.params("dispenseInPlace"), + {"flowRate": 7.56, "volume": 10, "pipetteId": "left-pipette-id", "pushOut": False}, + ) - @patch("ot_api.health.home") - async def test_home_calls_health_home(self, mock_home): + async def test_a_motion_command_never_goes_through_ot_apis_capped_wrapper(self): + """``ot_api``'s ``@command`` decorator hard-codes a 30s ceiling and takes no + ``timeout`` kwarg, so anything routed through it ignores ``command_timeout``.""" + with patch("ot_api.lh.aspirate_in_place") as capped_wrapper: + await self.lh.pick_up_tips(self.tip_rack["A1"]) + self.plate.get_well("A1").tracker.set_volume(10) + await self.lh.aspirate(self.plate["A1"], vols=[10]) + + capped_wrapper.assert_not_called() + self.assertIn("aspirateInPlace", self.robot.command_types()) + + async def test_a_command_may_outlast_a_single_requests_budget(self): + """The request budget bounds one exchange; the command budget bounds the motion. + A move that inherited the request budget could never wait out a real motion.""" + answers = ["running", "running", "running", "succeeded"] + + def still_moving(command_id, **kwargs): + return {"data": {"status": answers.pop(0), "result": {}}} + + self.backend.request_timeout = 0.05 + self.backend.command_timeout = 5.0 + self.backend.status_poll_interval = 0.05 + with patch("ot_api.runs.get_command", side_effect=still_moving): + started = time.monotonic() + await self.backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + + self.assertGreater(time.monotonic() - started, self.backend.request_timeout) + self.assertEqual(answers, []) + + async def test_home_calls_health_home(self): """home() issues exactly one ot_api.health.home() call.""" - await self.backend.home() + with patch("ot_api.health.home") as mock_home: + 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): + async def test_list_connected_modules_passthrough(self): """list_connected_modules() returns ot_api.modules.list_connected_modules() verbatim.""" - mock_modules.return_value = [{"id": "tempdeck"}] - result = await self.backend.list_connected_modules() + with patch("ot_api.modules.list_connected_modules") as mock_modules: + 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") + async def test_stop_halts_the_run_then_releases_it(self): + """stop() halts the run before deleting it: deleting a run the robot is still + working through leaves it working.""" + with patch("ot_api.requestor.post") as mock_post, patch("ot_api.requestor.delete") as mock_del: + await self.backend.stop() + + mock_post.assert_called_once_with("/runs/run-id/actions", {"data": {"actionType": "stop"}}) + mock_del.assert_called_once_with("/runs/run-id") 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, - ): + async def test_tip_drop_to_trash_uses_addressable_area(self): """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 + area (moveToAddressableAreaForDropTip + dropTipInPlace), not dropTip.""" 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() + types = self.robot.command_types() + self.assertEqual(types.count("moveToAddressableAreaForDropTip"), 1) + self.assertEqual(types.count("dropTipInPlace"), 1) + self.assertNotIn("dropTip", types) def _make_backend_with_pipettes(left_name="p300_single_gen2", right_name="p20_single_gen2"): @@ -434,3 +438,349 @@ 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) + + +class OpentronsBackendTimeoutTests(unittest.IsolatedAsyncioTestCase): + """A robot that stops answering must not hang the process or freeze the event loop. + + ``ot_api`` calls ``urlopen`` with no socket timeout, so an unanswered request blocks + its thread for good. These pin that the backend stops waiting on its own, that the + rest of the process keeps running while it waits, and that giving up leaves neither + the loop nor the robot in a state the next caller can walk into. + """ + + def _backend(self, **kwargs: float) -> OpentronsOT2Backend: + backend = OpentronsOT2Backend(host="localhost", port=1338, **kwargs) + backend.left_pipette = {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"} + backend.right_pipette = None + return backend + + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + async def test_a_position_read_that_never_answers_fails_at_the_deadline( + self, mock_enqueue, mock_get_command + ): + mock_enqueue.return_value = "cmd-1" + released = threading.Event() + mock_get_command.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2) + started = time.monotonic() + try: + with self.assertRaises(RuntimeError) as caught: + await backend.get_channel_position(0) + finally: + released.set() # let the abandoned thread finish, so the suite can exit + + self.assertIsInstance(caught.exception.__cause__, TimeoutError) + self.assertLess(time.monotonic() - started, 5.0) + + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + async def test_a_slow_position_read_leaves_the_event_loop_free( + self, mock_enqueue, mock_get_command + ): + mock_enqueue.return_value = "cmd-1" + + def slow_answer(*args, **kwargs): + time.sleep(0.3) + return { + "data": {"status": "succeeded", "result": {"position": {"x": 1.0, "y": 2.0, "z": 3.0}}} + } + + mock_get_command.side_effect = slow_answer + + ticks = 0 + + async def tick(): + nonlocal ticks + while True: + await asyncio.sleep(0.01) + ticks += 1 + + ticker = asyncio.ensure_future(tick()) + try: + position = await self._backend().get_channel_position(0) + finally: + ticker.cancel() + + self.assertEqual(position, Coordinate(1.0, 2.0, 3.0)) + self.assertGreater(ticks, 1) # zero or one means the read blocked the loop + + @patch("ot_api.health.home") + async def test_a_robot_command_gets_the_longer_budget(self, mock_home): + mock_home.side_effect = lambda *a, **kw: time.sleep(0.3) + + # A move outlasts the plain request budget on purpose: it waits on the motion. + await self._backend(request_timeout=0.05, command_timeout=5.0).home() + + mock_home.assert_called_once() + + @patch("ot_api.modules.list_connected_modules") + async def test_a_plain_read_gets_the_shorter_budget(self, mock_list): + released = threading.Event() + mock_list.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2, command_timeout=60.0) + started = time.monotonic() + try: + with self.assertRaises(TimeoutError): + await backend.list_connected_modules() + finally: + released.set() + + self.assertLess(time.monotonic() - started, 5.0) + + @patch("ot_api.modules.list_connected_modules") + async def test_a_request_queued_behind_another_still_gives_up_at_its_own_budget(self, mock_list): + """The budget is what a caller is told a read costs. Starting the clock only after + the queue lets a 7s read block for as long as whatever is ahead of it takes.""" + released = threading.Event() + mock_list.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.3) + holder = asyncio.ensure_future(backend.list_connected_modules()) + await asyncio.sleep(0.05) # let the holder take the lock + + started = time.monotonic() + try: + with self.assertRaises(TimeoutError): + await asyncio.wait_for(backend.list_connected_modules(), timeout=5.0) + elapsed = time.monotonic() - started + finally: + released.set() + holder.cancel() + + self.assertLess(elapsed, 2.0) + + @patch("ot_api.modules.list_connected_modules") + async def test_an_abandoned_request_does_not_hold_the_loops_default_executor(self, mock_list): + """``asyncio.run()`` joins the default executor's threads on the way out, so a + request parked there moves the hang from mid-command to shutdown, and burns a + pool slot every other backend in the process shares.""" + released = threading.Event() + mock_list.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2) + try: + with self.assertRaises(TimeoutError): + await backend.list_connected_modules() + loop = asyncio.get_running_loop() + await asyncio.wait_for(loop.shutdown_default_executor(), timeout=3.0) + finally: + released.set() + + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_a_timed_out_move_stops_the_run_and_refuses_the_next_command( + self, mock_enqueue, mock_get_command, mock_post + ): + """Giving up on the wait leaves the command in the robot's queue, where it still + runs. A caller who retries would make the pipette aspirate twice from one well.""" + mock_enqueue.return_value = "cmd-1" + released = threading.Event() + mock_get_command.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2, command_timeout=0.2) + try: + with self.assertRaises(TimeoutError): + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + + mock_post.assert_called_once_with("/runs/run-id/actions", {"data": {"actionType": "stop"}}) + + with self.assertRaises(RuntimeError) as refused: + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + finally: + released.set() + + self.assertIn("setup()", str(refused.exception)) + self.assertEqual(mock_enqueue.call_count, 1) # the second attempt reached no robot + + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_a_command_that_finished_during_the_last_sleep_is_not_a_timeout( + self, mock_enqueue, mock_get_command, mock_post + ): + """The read after the final sleep is the one that sees a command the robot has + just finished. Handing it the sliver of deadline that is left cannot answer, and + then a move that worked halts the robot and latches the refusal.""" + mock_enqueue.return_value = "cmd-1" + answers = ["running"] + + def status(*args, **kwargs): + state = answers.pop(0) if answers else "succeeded" + return {"data": {"status": state, "result": {}}} + + mock_get_command.side_effect = status + + # A poll interval wider than the budget puts the whole remainder into one sleep, + # so the read that follows it lands exactly on the deadline. + backend = self._backend(request_timeout=5.0, command_timeout=0.2, status_poll_interval=1.0) + + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + + mock_post.assert_not_called() # nothing halted a run that completed + await backend.move_pipette_head(Coordinate(4.0, 5.0, 6.0), pipette_id="left") + self.assertEqual(mock_enqueue.call_count, 2) # and nothing latched the refusal + + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_a_status_read_that_never_answers_does_not_end_the_command( + self, mock_enqueue, mock_get_command, mock_post + ): + """A lost GET says nothing about the motion. Ending the command on one takes the + abort decision away from whoever asked for the longer budget: with a ten-minute + mix and a seven-second request budget, the driver would fire at eight seconds.""" + mock_enqueue.return_value = "cmd-1" + released = threading.Event() + reads = [] + + def status(*args, **kwargs): + reads.append(1) + if len(reads) == 1: + released.wait() # this one read never comes back + return {"data": {"status": "succeeded", "result": {}}} + + mock_get_command.side_effect = status + + backend = self._backend(request_timeout=0.2, command_timeout=5.0, status_poll_interval=0.01) + try: + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + finally: + released.set() + + self.assertEqual(len(reads), 2) # it polled again rather than giving up + mock_post.assert_not_called() # and nothing halted a run that was still running + await backend.move_pipette_head(Coordinate(4.0, 5.0, 6.0), pipette_id="left") + + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_a_slow_position_read_does_not_halt_the_robot( + self, mock_enqueue, mock_get_command, mock_post + ): + """savePosition moves nothing, so giving up on it leaves no motion outstanding. + Halting the run and refusing everything after would cost more than the timeout.""" + mock_enqueue.return_value = "cmd-1" + released = threading.Event() + mock_get_command.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2) + try: + with self.assertRaises(RuntimeError): + await backend.get_channel_position(0) + + mock_post.assert_not_called() + + # still usable: the refusal did not latch + with self.assertRaises(RuntimeError) as second: + await backend.get_channel_position(0) + finally: + released.set() + + self.assertNotIn("setup()", str(second.exception)) + self.assertEqual(mock_enqueue.call_count, 2) + + @patch("ot_api.health.home") + @patch("ot_api.health.get") + @patch("ot_api.lh.add_mounted_pipettes") + @patch("ot_api.runs.create") + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_a_setup_that_cannot_create_a_run_leaves_the_refusal_in_place( + self, + mock_enqueue, + mock_get_command, + mock_post, + mock_create, + mock_pipettes, + mock_health, + mock_home, + ): + """A fresh run is what the stale command cannot execute in, so until one exists + there is nothing to lift the refusal. The robot-server answering 409 while it + still holds the old run is the likeliest reason an operator is here at all.""" + mock_enqueue.return_value = "cmd-1" + released = threading.Event() + mock_get_command.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2, command_timeout=0.2) + try: + with self.assertRaises(TimeoutError): + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + + mock_create.side_effect = RuntimeError("RunConflictError") + with self.assertRaises(RuntimeError): + await backend.setup() + + with self.assertRaises(RuntimeError) as refused: + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + finally: + released.set() + + self.assertIn("setup()", str(refused.exception)) + self.assertEqual(mock_enqueue.call_count, 1) # nothing went into the stale run + + @patch("ot_api.health.home") + @patch("ot_api.health.get") + @patch("ot_api.lh.add_mounted_pipettes") + @patch("ot_api.runs.create") + @patch("ot_api.requestor.post") + @patch("ot_api.runs.get_command") + @patch("ot_api.runs.enqueue_command") + @patch("ot_api.run_id", "run-id", create=True) + async def test_setup_is_the_way_back_from_a_latched_refusal( + self, + mock_enqueue, + mock_get_command, + mock_post, + mock_create, + mock_pipettes, + mock_health, + mock_home, + ): + """The refusal names setup(); this is what makes that a real instruction.""" + mock_enqueue.return_value = "cmd-1" + mock_create.return_value = "run-2" + mock_pipettes.return_value = _PIPETTES + mock_health.side_effect = _mock_health_get + released = threading.Event() + mock_get_command.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2, command_timeout=0.2) + try: + with self.assertRaises(TimeoutError): + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + finally: + released.set() + + mock_get_command.side_effect = lambda *a, **kw: {"data": {"status": "succeeded", "result": {}}} + await backend.setup() + + await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left") + + @patch("ot_api.modules.list_connected_modules") + async def test_a_timeout_names_the_call_and_a_readable_budget(self, mock_list): + """An operator debugging a lossy OT-2 reads this line; the proxy the backend logs + through must not be what it names.""" + released = threading.Event() + mock_list.side_effect = lambda *a, **kw: released.wait() + + backend = self._backend(request_timeout=0.2) + try: + with self.assertRaises(TimeoutError) as caught: + await backend.list_connected_modules() + finally: + released.set() + + self.assertIn("modules.list_connected_modules", str(caught.exception)) + self.assertIn("0.2s", str(caught.exception)) diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py index 628fd67113e..98da141b2b0 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py +++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py @@ -3,6 +3,8 @@ 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. +Robot commands arrive as ``runs.enqueue_command(command_type, params)``, which is +what the backend puts on the wire, and each is answered "succeeded" on the poll. This mirrors how ``STARChatterboxBackend`` dry-runs ``STARBackend``: only the transport is replaced, so all the real high-level logic (pipette selection, tip @@ -17,6 +19,9 @@ from pylabrobot.legacy.liquid_handling.backends.backend import LiquidHandlerBackend from pylabrobot.legacy.liquid_handling.backends.opentrons_backend import ( _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + DEFAULT_COMMAND_TIMEOUT, + DEFAULT_REQUEST_TIMEOUT, + DEFAULT_STATUS_POLL_INTERVAL, OpentronsOT2Backend, ) @@ -55,18 +60,35 @@ 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. + ``runs.enqueue_command``/``runs.get_command``, ``lh.add_mounted_pipettes``, + ``health.get``, ``labware.define`` and ``modules.list_connected_modules`` return + canned data; everything else is recorded and returns ``None``. ``run_id`` stays + ``None`` so ``stop()`` skips the release requests. + + A recorded ``moveToCoordinates`` updates the position ``savePosition`` reports, + so a dry run that reads a channel back gets where it just sent it rather than + the origin every time. """ def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool = True): self.calls: List[Tuple[str, tuple, dict]] = [] + self.commands: List[Tuple[str, dict]] = [] # (command_type, params) in send order self.run_id: Optional[str] = None self._verbose = verbose - self.runs = _RecordingNamespace(self, "runs", {"create": lambda: "chatterbox-run"}) + self.position: Dict[str, float] = {"x": 0.0, "y": 0.0, "z": 0.0} + + self.runs = _RecordingNamespace( + self, + "runs", + { + "create": lambda: "chatterbox-run", + "enqueue_command": lambda: "chatterbox-command", + "get_command": lambda: { + "data": {"status": "succeeded", "result": {"position": dict(self.position)}} + }, + }, + ) self.health = _RecordingNamespace(self, "health", {"get": lambda: {"api_version": api_version}}) self.labware = _RecordingNamespace( self, "labware", {"define": lambda: {"data": {"definitionUri": "pylabrobot/chatterbox/1"}}} @@ -76,10 +98,7 @@ def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool 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}}}}, - }, + {"add_mounted_pipettes": lambda: (left_pipette, right_pipette)}, ) def log(self, qualified: str, args: tuple, kwargs: dict): @@ -90,6 +109,17 @@ def log(self, qualified: str, args: tuple, kwargs: dict): logger.log(LOG_LEVEL_IO, "%s", rendered) if self._verbose: print(rendered) + if qualified == "runs.enqueue_command" and len(args) >= 2: + self._track_command(args[0], args[1]) + + def _track_command(self, command_type: str, params: dict) -> None: + """Record one robot command, and follow a move so ``savePosition`` reports it.""" + self.commands.append((command_type, dict(params))) + if command_type != "moveToCoordinates": + return + for axis, moved_to in params.get("coordinates", {}).items(): + if moved_to is not None: + self.position[axis] = moved_to def __getattr__(self, name: str): # top-level functions the backend calls directly: set_host, set_port, set_run @@ -108,7 +138,8 @@ class OpentronsOT2ChatterboxBackend(OpentronsOT2Backend): 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`. + printed; :attr:`commands` collects the robot commands and :attr:`calls` every + ``ot_api`` call underneath them. Example: >>> from pylabrobot.legacy.liquid_handling import LiquidHandler @@ -126,6 +157,9 @@ def __init__( port: int = 31950, api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, verbose: bool = True, + request_timeout: float = DEFAULT_REQUEST_TIMEOUT, + command_timeout: float = DEFAULT_COMMAND_TIMEOUT, + status_poll_interval: float = DEFAULT_STATUS_POLL_INTERVAL, ): """Initialize the chatterbox. @@ -135,6 +169,9 @@ 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. + request_timeout: how long one request/response with the robot may take, in seconds. + command_timeout: how long a command that moves the robot may take, in seconds. + status_poll_interval: delay between two reads of a running command's status. """ # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly. LiquidHandlerBackend.__init__(self) @@ -149,6 +186,7 @@ def __init__( self._right_pipette_name = right_pipette_name self.host = host self.port = port + self._init_wire_state(request_timeout, command_timeout, status_poll_interval) left = ( {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None @@ -166,10 +204,15 @@ def __init__( 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.""" + def calls(self) -> List[Tuple[str, tuple, dict]]: + """Recorded ``(qualified_name, args, kwargs)`` for every ``ot_api`` call issued.""" return cast(List[Tuple[str, tuple, dict]], self._ot.calls) + @property + def commands(self) -> List[Tuple[str, dict]]: + """Recorded ``(command_type, params)`` for every robot command issued so far.""" + return cast(List[Tuple[str, dict]], self._ot.commands) + def serialize(self) -> dict: return { **LiquidHandlerBackend.serialize(self), diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py index 1866ead93f5..78c5b6a1bdb 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py +++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -11,13 +11,13 @@ OpentronsOT2ChatterboxBackend, OpentronsOT2Simulator, ) -from pylabrobot.resources import set_tip_tracking, set_volume_tracking +from pylabrobot.resources import Coordinate, 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 -def _names(backend: OpentronsOT2ChatterboxBackend): - return [call[0] for call in backend.commands] +def _command_types(backend: OpentronsOT2ChatterboxBackend): + return [command_type for command_type, _params in backend.commands] class OpentronsChatterboxTests(unittest.IsolatedAsyncioTestCase): @@ -49,28 +49,55 @@ async def test_setup_resolves_two_channels_without_ot_api(self): assert self.backend.left_pipette is not None and self.backend.right_pipette is not None self.assertEqual(self.backend.left_pipette["name"], "p20_single_gen2") - async def test_full_protocol_records_one_wire_call_per_operation(self): - """A pickup -> aspirate -> dispense -> trash-discard records exactly one - wire call each, via the real backend logic.""" + async def test_a_channel_reads_back_where_the_dry_run_last_moved_it(self): + """savePosition reports the recorded move, so a dry run can drive a channel.""" + await self.backend.move_channel_to(0, x=50.0, y=60.0, z=70.0) + + self.assertEqual(await self.backend.get_channel_position(0), Coordinate(50.0, 60.0, 70.0)) + + await self.backend.move_channel_x(0, x=15.0) + + self.assertEqual(await self.backend.get_channel_position(0), Coordinate(15.0, 60.0, 70.0)) + + async def test_full_protocol_sends_one_robot_command_per_operation(self): + """A pickup -> aspirate -> dispense -> trash-discard puts exactly one command + of each kind on the wire, via the real backend logic.""" self.plate.get_well("A1").tracker.set_volume(15) await self.lh.pick_up_tips(self.tips["A1"]) await self.lh.aspirate(self.plate["A1"], vols=[10]) await self.lh.dispense(self.plate["B1"], vols=[10]) await self.lh.discard_tips() - names = _names(self.backend) - self.assertEqual(names.count("lh.pick_up_tip"), 1) - self.assertEqual(names.count("lh.aspirate_in_place"), 1) - self.assertEqual(names.count("lh.dispense_in_place"), 1) + types = _command_types(self.backend) + self.assertEqual(types.count("pickUpTip"), 1) + self.assertEqual(types.count("aspirateInPlace"), 1) + self.assertEqual(types.count("dispenseInPlace"), 1) # api_version defaults to 7.1.0, so the discard routes through the trash addressable area - self.assertEqual(names.count("lh.move_to_addressable_area_for_drop_tip"), 1) - self.assertEqual(names.count("lh.drop_tip_in_place"), 1) + self.assertEqual(types.count("moveToAddressableAreaForDropTip"), 1) + self.assertEqual(types.count("dropTipInPlace"), 1) + + async def test_a_pick_up_names_the_well_and_the_pipette_the_robot_needs(self): + """The backend builds the wire params itself now, so they are pinned here.""" + await self.lh.pick_up_tips(self.tips["A1"]) + + params = dict(self.backend.commands)["pickUpTip"] + self.assertEqual(params["labwareId"], self.backend.get_ot_name("tips")) + tip_spot = self.tips.get_item("A1") + self.assertEqual(params["wellName"], self.backend.get_ot_name(tip_spot.name)) + assert self.backend.left_pipette is not None + self.assertEqual(params["pipetteId"], self.backend.left_pipette["pipetteId"]) + self.assertEqual(params["wellLocation"]["origin"], "bottom") def test_unknown_pipette_name_raises(self): """An unrecognised pipette name is rejected at construction.""" with self.assertRaises(ValueError): OpentronsOT2ChatterboxBackend(left_pipette_name="not_a_pipette") + def test_the_chatterbox_checks_the_budgets_the_real_backend_checks(self): + """It skips the real __init__, so the two share one helper rather than drifting.""" + with self.assertRaises(ValueError): + OpentronsOT2ChatterboxBackend(status_poll_interval=0.0, verbose=False) + def test_serialize_includes_pipettes(self): """serialize() captures the mounted-pipette names (None for an empty mount).""" backend = OpentronsOT2ChatterboxBackend( diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py index 262c4d3ef60..32c666ef803 100644 --- a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py +++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py @@ -109,7 +109,7 @@ async def stop(self): self.right_pipette_has_tip = False logger.info("OpentronsOT2Simulator stopped.") - def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]: + async def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]: pipette_id = self._pipette_id_for_channel(channel) return pipette_id, self._positions.get(pipette_id, Coordinate.zero())