diff --git a/docs/api/pylabrobot.opentrons.rst b/docs/api/pylabrobot.opentrons.rst new file mode 100644 index 00000000000..e7e6ace9b04 --- /dev/null +++ b/docs/api/pylabrobot.opentrons.rst @@ -0,0 +1,15 @@ +.. currentmodule:: pylabrobot.opentrons + +pylabrobot.opentrons package +============================= + +.. currentmodule:: pylabrobot.opentrons.ot2 + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + OpentronsOT2 + OT2Pipette + OpentronsOT2Error diff --git a/docs/api/pylabrobot.rst b/docs/api/pylabrobot.rst index 206ff03dd1e..8251ebc05b6 100644 --- a/docs/api/pylabrobot.rst +++ b/docs/api/pylabrobot.rst @@ -33,6 +33,7 @@ Manufacturers pylabrobot.kbiosystems pylabrobot.mettler_toledo pylabrobot.molecular_devices + pylabrobot.opentrons pylabrobot.qinstruments pylabrobot.sartorius pylabrobot.thermo_fisher diff --git a/docs/user_guide/index.md b/docs/user_guide/index.md index c058aa8dca0..b9c257c5b08 100644 --- a/docs/user_guide/index.md +++ b/docs/user_guide/index.md @@ -41,6 +41,7 @@ kbioscience/index kbiosystems/index mettler_toledo/index molecular_devices/index +opentrons/index qinstruments/index sartorius/index thermo_fisher/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..4f0b3f8b19a --- /dev/null +++ b/docs/user_guide/opentrons/ot2/hello-world.ipynb @@ -0,0 +1,354 @@ +{ + "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", + "```{warning}\n", + "This new-architecture driver has NOT been tested against hardware in PyLabRobot. `setup()` logs a warning to that effect. Keep clear of the deck whenever the robot can move. If you verify it on your OT-2, please open a PR to remove the warning.\n", + "```\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 OpentronsOT2\n", + "from pylabrobot.resources import OTDeck\n", + "\n", + "deck = OTDeck()\n", + "ot2 = OpentronsOT2(host=\"ot2.local\", deck=deck)" + ] + }, + { + "cell_type": "markdown", + "id": "ot2-setup-md", + "metadata": {}, + "source": [ + "## Connect\n", + "\n", + "`setup()` creates an Opentrons run, discovers the mounted pipettes, reads the robot API version, and homes the robot." + ] + }, + { + "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.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 reported HTTP API version." + ] + }, + { + "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 5a21e5f51aa..051568658c1 100644 --- a/pylabrobot/io/__init__.py +++ b/pylabrobot/io/__init__.py @@ -1,4 +1,5 @@ from .capture import start_capture, stop_capture +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..645de782983 --- /dev/null +++ b/pylabrobot/opentrons/__init__.py @@ -0,0 +1 @@ +from .ot2 import OT2Pipette, OpentronsOT2, OpentronsOT2Error diff --git a/pylabrobot/opentrons/ot2/__init__.py b/pylabrobot/opentrons/ot2/__init__.py new file mode 100644 index 00000000000..645de782983 --- /dev/null +++ b/pylabrobot/opentrons/ot2/__init__.py @@ -0,0 +1 @@ +from .ot2 import OT2Pipette, OpentronsOT2, OpentronsOT2Error diff --git a/pylabrobot/opentrons/ot2/ot2.py b/pylabrobot/opentrons/ot2/ot2.py new file mode 100644 index 00000000000..c2298a07b74 --- /dev/null +++ b/pylabrobot/opentrons/ot2/ot2.py @@ -0,0 +1,866 @@ +from __future__ import annotations + +import asyncio +import logging +import math +import re +import time +import uuid +from dataclasses import dataclass +from typing import Any, Dict, List, Literal, Optional, Tuple, cast + +from pylabrobot import utils +from pylabrobot.io.http import HTTP +from pylabrobot.resources.container import Container +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, TipSpot +from pylabrobot.resources.tip_tracker import does_tip_tracking +from pylabrobot.resources.volume_tracker import does_volume_tracking + +logger = logging.getLogger(__name__) + +Mount = Literal["left", "right"] + +_OT_DECK_IS_ADDRESSABLE_AREA_VERSION = "7.1.0" + + +class OpentronsOT2Error(RuntimeError): + """An error reported by the OT-2 HTTP API.""" + + +@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}, +} + +_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 _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())) + return tuple(parts) + + +def _version_at_least(version: str, required: str) -> bool: + actual = _version_tuple(version) + minimum = _version_tuple(required) + width = max(len(actual), len(minimum)) + return actual + (0,) * (width - len(actual)) >= minimum + (0,) * (width - len(minimum)) + + +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") + + +class OT2Pipette: + """A pipette mounted on an OT-2 carriage. + + Instances are discovered and created by :meth:`OpentronsOT2.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: OpentronsOT2, + 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.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 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_single_channel(self) -> None: + if self.channels != 1: + raise NotImplementedError( + f"{self.name} has {self.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") + + params: Dict[str, Any] = { + "pipetteId": self.pipette_id, + "coordinates": {"x": location.x, "y": location.y, "z": location.z}, + "forceDirect": force_direct, + } + if minimum_z_height is not None: + params["minimumZHeight"] = minimum_z_height + if speed is not None: + params["speed"] = speed + await self.robot._enqueue_command("moveToCoordinates", params) + + async def move_to( + self, + location: Coordinate, + speed: Optional[float] = None, + minimum_z_height: Optional[float] = None, + force_direct: bool = False, + ) -> None: + """Move the pipette's nozzle or mounted tip to an absolute robot-frame coordinate.""" + async with self.robot._operation_lock: + await self._move_to( + location=location, + speed=speed, + minimum_z_height=minimum_z_height, + force_direct=force_direct, + ) + + async def pick_up_tip( + self, + tip_spot: TipSpot, + offset: Optional[Coordinate] = None, + ) -> None: + """Pick up one tip from a tip rack.""" + 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: + async with self.robot._operation_lock: + await self.robot._assign_tip_rack(tip_spot.parent, tip) + await self.robot._enqueue_command( + "pickUpTip", + { + "labwareId": self.robot._ot_name(tip_spot.parent.name), + "wellName": self.robot._well_name(tip_spot), + "wellLocation": { + "origin": "bottom", + "offset": { + "x": offset.x, + "y": offset.y, + "z": offset.z + tip.total_tip_length, + }, + }, + "pipetteId": self.pipette_id, + }, + ) + except Exception: + if tracked: + tip_spot.tracker.rollback() + raise + + if tracked: + tip_spot.tracker.commit() + self._tip = tip + self._tip_origin = tip_spot + + async def drop_tip( + self, + tip_spot: TipSpot, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Drop the mounted tip into a tip-rack position.""" + 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: + async with self.robot._operation_lock: + await self.robot._assign_tip_rack(tip_spot.parent, tip) + await self.robot._enqueue_command( + "dropTip", + { + "labwareId": self.robot._ot_name(tip_spot.parent.name), + "wellName": self.robot._well_name(tip_spot), + "wellLocation": { + "origin": "bottom", + "offset": {"x": offset.x, "y": offset.y, "z": offset.z + 10}, + }, + "pipetteId": self.pipette_id, + }, + ) + except Exception: + if tracked: + tip_spot.tracker.rollback() + raise + + if tracked: + tip_spot.tracker.commit() + self._tip = None + self._tip_origin = None + + async def return_tip( + self, + offset: Optional[Coordinate] = None, + allow_nonzero_volume: bool = False, + ) -> None: + """Return the mounted tip to the position it came from.""" + 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 the mounted tip into the OT-2's fixed trash.""" + 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) + + async with self.robot._operation_lock: + if self.robot.api_version is None: + raise RuntimeError("OT-2 API version is unavailable; call setup() first") + if _version_at_least( + self.robot.api_version, + _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + ): + await self.robot._enqueue_command( + "moveToAddressableAreaForDropTip", + { + "pipetteId": self.pipette_id, + "addressableAreaName": "fixedTrash", + "offset": {"x": offset.x, "y": offset.y, "z": offset.z + 10}, + "alternateDropLocation": False, + }, + ) + await self.robot._enqueue_command( + "dropTipInPlace", + {"pipetteId": self.pipette_id}, + ) + else: + await self.robot._enqueue_command( + "dropTip", + { + "labwareId": "fixedTrash", + "wellName": "A1", + "wellLocation": { + "origin": "bottom", + "offset": {"x": offset.x, "y": offset.y, "z": offset.z + 10}, + }, + "pipetteId": self.pipette_id, + }, + ) + + self._tip = None + self._tip_origin = None + + 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_in_place(self, volume: float, flow_rate: float) -> None: + await self.robot._enqueue_command( + "aspirateInPlace", + {"flowRate": flow_rate, "volume": volume, "pipetteId": self.pipette_id}, + ) + + async def _dispense_in_place(self, volume: float, flow_rate: float) -> None: + await self.robot._enqueue_command( + "dispenseInPlace", + { + "flowRate": flow_rate, + "volume": volume, + "pipetteId": self.pipette_id, + "pushOut": 0.0, + }, + ) + + 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.""" + 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) + + tracked = does_volume_tracking() + if tracked: + if not container.tracker.is_disabled: + container.tracker.remove_liquid(volume) + tip.tracker.add_liquid(volume) + + try: + async with self.robot._operation_lock: + await self._move_to( + location, + minimum_z_height=self.robot.traversal_height, + ) + await self._aspirate_in_place(volume, flow_rate) + await self._move_to( + Coordinate(location.x, location.y, self.robot.traversal_height), + minimum_z_height=self.robot.traversal_height, + ) + except Exception: + if tracked: + if not container.tracker.is_disabled: + container.tracker.rollback() + tip.tracker.rollback() + raise + + if tracked: + if not container.tracker.is_disabled: + container.tracker.commit() + tip.tracker.commit() + + 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.""" + 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) + + tracked = does_volume_tracking() + if tracked: + tip.tracker.remove_liquid(volume) + if not container.tracker.is_disabled: + container.tracker.add_liquid(volume) + + try: + async with self.robot._operation_lock: + await self._move_to( + location, + minimum_z_height=self.robot.traversal_height, + ) + await self._dispense_in_place(volume, flow_rate) + await self._move_to( + Coordinate(location.x, location.y, self.robot.traversal_height), + minimum_z_height=self.robot.traversal_height, + ) + except Exception: + if tracked: + tip.tracker.rollback() + if not container.tracker.is_disabled: + container.tracker.rollback() + raise + + if tracked: + tip.tracker.commit() + if not container.tracker.is_disabled: + container.tracker.commit() + + 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 client-side aspiration and dispense cycles.""" + self._require_single_channel() + 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) + async with self.robot._operation_lock: + await self._move_to(location, minimum_z_height=self.robot.traversal_height) + for _ in range(repetitions): + await self._aspirate_in_place(volume, aspiration_flow_rate) + await self._dispense_in_place(volume, dispense_flow_rate) + await self._move_to( + Coordinate(location.x, location.y, self.robot.traversal_height), + minimum_z_height=self.robot.traversal_height, + ) + + +class OpentronsOT2: + """Opentrons OT-2 liquid-handling robot controlled through its HTTP API. + + The OT-2's mounted pipettes are discovered during :meth:`setup` and exposed as + :attr:`left_pipette` and :attr:`right_pipette` objects. + """ + + 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, + ): + 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": "3"}, + timeout=command_timeout, + ) + + self.api_version: Optional[str] = None + self.left_pipette: Optional[OT2Pipette] = None + self.right_pipette: Optional[OT2Pipette] = None + self._run_id: Optional[str] = None + self._tip_racks: Dict[str, int] = {} + self._plr_name_to_ot_name: Dict[str, str] = {} + 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] + + async def setup(self, skip_home: bool = False) -> None: + """Connect, create an OT run, discover pipettes, and optionally home.""" + logger.warning( + "OpentronsOT2 has NOT been tested against hardware in the new PyLabRobot architecture. " + "Please make a PR to remove this message if you have verified it on your hardware." + ) + if self._run_id is not None: + raise RuntimeError("The OT-2 is already set up") + + await self.io.setup() + try: + run = await self.io.request("POST", "/runs") + self._run_id = cast(str, run["data"]["id"]) + mounted = await self.io.request("GET", "/pipettes") + self.left_pipette = await self._load_mounted_pipette("left", mounted) + self.right_pipette = await self._load_mounted_pipette("right", mounted) + health = await self.io.request("GET", "/health") + self.api_version = cast(str, health["api_version"]) + if not skip_home: + await self.home() + except Exception: + await self._cancel_run() + self._clear_run_state() + await self.io.stop() + raise + + async def stop(self) -> None: + """Cancel the active OT run and close the HTTP transport.""" + try: + await self._cancel_run() + finally: + self._clear_run_state() + await self.io.stop() + + def _clear_run_state(self) -> None: + self._run_id = None + self.api_version = None + self.left_pipette = None + self.right_pipette = None + self._tip_racks = {} + self._plr_name_to_ot_name = {} + + async def _cancel_run(self) -> None: + if self._run_id is None: + return + requests = ( + ( + "POST", + f"/runs/{self._run_id}/actions", + {"data": {"actionType": "stop"}}, + ), + ("POST", f"/runs/{self._run_id}/cancel", None), + ("POST", f"/runs/{self._run_id}/actions/cancel", None), + ("DELETE", f"/runs/{self._run_id}", None), + ) + for method, path, data in requests: + try: + await self.io.request(method, path, data) + return + except Exception as error: # noqa: BLE001 - firmware versions expose different routes + logger.debug("OT-2 run cancellation through %s failed: %s", path, error) + logger.warning("Could not cancel OT-2 run %s", self._run_id) + + async def _load_mounted_pipette( + self, + mount: Mount, + mounted: Dict[str, Any], + ) -> Optional[OT2Pipette]: + pipette_name = mounted[mount]["name"] + if pipette_name is None: + return None + if pipette_name not in _PIPETTE_SPECS: + raise ValueError(f"Unsupported OT-2 pipette {pipette_name!r} on the {mount} mount") + result = await self._enqueue_command( + "loadPipette", + {"pipetteName": pipette_name, "mount": mount}, + ) + return OT2Pipette( + robot=self, + mount=mount, + name=cast(str, pipette_name), + pipette_id=cast(str, result["pipetteId"]), + ) + + async def _enqueue_command( + self, + command_type: str, + params: Dict[str, Any], + intent: Literal["setup", "protocol"] = "setup", + ) -> Dict[str, Any]: + if self._run_id is None: + raise RuntimeError("The OT-2 is not set up") + response = await self.io.request( + "POST", + f"/runs/{self._run_id}/commands", + { + "data": { + "commandType": command_type, + "params": params, + "intent": intent, + } + }, + ) + command_id = cast(str, response["data"]["id"]) + deadline = time.monotonic() + self.command_timeout + while True: + response = await self.io.request( + "GET", + f"/runs/{self._run_id}/commands/{command_id}", + ) + data = cast(Dict[str, Any], response["data"]) + status = data["status"] + if status == "succeeded": + return cast(Dict[str, Any], data.get("result", {})) + if status == "failed": + error = cast(Dict[str, Any], data.get("error", {})) + error_type = error.get("errorType", "unknown") + detail = error.get("detail", "no detail returned") + raise OpentronsOT2Error(f"{command_type} failed with {error_type}: {detail}") + if status not in {"queued", "running"}: + raise OpentronsOT2Error(f"{command_type} returned unexpected command status {status!r}") + if time.monotonic() >= deadline: + raise TimeoutError(f"Timed out waiting for OT-2 command {command_type!r}") + await asyncio.sleep(self.command_poll_interval) + + async def home(self) -> None: + """Home the OT-2 gantry and pipette axes.""" + if self._run_id is None: + raise RuntimeError("The OT-2 is not set up") + async with self._operation_lock: + await self.io.request("POST", "/robot/home", {"target": "robot"}) + + async def list_connected_modules(self) -> List[Dict[str, Any]]: + """Return modules connected to the OT-2.""" + if self._run_id is None: + raise RuntimeError("The OT-2 is not set up") + response = await self.io.request("GET", "/modules") + return cast(List[Dict[str, Any]], response["data"]) + + def _ot_name(self, plr_resource_name: str) -> str: + if plr_resource_name not in self._plr_name_to_ot_name: + self._plr_name_to_ot_name[plr_resource_name] = uuid.uuid4().hex + return self._plr_name_to_ot_name[plr_resource_name] + + @staticmethod + def _well_name(tip_spot: TipSpot) -> str: + """Return the rack-local Opentrons well identifier for a tip spot.""" + if not isinstance(tip_spot.parent, TipRack): + raise ValueError("tip_spot must be assigned to a tip rack") + return tip_spot.parent.get_child_identifier(tip_spot) + + async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip) -> None: + if tip_rack.name in self._tip_racks: + return + 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") + + official_load_name = _OFFICIAL_TIP_RACKS.get(tip_rack.model or "") + if official_load_name is not None: + namespace, load_name, version = "opentrons", official_load_name, 1 + else: + tip_spots = tip_rack.get_all_items() + well_names = { + tip_spot.name: tip_rack.get_child_identifier(tip_spot) for tip_spot in tip_spots + } + definition = { + "schemaVersion": 2, + "version": 1, + "namespace": "pylabrobot", + "metadata": { + "displayName": self._ot_name(tip_rack.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": self._ot_name(tip_rack.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": {}, + } + ], + } + response = await self.io.request( + "POST", + f"/runs/{self._run_id}/labware_definitions", + {"data": definition}, + ) + namespace, load_name, version_text = cast( + str, + response["data"]["definitionUri"], + ).split("/") + version = int(version_text) + await self._enqueue_command( + "loadLabware", + { + "location": {"slotName": str(slot)}, + "loadName": load_name, + "namespace": namespace, + "version": version, + "labwareId": self._ot_name(tip_rack.name), + "displayName": self._ot_name(tip_rack.name), + }, + ) + self._tip_racks[tip_rack.name] = slot + + 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..681ec478ef2 --- /dev/null +++ b/pylabrobot/opentrons/ot2/ot2_tests.py @@ -0,0 +1,326 @@ +import unittest +from typing import Any, Dict, List, Optional, Tuple + +from pylabrobot.io.http import HTTP +from pylabrobot.opentrons.ot2.ot2 import OpentronsOT2, OpentronsOT2Error, _version_at_least +from pylabrobot.resources import Coordinate, set_tip_tracking, set_volume_tracking +from pylabrobot.resources.celltreat import celltreat_96_wellplate_350uL_Fb +from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul + + +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.stop_action_supported = True + self.started = False + + 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]: + self.calls.append((method, path, data)) + if method == "POST" and path == "/runs": + return {"data": {"id": "run-id"}} + 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 {"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 RuntimeError("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"} + self.command_results[command_id] = { + "commandType": command["commandType"], + "result": result, + } + 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: + 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 OpentronsOT2Tests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + set_tip_tracking(True) + set_volume_tracking(True) + self.io = FakeHTTP() + self.deck = OTDeck() + self.robot = OpentronsOT2( + 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_id 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.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(), [{"id": "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"), 4) + 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) + + 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(OpentronsOT2Error, "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_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.api_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) + + +class OpentronsOT2MultiChannelTests(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 = OpentronsOT2(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.channels, 8) + + with self.assertRaisesRegex(NotImplementedError, "Multi-channel"): + await robot.left_pipette.pick_up_tip(tips.get_item("A1")) + + await robot.stop() + + +class OpentronsVersionTests(unittest.TestCase): + def test_version_comparison_is_numeric(self) -> None: + self.assertTrue(_version_at_least("7.10.0", "7.1.0")) + self.assertTrue(_version_at_least("10.0.0", "7.1.0")) + self.assertFalse(_version_at_least("7.0.9", "7.1.0")) + + +if __name__ == "__main__": + unittest.main()