diff --git a/docs/_static/devices.json b/docs/_static/devices.json index 15caed83b90..1335c70215c 100644 --- a/docs/_static/devices.json +++ b/docs/_static/devices.json @@ -1713,9 +1713,10 @@ "liquid handling" ], "status": "mostly", - "api": "pylabrobot.legacy.liquid_handling.backends.OpentronsOT2Backend", - "api_version": "v0", - "code_slug": "legacy/liquid_handling/backends", + "api": "pylabrobot.opentrons.OT2", + "api_version": "v1", + "code_slug": "opentrons/ot2", + "doc_slug": "opentrons/ot2/hello-world", "manager": "https://discuss.pylabrobot.org/u/rickwierenga", "oem": "https://opentrons.com/products/ot-2-robot" }, diff --git a/docs/api/pylabrobot.opentrons.rst b/docs/api/pylabrobot.opentrons.rst new file mode 100644 index 00000000000..22ade5ce846 --- /dev/null +++ b/docs/api/pylabrobot.opentrons.rst @@ -0,0 +1,31 @@ +.. currentmodule:: pylabrobot.opentrons + +pylabrobot.opentrons package +============================= + +``OT2`` owns the connection, active run, deck, and mounted pipettes. Use +``connect()`` for health and discovery queries without starting a run or moving +the robot; ``setup()`` also loads the pipettes into a run and optionally homes. +Queries return fresh values without changing the driver's session state. + +``OpentronsAPI`` contains the named HTTP endpoints and response parsing. +``OpentronsRun`` provides command primitives and waits for their completion. +Both use the device's ``pylabrobot.io.HTTP`` transport. Pipette operations own +resource tracking and retraction to traversal height. + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + OT2 + OT2Pipette + OpentronsAPI + OpentronsRun + RobotInfo + MountedPipette + ModuleInfo + OpentronsError + OpentronsCommandError + OpentronsCommandTimeout + OpentronsProtocolError diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index 9a842454106..394622fb707 100644 --- a/docs/api/pylabrobot.rst +++ b/docs/api/pylabrobot.rst @@ -35,6 +35,7 @@ Manufacturers pylabrobot.mettler_toledo pylabrobot.micronic pylabrobot.molecular_devices + pylabrobot.opentrons pylabrobot.qinstruments pylabrobot.revvity pylabrobot.sartorius diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md index 4b3546dde2b..3bbce5af02e 100644 --- a/docs/user_guide/index.md +++ b/docs/user_guide/index.md @@ -43,6 +43,7 @@ kbiosystems/index mettler_toledo/index micronic/index molecular_devices/index +opentrons/index qinstruments/index revvity/index sartorius/index diff --git a/docs/user_guide/opentrons/index.md b/docs/user_guide/opentrons/index.md new file mode 100644 index 00000000000..bb41aef2aac --- /dev/null +++ b/docs/user_guide/opentrons/index.md @@ -0,0 +1,7 @@ +# Opentrons + +```{toctree} +:maxdepth: 1 + +ot2/hello-world +``` diff --git a/docs/user_guide/opentrons/ot2/hello-world.ipynb b/docs/user_guide/opentrons/ot2/hello-world.ipynb new file mode 100644 index 00000000000..e78d333009b --- /dev/null +++ b/docs/user_guide/opentrons/ot2/hello-world.ipynb @@ -0,0 +1,352 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "ot2-title", + "metadata": {}, + "source": [ + "# Opentrons OT-2\n", + "\n", + "The OT-2 is a two-mount liquid-handling robot. PyLabRobot discovers the pipette on each mount and exposes it as a real object, so operations read as `pipette.pick_up_tip(...)`, `pipette.aspirate(...)`, and `pipette.dispense(...)`.\n", + "\n", + "| Property | Value |\n", + "|---|---|\n", + "| Communication | JSON over HTTP |\n", + "| Default address | Robot hostname or IP, port `31950` |\n", + "| Pipette mounts | Left and right |\n", + "| Supported liquid operations | Single-channel GEN1 and GEN2 pipettes |\n", + "| Deck | 12 slots; slot 12 contains fixed trash by default |\n", + "\n", + "The OT-2 exposes a run-command HTTP API. PyLabRobot creates a run during `setup()`, sends one command at a time, and waits for each command to succeed or fail before continuing." + ] + }, + { + "cell_type": "markdown", + "id": "ot2-physical", + "metadata": {}, + "source": [ + "## Physical setup\n", + "\n", + "1. Install the pipettes and remove any tips already attached to their nozzles.\n", + "2. In the Opentrons App, complete deck calibration, pipette-offset calibration, and tip-length calibration for the exact Opentrons tip rack you will use.\n", + "3. Put the computer and OT-2 on the same network.\n", + "4. Find the robot's hostname or IP in the Opentrons App. A hostname such as `ot2.local` may also work on your network.\n", + "5. Keep the deck clear until the labware layout below matches the physical deck." + ] + }, + { + "cell_type": "markdown", + "id": "ot2-create-md", + "metadata": {}, + "source": [ + "## Create the robot\n", + "\n", + "Create the deck first and pass it to the robot. Replace `ot2.local` with your robot's hostname or IP address, without `http://`." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-create", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.opentrons import OT2\n", + "from pylabrobot.resources import OTDeck\n", + "\n", + "deck = OTDeck()\n", + "ot2 = OT2(host=\"ot2.local\", deck=deck)" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-setup-md", + "metadata": {}, + "source": [ + "## Set up the robot\n", + "\n", + "`setup()` reads the robot software version and mounted pipettes, creates an Opentrons run, loads the pipettes, and homes the robot.\n", + "\n", + "For inspection without a run or motion, use `await ot2.connect()` followed by `await ot2.get_health()` or `await ot2.get_mounted_pipettes()`. These queries return fresh readings without changing the driver's session or tracked state. `await ot2.stop()` closes that connection; `await ot2.setup()` can also continue from it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-setup", + "metadata": {}, + "outputs": [], + "source": [ + "await ot2.setup()" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-pipettes-md", + "metadata": {}, + "source": [ + "## Inspect the pipettes\n", + "\n", + "The left and right mount are either an `OT2Pipette` or `None`. This notebook uses the first mounted single-channel pipette." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-pipettes", + "metadata": {}, + "outputs": [], + "source": [ + "print(\"Left:\", ot2.left_pipette.name if ot2.left_pipette else None)\n", + "print(\"Right:\", ot2.right_pipette.name if ot2.right_pipette else None)\n", + "\n", + "pipette = next((p for p in ot2.pipettes if p.num_channels == 1), None)\n", + "assert pipette is not None, \"This example needs a mounted single-channel pipette\"" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-deck-md", + "metadata": {}, + "source": [ + "## Model the physical deck\n", + "\n", + "Choose a tip rack that exactly matches the physical rack and discovered pipette, place it in slot 1, and place the plate in slot 2. Make the physical deck match this layout before continuing. The standard Opentrons rack definitions below preserve the rack identity used by the robot's tip-length calibration. Tracking is enabled so PyLabRobot checks tip and liquid state around each robot command." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-deck", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.resources import set_tip_tracking, set_volume_tracking\n", + "from pylabrobot.resources.celltreat import celltreat_96_wellplate_350uL_Fb\n", + "from pylabrobot.resources.opentrons import (\n", + " opentrons_96_filtertiprack_10ul,\n", + " opentrons_96_filtertiprack_20ul,\n", + " opentrons_96_filtertiprack_200ul,\n", + " opentrons_96_filtertiprack_1000ul,\n", + " opentrons_96_tiprack_300ul,\n", + ")\n", + "\n", + "set_tip_tracking(True)\n", + "set_volume_tracking(True)\n", + "\n", + "tip_rack_factory = {\n", + " 10: opentrons_96_filtertiprack_10ul,\n", + " 20: opentrons_96_filtertiprack_20ul,\n", + " 50: opentrons_96_filtertiprack_200ul,\n", + " 300: opentrons_96_tiprack_300ul,\n", + " 1000: opentrons_96_filtertiprack_1000ul,\n", + "}[pipette.maximum_volume]\n", + "\n", + "tips = tip_rack_factory(name=\"tips\")\n", + "plate = celltreat_96_wellplate_350uL_Fb(name=\"plate\")\n", + "deck.assign_child_at_slot(tips, slot=1)\n", + "deck.assign_child_at_slot(plate, slot=2)\n", + "\n", + "transfer_volume = max(pipette.minimum_volume, min(20, pipette.maximum_volume))\n", + "test_liquid_volume = max(100, transfer_volume * 2)\n", + "plate.get_well(\"A1\").tracker.set_volume(test_liquid_volume)\n", + "print(f\"Before continuing, manually add {test_liquid_volume:g} µL of water to plate well A1.\")" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-pickup-md", + "metadata": {}, + "source": [ + "## Pick up a tip\n", + "\n", + "Pause here and add the printed amount of water to plate well A1. Verify that the matching tip rack is physically in slot 1, the plate is in slot 2, and tip A1 is present. This is the first operation after homing that approaches labware." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-pickup", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.pick_up_tip(tips.get_item(\"A1\"))" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-mix-md", + "metadata": {}, + "source": [ + "## Mix\n", + "\n", + "`mix()` moves to 1 mm above the well bottom, performs the requested aspiration/dispense cycles client-side, then returns to traversal height." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-mix", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.mix(\n", + " plate.get_well(\"A1\"), volume=transfer_volume, repetitions=3, liquid_height=1\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-aspirate-md", + "metadata": {}, + "source": [ + "## Aspirate\n", + "\n", + "Aspirate from the cavity bottom plus `liquid_height`. You can also pass a `Coordinate` offset to compensate for a carefully measured positional calibration difference." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-aspirate", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.aspirate(plate.get_well(\"A1\"), volume=transfer_volume, liquid_height=1)" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-dispense-md", + "metadata": {}, + "source": [ + "## Dispense\n", + "\n", + "Dispense the tracked liquid into another well. The pipette returns to the configured traversal height after the operation." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-dispense", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.dispense(plate.get_well(\"B1\"), volume=transfer_volume, liquid_height=1)" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-return-md", + "metadata": {}, + "source": [ + "## Return the tip\n", + "\n", + "`return_tip()` uses the recorded pickup origin and restores the tip-rack tracker after the robot command succeeds." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-return", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.return_tip()" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-pickup-two-md", + "metadata": {}, + "source": [ + "## Pick up another tip\n", + "\n", + "Pick up a fresh tip to demonstrate disposal in the fixed trash." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-pickup-two", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.pick_up_tip(tips.get_item(\"A2\"))" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-discard-md", + "metadata": {}, + "source": [ + "## Discard the tip\n", + "\n", + "`discard_tip()` uses the fixed-trash command sequence appropriate for the robot's software version recorded during setup." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-discard", + "metadata": {}, + "outputs": [], + "source": [ + "await pipette.discard_tip()" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-home-md", + "metadata": {}, + "source": [ + "## Home\n", + "\n", + "Home the gantry and pipette axes when you need to return the robot to its reference state." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-home", + "metadata": {}, + "outputs": [], + "source": [ + "await ot2.home()" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-stop-md", + "metadata": {}, + "source": [ + "## Teardown\n", + "\n", + "Always run `stop()`, including after an error. It stops the active Opentrons run, making the robot available to the Opentrons App again, and closes the HTTP transport." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ot2-stop", + "metadata": {}, + "outputs": [], + "source": [ + "await ot2.stop()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.11.0" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/pylabrobot/io/__init__.py b/pylabrobot/io/__init__.py index 3c6079b0673..48119621ba6 100644 --- a/pylabrobot/io/__init__.py +++ b/pylabrobot/io/__init__.py @@ -1,5 +1,6 @@ from .capture import start_capture, stop_capture from .command_line import CommandLineResult, CommandLineTransport, CommandLineValidator +from .http import HTTP, HTTPError from .socket import Socket, SocketValidator from .validation import end_validation, validate from .validation_utils import LOG_LEVEL_IO diff --git a/pylabrobot/io/http.py b/pylabrobot/io/http.py index e69de29bb2d..b905fbf5f52 100644 --- a/pylabrobot/io/http.py +++ b/pylabrobot/io/http.py @@ -0,0 +1,153 @@ +import asyncio +import json +import logging +import urllib.error +import urllib.parse +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from functools import partial +from typing import Any, Dict, Mapping, Optional + +from pylabrobot.io.capture import Command, capturer, get_capture_or_validation_active +from pylabrobot.io.validation_utils import LOG_LEVEL_IO + +logger = logging.getLogger(__name__) + + +class HTTPError(RuntimeError): + """An HTTP response outside the 2xx range.""" + + def __init__(self, method: str, url: str, status: int, body: str): + self.method = method + self.url = url + self.status = status + self.body = body + super().__init__(f"{method} {url} returned HTTP {status}: {body}") + + +@dataclass +class HTTPCommand(Command): + """One JSON HTTP request and its decoded response.""" + + path: str + request: Optional[str] + response: str + + def __init__( + self, + device_id: str, + method: str, + path: str, + request: Optional[str], + response: str, + ): + super().__init__(module="http", device_id=device_id, action=method) + self.path = path + self.request = request + self.response = response + + +class HTTP: + """Asynchronous JSON-over-HTTP transport. + + The standard-library HTTP client is blocking, so requests run on a private + single-thread executor. The executor and a request lock keep a device's + request/response stream ordered without blocking the asyncio event loop. + """ + + def __init__( + self, + human_readable_device_name: str, + base_url: str, + headers: Optional[Mapping[str, str]] = None, + timeout: float = 30.0, + ): + if get_capture_or_validation_active(): + raise RuntimeError("Cannot create a new HTTP object while capture or validation is active") + if timeout <= 0: + raise ValueError("timeout must be greater than zero") + + self.human_readable_device_name = human_readable_device_name + self.base_url = base_url.rstrip("/") + self.headers = dict(headers or {}) + self.timeout = timeout + self._executor: Optional[ThreadPoolExecutor] = None + self._request_lock = asyncio.Lock() + + async def setup(self) -> None: + if self._executor is None: + self._executor = ThreadPoolExecutor(max_workers=1) + + async def stop(self) -> None: + if self._executor is not None: + self._executor.shutdown(wait=True) + self._executor = None + + def _make_request( + self, + method: str, + path: str, + data: Optional[Dict[str, Any]], + ) -> Dict[str, Any]: + url = urllib.parse.urljoin(f"{self.base_url}/", path.lstrip("/")) + headers = self.headers.copy() + body = None + if data is not None: + body = json.dumps(data).encode("utf-8") + headers["Content-Type"] = "application/json" + + request = urllib.request.Request(url, headers=headers, data=body, method=method) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + response_body = response.read() + except urllib.error.HTTPError as error: + error_body = error.read().decode("utf-8", errors="replace") + raise HTTPError(method, url, error.code, error_body) from error + + if response_body == b"": + return {} + return dict(json.loads(response_body.decode("utf-8"))) + + async def request( + self, + method: str, + path: str, + data: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Send a JSON request and return the decoded JSON object.""" + if self._executor is None: + raise RuntimeError( + f"HTTP transport for '{self.human_readable_device_name}' is not set up; call setup() first" + ) + + normalized_method = method.upper() + request_json = json.dumps(data, sort_keys=True) if data is not None else None + logger.log( + LOG_LEVEL_IO, + "[%s] %s %s %s", + self.base_url, + normalized_method, + path, + request_json or "", + ) + + async with self._request_lock: + loop = asyncio.get_running_loop() + response = await loop.run_in_executor( + self._executor, + partial(self._make_request, normalized_method, path, data), + ) + + response_json = json.dumps(response, sort_keys=True) + logger.log(LOG_LEVEL_IO, "[%s] response %s", self.base_url, response_json) + capturer.record( + HTTPCommand( + device_id=self.base_url, + method=normalized_method, + path=path, + request=request_json, + response=response_json, + ) + ) + return response diff --git a/pylabrobot/io/http_tests.py b/pylabrobot/io/http_tests.py new file mode 100644 index 00000000000..e97b68204e5 --- /dev/null +++ b/pylabrobot/io/http_tests.py @@ -0,0 +1,67 @@ +import io +import json +import unittest +import urllib.error +from email.message import Message +from unittest.mock import patch + +from pylabrobot.io.http import HTTP, HTTPError + + +class _Response: + def __init__(self, body: bytes): + self.body = body + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + return False + + def read(self) -> bytes: + return self.body + + +class HTTPTests(unittest.IsolatedAsyncioTestCase): + async def test_request_sends_and_decodes_json_off_event_loop(self) -> None: + transport = HTTP( + human_readable_device_name="test device", + base_url="http://device.local:1234", + headers={"X-API-Version": "3"}, + ) + await transport.setup() + with patch( + "urllib.request.urlopen", return_value=_Response(b'{"data":{"id":"run"}}') + ) as urlopen: + response = await transport.request("post", "/runs", {"value": 1}) + await transport.stop() + + self.assertEqual(response, {"data": {"id": "run"}}) + request = urlopen.call_args.args[0] + self.assertEqual(request.full_url, "http://device.local:1234/runs") + self.assertEqual(request.method, "POST") + self.assertEqual(json.loads(request.data), {"value": 1}) + self.assertEqual(request.headers["X-api-version"], "3") + self.assertEqual(request.headers["Content-type"], "application/json") + + async def test_http_error_includes_response_body(self) -> None: + transport = HTTP( + human_readable_device_name="test device", + base_url="http://device.local:1234", + ) + await transport.setup() + error = urllib.error.HTTPError( + url="http://device.local:1234/runs", + code=400, + msg="Bad Request", + hdrs=Message(), + fp=io.BytesIO(b'{"message":"bad run"}'), + ) + with patch("urllib.request.urlopen", side_effect=error): + with self.assertRaisesRegex(HTTPError, "bad run"): + await transport.request("POST", "/runs") + await transport.stop() + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/opentrons/__init__.py b/pylabrobot/opentrons/__init__.py new file mode 100644 index 00000000000..1b16becdd93 --- /dev/null +++ b/pylabrobot/opentrons/__init__.py @@ -0,0 +1,10 @@ +from .api import OpentronsAPI +from .errors import ( + OpentronsCommandError, + OpentronsCommandTimeout, + OpentronsError, + OpentronsProtocolError, +) +from .ot2 import OT2, OT2Pipette +from .run import OpentronsRun +from .types import ModuleInfo, MountedPipette, RobotInfo diff --git a/pylabrobot/opentrons/api.py b/pylabrobot/opentrons/api.py new file mode 100644 index 00000000000..d3ca1846ae6 --- /dev/null +++ b/pylabrobot/opentrons/api.py @@ -0,0 +1,123 @@ +"""Named robot-server endpoints over a caller-owned PLR HTTP transport.""" + +from typing import Any, Dict, List, Literal, Optional, Tuple + +from pylabrobot.io.http import HTTP, HTTPError +from pylabrobot.opentrons.errors import OpentronsError, OpentronsProtocolError +from pylabrobot.opentrons.types import ( + CommandInfo, + LabwareIdentity, + ModuleInfo, + Mount, + MountedPipette, + RobotInfo, + RunInfo, + _object, + _optional_string, + _string, +) + +HTTP_API_VERSION = "3" + + +def _data_list(response: Dict[str, Any]) -> List[Dict[str, Any]]: + data = response.get("data") + if not isinstance(data, list): + raise OpentronsProtocolError("Expected a list in the response's 'data' field") + return [_object(item) for item in data] + + +class OpentronsAPI: + """HTTP endpoint encoding and decoding, without device or selected-run state. + + The caller opens and closes ``io``. Queries return independent readings; + commands return server receipts without attaching them to a device. + """ + + def __init__(self, io: HTTP) -> None: + self._io = io + + async def get_health(self) -> RobotInfo: + return RobotInfo.from_response(await self._io.request("GET", "/health")) + + async def get_mounted_pipettes(self) -> Tuple[MountedPipette, ...]: + """Read the OT-2 mount endpoint without loading instruments into a run.""" + response = await self._io.request("GET", "/pipettes") + pipettes = [] + mounts: Tuple[Mount, ...] = ("left", "right") + for mount in mounts: + data = _object(response.get(mount)) + name = _optional_string(data, "name") + if name is not None: + pipettes.append( + MountedPipette(mount, name, _optional_string(data, "model"), _optional_string(data, "id")) + ) + return tuple(pipettes) + + async def get_connected_modules(self) -> Tuple[ModuleInfo, ...]: + response = await self._io.request("GET", "/modules") + return tuple(ModuleInfo.from_response(data) for data in _data_list(response)) + + async def get_runs(self) -> Tuple[RunInfo, ...]: + response = await self._io.request("GET", "/runs") + return tuple(RunInfo.from_response(data) for data in _data_list(response)) + + async def get_run(self, run_id: str) -> RunInfo: + response = await self._io.request("GET", f"/runs/{run_id}") + return RunInfo.from_response(_object(response.get("data"))) + + async def create_run(self) -> RunInfo: + response = await self._io.request("POST", "/runs") + return RunInfo.from_response(_object(response.get("data"))) + + async def stop_run(self, run_id: str) -> None: + """Stop a run, falling back only when a firmware endpoint is unsupported.""" + requests = ( + ("POST", f"/runs/{run_id}/actions", {"data": {"actionType": "stop"}}), + ("POST", f"/runs/{run_id}/cancel", None), + ("POST", f"/runs/{run_id}/actions/cancel", None), + ("DELETE", f"/runs/{run_id}", None), + ) + last_error: Optional[Exception] = None + for method, path, data in requests: + try: + await self._io.request(method, path, data) + return + except HTTPError as error: + last_error = error + if error.status not in {404, 405}: + break + except Exception as error: + last_error = error + break + raise OpentronsError( + f"Could not cancel Opentrons run {run_id}; state is retained for retry" + ) from (last_error) + + async def home(self) -> None: + """Home the OT-2 gantry and pipette axes through the robot endpoint.""" + await self._io.request("POST", "/robot/home", {"target": "robot"}) + + async def submit_command( + self, + run_id: str, + command_type: str, + params: Dict[str, Any], + intent: Literal["setup", "protocol"] = "setup", + ) -> str: + response = await self._io.request( + "POST", + f"/runs/{run_id}/commands", + {"data": {"commandType": command_type, "params": params, "intent": intent}}, + ) + return _string(_object(response.get("data")), "id") + + async def get_command(self, run_id: str, command_id: str) -> CommandInfo: + response = await self._io.request("GET", f"/runs/{run_id}/commands/{command_id}") + return CommandInfo.from_response(_object(response.get("data"))) + + async def define_labware(self, run_id: str, definition: Dict[str, Any]) -> LabwareIdentity: + response = await self._io.request( + "POST", f"/runs/{run_id}/labware_definitions", {"data": definition} + ) + return LabwareIdentity.from_uri(_string(_object(response.get("data")), "definitionUri")) diff --git a/pylabrobot/opentrons/api_tests.py b/pylabrobot/opentrons/api_tests.py new file mode 100644 index 00000000000..a21f3d19057 --- /dev/null +++ b/pylabrobot/opentrons/api_tests.py @@ -0,0 +1,117 @@ +import unittest +from dataclasses import FrozenInstanceError +from typing import Any, Dict, List +from unittest.mock import AsyncMock, call + +from pylabrobot.io.http import HTTP, HTTPError +from pylabrobot.opentrons.api import OpentronsAPI +from pylabrobot.opentrons.errors import OpentronsError, OpentronsProtocolError +from pylabrobot.opentrons.types import LabwareIdentity, ModuleInfo, MountedPipette, RunInfo + + +class OpentronsAPITests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.io = AsyncMock(spec=HTTP) + self.api = OpentronsAPI(self.io) + + async def test_health_is_a_fresh_value_without_changing_client_state(self) -> None: + response = { + "name": "OT2sterile", + "robot_model": "OT-2 Standard", + "api_version": "8.7.0", + "fw_version": "v1.1.0", + "robot_serial": None, + } + self.io.request.return_value = response + state = vars(self.api).copy() + first = await self.api.get_health() + response["api_version"] = "8.8.1" + second = await self.api.get_health() + self.assertEqual(first.software_version, "8.7.0") + self.assertEqual(second.software_version, "8.8.1") + self.assertEqual(first.model, "OT-2 Standard") + self.assertEqual(vars(self.api), state) + with self.assertRaises(FrozenInstanceError): + setattr(first, "name", "replacement") + self.io.setup.assert_not_called() + + async def test_mount_discovery_handles_an_empty_mount_without_a_name_key(self) -> None: + self.io.request.return_value = { + "left": {"id": "physical-serial", "model": "p20_single_v2.2", "name": "p20_single_gen2"}, + "right": {"mount_axis": "a", "plunger_axis": "c"}, + } + pipettes = await self.api.get_mounted_pipettes() + self.assertEqual( + pipettes, (MountedPipette("left", "p20_single_gen2", "p20_single_v2.2", "physical-serial"),) + ) + self.io.request.assert_awaited_once_with("GET", "/pipettes") + + async def test_run_creation_does_not_select_or_replace_a_run(self) -> None: + self.io.request.side_effect = [ + {"data": {"id": "first", "status": "idle"}}, + {"data": {"id": "second", "status": "idle"}}, + ] + state = vars(self.api).copy() + first, second = await self.api.create_run(), await self.api.create_run() + self.assertEqual((first.id, second.id), ("first", "second")) + self.assertEqual(vars(self.api), state) + self.assertEqual(self.io.request.await_args_list, [call("POST", "/runs")] * 2) + + async def test_queries_unpack_run_and_module_results(self) -> None: + self.io.request.side_effect = [ + {"data": [{"id": "run", "status": "idle"}]}, + {"data": {"id": "run", "status": "running"}}, + {"data": [{"id": "module", "moduleType": "temperatureModuleType"}]}, + ] + self.assertEqual(await self.api.get_runs(), (RunInfo("run", "idle"),)) + self.assertEqual(await self.api.get_run("run"), RunInfo("run", "running")) + self.assertEqual( + await self.api.get_connected_modules(), (ModuleInfo("module", "temperatureModuleType"),) + ) + + async def test_labware_definition_receipt_is_parsed(self) -> None: + self.io.request.return_value = {"data": {"definitionUri": "pylabrobot/rack/1"}} + definition = {"schemaVersion": 2} + self.assertEqual( + await self.api.define_labware("run", definition), LabwareIdentity("pylabrobot", "rack", 1) + ) + self.io.request.assert_awaited_once_with( + "POST", "/runs/run/labware_definitions", {"data": definition} + ) + + async def test_invalid_receipts_are_rejected_before_an_id_can_be_used(self) -> None: + responses: List[Dict[str, Any]] = [{}, {"data": {}}, {"data": {"id": None}}] + for response in responses: + with self.subTest(response=response): + self.io.request.return_value = response + with self.assertRaises(OpentronsProtocolError): + await self.api.create_run() + + async def test_command_status_accepts_absent_or_null_results(self) -> None: + for data in ({"status": "queued"}, {"status": "running", "result": None, "error": None}): + with self.subTest(data=data): + self.io.request.return_value = {"data": data} + command = await self.api.get_command("run", "command") + self.assertEqual(command.result, {}) + self.assertEqual(command.error, {}) + + async def test_stop_uses_legacy_route_only_for_unsupported_endpoint(self) -> None: + self.io.request.side_effect = [HTTPError("POST", "/actions", 404, "unsupported"), {}] + await self.api.stop_run("run") + self.assertEqual( + self.io.request.await_args_list, + [ + call("POST", "/runs/run/actions", {"data": {"actionType": "stop"}}), + call("POST", "/runs/run/cancel", None), + ], + ) + + async def test_stop_does_not_mask_a_server_or_connection_failure_with_a_fallback(self) -> None: + for failure in (HTTPError("POST", "/actions", 500, "server error"), TimeoutError("timeout")): + with self.subTest(failure=failure): + self.io.request.reset_mock() + self.io.request.side_effect = failure + with self.assertRaises(OpentronsError) as raised: + await self.api.stop_run("run") + self.assertIs(raised.exception.__cause__, failure) + self.io.request.assert_awaited_once() diff --git a/pylabrobot/opentrons/errors.py b/pylabrobot/opentrons/errors.py new file mode 100644 index 00000000000..497e80fc386 --- /dev/null +++ b/pylabrobot/opentrons/errors.py @@ -0,0 +1,36 @@ +"""Errors from the Opentrons robot-server protocol.""" + + +class OpentronsError(RuntimeError): + """An Opentrons operation could not be completed.""" + + +class OpentronsProtocolError(OpentronsError): + """A robot-server response does not match the expected protocol.""" + + +class OpentronsCommandError(OpentronsError): + """A command failed, with its server identifiers available for inspection.""" + + def __init__( + self, run_id: str, command_id: str, command_type: str, error_type: str, detail: str + ) -> None: + self.run_id = run_id + self.command_id = command_id + self.command_type = command_type + self.error_type = error_type + self.detail = detail + super().__init__(f"{command_type} failed with {error_type}: {detail}") + + +class OpentronsCommandTimeout(TimeoutError): + """The command's completion is unknown; its ID can be used to query the server.""" + + def __init__(self, run_id: str, command_id: str, command_type: str) -> None: + self.run_id = run_id + self.command_id = command_id + self.command_type = command_type + super().__init__( + f"Timed out waiting for Opentrons command {command_type!r} " + f"({command_id} in run {run_id}); completion is unknown" + ) diff --git a/pylabrobot/opentrons/labware.py b/pylabrobot/opentrons/labware.py new file mode 100644 index 00000000000..784e1ea94f9 --- /dev/null +++ b/pylabrobot/opentrons/labware.py @@ -0,0 +1,155 @@ +"""PLR labware conversion and bindings scoped to one Opentrons run.""" + +import math +import uuid +from dataclasses import dataclass +from typing import Any, Dict, Optional, cast + +from pylabrobot import utils +from pylabrobot.opentrons.run import OpentronsRun +from pylabrobot.opentrons.types import LabwareIdentity +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.resource import Resource +from pylabrobot.resources.tip import Tip +from pylabrobot.resources.tip_rack import TipRack + +_OFFICIAL_TIP_RACKS = { + "Opentrons OT-2 96 Filter Tip Rack 10 µL": "opentrons_96_filtertiprack_10ul", + "Opentrons OT-2 96 Filter Tip Rack 20 µL": "opentrons_96_filtertiprack_20ul", + "Opentrons OT-2 96 Filter Tip Rack 200 µL": "opentrons_96_filtertiprack_200ul", + "Opentrons OT-2 96 Filter Tip Rack 1000 µL": "opentrons_96_filtertiprack_1000ul", + "Opentrons OT-2 96 Tip Rack 10 µL": "opentrons_96_tiprack_10ul", + "Opentrons OT-2 96 Tip Rack 20 µL": "opentrons_96_tiprack_20ul", + "Opentrons OT-2 96 Tip Rack 300 µL": "opentrons_96_tiprack_300ul", + "Opentrons OT-2 96 Tip Rack 1000 µL": "opentrons_96_tiprack_1000ul", +} + + +def official_tip_rack_identity(tip_rack: TipRack) -> Optional[LabwareIdentity]: + """Look up a rack's official definition identity for its calibration data.""" + load_name = _OFFICIAL_TIP_RACKS.get(tip_rack.model or "") + return LabwareIdentity("opentrons", load_name, 1) if load_name is not None else None + + +def build_tip_rack_definition(tip_rack: TipRack, tip: Tip, load_name: str) -> Dict[str, Any]: + """Build a definition from PLR geometry without loading or modifying the rack.""" + tip_spots = tip_rack.get_all_items() + well_names = {spot.name: tip_rack.get_child_identifier(spot) for spot in tip_spots} + definition = { + "schemaVersion": 2, + "version": 1, + "namespace": "pylabrobot", + "metadata": { + "displayName": load_name, + "displayCategory": "tipRack", + "displayVolumeUnits": "µL", + }, + "brand": {"brand": "unknown"}, + "parameters": { + "format": ( + "96Standard" + if (tip_rack.num_items_x, tip_rack.num_items_y) == (12, 8) + else "384Standard" + if (tip_rack.num_items_x, tip_rack.num_items_y) == (24, 16) + else "irregular" + ), + "isTiprack": True, + "tipLength": tip.total_tip_length, + "tipOverlap": tip.fitting_depth, + "loadName": load_name, + "isMagneticModuleCompatible": False, + }, + "ordering": utils.reshape_2d( + [well_names[tip_spot.name] for tip_spot in tip_spots], + (tip_rack.num_items_x, tip_rack.num_items_y), + ), + "cornerOffsetFromSlot": { + "x": 0, + "y": 0, + "z": 0, + }, + "dimensions": { + "xDimension": tip_rack.get_absolute_size_x(), + "yDimension": tip_rack.get_absolute_size_y(), + "zDimension": tip_rack.get_absolute_size_z(), + }, + "wells": { + well_names[child.name]: { + "depth": tip.total_tip_length, + "x": cast(Coordinate, child.location).x + child.get_absolute_size_x() / 2, + "y": cast(Coordinate, child.location).y + child.get_absolute_size_y() / 2, + "z": cast(Coordinate, child.location).z, + "shape": "circular", + "diameter": math.hypot( + child.get_absolute_size_x(), + child.get_absolute_size_y(), + ), + "totalLiquidVolume": tip.maximal_volume, + } + for child in tip_rack.children + }, + "groups": [ + { + "wells": [well_names[tip_spot.name] for tip_spot in tip_spots], + "metadata": {}, + } + ], + } + return definition + + +@dataclass(frozen=True) +class LabwareBinding: + """A resource's server identity and location within one run.""" + + resource: Resource + labware_id: str + slot: str + identity: LabwareIdentity + + +class LabwareRegistry: + """Record successful labware loads independently of the device's deck model.""" + + def __init__(self, run: OpentronsRun) -> None: + self._run = run + self._bindings: Dict[int, LabwareBinding] = {} + + def is_loaded(self, resource: Resource) -> bool: + return id(resource) in self._bindings + + def get(self, resource: Resource) -> LabwareBinding: + """Look up an existing binding without uploading, loading, or allocating an ID.""" + return self._bindings[id(resource)] + + async def load( + self, + resource: Resource, + slot: str, + identity: LabwareIdentity, + definition: Optional[Dict[str, Any]] = None, + ) -> None: + """Load a resource and record its binding after confirmed success. + + Repeated loads at the same slot are a no-op. A changed slot is rejected + until an explicit server-side relocation has been implemented. + """ + self._run._require_active() + if self.is_loaded(resource): + binding = self.get(resource) + if slot != binding.slot: + raise ValueError( + f"Labware {resource.name!r} is loaded in slot {binding.slot}; " + f"it cannot be used in slot {slot} during the same run" + ) + if identity != binding.identity: + raise ValueError( + f"Labware {resource.name!r} already has a different definition in this run" + ) + return + + if definition is not None: + identity = await self._run.define_labware(definition) + labware_id = uuid.uuid4().hex + await self._run.load_labware(identity, slot, labware_id, resource.name) + self._bindings[id(resource)] = LabwareBinding(resource, labware_id, slot, identity) diff --git a/pylabrobot/opentrons/labware_tests.py b/pylabrobot/opentrons/labware_tests.py new file mode 100644 index 00000000000..310ec9cc9b2 --- /dev/null +++ b/pylabrobot/opentrons/labware_tests.py @@ -0,0 +1,70 @@ +import unittest +from unittest.mock import AsyncMock + +from pylabrobot.opentrons.labware import LabwareRegistry, build_tip_rack_definition +from pylabrobot.opentrons.run import OpentronsRun +from pylabrobot.opentrons.types import LabwareIdentity +from pylabrobot.resources.opentrons import opentrons_96_filtertiprack_20ul + + +class LabwareRegistryTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.protocol_run = AsyncMock(spec=OpentronsRun) + self.registry = LabwareRegistry(self.protocol_run) + self.rack = opentrons_96_filtertiprack_20ul("rack") + self.identity = LabwareIdentity("opentrons", "opentrons_96_filtertiprack_20ul", 1) + + async def test_lookup_does_not_load_or_allocate_missing_labware(self) -> None: + self.assertFalse(self.registry.is_loaded(self.rack)) + with self.assertRaises(KeyError): + self.registry.get(self.rack) + self.protocol_run.load_labware.assert_not_awaited() + self.protocol_run.define_labware.assert_not_awaited() + + async def test_successful_load_is_recorded_once_and_changed_location_is_rejected(self) -> None: + await self.registry.load(self.rack, "1", self.identity) + binding = self.registry.get(self.rack) + await self.registry.load(self.rack, "1", self.identity) + self.assertIs(self.registry.get(self.rack), binding) + self.protocol_run.load_labware.assert_awaited_once() + with self.assertRaisesRegex(ValueError, "loaded in slot 1"): + await self.registry.load(self.rack, "5", self.identity) + self.protocol_run.load_labware.assert_awaited_once() + self.assertEqual(binding.slot, "1") + + async def test_failed_load_does_not_publish_a_binding(self) -> None: + self.protocol_run.load_labware.side_effect = RuntimeError("load failed") + with self.assertRaisesRegex(RuntimeError, "load failed"): + await self.registry.load(self.rack, "1", self.identity) + self.assertFalse(self.registry.is_loaded(self.rack)) + + async def test_distinct_resource_objects_do_not_inherit_each_others_binding(self) -> None: + await self.registry.load(self.rack, "1", self.identity) + replacement = opentrons_96_filtertiprack_20ul("rack") + self.assertFalse(self.registry.is_loaded(replacement)) + with self.assertRaises(KeyError): + self.registry.get(replacement) + + async def test_custom_load_uses_the_definition_receipt(self) -> None: + self.protocol_run.define_labware.return_value = LabwareIdentity("pylabrobot", "uploaded", 2) + definition = build_tip_rack_definition(self.rack, self.rack.get_item("A1").get_tip(), "custom") + await self.registry.load(self.rack, "5", self.identity, definition) + binding = self.registry.get(self.rack) + self.assertEqual(binding.identity, LabwareIdentity("pylabrobot", "uploaded", 2)) + self.protocol_run.load_labware.assert_awaited_once_with( + binding.identity, "5", binding.labware_id, "rack" + ) + + +class LabwareConversionTests(unittest.TestCase): + def test_definition_building_does_not_change_the_resource_or_its_tips(self) -> None: + rack = opentrons_96_filtertiprack_20ul("rack") + tip = rack.get_item("A1").get_tip() + before = [spot.serialize_state() for spot in rack.get_all_items()] + counters = [spot._tip_counter for spot in rack.get_all_items()] + first = build_tip_rack_definition(rack, tip, "custom") + second = build_tip_rack_definition(rack, tip, "custom") + self.assertEqual(first, second) + self.assertEqual([spot.serialize_state() for spot in rack.get_all_items()], before) + self.assertEqual([spot._tip_counter for spot in rack.get_all_items()], counters) + self.assertIs(rack.get_item("A1").get_tip(), tip) diff --git a/pylabrobot/opentrons/ot2/__init__.py b/pylabrobot/opentrons/ot2/__init__.py new file mode 100644 index 00000000000..b59813e6849 --- /dev/null +++ b/pylabrobot/opentrons/ot2/__init__.py @@ -0,0 +1,2 @@ +from .ot2 import OT2 +from .pipette import OT2Pipette diff --git a/pylabrobot/opentrons/ot2/ot2.py b/pylabrobot/opentrons/ot2/ot2.py new file mode 100644 index 00000000000..7cb86eff4a2 --- /dev/null +++ b/pylabrobot/opentrons/ot2/ot2.py @@ -0,0 +1,212 @@ +"""OT-2 lifecycle, deck ownership, and physical pipette discovery.""" + +import asyncio +import math +import uuid +from typing import List, Optional, Tuple + +from pylabrobot.io.http import HTTP +from pylabrobot.opentrons.api import HTTP_API_VERSION, OpentronsAPI +from pylabrobot.opentrons.labware import ( + LabwareRegistry, + build_tip_rack_definition, + official_tip_rack_identity, +) +from pylabrobot.opentrons.ot2.pipette import _PIPETTE_SPECS, OT2Pipette +from pylabrobot.opentrons.run import OpentronsRun +from pylabrobot.opentrons.types import ( + LabwareIdentity, + ModuleInfo, + MountedPipette, + RobotInfo, + RunInfo, +) +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.opentrons import OT2RobotGeometry, OTDeck +from pylabrobot.resources.tip import Tip +from pylabrobot.resources.tip_rack import TipRack + + +class OT2: + """Opentrons OT-2 controlled through its robot-server HTTP API. + + ``connect()`` opens the transport for queries. ``setup()`` also creates a + run, binds mounted pipettes, and optionally homes the robot. Physical + operations are exposed on ``left_pipette`` and ``right_pipette``. + """ + + def __init__( + self, + host: str, + port: int = 31950, + deck: Optional[OTDeck] = None, + traversal_height: float = 120, + command_timeout: float = 30, + command_poll_interval: float = 0.05, + io: Optional[HTTP] = None, + ) -> None: + if "://" in host: + raise ValueError("host must be a hostname or IP address without a URL scheme") + if not 1 <= port <= 65535: + raise ValueError("port must be between 1 and 65535") + if not math.isfinite(traversal_height) or traversal_height < 0: + raise ValueError("traversal_height must be finite and non-negative") + if not math.isfinite(command_timeout) or command_timeout <= 0: + raise ValueError("command_timeout must be finite and greater than zero") + if not math.isfinite(command_poll_interval) or command_poll_interval < 0: + raise ValueError("command_poll_interval must be finite and non-negative") + + self.host = host + self.port = port + self.deck = deck or OTDeck() + self.geometry = OT2RobotGeometry() + self.traversal_height = traversal_height + self.command_timeout = command_timeout + self.command_poll_interval = command_poll_interval + self.io = io or HTTP( + human_readable_device_name="Opentrons OT-2", + base_url=f"http://{host}:{port}", + headers={"Opentrons-Version": HTTP_API_VERSION}, + timeout=command_timeout, + ) + self._api = OpentronsAPI(self.io) + self._connected = False + self._run: Optional[OpentronsRun] = None + self._labware: Optional[LabwareRegistry] = None + self.left_pipette: Optional[OT2Pipette] = None + self.right_pipette: Optional[OT2Pipette] = None + self._operation_lock = asyncio.Lock() + + @property + def pipettes(self) -> List[OT2Pipette]: + """Mounted pipettes, left first.""" + return [p for p in (self.left_pipette, self.right_pipette) if p is not None] + + @property + def software_version(self) -> Optional[str]: + """Software version used to configure the active run, or None before setup.""" + return self._run.software_version if self._run is not None else None + + async def connect(self) -> None: + """Open the transport for queries, without creating a run or moving the robot.""" + async with self._operation_lock: + await self._connect() + + async def _connect(self) -> None: + if not self._connected: + await self.io.setup() + self._connected = True + + def _require_connected(self) -> None: + if not self._connected: + raise RuntimeError("The OT-2 is not connected; call connect() or setup() first") + + async def get_health(self) -> RobotInfo: + """Return a fresh health reading without changing device or run state.""" + self._require_connected() + return await self._api.get_health() + + async def get_mounted_pipettes(self) -> Tuple[MountedPipette, ...]: + """Read physical pipette identities without loading or replacing pipette objects.""" + self._require_connected() + return await self._api.get_mounted_pipettes() + + async def list_connected_modules(self) -> Tuple[ModuleInfo, ...]: + """Read connected module identities without loading modules into the run.""" + self._require_connected() + return await self._api.get_connected_modules() + + async def get_runs(self) -> Tuple[RunInfo, ...]: + """Read server runs without selecting, stopping, or replacing the active run.""" + self._require_connected() + return await self._api.get_runs() + + async def setup(self, skip_home: bool = False) -> None: + """Connect, discover and bind pipettes to a run, and optionally home.""" + async with self._operation_lock: + if self._run is not None: + raise RuntimeError("The OT-2 is already set up") + await self._connect() + try: + health = await self.get_health() + mounted = await self.get_mounted_pipettes() + for pipette in mounted: + if pipette.name not in _PIPETTE_SPECS: + raise ValueError(f"Unsupported OT-2 pipette {pipette.name!r} on {pipette.mount}") + receipt = await self._api.create_run() + self._run = OpentronsRun( + self._api, + receipt.id, + health.software_version, + command_timeout=self.command_timeout, + command_poll_interval=self.command_poll_interval, + ) + self._labware = LabwareRegistry(self._run) + await self._load_pipettes(mounted) + if not skip_home: + await self._api.home() + except BaseException: + await self._stop() + raise + + async def _load_pipettes(self, mounted: Tuple[MountedPipette, ...]) -> None: + run = self._require_run() + for pipette in mounted: + pipette_id = await run.load_pipette(pipette.name, pipette.mount) + bound = OT2Pipette(self, pipette.mount, pipette.name, pipette_id) + if pipette.mount == "left": + self.left_pipette = bound + else: + self.right_pipette = bound + + async def stop(self) -> None: + """Stop the run and close the transport, retaining state if run shutdown fails.""" + async with self._operation_lock: + await self._stop() + + async def _stop(self) -> None: + if self._run is not None: + await self._run.stop() + self._run = None + self._labware = None + self.left_pipette = None + self.right_pipette = None + if self._connected: + await self.io.stop() + self._connected = False + + def _require_run(self) -> OpentronsRun: + if self._run is None or not self._run.active: + raise RuntimeError("The OT-2 is not set up") + return self._run + + def _require_labware(self) -> LabwareRegistry: + self._require_run() + if self._labware is None: + raise RuntimeError("The OT-2 labware registry is unavailable") + return self._labware + + async def home(self) -> None: + """Home the gantry and pipette axes.""" + async with self._operation_lock: + self._require_run() + await self._api.home() + + async def _load_tip_rack(self, tip_rack: TipRack, tip: Tip) -> None: + slot = self.deck.get_slot(tip_rack) + if slot is None: + raise ValueError("tip rack must be assigned directly to an OT-2 deck slot") + registry = self._require_labware() + definition = None + identity: Optional[LabwareIdentity] + if registry.is_loaded(tip_rack): + identity = registry.get(tip_rack).identity + else: + identity = official_tip_rack_identity(tip_rack) + if identity is None: + identity = LabwareIdentity("pylabrobot", uuid.uuid4().hex, 1) + definition = build_tip_rack_definition(tip_rack, tip, identity.load_name) + await registry.load(tip_rack, str(slot), identity, definition) + + def _deck_to_robot_frame(self, location: Coordinate) -> Coordinate: + return location - self.deck.slot_locations[0] diff --git a/pylabrobot/opentrons/ot2/ot2_tests.py b/pylabrobot/opentrons/ot2/ot2_tests.py new file mode 100644 index 00000000000..73d56c4184f --- /dev/null +++ b/pylabrobot/opentrons/ot2/ot2_tests.py @@ -0,0 +1,895 @@ +import asyncio +import unittest +from typing import Any, Dict, List, Optional, Tuple + +from pylabrobot.io.http import HTTP, HTTPError +from pylabrobot.opentrons import OT2, OpentronsError +from pylabrobot.opentrons.types import ModuleInfo +from pylabrobot.resources import Coordinate, set_tip_tracking, set_volume_tracking +from pylabrobot.resources.celltreat import celltreat_96_wellplate_350uL_Fb +from pylabrobot.resources.errors import TooLittleLiquidError, TooLittleVolumeError +from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul + + +class FakeHTTP(HTTP): + """In-memory OT-2 HTTP API with successful commands by default.""" + + def __init__( + self, + left_pipette_name: Optional[str] = "p20_single_gen2", + right_pipette_name: Optional[str] = None, + api_version: str = "7.1.0", + ): + self.left_pipette_name = left_pipette_name + self.right_pipette_name = right_pipette_name + self.api_version = api_version + self.calls: List[Tuple[str, str, Optional[Dict[str, Any]]]] = [] + self.commands: List[Dict[str, Any]] = [] + self.command_results: Dict[str, Dict[str, Any]] = {} + self.fail_command_type: Optional[str] = None + self.fail_command_occurrence = 1 + self.stop_action_supported = True + self.stop_requests_fail = False + self.started = False + self.saved_position = {"x": 30.25, "y": 40.5, "z": 20.0} + + async def setup(self) -> None: + self.started = True + + async def stop(self) -> None: + self.started = False + + async def request( + self, + method: str, + path: str, + data: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + await asyncio.sleep(0) + self.calls.append((method, path, data)) + if self.stop_requests_fail and (path.endswith(("/actions", "/cancel")) or method == "DELETE"): + raise RuntimeError("stop request rejected") + if method == "POST" and path == "/runs": + return {"data": {"id": "run-id"}} + if method == "GET" and path == "/runs": + return {"data": [{"id": "run-id", "status": "idle"}]} + if method == "GET" and path == "/runs/run-id": + return {"data": {"id": "run-id", "status": "stopped"}} + if method == "GET" and path == "/pipettes": + return { + "left": {"name": self.left_pipette_name}, + "right": {"name": self.right_pipette_name}, + } + if method == "GET" and path == "/health": + return {"name": "test-ot2", "robot_model": "OT-2 Standard", "api_version": self.api_version} + if method == "POST" and path == "/robot/home": + return {"data": {}} + if method == "GET" and path == "/modules": + return {"data": [{"id": "temperature-module"}]} + if method == "POST" and path == "/runs/run-id/actions": + if data != {"data": {"actionType": "stop"}}: + raise AssertionError(f"Unexpected stop action: {data}") + if not self.stop_action_supported: + raise HTTPError(method, path, 404, "stop action is unsupported") + return {"data": {}} + if method == "POST" and path == "/runs/run-id/cancel": + return {"data": {}} + if method == "POST" and path == "/runs/run-id/labware_definitions": + return {"data": {"definitionUri": "pylabrobot/fake-tip-rack/1"}} + if method == "POST" and path == "/runs/run-id/commands": + assert data is not None + command = data["data"] + self.commands.append(command) + command_id = f"command-{len(self.commands)}" + result: Dict[str, Any] = {} + if command["commandType"] == "loadPipette": + result = {"pipetteId": f"{command['params']['mount']}-pipette-id"} + elif command["commandType"] == "savePosition": + result = {"positionId": "position-id", "position": self.saved_position.copy()} + self.command_results[command_id] = { + "commandType": command["commandType"], + "result": result, + "occurrence": sum( + previous["commandType"] == command["commandType"] for previous in self.commands + ), + } + return {"data": {"id": command_id}} + if method == "GET" and path.startswith("/runs/run-id/commands/"): + command_id = path.rsplit("/", 1)[-1] + command = self.command_results[command_id] + if ( + command["commandType"] == self.fail_command_type + and command["occurrence"] >= self.fail_command_occurrence + ): + return { + "data": { + "status": "failed", + "error": {"errorType": "hardware", "detail": "simulated failure"}, + } + } + return {"data": {"status": "succeeded", "result": command["result"]}} + raise AssertionError(f"Unexpected HTTP request: {method} {path} {data}") + + +class OT2Tests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + set_tip_tracking(True) + set_volume_tracking(True) + self.io = FakeHTTP() + self.deck = OTDeck() + self.robot = OT2( + host="ot2.local", + deck=self.deck, + command_poll_interval=0, + io=self.io, + ) + await self.robot.setup() + self.tips = opentrons_96_filtertiprack_20ul(name="tips") + self.tips.model = None + self.deck.assign_child_at_slot(self.tips, slot=1) + self.plate = celltreat_96_wellplate_350uL_Fb(name="plate") + self.deck.assign_child_at_slot(self.plate, slot=2) + + async def asyncTearDown(self) -> None: + if self.robot._run is not None: + await self.robot.stop() + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_setup_discovers_real_pipette_objects_and_homes(self) -> None: + self.assertTrue(self.io.started) + self.assertIsNotNone(self.robot.left_pipette) + assert self.robot.left_pipette is not None + self.assertEqual(self.robot.left_pipette.mount, "left") + self.assertEqual(self.robot.left_pipette.name, "p20_single_gen2") + self.assertEqual(self.robot.left_pipette.num_channels, 1) + self.assertIsNone(self.robot.right_pipette) + self.assertIn(("POST", "/robot/home", {"target": "robot"}), self.io.calls) + self.assertEqual(await self.robot.list_connected_modules(), (ModuleInfo("temperature-module"),)) + + async def test_full_single_channel_protocol_updates_trackers_and_commands(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + source = self.plate.get_well("A1") + destination = self.plate.get_well("B1") + source.tracker.set_volume(15) + + await pipette.pick_up_tip(self.tips.get_item("A1")) + await pipette.aspirate(source, volume=10) + await pipette.dispense(destination, volume=10) + await pipette.discard_tip() + + self.assertFalse(self.tips.get_item("A1").has_tip()) + self.assertAlmostEqual(source.tracker.get_used_volume(), 5) + self.assertAlmostEqual(destination.tracker.get_used_volume(), 10) + self.assertFalse(pipette.has_tip) + + command_types = [command["commandType"] for command in self.io.commands] + self.assertEqual(command_types.count("loadLabware"), 1) + self.assertEqual(command_types.count("pickUpTip"), 1) + self.assertEqual(command_types.count("aspirateInPlace"), 1) + self.assertEqual(command_types.count("dispenseInPlace"), 1) + self.assertEqual(command_types.count("moveToCoordinates"), 6) + self.assertEqual(command_types.count("moveToAddressableAreaForDropTip"), 1) + self.assertEqual(command_types.count("dropTipInPlace"), 1) + + definition_request = next( + data + for method, path, data in self.io.calls + if method == "POST" and path.endswith("/labware_definitions") + ) + assert definition_request is not None + definition = definition_request["data"] + self.assertEqual(definition["ordering"][0][0], "A1") + self.assertIn("A1", definition["wells"]) + self.assertEqual(definition["groups"][0]["metadata"], {}) + self.assertEqual(definition["cornerOffsetFromSlot"], {"x": 0, "y": 0, "z": 0}) + self.assertEqual( + definition["wells"]["A1"]["depth"], + definition["parameters"]["tipLength"], + ) + + load_labware = next( + command for command in self.io.commands if command["commandType"] == "loadLabware" + ) + self.assertIsInstance(load_labware["params"]["version"], int) + + pick_up_tip = next( + command for command in self.io.commands if command["commandType"] == "pickUpTip" + ) + self.assertEqual(pick_up_tip["params"]["wellName"], "A1") + + move_to_trash = next( + command + for command in self.io.commands + if command["commandType"] == "moveToAddressableAreaForDropTip" + ) + self.assertEqual(move_to_trash["params"]["offset"], {"x": 0, "y": 0, "z": 10}) + self.assertNotIn("wellLocation", move_to_trash["params"]) + + aspirate = next( + command for command in self.io.commands if command["commandType"] == "aspirateInPlace" + ) + dispense = next( + command for command in self.io.commands if command["commandType"] == "dispenseInPlace" + ) + self.assertEqual(aspirate["params"]["flowRate"], 3.78) + self.assertEqual(dispense["params"]["flowRate"], 7.56) + self.assertEqual(dispense["params"]["pushOut"], 0.0) + + async def test_return_tip_restores_its_origin(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + origin = self.tips.get_item("A1") + + await pipette.pick_up_tip(origin) + await pipette.return_tip() + + self.assertTrue(origin.has_tip()) + self.assertFalse(pipette.has_tip) + command_types = [command["commandType"] for command in self.io.commands] + self.assertEqual(command_types.count("loadLabware"), 1) + self.assertEqual(command_types.count("dropTip"), 1) + self.assertEqual(command_types[-3:], ["dropTip", "savePosition", "moveToCoordinates"]) + + async def test_pickup_retracts_with_tip_state_committed(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + origin = self.tips.get_item("A1") + + await pipette.pick_up_tip(origin) + + pickup, save, retract = self.io.commands[-3:] + self.assertEqual(pickup["commandType"], "pickUpTip") + self.assertEqual(save["commandType"], "savePosition") + self.assertEqual(save["params"], {"pipetteId": pipette.pipette_id}) + self.assertEqual(retract["commandType"], "moveToCoordinates") + self.assertEqual(retract["params"]["coordinates"], {"x": 30.25, "y": 40.5, "z": 120}) + self.assertTrue(retract["params"]["forceDirect"]) + self.assertTrue(pipette.has_tip) + self.assertFalse(origin.has_tip()) + + async def test_failed_retraction_preserves_completed_pickup(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + origin = self.tips.get_item("A1") + for failed_command in ("savePosition", "moveToCoordinates"): + with self.subTest(failed_command=failed_command): + self.io.fail_command_type = failed_command + tip = origin.get_tip() + + with self.assertRaisesRegex(OpentronsError, failed_command): + await pipette.pick_up_tip(origin) + + self.assertIs(pipette.tip, tip) + self.assertFalse(origin.has_tip()) + self.io.fail_command_type = None + await pipette.return_tip() + self.assertIs(origin.get_tip(), tip) + + async def test_move_to_retracts_without_lowering_an_already_high_tip(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + for z in (10, 150): + with self.subTest(z=z): + self.io.saved_position = {"x": 100, "y": 200, "z": z} + command_count = len(self.io.commands) + + await asyncio.wait_for(pipette.move_to(Coordinate(100, 200, z)), timeout=1) + + commands = self.io.commands[command_count:] + self.assertEqual(commands[0]["commandType"], "moveToCoordinates") + self.assertEqual(commands[0]["params"]["coordinates"], self.io.saved_position) + self.assertEqual(commands[1]["commandType"], "savePosition") + if z < self.robot.traversal_height: + self.assertEqual(len(commands), 3) + self.assertEqual(commands[2]["commandType"], "moveToCoordinates") + self.assertEqual(commands[2]["params"]["coordinates"], {"x": 100, "y": 200, "z": 120}) + self.assertTrue(commands[2]["params"]["forceDirect"]) + else: + self.assertEqual(len(commands), 2) + + async def test_drop_tip_retracts_vertically_from_reported_position(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + self.tips.set_tip_state({"A12": False}) + self.robot.traversal_height = 140 + + await pipette.drop_tip(self.tips.get_item("A12")) + + drop, save, retract = self.io.commands[-3:] + self.assertEqual(drop["commandType"], "dropTip") + self.assertEqual(drop["params"]["wellName"], "A12") + self.assertEqual(save["commandType"], "savePosition") + self.assertEqual(save["params"], {"pipetteId": pipette.pipette_id}) + self.assertEqual(retract["commandType"], "moveToCoordinates") + self.assertEqual(retract["params"]["pipetteId"], pipette.pipette_id) + self.assertEqual(retract["params"]["coordinates"], {"x": 30.25, "y": 40.5, "z": 140}) + self.assertTrue(retract["params"]["forceDirect"]) + self.assertTrue(self.tips.get_item("A12").has_tip()) + self.assertFalse(pipette.has_tip) + + async def test_discard_tip_retracts_for_both_trash_apis(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + for tip_index, version in enumerate(("6.3.0", "7.1.0")): + with self.subTest(version=version): + await self.robot.stop() + self.io.api_version = version + await self.robot.setup(skip_home=True) + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item(tip_index)) + + await pipette.discard_tip() + + drop, save, retract = self.io.commands[-3:] + expected_drop = "dropTip" if version == "6.3.0" else "dropTipInPlace" + self.assertEqual(drop["commandType"], expected_drop) + self.assertEqual(save["commandType"], "savePosition") + self.assertEqual(retract["commandType"], "moveToCoordinates") + self.assertEqual(retract["params"]["coordinates"], {"x": 30.25, "y": 40.5, "z": 120}) + self.assertTrue(retract["params"]["forceDirect"]) + self.assertFalse(pipette.has_tip) + + async def test_drop_does_not_lower_a_nozzle_above_traversal_height(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + self.io.saved_position["z"] = self.robot.traversal_height + 10 + command_count = len(self.io.commands) + + await pipette.return_tip() + + self.assertEqual( + [command["commandType"] for command in self.io.commands[command_count:]], + ["dropTip", "savePosition"], + ) + self.assertFalse(pipette.has_tip) + + async def test_failed_drop_preserves_tip_and_does_not_retract(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + tip = pipette.tip + self.io.fail_command_type = "dropTip" + command_count = len(self.io.commands) + + with self.assertRaisesRegex(OpentronsError, "dropTip"): + await pipette.return_tip() + + self.assertIs(pipette.tip, tip) + self.assertFalse(self.tips.get_item("A1").has_tip()) + self.assertEqual( + [command["commandType"] for command in self.io.commands[command_count:]], ["dropTip"] + ) + + async def test_failed_retraction_preserves_completed_tip_drop(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + for tip_index, (operation, failed_command, returned) in enumerate( + ( + (pipette.return_tip, "savePosition", True), + (pipette.return_tip, "moveToCoordinates", True), + (pipette.discard_tip, "savePosition", False), + (pipette.discard_tip, "moveToCoordinates", False), + ) + ): + with self.subTest(operation=operation.__name__, failed_command=failed_command): + self.io.fail_command_type = None + origin = self.tips.get_item(tip_index) + await pipette.pick_up_tip(origin) + self.io.fail_command_type = failed_command + + with self.assertRaisesRegex(OpentronsError, failed_command): + await operation() + + self.assertFalse(pipette.has_tip) + self.assertIsNone(pipette.tip) + self.assertEqual(origin.has_tip(), returned) + with self.assertRaisesRegex(RuntimeError, "origin is unknown"): + await pipette.return_tip() + + async def test_official_tip_rack_uses_builtin_definition_for_tip_length_calibration(self) -> None: + tips = opentrons_96_filtertiprack_20ul(name="official_tips") + self.deck.assign_child_at_slot(tips, slot=3) + pipette = self.robot.left_pipette + assert pipette is not None + definition_request_count = len( + [ + path + for method, path, _ in self.io.calls + if method == "POST" and path.endswith("definitions") + ] + ) + + await pipette.pick_up_tip(tips.get_item("A1")) + + self.assertEqual( + len( + [ + path + for method, path, _ in self.io.calls + if method == "POST" and path.endswith("definitions") + ] + ), + definition_request_count, + ) + load_labware = next( + command for command in reversed(self.io.commands) if command["commandType"] == "loadLabware" + ) + self.assertEqual(load_labware["params"]["namespace"], "opentrons") + self.assertEqual( + load_labware["params"]["loadName"], + "opentrons_96_filtertiprack_20ul", + ) + self.assertEqual(load_labware["params"]["version"], 1) + + async def test_failed_aspiration_rolls_back_volume_trackers(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + source = self.plate.get_well("A1") + source.tracker.set_volume(15) + await pipette.pick_up_tip(self.tips.get_item("A1")) + self.io.fail_command_type = "aspirateInPlace" + + with self.assertRaisesRegex(OpentronsError, "simulated failure"): + await pipette.aspirate(source, volume=10) + + self.assertAlmostEqual(source.tracker.get_used_volume(), 15) + assert pipette.tip is not None + self.assertAlmostEqual(pipette.tip.tracker.get_used_volume(), 0) + + async def test_liquid_operations_retract_from_reported_position(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + well = self.plate.get_well("A1") + for operation, initial_tip in ((pipette.aspirate, 0), (pipette.dispense, 10)): + with self.subTest(operation=operation.__name__): + well.tracker.set_volume(30) + pipette.tip.tracker.set_volume(initial_tip) + + await operation(well, volume=10) + + save, retract = self.io.commands[-2:] + self.assertEqual(save["commandType"], "savePosition") + self.assertEqual(save["params"], {"pipetteId": pipette.pipette_id}) + self.assertEqual(retract["commandType"], "moveToCoordinates") + self.assertEqual(retract["params"]["coordinates"], {"x": 30.25, "y": 40.5, "z": 120}) + self.assertTrue(retract["params"]["forceDirect"]) + + async def test_rejected_aspiration_preserves_both_volume_trackers(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + pipette.tip.tracker.set_volume(15) + source = self.plate.get_well("A1") + source.tracker.set_volume(30) + command_count = len(self.io.commands) + + with self.assertRaises(TooLittleVolumeError): + await pipette.aspirate(source, volume=10) + + self.assertEqual(len(self.io.commands), command_count) + self.assertEqual(source.tracker.get_used_volume(), 30) + self.assertEqual(source.tracker.volume, 30) + self.assertEqual(pipette.tip.tracker.get_used_volume(), 15) + self.assertEqual(pipette.tip.tracker.volume, 15) + + async def test_rejected_dispense_preserves_both_volume_trackers(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + pipette.tip.tracker.set_volume(10) + destination = self.plate.get_well("A1") + destination.tracker.set_volume(destination.tracker.max_volume - 5) + initial_volume = destination.tracker.get_used_volume() + command_count = len(self.io.commands) + + with self.assertRaises(TooLittleVolumeError): + await pipette.dispense(destination, volume=10) + + self.assertEqual(len(self.io.commands), command_count) + self.assertEqual(destination.tracker.get_used_volume(), initial_volume) + self.assertEqual(destination.tracker.volume, initial_volume) + self.assertEqual(pipette.tip.tracker.get_used_volume(), 10) + self.assertEqual(pipette.tip.tracker.volume, 10) + + async def test_failed_retraction_preserves_completed_transfer(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + well = self.plate.get_well("A1") + self.io.fail_command_type = "moveToCoordinates" + for operation, initial_tip, expected_tip, expected_well in ( + (pipette.aspirate, 0, 10, 20), + (pipette.dispense, 10, 0, 40), + ): + with self.subTest(operation=operation.__name__): + well.tracker.set_volume(30) + pipette.tip.tracker.set_volume(initial_tip) + move_count = sum( + command["commandType"] == "moveToCoordinates" for command in self.io.commands + ) + self.io.fail_command_occurrence = move_count + 2 + + with self.assertRaisesRegex(OpentronsError, "moveToCoordinates"): + await operation(well, volume=10) + + self.assertEqual(well.tracker.get_used_volume(), expected_well) + self.assertEqual(well.tracker.volume, expected_well) + self.assertEqual(pipette.tip.tracker.get_used_volume(), expected_tip) + self.assertEqual(pipette.tip.tracker.volume, expected_tip) + + async def test_moved_or_unassigned_loaded_rack_is_rejected_before_a_command(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + await pipette.return_tip() + self.deck.unassign_child_resource(self.tips) + command_count = len(self.io.commands) + + with self.assertRaisesRegex(ValueError, "assigned directly"): + await pipette.pick_up_tip(self.tips.get_item("A1")) + self.deck.assign_child_at_slot(self.tips, slot=5) + with self.assertRaisesRegex(ValueError, "loaded in slot 1"): + await pipette.pick_up_tip(self.tips.get_item("A1")) + + self.assertEqual(len(self.io.commands), command_count) + self.assertTrue(self.tips.get_item("A1").has_tip()) + self.assertFalse(pipette.has_tip) + + async def test_drop_into_moved_rack_preserves_tip_state(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + self.deck.unassign_child_resource(self.tips) + self.deck.assign_child_at_slot(self.tips, slot=5) + command_count = len(self.io.commands) + + with self.assertRaisesRegex(ValueError, "loaded in slot 1"): + await pipette.return_tip() + + self.assertEqual(len(self.io.commands), command_count) + self.assertFalse(self.tips.get_item("A1").has_tip()) + self.assertTrue(pipette.has_tip) + + async def test_concurrent_pickups_only_pick_up_one_tip(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + results = await asyncio.wait_for( + asyncio.gather( + pipette.pick_up_tip(self.tips.get_item("A1")), + pipette.pick_up_tip(self.tips.get_item("A2")), + return_exceptions=True, + ), + timeout=1, + ) + + self.assertIsNone(results[0]) + self.assertIsInstance(results[1], RuntimeError) + self.assertIn("already has a tip", str(results[1])) + self.assertEqual(sum(command["commandType"] == "pickUpTip" for command in self.io.commands), 1) + self.assertTrue(pipette.has_tip) + self.assertFalse(self.tips.get_item("A1").has_tip()) + self.assertTrue(self.tips.get_item("A2").has_tip()) + + async def test_concurrent_transfer_checks_state_after_preceding_operation(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + source = self.plate.get_well("A1") + destination = self.plate.get_well("B1") + source.tracker.set_volume(15) + + await asyncio.wait_for( + asyncio.gather( + pipette.aspirate(source, volume=10), + pipette.dispense(destination, volume=10), + ), + timeout=1, + ) + + self.assertEqual(source.tracker.get_used_volume(), 5) + self.assertEqual(destination.tracker.get_used_volume(), 10) + assert pipette.tip is not None + self.assertEqual(pipette.tip.tracker.get_used_volume(), 0) + + async def test_concurrent_return_and_discard_only_drop_once(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + results = await asyncio.wait_for( + asyncio.gather(pipette.return_tip(), pipette.discard_tip(), return_exceptions=True), + timeout=1, + ) + + self.assertIsNone(results[0]) + self.assertIsInstance(results[1], RuntimeError) + self.assertIn("does not have a tip", str(results[1])) + self.assertFalse(pipette.has_tip) + self.assertTrue(self.tips.get_item("A1").has_tip()) + self.assertEqual( + sum(command["commandType"] in {"dropTip", "dropTipInPlace"} for command in self.io.commands), + 1, + ) + + async def test_mix_rejects_insufficient_liquid_or_tip_capacity_before_moving(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + source = self.plate.get_well("A1") + command_count = len(self.io.commands) + for source_volume, tip_volume, error in ( + (0, 0, TooLittleLiquidError), + (30, 15, TooLittleVolumeError), + ): + with self.subTest(source_volume=source_volume, tip_volume=tip_volume): + source.tracker.set_volume(source_volume) + pipette.tip.tracker.set_volume(tip_volume) + with self.assertRaises(error): + await pipette.mix(source, volume=10, repetitions=3) + self.assertEqual(len(self.io.commands), command_count) + self.assertEqual(source.tracker.get_used_volume(), source_volume) + self.assertEqual(pipette.tip.tracker.get_used_volume(), tip_volume) + + async def test_mix_tracks_each_transfer_when_dispense_fails(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + source = self.plate.get_well("A1") + source.tracker.set_volume(15) + self.io.fail_command_type = "dispenseInPlace" + self.io.fail_command_occurrence = 2 + + with self.assertRaisesRegex(OpentronsError, "dispenseInPlace"): + await pipette.mix(source, volume=10, repetitions=3) + + self.assertEqual(source.tracker.get_used_volume(), 5) + self.assertEqual(source.tracker.volume, 5) + assert pipette.tip is not None + self.assertEqual(pipette.tip.tracker.get_used_volume(), 10) + self.assertEqual(pipette.tip.tracker.volume, 10) + + async def test_mix_preserves_volume_after_completed_cycles(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + await pipette.pick_up_tip(self.tips.get_item("A1")) + assert pipette.tip is not None + source = self.plate.get_well("A1") + source.tracker.set_volume(15) + pipette.tip.tracker.set_volume(5) + command_count = len(self.io.commands) + + await pipette.mix(source, volume=10, repetitions=3) + + self.assertEqual(source.tracker.get_used_volume(), 15) + self.assertEqual(pipette.tip.tracker.get_used_volume(), 5) + self.assertEqual( + sum(command["commandType"] == "aspirateInPlace" for command in self.io.commands), 3 + ) + self.assertEqual( + sum(command["commandType"] == "dispenseInPlace" for command in self.io.commands), 3 + ) + self.assertEqual( + [command["commandType"] for command in self.io.commands[command_count:]], + ["moveToCoordinates"] + + ["aspirateInPlace", "dispenseInPlace"] * 3 + + ["savePosition", "moveToCoordinates"], + ) + retract = self.io.commands[-1] + self.assertEqual(retract["params"]["coordinates"], {"x": 30.25, "y": 40.5, "z": 120}) + self.assertTrue(retract["params"]["forceDirect"]) + + async def test_unreachable_move_is_rejected_before_an_http_command(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + command_count = len(self.io.commands) + + with self.assertRaisesRegex(ValueError, "reachable"): + await pipette.move_to(Coordinate(500, 0, 10)) + + self.assertEqual(len(self.io.commands), command_count) + + async def test_negative_z_move_is_rejected_before_an_http_command(self) -> None: + pipette = self.robot.left_pipette + assert pipette is not None + command_count = len(self.io.commands) + + with self.assertRaisesRegex(ValueError, "non-negative"): + await pipette.move_to(Coordinate(10, 10, -1)) + + self.assertEqual(len(self.io.commands), command_count) + + async def test_stop_cancels_run_and_clears_discovered_state(self) -> None: + await self.robot.stop() + self.assertFalse(self.io.started) + self.assertIsNone(self.robot.left_pipette) + self.assertIsNone(self.robot.software_version) + self.assertIn( + ("POST", "/runs/run-id/actions", {"data": {"actionType": "stop"}}), + self.io.calls, + ) + + async def test_stop_falls_back_for_older_robot_software(self) -> None: + self.io.stop_action_supported = False + + await self.robot.stop() + + self.assertIn(("POST", "/runs/run-id/cancel", None), self.io.calls) + + async def test_failed_stop_retains_state_and_transport_for_retry(self) -> None: + pipette = self.robot.left_pipette + run_id = self.robot._run + self.io.stop_requests_fail = True + + with self.assertRaisesRegex(OpentronsError, "Could not cancel") as error: + await self.robot.stop() + + self.assertIsInstance(error.exception.__cause__, RuntimeError) + self.assertEqual(self.robot._run, run_id) + self.assertIs(self.robot.left_pipette, pipette) + self.assertTrue(self.io.started) + self.io.stop_requests_fail = False + await self.robot.stop() + self.assertIsNone(self.robot._run) + self.assertIsNone(self.robot.left_pipette) + self.assertFalse(self.io.started) + + async def test_failed_setup_cleanup_retains_run_for_stop_retry(self) -> None: + await self.robot.stop() + self.io.fail_command_type = "loadPipette" + self.io.stop_requests_fail = True + + with self.assertRaisesRegex(OpentronsError, "Could not cancel"): + await self.robot.setup() + + assert self.robot._run is not None + self.assertEqual(self.robot._run.id, "run-id") + self.assertTrue(self.io.started) + self.io.stop_requests_fail = False + await self.robot.stop() + self.assertIsNone(self.robot._run) + self.assertFalse(self.io.started) + + +class OT2MultiChannelTests(unittest.IsolatedAsyncioTestCase): + async def test_multi_channel_is_modeled_but_not_mistracked_as_one_tip(self) -> None: + io = FakeHTTP(left_pipette_name="p20_multi_gen2") + deck = OTDeck() + robot = OT2(host="ot2.local", deck=deck, command_poll_interval=0, io=io) + await robot.setup(skip_home=True) + tips = opentrons_96_filtertiprack_20ul(name="tips") + deck.assign_child_at_slot(tips, slot=1) + assert robot.left_pipette is not None + self.assertEqual(robot.left_pipette.num_channels, 8) + + with self.assertRaisesRegex(NotImplementedError, "Multi-channel"): + await robot.left_pipette.pick_up_tip(tips.get_item("A1")) + + await robot.stop() + + +class OT2ArchitectureTests(unittest.IsolatedAsyncioTestCase): + async def test_connection_allows_queries_without_creating_a_run(self) -> None: + io = FakeHTTP() + robot = OT2("ot2.local", io=io) + await robot.connect() + try: + self.assertEqual((await robot.get_health()).software_version, "7.1.0") + self.assertEqual((await robot.get_mounted_pipettes())[0].name, "p20_single_gen2") + self.assertEqual((await robot.get_runs())[0].id, "run-id") + self.assertEqual(await robot.list_connected_modules(), (ModuleInfo("temperature-module"),)) + self.assertIsNone(robot._run) + self.assertIsNone(robot.software_version) + self.assertEqual(robot.pipettes, []) + finally: + await robot.stop() + self.assertTrue(all(method == "GET" for method, _, _ in io.calls)) + self.assertFalse(io.started) + + async def test_queries_do_not_reconfigure_an_active_session(self) -> None: + io = FakeHTTP() + robot = OT2("ot2.local", io=io) + await robot.setup(skip_home=True) + try: + state = vars(robot).copy() + io.api_version = "8.7.0" + io.left_pipette_name = "p300_single_gen2" + command_count = len(io.commands) + self.assertEqual((await robot.get_health()).software_version, "8.7.0") + self.assertEqual((await robot.get_mounted_pipettes())[0].name, "p300_single_gen2") + self.assertEqual(robot.software_version, "7.1.0") + self.assertEqual(vars(robot), state) + self.assertEqual(len(io.commands), command_count) + finally: + await robot.stop() + + async def test_reconnect_invalidates_old_pipette_and_labware_bindings(self) -> None: + io = FakeHTTP() + robot = OT2("ot2.local", io=io) + await robot.setup(skip_home=True) + old_pipette, old_registry = robot.left_pipette, robot._labware + assert old_pipette is not None + await robot.stop() + await robot.setup(skip_home=True) + try: + self.assertIsNot(robot._labware, old_registry) + command_count = len(io.commands) + with self.assertRaisesRegex(RuntimeError, "earlier OT-2 run"): + await old_pipette.move_to(Coordinate(100, 100, 120)) + self.assertEqual(len(io.commands), command_count) + assert robot.left_pipette is not None + await robot.left_pipette.move_to(Coordinate(100, 100, 120)) + finally: + await robot.stop() + + async def test_two_robots_own_independent_sessions_and_connections(self) -> None: + left_io = FakeHTTP(api_version="6.3.0") + right_io = FakeHTTP(api_version="8.7.0") + left = OT2("first.local", io=left_io) + right = OT2("second.local", io=right_io) + await asyncio.gather(left.setup(skip_home=True), right.setup(skip_home=True)) + try: + self.assertIsNot(left._run, right._run) + self.assertIsNot(left._labware, right._labware) + self.assertEqual((left.software_version, right.software_version), ("6.3.0", "8.7.0")) + await left.stop() + self.assertTrue(right_io.started) + assert right.left_pipette is not None + await right.left_pipette.move_to(Coordinate(100, 100, 120)) + self.assertEqual([c["commandType"] for c in left_io.commands], ["loadPipette"]) + finally: + await asyncio.gather(left.stop(), right.stop()) + + async def test_unsupported_pipette_does_not_create_a_run_or_home(self) -> None: + io = FakeHTTP(left_pipette_name="unsupported") + robot = OT2("ot2.local", io=io) + with self.assertRaisesRegex(ValueError, "Unsupported OT-2 pipette"): + await robot.setup() + self.assertIsNone(robot._run) + self.assertFalse(io.started) + self.assertTrue(all(method == "GET" for method, _, _ in io.calls)) + + async def test_stop_waits_for_the_complete_pipette_operation(self) -> None: + entered, release = asyncio.Event(), asyncio.Event() + + class PausingHTTP(FakeHTTP): + async def request( + self, method: str, path: str, data: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: + if data is not None and data.get("data", {}).get("commandType") == "moveToCoordinates": + entered.set() + await release.wait() + return await super().request(method, path, data) + + io = PausingHTTP() + robot = OT2("ot2.local", io=io) + await robot.setup(skip_home=True) + assert robot.left_pipette is not None + movement = asyncio.create_task(robot.left_pipette.move_to(Coordinate(100, 100, 100))) + await asyncio.wait_for(entered.wait(), timeout=1) + stop = asyncio.create_task(robot.stop()) + try: + await asyncio.sleep(0) + self.assertFalse(stop.done()) + self.assertTrue(io.started) + finally: + release.set() + await asyncio.wait_for(asyncio.gather(movement, stop), timeout=1) + self.assertFalse(io.started) + self.assertEqual( + [command["commandType"] for command in io.commands[-3:]], + ["moveToCoordinates", "savePosition", "moveToCoordinates"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/opentrons/ot2/pipette.py b/pylabrobot/opentrons/ot2/pipette.py new file mode 100644 index 00000000000..3b6507677e9 --- /dev/null +++ b/pylabrobot/opentrons/ot2/pipette.py @@ -0,0 +1,475 @@ +from __future__ import annotations + +import math +from contextlib import contextmanager +from dataclasses import dataclass +from typing import TYPE_CHECKING, Dict, Iterator, Optional + +from pylabrobot.opentrons.types import Mount +from pylabrobot.resources.container import Container +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.tip import Tip +from pylabrobot.resources.tip_rack import TipRack, TipSpot +from pylabrobot.resources.tip_tracker import does_tip_tracking +from pylabrobot.resources.volume_tracker import VolumeTracker, does_volume_tracking + +if TYPE_CHECKING: + from pylabrobot.opentrons.ot2.ot2 import OT2 + + +@dataclass(frozen=True) +class _PipetteSpec: + minimum_volume: float + maximum_volume: float + channels: int + default_aspiration_flow_rate: float + default_dispense_flow_rate: float + + +_PIPETTE_SPECS = { + "p10_single": _PipetteSpec(1, 10, 1, 5, 10), + "p10_multi": _PipetteSpec(1, 10, 8, 5, 10), + "p20_single_gen2": _PipetteSpec(1, 20, 1, 3.78, 7.56), + "p20_multi_gen2": _PipetteSpec(1, 20, 8, 7.6, 7.6), + "p50_single": _PipetteSpec(5, 50, 1, 25, 50), + "p50_multi": _PipetteSpec(5, 50, 8, 25, 50), + "p300_single": _PipetteSpec(30, 300, 1, 150, 300), + "p300_multi": _PipetteSpec(30, 300, 8, 150, 300), + "p300_single_gen2": _PipetteSpec(20, 300, 1, 46.43, 92.86), + "p300_multi_gen2": _PipetteSpec(20, 300, 8, 94, 94), + "p1000_single": _PipetteSpec(100, 1000, 1, 500, 1000), + "p1000_single_gen2": _PipetteSpec(100, 1000, 1, 137.35, 274.7), +} + + +_COMPATIBLE_TIP_CAPACITIES: Dict[float, set] = { + 10: {10}, + 20: {10, 20}, + 50: {200}, + 300: {200, 300}, + 1000: {1000}, +} + + +def _require_finite_coordinate(name: str, coordinate: Coordinate) -> None: + if not all(math.isfinite(axis) for axis in coordinate): + raise ValueError(f"{name} coordinates must be finite") + + +@contextmanager +def _track_liquid_transfer( + source: VolumeTracker, destination: VolumeTracker, volume: float +) -> Iterator[None]: + """Track one liquid transfer, committing when its command succeeds.""" + trackers = [ + tracker + for tracker in (source, destination) + if does_volume_tracking() and not tracker.is_disabled + ] + try: + if source in trackers: + source.remove_liquid(volume) + if destination in trackers: + destination.add_liquid(volume) + yield + except BaseException: + for tracker in trackers: + tracker.rollback() + raise + else: + for tracker in trackers: + tracker.commit() + + +class OT2Pipette: + """A pipette mounted on an OT-2 carriage. + + Instances are discovered and created by :meth:`OT2.setup`. Single-channel + pipettes expose tip, liquid, and motion operations. Multi-channel pipettes are represented + accurately, but their liquid operations are rejected until all eight tip and volume trackers + can be updated atomically. + """ + + def __init__( + self, + robot: OT2, + mount: Mount, + name: str, + pipette_id: str, + ): + try: + spec = _PIPETTE_SPECS[name] + except KeyError as error: + raise ValueError(f"Unsupported OT-2 pipette {name!r}") from error + + self.robot = robot + self._run = robot._require_run() + self.mount = mount + self.name = name + self.pipette_id = pipette_id + self._spec = spec + self._tip: Optional[Tip] = None + self._tip_origin: Optional[TipSpot] = None + + @property + def minimum_volume(self) -> float: + """Minimum supported transfer volume, in µL.""" + return self._spec.minimum_volume + + @property + def maximum_volume(self) -> float: + """Maximum supported transfer volume, in µL.""" + return self._spec.maximum_volume + + @property + def num_channels(self) -> int: + """Number of nozzles on the pipette.""" + return self._spec.channels + + @property + def has_tip(self) -> bool: + """Whether the pipette holds a tip according to commands issued by this object.""" + return self._tip is not None + + @property + def tip(self) -> Optional[Tip]: + """The mounted tip, or ``None`` when no tip is mounted.""" + return self._tip + + def _require_active(self) -> None: + if self.robot._require_run() is not self._run: + raise RuntimeError("This pipette belongs to an earlier OT-2 run; use the current pipette") + + def _require_single_channel(self) -> None: + if self.num_channels != 1: + raise NotImplementedError( + f"{self.name} has {self.num_channels} channels. Multi-channel liquid operations are not " + "implemented yet." + ) + + def _require_tip(self) -> Tip: + if self._tip is None: + raise RuntimeError(f"The {self.mount} pipette does not have a tip") + return self._tip + + def _validate_volume(self, volume: float) -> float: + volume = float(volume) + if not self.minimum_volume <= volume <= self.maximum_volume: + raise ValueError( + f"volume must be between {self.minimum_volume:g} and {self.maximum_volume:g} µL " + f"for {self.name}" + ) + return volume + + def can_use_tip(self, tip: Tip) -> bool: + """Whether the tip capacity is supported by this pipette.""" + return tip.maximal_volume in _COMPATIBLE_TIP_CAPACITIES[self.maximum_volume] + + async def _move_to( + self, + location: Coordinate, + speed: Optional[float] = None, + minimum_z_height: Optional[float] = None, + force_direct: bool = False, + ) -> None: + _require_finite_coordinate("location", location) + if location.z < 0: + raise ValueError("location.z must be non-negative") + if not self.robot.geometry.can_reach_position(self.mount, location): + bounds = self.robot.geometry.single_channel_reach(self.mount) + raise ValueError( + f"{location} is outside the {self.mount} mount's reachable x/y region {bounds}" + ) + if speed is not None and (not math.isfinite(speed) or speed <= 0): + raise ValueError("speed must be finite and greater than zero") + if minimum_z_height is not None and ( + not math.isfinite(minimum_z_height) or minimum_z_height < 0 + ): + raise ValueError("minimum_z_height must be finite and non-negative") + + await self._run.move_to( + self.pipette_id, + location, + speed=speed, + minimum_z_height=minimum_z_height, + force_direct=force_direct, + ) + + async def move_to( + self, + location: Coordinate, + speed: Optional[float] = None, + minimum_z_height: Optional[float] = None, + force_direct: bool = False, + ) -> None: + """Move to an absolute robot-frame coordinate, then retract to at least traversal height.""" + async with self.robot._operation_lock: + self._require_active() + await self._move_to( + location=location, + speed=speed, + minimum_z_height=minimum_z_height, + force_direct=force_direct, + ) + await self._retract_to_traversal_height() + + async def pick_up_tip( + self, + tip_spot: TipSpot, + offset: Optional[Coordinate] = None, + ) -> None: + """Pick up one tip from a tip rack and retract to at least traversal height.""" + async with self.robot._operation_lock: + self._require_active() + self._require_single_channel() + if self._tip is not None: + raise RuntimeError(f"The {self.mount} pipette already has a tip") + if not isinstance(tip_spot.parent, TipRack): + raise ValueError("tip_spot must be assigned to a tip rack") + + tip = tip_spot.get_tip() + if not self.can_use_tip(tip): + raise ValueError(f"{self.name} cannot use a {tip.maximal_volume:g} µL-capacity tip") + offset = offset or Coordinate.zero() + _require_finite_coordinate("offset", offset) + tracked = does_tip_tracking() and not tip_spot.tracker.is_disabled + if tracked: + tip_spot.tracker.remove_tip(commit=False) + + try: + await self.robot._load_tip_rack(tip_spot.parent, tip) + binding = self.robot._require_labware().get(tip_spot.parent) + await self._run.pick_up_tip( + self.pipette_id, + binding.labware_id, + tip_spot.parent.get_child_identifier(tip_spot), + offset + Coordinate(z=tip.total_tip_length), + ) + except Exception: + if tracked: + tip_spot.tracker.rollback() + raise + + if tracked: + tip_spot.tracker.commit() + self._tip = tip + self._tip_origin = tip_spot + await self._retract_to_traversal_height() + + async def drop_tip( + self, + tip_spot: TipSpot, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Drop into a tip-rack position and retract vertically to at least traversal height.""" + async with self.robot._operation_lock: + self._require_active() + await self._drop_tip(tip_spot, offset, allow_nonzero_volume) + + async def _drop_tip( + self, + tip_spot: TipSpot, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Drop a tip while the robot's operation lock is held.""" + self._require_single_channel() + tip = self._require_tip() + if not isinstance(tip_spot.parent, TipRack): + raise ValueError("tip_spot must be assigned to a tip rack") + if does_volume_tracking() and tip.tracker.get_used_volume() > 0 and not allow_nonzero_volume: + raise ValueError("The mounted tip still contains liquid") + + offset = offset or Coordinate.zero() + _require_finite_coordinate("offset", offset) + tracked = does_tip_tracking() and not tip_spot.tracker.is_disabled + if tracked: + tip_spot.tracker.add_tip(tip, origin=tip_spot, commit=False) + + try: + await self.robot._load_tip_rack(tip_spot.parent, tip) + binding = self.robot._require_labware().get(tip_spot.parent) + await self._run.drop_tip( + self.pipette_id, + binding.labware_id, + tip_spot.parent.get_child_identifier(tip_spot), + offset + Coordinate(z=10), + ) + except Exception: + if tracked: + tip_spot.tracker.rollback() + raise + + if tracked: + tip_spot.tracker.commit() + self._tip = None + self._tip_origin = None + await self._retract_to_traversal_height() + + async def _retract_to_traversal_height(self) -> None: + """Raise the nozzle or mounted tip from its reported position under the operation lock.""" + position = await self._run.get_position(self.pipette_id) + _require_finite_coordinate("position", position) + if position.z < self.robot.traversal_height: + await self._move_to( + Coordinate(position.x, position.y, self.robot.traversal_height), + force_direct=True, + ) + + async def return_tip( + self, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Return the mounted tip to its pickup position and retract to at least traversal height.""" + async with self.robot._operation_lock: + self._require_active() + if self._tip_origin is None: + raise RuntimeError("The mounted tip's origin is unknown") + await self._drop_tip( + self._tip_origin, + offset=offset, + allow_nonzero_volume=allow_nonzero_volume, + ) + + async def discard_tip( + self, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Discard into fixed trash and retract vertically to at least traversal height.""" + async with self.robot._operation_lock: + self._require_active() + self._require_single_channel() + tip = self._require_tip() + if does_volume_tracking() and tip.tracker.get_used_volume() > 0 and not allow_nonzero_volume: + raise ValueError("The mounted tip still contains liquid") + offset = offset or Coordinate.zero() + _require_finite_coordinate("offset", offset) + + await self._run.discard_tip_in_fixed_trash(self.pipette_id, offset + Coordinate(z=10)) + + self._tip = None + self._tip_origin = None + await self._retract_to_traversal_height() + + def _liquid_location( + self, + container: Container, + offset: Coordinate, + liquid_height: float, + ) -> Coordinate: + _require_finite_coordinate("offset", offset) + if not math.isfinite(liquid_height) or liquid_height < 0: + raise ValueError("liquid_height must be finite and non-negative") + location = container.get_location_wrt( + self.robot.deck, + "c", + "c", + "cavity_bottom", + ) + return self.robot._deck_to_robot_frame(location + offset + Coordinate(z=liquid_height)) + + async def aspirate( + self, + container: Container, + volume: float, + flow_rate: Optional[float] = None, + liquid_height: float = 0, + offset: Optional[Coordinate] = None, + ) -> None: + """Aspirate liquid from a container and return to traversal height.""" + async with self.robot._operation_lock: + self._require_active() + self._require_single_channel() + tip = self._require_tip() + volume = self._validate_volume(volume) + flow_rate = self._spec.default_aspiration_flow_rate if flow_rate is None else float(flow_rate) + if not math.isfinite(flow_rate) or flow_rate <= 0: + raise ValueError("flow_rate must be finite and greater than zero") + offset = offset or Coordinate.zero() + location = self._liquid_location(container, offset, liquid_height) + + with _track_liquid_transfer(container.tracker, tip.tracker, volume): + await self._move_to( + location, + minimum_z_height=self.robot.traversal_height, + ) + await self._run.aspirate_in_place(self.pipette_id, volume, flow_rate) + await self._retract_to_traversal_height() + + async def dispense( + self, + container: Container, + volume: float, + flow_rate: Optional[float] = None, + liquid_height: float = 0, + offset: Optional[Coordinate] = None, + ) -> None: + """Dispense liquid into a container and return to traversal height.""" + async with self.robot._operation_lock: + self._require_active() + self._require_single_channel() + tip = self._require_tip() + volume = self._validate_volume(volume) + flow_rate = self._spec.default_dispense_flow_rate if flow_rate is None else float(flow_rate) + if not math.isfinite(flow_rate) or flow_rate <= 0: + raise ValueError("flow_rate must be finite and greater than zero") + offset = offset or Coordinate.zero() + location = self._liquid_location(container, offset, liquid_height) + + with _track_liquid_transfer(tip.tracker, container.tracker, volume): + await self._move_to( + location, + minimum_z_height=self.robot.traversal_height, + ) + await self._run.dispense_in_place(self.pipette_id, volume, flow_rate) + await self._retract_to_traversal_height() + + async def mix( + self, + container: Container, + volume: float, + repetitions: int, + aspiration_flow_rate: Optional[float] = None, + dispense_flow_rate: Optional[float] = None, + liquid_height: float = 0, + offset: Optional[Coordinate] = None, + ) -> None: + """Mix in place using aspiration and dispense cycles, then retract to traversal height.""" + async with self.robot._operation_lock: + self._require_active() + self._require_single_channel() + tip = self._require_tip() + volume = self._validate_volume(volume) + if repetitions < 1: + raise ValueError("repetitions must be at least 1") + aspiration_flow_rate = ( + self._spec.default_aspiration_flow_rate + if aspiration_flow_rate is None + else float(aspiration_flow_rate) + ) + dispense_flow_rate = ( + self._spec.default_dispense_flow_rate + if dispense_flow_rate is None + else float(dispense_flow_rate) + ) + if ( + not math.isfinite(aspiration_flow_rate) + or not math.isfinite(dispense_flow_rate) + or aspiration_flow_rate <= 0 + or dispense_flow_rate <= 0 + ): + raise ValueError("flow rates must be finite and greater than zero") + + offset = offset or Coordinate.zero() + location = self._liquid_location(container, offset, liquid_height) + for repetition in range(repetitions): + with _track_liquid_transfer(container.tracker, tip.tracker, volume): + if repetition == 0: + await self._move_to(location, minimum_z_height=self.robot.traversal_height) + await self._run.aspirate_in_place(self.pipette_id, volume, aspiration_flow_rate) + with _track_liquid_transfer(tip.tracker, container.tracker, volume): + await self._run.dispense_in_place(self.pipette_id, volume, dispense_flow_rate) + await self._retract_to_traversal_height() diff --git a/pylabrobot/opentrons/run.py b/pylabrobot/opentrons/run.py new file mode 100644 index 00000000000..7e40b6069ad --- /dev/null +++ b/pylabrobot/opentrons/run.py @@ -0,0 +1,234 @@ +"""Run-scoped Opentrons command primitives and completion handling.""" + +import asyncio +import math +import re +import time +from typing import Any, Dict, Optional, Tuple + +from pylabrobot.io.http import HTTPError +from pylabrobot.opentrons.api import OpentronsAPI +from pylabrobot.opentrons.errors import ( + OpentronsCommandError, + OpentronsCommandTimeout, + OpentronsError, + OpentronsProtocolError, +) +from pylabrobot.opentrons.types import LabwareIdentity, Mount, _object, _string +from pylabrobot.resources.coordinate import Coordinate + + +def _version_tuple(version: str) -> Tuple[int, ...]: + parts = [] + for part in version.split("."): + match = re.match(r"\d+", part) + if match is None: + break + parts.append(int(match.group())) + if not parts: + raise ValueError(f"Unrecognized robot software version {version!r}") + return tuple(parts) + + +def _version_at_least(version: str, required: str) -> bool: + actual, minimum = _version_tuple(version), _version_tuple(required) + width = max(len(actual), len(minimum)) + return actual + (0,) * (width - len(actual)) >= minimum + (0,) * (width - len(minimum)) + + +def _coordinates(location: Coordinate) -> Dict[str, float]: + return {"x": location.x, "y": location.y, "z": location.z} + + +def _well_params( + pipette_id: str, labware_id: str, well_name: str, offset: Coordinate +) -> Dict[str, Any]: + return { + "pipetteId": pipette_id, + "labwareId": labware_id, + "wellName": well_name, + "wellLocation": {"origin": "bottom", "offset": _coordinates(offset)}, + } + + +class OpentronsRun: + """One server run, bound to one client and a fixed software version. + + Primitives await confirmed completion and do not modify PLR resource trackers + or add pipette retractions. The owning device sequences those operations. + """ + + def __init__( + self, + api: OpentronsAPI, + run_id: str, + software_version: str, + command_timeout: float = 30, + command_poll_interval: float = 0.05, + ) -> None: + if not math.isfinite(command_timeout) or command_timeout <= 0: + raise ValueError("command_timeout must be finite and greater than zero") + if not math.isfinite(command_poll_interval) or command_poll_interval < 0: + raise ValueError("command_poll_interval must be finite and non-negative") + self._api = api + self._id = run_id + self._software_version = software_version + self.command_timeout = command_timeout + self.command_poll_interval = command_poll_interval + self._active = True + + @property + def id(self) -> str: + return self._id + + @property + def software_version(self) -> str: + return self._software_version + + @property + def active(self) -> bool: + return self._active + + def _require_active(self) -> None: + if not self._active: + raise RuntimeError(f"Opentrons run {self.id} has stopped") + + async def stop(self) -> None: + """Wait for this run to stop; retain active state until shutdown is confirmed.""" + if self._active: + await self._api.stop_run(self.id) + deadline = time.monotonic() + self.command_timeout + while True: + try: + status = (await self._api.get_run(self.id)).status + except HTTPError as error: + if error.status == 404: + break # The legacy stop route can delete the run. + raise + if status == "stopped": + break + if status is None: + raise OpentronsProtocolError(f"Missing status while waiting for run {self.id} to stop") + if time.monotonic() >= deadline: + raise OpentronsError( + f"Timed out waiting for run {self.id} to stop (status {status!r}); " + "state is retained for retry" + ) + await asyncio.sleep(self.command_poll_interval) + self._active = False + + async def _execute(self, command_type: str, params: Dict[str, Any]) -> Dict[str, Any]: + self._require_active() + command_id = await self._api.submit_command(self.id, command_type, params) + deadline = time.monotonic() + self.command_timeout + while True: + try: + command = await self._api.get_command(self.id, command_id) + except TimeoutError as error: + raise OpentronsCommandTimeout(self.id, command_id, command_type) from error + if command.status == "succeeded": + return command.result + if command.status == "failed": + raise OpentronsCommandError( + self.id, + command_id, + command_type, + command.error.get("errorType", "unknown"), + command.error.get("detail", "no detail returned"), + ) + if command.status not in {"queued", "running"}: + raise OpentronsProtocolError( + f"{command_type} ({command_id} in run {self.id}) returned " + f"unexpected command status {command.status!r}" + ) + if time.monotonic() >= deadline: + raise OpentronsCommandTimeout(self.id, command_id, command_type) + await asyncio.sleep(self.command_poll_interval) + + async def load_pipette(self, name: str, mount: Mount) -> str: + result = await self._execute("loadPipette", {"pipetteName": name, "mount": mount}) + return _string(result, "pipetteId") + + async def define_labware(self, definition: Dict[str, Any]) -> LabwareIdentity: + self._require_active() + return await self._api.define_labware(self.id, definition) + + async def load_labware( + self, identity: LabwareIdentity, slot: str, labware_id: str, display_name: str + ) -> None: + await self._execute( + "loadLabware", + { + "location": {"slotName": slot}, + "loadName": identity.load_name, + "namespace": identity.namespace, + "version": identity.version, + "labwareId": labware_id, + "displayName": display_name, + }, + ) + + async def pick_up_tip( + self, pipette_id: str, labware_id: str, well_name: str, offset: Coordinate + ) -> None: + await self._execute("pickUpTip", _well_params(pipette_id, labware_id, well_name, offset)) + + async def drop_tip( + self, pipette_id: str, labware_id: str, well_name: str, offset: Coordinate + ) -> None: + await self._execute("dropTip", _well_params(pipette_id, labware_id, well_name, offset)) + + async def move_to( + self, + pipette_id: str, + location: Coordinate, + speed: Optional[float] = None, + minimum_z_height: Optional[float] = None, + force_direct: bool = False, + ) -> None: + params: Dict[str, Any] = { + "pipetteId": pipette_id, + "coordinates": _coordinates(location), + "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._execute("moveToCoordinates", params) + + async def get_position(self, pipette_id: str) -> Coordinate: + """Read the nozzle or mounted tip's critical point through savePosition.""" + result = await self._execute("savePosition", {"pipetteId": pipette_id}) + position = _object(result.get("position")) + axes = [position.get(axis) for axis in ("x", "y", "z")] + if not all(isinstance(axis, (float, int)) and math.isfinite(axis) for axis in axes): + raise OpentronsProtocolError(f"Invalid position in savePosition result: {position!r}") + return Coordinate(x=position["x"], y=position["y"], z=position["z"]) + + async def aspirate_in_place(self, pipette_id: str, volume: float, flow_rate: float) -> None: + await self._execute( + "aspirateInPlace", {"pipetteId": pipette_id, "volume": volume, "flowRate": flow_rate} + ) + + async def dispense_in_place(self, pipette_id: str, volume: float, flow_rate: float) -> None: + await self._execute( + "dispenseInPlace", + {"pipetteId": pipette_id, "volume": volume, "flowRate": flow_rate, "pushOut": 0.0}, + ) + + async def discard_tip_in_fixed_trash(self, pipette_id: str, offset: Coordinate) -> None: + """Drop in OT-2 fixed trash using the command form supported by this run's server.""" + if _version_at_least(self.software_version, "7.1.0"): + await self._execute( + "moveToAddressableAreaForDropTip", + { + "pipetteId": pipette_id, + "addressableAreaName": "fixedTrash", + "offset": _coordinates(offset), + "alternateDropLocation": False, + }, + ) + await self._execute("dropTipInPlace", {"pipetteId": pipette_id}) + else: + await self.drop_tip(pipette_id, "fixedTrash", "A1", offset) diff --git a/pylabrobot/opentrons/run_tests.py b/pylabrobot/opentrons/run_tests.py new file mode 100644 index 00000000000..0cc906b56eb --- /dev/null +++ b/pylabrobot/opentrons/run_tests.py @@ -0,0 +1,170 @@ +import unittest +from unittest.mock import AsyncMock, call, patch + +from pylabrobot.io.http import HTTP, HTTPError +from pylabrobot.opentrons.api import OpentronsAPI +from pylabrobot.opentrons.errors import ( + OpentronsCommandError, + OpentronsCommandTimeout, + OpentronsError, + OpentronsProtocolError, +) +from pylabrobot.opentrons.run import OpentronsRun, _version_at_least +from pylabrobot.resources import Coordinate + + +class OpentronsRunTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.io = AsyncMock(spec=HTTP) + self.api = OpentronsAPI(self.io) + self.protocol_run = OpentronsRun(self.api, "run", "8.7.0", command_poll_interval=0) + + async def test_named_command_waits_for_success_and_returns_only_its_result(self) -> None: + self.io.request.side_effect = [ + {"data": {"id": "load"}}, + {"data": {"status": "queued"}}, + {"data": {"status": "running"}}, + {"data": {"status": "succeeded", "result": {"pipetteId": "loaded-pipette"}}}, + ] + self.assertEqual( + await self.protocol_run.load_pipette("p20_single_gen2", "left"), "loaded-pipette" + ) + self.assertEqual( + self.io.request.await_args_list, + [ + call( + "POST", + "/runs/run/commands", + { + "data": { + "commandType": "loadPipette", + "params": {"pipetteName": "p20_single_gen2", "mount": "left"}, + "intent": "setup", + } + }, + ), + ] + + [call("GET", "/runs/run/commands/load")] * 3, + ) + + async def test_command_failure_preserves_server_identifiers(self) -> None: + self.io.request.side_effect = [ + {"data": {"id": "aspirate"}}, + {"data": {"status": "failed", "error": {"errorType": "hardware", "detail": "Motor stalled"}}}, + ] + with self.assertRaises(OpentronsCommandError) as raised: + await self.protocol_run.aspirate_in_place("pipette", 10, 3) + error = raised.exception + self.assertEqual( + (error.run_id, error.command_id, error.command_type), ("run", "aspirate", "aspirateInPlace") + ) + self.assertEqual((error.error_type, error.detail), ("hardware", "Motor stalled")) + + async def test_timeout_keeps_command_id_and_never_resubmits(self) -> None: + self.io.request.side_effect = [ + {"data": {"id": "slow-command"}}, + {"data": {"status": "running"}}, + ] + with patch("pylabrobot.opentrons.run.time") as clock: + clock.monotonic.side_effect = [0, 31] + with self.assertRaises(OpentronsCommandTimeout) as raised: + await self.protocol_run.aspirate_in_place("pipette", 10, 3) + self.assertEqual(raised.exception.run_id, "run") + self.assertEqual(raised.exception.command_id, "slow-command") + self.assertEqual(self.io.request.await_count, 2) + + async def test_unexpected_status_and_missing_position_are_not_success(self) -> None: + for response in ( + {"status": "cancelled"}, + {"status": "succeeded", "result": {}}, + {"status": "succeeded", "result": {"position": {"x": 1, "y": 2, "z": None}}}, + ): + with self.subTest(response=response): + self.io.request.side_effect = [{"data": {"id": "position"}}, {"data": response}] + with self.assertRaises(OpentronsProtocolError): + await self.protocol_run.get_position("pipette") + + async def test_poll_transport_timeout_preserves_the_submitted_command_id(self) -> None: + failure = TimeoutError("HTTP response timed out") + self.io.request.side_effect = [{"data": {"id": "pending"}}, failure] + with self.assertRaises(OpentronsCommandTimeout) as raised: + await self.protocol_run.aspirate_in_place("pipette", 10, 3) + self.assertEqual(raised.exception.command_id, "pending") + self.assertIs(raised.exception.__cause__, failure) + self.assertEqual(self.io.request.await_count, 2) + + async def test_position_read_does_not_move_or_cache_the_position(self) -> None: + self.io.request.side_effect = [ + {"data": {"id": "position"}}, + { + "data": { + "status": "succeeded", + "result": {"positionId": "saved", "position": {"x": 1, "y": 2, "z": 120}}, + } + }, + ] + state = vars(self.protocol_run).copy() + self.assertEqual(await self.protocol_run.get_position("pipette"), Coordinate(1, 2, 120)) + self.assertEqual(vars(self.protocol_run), state) + self.assertEqual(self.io.request.await_count, 2) + + async def test_stopped_run_cannot_issue_commands_and_failed_stop_can_be_retried(self) -> None: + self.io.request.side_effect = OSError("connection lost") + with self.assertRaises(OpentronsError): + await self.protocol_run.stop() + self.assertTrue(self.protocol_run.active) + self.io.request.side_effect = [ + {}, + {"data": {"id": "run", "status": "stop-requested"}}, + {"data": {"id": "run", "status": "stopped"}}, + ] + self.io.request.reset_mock() + await self.protocol_run.stop() + self.assertFalse(self.protocol_run.active) + self.assertEqual( + self.io.request.await_args_list, + [ + call("POST", "/runs/run/actions", {"data": {"actionType": "stop"}}), + call("GET", "/runs/run"), + call("GET", "/runs/run"), + ], + ) + self.io.request.reset_mock() + await self.protocol_run.stop() + with self.assertRaisesRegex(RuntimeError, "has stopped"): + await self.protocol_run.move_to("pipette", Coordinate(1, 2, 3)) + self.io.request.assert_not_awaited() + + async def test_stop_timeout_retains_the_run_until_a_retry_confirms_shutdown(self) -> None: + self.io.request.side_effect = [ + {}, + {"data": {"id": "run", "status": "stop-requested"}}, + ] + with patch("pylabrobot.opentrons.run.time") as clock: + clock.monotonic.side_effect = [0, 31] + with self.assertRaisesRegex(OpentronsError, "Timed out waiting for run run to stop"): + await self.protocol_run.stop() + self.assertTrue(self.protocol_run.active) + self.io.request.side_effect = [{}, {"data": {"id": "run", "status": "stopped"}}] + await self.protocol_run.stop() + self.assertFalse(self.protocol_run.active) + + async def test_stop_accepts_a_run_removed_by_a_legacy_endpoint(self) -> None: + self.io.request.side_effect = [{}, HTTPError("GET", "/runs/run", 404, "not found")] + await self.protocol_run.stop() + self.assertFalse(self.protocol_run.active) + + async def test_stop_query_failure_keeps_the_run_active(self) -> None: + self.io.request.side_effect = [{}, HTTPError("GET", "/runs/run", 500, "server error")] + with self.assertRaises(HTTPError): + await self.protocol_run.stop() + self.assertTrue(self.protocol_run.active) + + +class OpentronsVersionTests(unittest.TestCase): + def test_versions_are_compared_numerically(self) -> None: + for version in ("7.10.0", "10.0.0", "7.1", "7.1.0-beta"): + self.assertTrue(_version_at_least(version, "7.1.0")) + self.assertFalse(_version_at_least("7.0.9", "7.1.0")) + with self.assertRaises(ValueError): + _version_at_least("unknown", "7.1.0") diff --git a/pylabrobot/opentrons/types.py b/pylabrobot/opentrons/types.py new file mode 100644 index 00000000000..86c8d842c46 --- /dev/null +++ b/pylabrobot/opentrons/types.py @@ -0,0 +1,126 @@ +"""Values exchanged with the Opentrons robot-server API.""" + +from dataclasses import dataclass +from typing import Any, Dict, Literal, Optional + +from pylabrobot.opentrons.errors import OpentronsProtocolError + +Mount = Literal["left", "right"] + + +def _object(value: Any) -> Dict[str, Any]: + if not isinstance(value, dict): + raise OpentronsProtocolError(f"Expected a JSON object, got {value!r}") + return value + + +def _string(data: Dict[str, Any], key: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value: + raise OpentronsProtocolError(f"Expected a non-empty string for {key!r}, got {value!r}") + return value + + +def _optional_string(data: Dict[str, Any], key: str) -> Optional[str]: + if data.get(key) is None: + return None + return _string(data, key) + + +@dataclass(frozen=True) +class RobotInfo: + """A health reading. Software version is distinct from the HTTP API version header.""" + + name: str + model: str + software_version: str + firmware_version: Optional[str] = None + serial_number: Optional[str] = None + + @classmethod + def from_response(cls, data: Dict[str, Any]) -> "RobotInfo": + return cls( + name=_string(data, "name"), + model=_string(data, "robot_model"), + software_version=_string(data, "api_version"), + firmware_version=_optional_string(data, "fw_version"), + serial_number=_optional_string(data, "robot_serial"), + ) + + +@dataclass(frozen=True) +class MountedPipette: + """A physical pipette, independent of any run's loaded pipette ID.""" + + mount: Mount + name: str + model: Optional[str] = None + serial_number: Optional[str] = None + + +@dataclass(frozen=True) +class ModuleInfo: + """The identity of a connected module.""" + + id: str + module_type: Optional[str] = None + model: Optional[str] = None + serial_number: Optional[str] = None + + @classmethod + def from_response(cls, data: Dict[str, Any]) -> "ModuleInfo": + return cls( + id=_string(data, "id"), + module_type=_optional_string(data, "moduleType"), + model=_optional_string(data, "moduleModel"), + serial_number=_optional_string(data, "serialNumber"), + ) + + +@dataclass(frozen=True) +class RunInfo: + """A run's identity and reported execution status.""" + + id: str + status: Optional[str] = None + + @classmethod + def from_response(cls, data: Dict[str, Any]) -> "RunInfo": + return cls(id=_string(data, "id"), status=_optional_string(data, "status")) + + +@dataclass(frozen=True) +class CommandInfo: + """One command status reading, with an unpacked result or error.""" + + status: str + result: Dict[str, Any] + error: Dict[str, Any] + + @classmethod + def from_response(cls, data: Dict[str, Any]) -> "CommandInfo": + return cls( + status=_string(data, "status"), + result=_object(data["result"]) if data.get("result") is not None else {}, + error=_object(data["error"]) if data.get("error") is not None else {}, + ) + + +@dataclass(frozen=True) +class LabwareIdentity: + """An Opentrons labware definition's namespace, load name, and revision.""" + + namespace: str + load_name: str + version: int + + @classmethod + def from_uri(cls, uri: str) -> "LabwareIdentity": + try: + namespace, load_name, version = uri.split("/") + revision = int(version) + if not namespace or not load_name or revision < 1: + raise ValueError + except ValueError as error: + raise OpentronsProtocolError(f"Invalid labware definition URI {uri!r}") from error + return cls(namespace, load_name, revision)