diff --git a/docs/api/pylabrobot.opentrons.rst b/docs/api/pylabrobot.opentrons.rst index 63a24cd2be1..7790b4afc02 100644 --- a/docs/api/pylabrobot.opentrons.rst +++ b/docs/api/pylabrobot.opentrons.rst @@ -3,6 +3,19 @@ pylabrobot.opentrons package ============================ +Flex +---- + +.. autosummary:: + :toctree: _autosummary + :nosignatures: + :recursive: + + OpentronsRobot + OpentronsFlex + OpentronsError + PipetteInfo + Temperature Module ------------------ diff --git a/docs/user_guide/opentrons/flex/hello-world.ipynb b/docs/user_guide/opentrons/flex/hello-world.ipynb new file mode 100644 index 00000000000..57377c18d3a --- /dev/null +++ b/docs/user_guide/opentrons/flex/hello-world.ipynb @@ -0,0 +1,229 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "flex-intro", + "metadata": {}, + "source": [ + "# Opentrons Flex\n", + "\n", + "The Opentrons Flex is a liquid handler with up to two independent pipette mounts (1-, 8-, or 96-channel), an optional gripper, and up to four temperature/heater-shaker/thermocycler modules. PyLabRobot talks to it over the robot's built-in HTTP robot-server API — the same server the Opentrons App uses — rather than through the Python Protocol API.\n", + "\n", + "| Property | Value |\n", + "|---|---|\n", + "| [OEM link](https://opentrons.com/flex) | |\n", + "| [HTTP API reference](https://docs.opentrons.com/v2/) | |\n", + "| Communication | HTTP, robot-server API, default port 31950 |\n", + "| PLR class | `OpentronsFlex` |\n", + "| PLR base class | `OpentronsRobot` (shared with a future OT-2 implementation) |\n", + "\n", + "```{warning}\n", + "This driver has NOT been tested against real hardware in PyLabRobot. The command\n", + "payloads follow the Opentrons HTTP robot-server API as documented and observed\n", + "on other Flex integrations. If you verify it on a Flex, please open a PR to\n", + "remove this warning.\n", + "```" + ] + }, + { + "cell_type": "markdown", + "id": "flex-transport-md", + "metadata": {}, + "source": [ + "## How it talks\n", + "\n", + "`OpentronsFlex` (and its shared base, `OpentronsRobot`) speaks HTTP to the robot-server running on the Flex itself, not the Python Protocol API. `setup()`:\n", + "\n", + "1. Opens an `httpx.AsyncClient` and checks `GET /health` to confirm the robot-server (not Jupyter/Python API mode) is reachable.\n", + "2. Creates an empty run with `POST /runs` — an empty run (no `protocolId`) lets PLR send interactive setup commands instead of a pre-baked protocol.\n", + "3. Discovers the mounted pipette via `GET /instruments` and loads it into the run (`loadPipette`).\n", + "4. Homes all axes.\n", + "\n", + "Every subsequent operation (`pickUpTip`, `aspirate`, `dispense`, `loadLabware`, ...) is POSTed as a command to that run (`POST /runs/{run_id}/commands`) and polled via `GET` until it succeeds or fails." + ] + }, + { + "cell_type": "markdown", + "id": "flex-physical-md", + "metadata": {}, + "source": [ + "## Physical setup / cabling\n", + "\n", + "Connect to the Flex over the network:\n", + "\n", + "- **Ethernet (recommended):** connect the Flex's ethernet port to the same network as your computer, or directly via a USB-C-to-ethernet adapter. Find the robot's IP address on its touchscreen under *Settings → Network*.\n", + "- **Wi-Fi:** configure Wi-Fi on the touchscreen; the IP address is shown in the same network settings screen.\n", + "\n", + "The robot-server listens on port `31950` by default. No USB driver or vendor SDK installation is required on the computer side — just `pip install pylabrobot[http]` (or ensure `httpx` is installed)." + ] + }, + { + "cell_type": "markdown", + "id": "flex-setup-md", + "metadata": {}, + "source": [ + "## Setup\n", + "\n", + "Construct `OpentronsFlex` with a `FlexDeck` and the robot's IP address. `setup()` connects, creates a run, and discovers the mounted pipette (`flex.pipette`)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-setup-code", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.opentrons import OpentronsFlex\n", + "from pylabrobot.resources.opentrons import FlexDeck\n", + "\n", + "flex = OpentronsFlex(deck=FlexDeck(), host=\"169.254.99.87\") # replace with your Flex's IP\n", + "await flex.setup()\n", + "\n", + "print(\"Pipette:\", flex.pipette.pipette_name, f\"({flex.pipette.channels} channels)\")" + ] + }, + { + "cell_type": "markdown", + "id": "flex-deck-md", + "metadata": {}, + "source": [ + "## Deck layout\n", + "\n", + "Load a Flex 50 uL tip rack and a Corning 96-well plate onto the deck. Corning labware isn't in the built-in Flex load-name map, so set `ot_load_name` explicitly to the Opentrons labware-library name." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-deck-code", + "metadata": {}, + "outputs": [], + "source": [ + "from pylabrobot.resources.corning import Cor_96_wellplate_360ul_Fb\n", + "from pylabrobot.resources.opentrons import flex_96_tiprack_50ul\n", + "\n", + "tiprack = flex_96_tiprack_50ul(name=\"tips_01\")\n", + "plate = Cor_96_wellplate_360ul_Fb(name=\"plate_01\")\n", + "plate.ot_load_name = \"corning_96_wellplate_360ul_flat\"\n", + "\n", + "flex.deck.assign_child_at_slot(tiprack, slot=\"C1\")\n", + "flex.deck.assign_child_at_slot(plate, slot=\"D1\")" + ] + }, + { + "cell_type": "markdown", + "id": "flex-picktips-md", + "metadata": {}, + "source": [ + "## Picking up tips\n", + "\n", + "`pick_up_tips` takes a list of `TipSpot`s and a matching `use_channels` list — here, channel 0 picks up the tip at well A1 of the rack. The op commits directly to the tip spot's tracker on the resource tree (`tiprack[\"A1\"]` now reports no tip)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-picktips-code", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.pick_up_tips(tiprack[\"A1\"], use_channels=[0])" + ] + }, + { + "cell_type": "markdown", + "id": "flex-aspirate-md", + "metadata": {}, + "source": [ + "## Aspirating\n", + "\n", + "The resource tree's volume tracker needs to know a well has liquid before you can aspirate from it — this driver has no liquid-level probe yet, so set the volume manually. Then aspirate 20 uL from well A1 of the plate on channel 0." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-aspirate-code", + "metadata": {}, + "outputs": [], + "source": [ + "plate[\"A1\"][0].tracker.set_volume(200.0)\n", + "plate[\"A1\"][0].tracker.commit()\n", + "\n", + "await flex.aspirate(plate[\"A1\"], vols=[20.0], use_channels=[0])" + ] + }, + { + "cell_type": "markdown", + "id": "flex-dispense-md", + "metadata": {}, + "source": [ + "## Dispensing\n", + "\n", + "Dispense the 20 uL into well A2 of the same plate, again on channel 0." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-dispense-code", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.dispense(plate[\"A2\"], vols=[20.0], use_channels=[0])" + ] + }, + { + "cell_type": "markdown", + "id": "flex-droptips-md", + "metadata": {}, + "source": [ + "## Dropping tips\n", + "\n", + "Return the tip to its original spot in the rack. Dropping into a `Trash` resource instead (e.g. `flex.deck.get_trash_area()`) routes through the Flex's addressable-area drop-tip commands and does not restore the tracker." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-droptips-code", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.drop_tips(tiprack[\"A1\"], use_channels=[0])" + ] + }, + { + "cell_type": "markdown", + "id": "flex-teardown-md", + "metadata": {}, + "source": [ + "## Teardown" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "flex-teardown-code", + "metadata": {}, + "outputs": [], + "source": [ + "await flex.stop()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "pygments_lexer": "ipython3" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/docs/user_guide/opentrons/index.md b/docs/user_guide/opentrons/index.md index afe03a1f052..97f9243525b 100644 --- a/docs/user_guide/opentrons/index.md +++ b/docs/user_guide/opentrons/index.md @@ -3,5 +3,6 @@ ```{toctree} :maxdepth: 1 +flex/hello-world temperature_module/hello-world ``` diff --git a/pylabrobot/opentrons/__init__.py b/pylabrobot/opentrons/__init__.py index 45ef90311f4..020711db75b 100644 --- a/pylabrobot/opentrons/__init__.py +++ b/pylabrobot/opentrons/__init__.py @@ -1,3 +1,6 @@ +from pylabrobot.opentrons.flex import OpentronsFlex +from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot, PipetteInfo + from .temperature_module import ( OpentronsTemperatureModuleDriver, OpentronsTemperatureModuleTemperatureBackend, diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py new file mode 100644 index 00000000000..7dd625be3ad --- /dev/null +++ b/pylabrobot/opentrons/flex.py @@ -0,0 +1,294 @@ +import logging +import uuid +from typing import Any, Dict, List, Literal, Optional, Sequence, cast + +from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot, PipetteInfo +from pylabrobot.resources import ( + Container, + Coordinate, + Resource, + TipSpot, + Trash, + does_tip_tracking, + does_volume_tracking, +) +from pylabrobot.resources.itemized_resource import ItemizedResource +from pylabrobot.resources.opentrons.flex_deck import FlexDeck +from pylabrobot.resources.tip import Tip + +logger = logging.getLogger(__name__) + +_OT_NAMESPACE = "opentrons" +_OT_VERSION = 1 + +# Flex-managed positioning flow-rate defaults (uL/s). +_DEFAULT_ASPIRATE_FLOW_RATE = 35.0 +_DEFAULT_DISPENSE_FLOW_RATE = 57.0 + +_TIP_RACK_MAP = { + "flex_96_tiprack_50ul": "opentrons_flex_96_tiprack_50ul", + "flex_96_tiprack_200ul": "opentrons_flex_96_tiprack_200ul", + "flex_96_tiprack_1000ul": "opentrons_flex_96_tiprack_1000ul", + "flex_96_tiprack_20ul": "opentrons_flex_96_tiprack_20ul", + "flex_96_filtertiprack_50ul": "opentrons_flex_96_filtertiprack_50ul", + "flex_96_filtertiprack_200ul": "opentrons_flex_96_filtertiprack_200ul", + "flex_96_filtertiprack_1000ul": "opentrons_flex_96_filtertiprack_1000ul", + "flex_96_filtertiprack_20ul": "opentrons_flex_96_filtertiprack_20ul", +} + + +class OpentronsFlex(OpentronsRobot): + """Opentrons Flex liquid handler (plain class, post-#1180 architecture). + + Tip ops commit state to the resource-tree trackers (``TipSpot.tracker``), + not to a private per-device copy. Only which physical channel holds which + tip is genuine device-local state (``self._channel_tips``). + """ + + def __init__(self, deck: FlexDeck, host: str, port: int = 31950) -> None: + super().__init__(host=host, port=port) + self.deck = deck + self._loaded_labware: Dict[str, str] = {} + self._channel_tips: List[Optional[Tip]] = [None] * 8 + + async def _model_setup(self) -> None: + await self.home() + + def _require_pipette(self) -> PipetteInfo: + """Return ``self.pipette``, narrowed to non-``None`` for mypy and callers. + + Raises if ``setup()`` (which discovers and loads the pipette) hasn't run. + """ + assert self.pipette is not None, "No pipette loaded. Call setup() first." + return self.pipette + + @staticmethod + def _require_itemized_parent(item: Resource) -> ItemizedResource: + """Return ``item.parent``, asserted to be an addressable-by-name container.""" + parent = item.parent + assert isinstance(parent, ItemizedResource), ( + f"'{item.name}' has no itemized parent resource (rack/plate)." + ) + return parent + + async def pick_up_tips( + self, + tip_spots: List[TipSpot], + use_channels: Optional[List[int]] = None, + offsets: Optional[List[Coordinate]] = None, + ) -> None: + use_channels = use_channels if use_channels is not None else list(range(len(tip_spots))) + pipette = self._require_pipette() + rack = self._require_itemized_parent(tip_spots[0]) + labware_id = await self._ensure_labware_loaded(rack) + well_name = rack.get_child_identifier(tip_spots[0]) + params: Dict[str, Any] = { + "pipetteId": pipette.pipette_id, + "labwareId": labware_id, + "wellName": well_name, + } + if offsets is not None and offsets[0] is not None: + offset = offsets[0] + params["wellLocation"] = { + "origin": "top", + "offset": {"x": offset.x, "y": offset.y, "z": offset.z}, + } + await self._execute_command("pickUpTip", params) + for ch, spot in zip(use_channels, tip_spots): + tip = spot.get_tip() + if does_tip_tracking(): + spot.tracker.remove_tip() + spot.tracker.commit() + self._channel_tips[ch] = tip + + async def drop_tips( + self, + tip_spots: List[TipSpot], + use_channels: Optional[List[int]] = None, + offsets: Optional[List[Coordinate]] = None, + ) -> None: + use_channels = use_channels if use_channels is not None else list(range(len(tip_spots))) + pipette = self._require_pipette() + target = tip_spots[0] + if isinstance(target, Trash) or isinstance(target.parent, Trash): + await self._execute_command( + "moveToAddressableAreaForDropTip", + { + "pipetteId": pipette.pipette_id, + "addressableAreaName": "movableTrashA3", + "alternateDropLocation": True, + }, + ) + await self._execute_command("dropTipInPlace", {"pipetteId": pipette.pipette_id}) + else: + rack = self._require_itemized_parent(target) + labware_id = await self._ensure_labware_loaded(rack) + well_name = rack.get_child_identifier(target) + params: Dict[str, Any] = { + "pipetteId": pipette.pipette_id, + "labwareId": labware_id, + "wellName": well_name, + } + if offsets is not None and offsets[0] is not None: + offset = offsets[0] + params["wellLocation"] = { + "origin": "top", + "offset": {"x": offset.x, "y": offset.y, "z": offset.z}, + } + await self._execute_command("dropTip", params) + for ch, spot in zip(use_channels, tip_spots): + tip = self._channel_tips[ch] + if ( + does_tip_tracking() + and tip is not None + and not isinstance(spot, Trash) + and not isinstance(spot.parent, Trash) + ): + spot.tracker.add_tip(tip) + spot.tracker.commit() + self._channel_tips[ch] = None + + async def aspirate( + self, + resources: Sequence[Container], + vols: List[float], + use_channels: Optional[List[int]] = None, + flow_rates: Optional[List[Optional[float]]] = None, + offsets: Optional[List[Optional[Coordinate]]] = None, + liquid_height: Optional[List[Optional[float]]] = None, + blow_out_air_volume: Optional[List[Optional[float]]] = None, + spread: Literal["wide", "tight", "custom"] = "wide", + ) -> None: + pipette = self._require_pipette() + parent = self._require_itemized_parent(resources[0]) + labware_id = await self._ensure_labware_loaded(parent) + well_name = parent.get_child_identifier(resources[0]) + flow_rate = (flow_rates[0] if flow_rates else None) or _DEFAULT_ASPIRATE_FLOW_RATE + params: Dict[str, Any] = { + "pipetteId": pipette.pipette_id, + "labwareId": labware_id, + "wellName": well_name, + "volume": vols[0], + "flowRate": flow_rate, + } + well_location = self._well_location(offsets, liquid_height) + if well_location is not None: + params["wellLocation"] = well_location + await self._execute_command("aspirate", params) + if does_volume_tracking(): + for well, vol in zip(resources, vols): + well.tracker.remove_liquid(vol) + well.tracker.commit() + + async def dispense( + self, + resources: Sequence[Container], + vols: List[float], + use_channels: Optional[List[int]] = None, + flow_rates: Optional[List[Optional[float]]] = None, + offsets: Optional[List[Optional[Coordinate]]] = None, + liquid_height: Optional[List[Optional[float]]] = None, + blow_out_air_volume: Optional[List[Optional[float]]] = None, + spread: Literal["wide", "tight", "custom"] = "wide", + ) -> None: + pipette = self._require_pipette() + parent = self._require_itemized_parent(resources[0]) + labware_id = await self._ensure_labware_loaded(parent) + well_name = parent.get_child_identifier(resources[0]) + flow_rate = (flow_rates[0] if flow_rates else None) or _DEFAULT_DISPENSE_FLOW_RATE + params: Dict[str, Any] = { + "pipetteId": pipette.pipette_id, + "labwareId": labware_id, + "wellName": well_name, + "volume": vols[0], + "flowRate": flow_rate, + } + well_location = self._well_location(offsets, liquid_height) + if well_location is not None: + params["wellLocation"] = well_location + await self._execute_command("dispense", params) + if does_volume_tracking(): + for well, vol in zip(resources, vols): + well.tracker.add_liquid(vol) + well.tracker.commit() + + @staticmethod + def _well_location( + offsets: Optional[List[Optional[Coordinate]]], + liquid_height: Optional[List[Optional[float]]], + ) -> Optional[dict]: + """Build the Flex ``wellLocation`` param from an offset and/or liquid height. + + Merges an explicit x/y/z offset with ``liquid_height`` (added to z); the + origin is always ``"bottom"``. Returns ``None`` if neither is given. + """ + offset = None + if offsets is not None and offsets[0] is not None: + o = offsets[0] + offset = {"x": o.x, "y": o.y, "z": o.z} + if liquid_height is not None and liquid_height[0] is not None: + offset = offset or {"x": 0, "y": 0, "z": 0} + offset["z"] += liquid_height[0] + if offset is None: + return None + return {"origin": "bottom", "offset": offset} + + async def _ensure_labware_loaded(self, resource: Resource) -> str: + """Load labware into the Flex run if not already loaded.""" + name = getattr(resource, "name", str(resource)) + if name in self._loaded_labware: + return self._loaded_labware[name] + + slot = self.deck.get_slot(resource) + if slot is None: + raise OpentronsError( + "Resource not on deck", + f"'{name}' is not on a deck slot. Use deck.assign_child_at_slot(resource, slot='C1').", + ) + + load_name = self._ot_load_name(resource) + labware_id = uuid.uuid4().hex[:12] + + result = await self._execute_command( + "loadLabware", + { + "loadName": load_name, + "location": {"slotName": slot}, + "namespace": _OT_NAMESPACE, + "version": _OT_VERSION, + "labwareId": labware_id, + "displayName": name, + }, + ) + labware_id = cast(str, result.get("result", {}).get("labwareId", labware_id)) + + self._loaded_labware[name] = labware_id + logger.info( + "Loaded labware '%s' at slot %s -> ID: %s (OT: %s)", + name, + slot, + labware_id, + load_name, + ) + return labware_id + + @staticmethod + def _ot_load_name(resource: Resource) -> str: + """Resolve a PLR resource to its Opentrons labware load name.""" + if hasattr(resource, "ot_load_name"): + return cast(str, resource.ot_load_name) + + name_lower = getattr(resource, "name", "").lower() + + for key, ot_name in _TIP_RACK_MAP.items(): + if key in name_lower: + return ot_name + + if name_lower.startswith("opentrons_"): + return name_lower + + raise OpentronsError( + "Cannot determine Opentrons load name", + f"'{name_lower}' — set resource.ot_load_name = 'opentrons_flex_96_tiprack_50ul' " + f"or use a standard Flex labware name.", + ) diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py new file mode 100644 index 00000000000..ad0081bfc3e --- /dev/null +++ b/pylabrobot/opentrons/robot.py @@ -0,0 +1,289 @@ +import abc +import asyncio +import logging +import time +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, cast + +try: + import httpx + + _HAS_HTTPX = True +except ImportError: + _HAS_HTTPX = False + +logger = logging.getLogger(__name__) + + +class OpentronsError(Exception): + def __init__(self, title: str, message: Optional[str] = None) -> None: + self.title, self.message = title, message + super().__init__(f"{title}: {message}" if message else title) + + +@dataclass +class PipetteInfo: + mount: str + pipette_name: str + pipette_model: str + pipette_id: str + channels: int + min_volume: float + max_volume: float + + +class OpentronsRobot(abc.ABC): + """Shared base for Opentrons HTTP robots (Flex, OT-2). + + Owns the httpx transport, the run/command protocol, and instrument discovery. + Subclasses implement the liquid-handling ops and any model-specific setup. + + Transport is an httpx AsyncClient held on the instance; PyLabRobot has no + pylabrobot.io HTTP transport yet, so the HTTP client lives here rather than + behind a pylabrobot.io primitive. + """ + + def __init__(self, host: str, port: int = 31950) -> None: + if not _HAS_HTTPX: + raise RuntimeError("httpx is required. Install with: pip install httpx") + self.host, self.port = host, port + self.base_url = f"http://{host}:{port}" + self._client: Optional["httpx.AsyncClient"] = None + self.run_id: Optional[str] = None + self.pipette: Optional[PipetteInfo] = None + self.api_version: Optional[str] = None + self.robot_model: Optional[str] = None + + async def setup(self) -> None: + logger.warning( + "OpentronsRobot has not been verified against real hardware; use with care " + "and report success so this warning can be removed." + ) + await self._connect() + await self._create_run() + self.pipette = await self._discover_pipette() + await self._model_setup() + + async def stop(self) -> None: + await self._cancel_run() + await self._disconnect() + + @abc.abstractmethod + async def _model_setup(self) -> None: + """Model-specific post-connection setup (home, load pipette id, etc.).""" + + # --- Connection Lifecycle --- + + async def _connect(self) -> None: + """Create HTTP session and verify connectivity. + + Sends a health check to confirm the robot is reachable and the robot + server is running (not in Jupyter/Python API mode). + """ + self._client = httpx.AsyncClient( + base_url=self.base_url, + timeout=30.0, + headers={"opentrons-version": "3"}, + ) + health = await self._get("/health") + self.api_version = health.get("api_version") + self.robot_model = health.get("robot_model", "") + robot_name = health.get("name", "unknown") + logger.info( + "Connected to robot '%s' at %s:%s (API %s, model: %s)", + robot_name, + self.host, + self.port, + self.api_version, + self.robot_model, + ) + + async def _disconnect(self) -> None: + """Close the HTTP session.""" + if self._client is not None: + await self._client.aclose() + self._client = None + + # --- Low-Level HTTP --- + + async def _get(self, path: str) -> Dict[str, Any]: + """HTTP GET, return parsed JSON.""" + assert self._client is not None, "Not connected. Call connect() first." + response = await self._client.get(path) + response.raise_for_status() + return cast(Dict[str, Any], response.json()) + + async def _post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """HTTP POST, return parsed JSON.""" + assert self._client is not None, "Not connected. Call connect() first." + response = await self._client.post(path, json=data or {}) + response.raise_for_status() + return cast(Dict[str, Any], response.json()) + + async def _delete(self, path: str) -> Dict[str, Any]: + """HTTP DELETE, return parsed JSON.""" + assert self._client is not None, "Not connected. Call connect() first." + response = await self._client.delete(path) + response.raise_for_status() + return cast(Dict[str, Any], response.json()) + + # --- Run Management --- + + async def _create_run(self) -> str: + """Create a new empty run on the robot. Returns the run ID. + + An empty run (no protocolId) allows sending setup commands + interactively, which is how PLR controls the robot. + """ + result = await self._post("/runs", {"data": {}}) + run_id = cast(str, result["data"]["id"]) + self.run_id = run_id + logger.info("Created run %s", self.run_id) + return run_id + + async def _cancel_run(self) -> None: + """Cancel the current run. Safe to call if no run is active.""" + if self.run_id is None: + return + try: + await self._post( + f"/runs/{self.run_id}/actions", + {"data": {"actionType": "stop"}}, + ) + except Exception: + try: + await self._delete(f"/runs/{self.run_id}") + except Exception: + pass + self.run_id = None + + # --- Command Execution --- + + async def _execute_command( + self, + command_type: str, + params: Dict[str, Any], + wait: bool = True, + timeout: float = 30.0, + ) -> Dict[str, Any]: + """Execute a command within the current run. + + Commands on the robot are asynchronous: the POST returns + immediately with status "queued". If ``wait=True`` (default), + this method polls until the command succeeds or fails. + + Args: + command_type: e.g., "home", "moveToCoordinates", + "aspirateInPlace", "pickUpTip", "loadLabware". + params: Command-specific parameters. + wait: If True, poll until completion. + timeout: Max seconds to wait. + + Returns: + The completed command data dict (includes "result" field). + + Raises: + RuntimeError: If the command fails or times out. + """ + assert self.run_id is not None, "No active run. Call create_run() first." + payload = { + "data": { + "commandType": command_type, + "params": params, + "intent": "setup", + } + } + result = await self._post(f"/runs/{self.run_id}/commands", payload) + cmd_data: Dict[str, Any] = result.get("data", {}) + + if not wait: + return cmd_data + + cmd_id = cmd_data.get("id", "") + if not cmd_id: + return cmd_data + + # Poll for completion + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + resp = await self._get(f"/runs/{self.run_id}/commands/{cmd_id}") + cmd_data = resp.get("data", {}) + status = cmd_data.get("status", "") + if status == "succeeded": + return cmd_data + elif status == "failed": + error = cmd_data.get("error", {}) + raise RuntimeError( + f"Opentrons command '{command_type}' failed: {error.get('detail', error)}" + ) + await asyncio.sleep(0.2) + + raise RuntimeError(f"Opentrons command '{command_type}' timed out after {timeout}s") + + # --- Instrument Discovery --- + + async def _get_instruments(self) -> Dict[str, Any]: + """Query mounted instruments (pipettes, gripper).""" + return await self._get("/instruments") + + def _parse_pipettes(self, instruments_data: Dict[str, Any]) -> List[PipetteInfo]: + """Parse the /instruments response into PipetteInfo objects. + + Uses actual data from the API (channels, min_volume, max_volume) + rather than guessing from pipette names. + """ + pipettes = [] + for instrument in instruments_data.get("data", []): + if instrument.get("instrumentType") != "pipette": + continue + pip_data = instrument.get("data", {}) + pipettes.append( + PipetteInfo( + mount=instrument.get("mount", "unknown"), + pipette_name=instrument.get("instrumentName", "unknown"), + pipette_model=instrument.get("instrumentModel", "unknown"), + pipette_id="", # set by _load_pipette() later + channels=pip_data.get("channels", 1), + min_volume=pip_data.get("min_volume", 1.0), + max_volume=pip_data.get("max_volume", 1000.0), + ) + ) + return pipettes + + # --- Pipette Loading --- + + async def _load_pipette(self, pipette_name: str, mount: str) -> str: + """Load a pipette into the current run. + + Returns the run-scoped pipette ID required by all subsequent + commands (pickUpTip, aspirateInPlace, moveToCoordinates, etc.). + Must be called after _create_run(). + """ + result = await self._execute_command( + "loadPipette", + {"pipetteName": pipette_name, "mount": mount}, + wait=True, + ) + pipette_id: str = result.get("result", {}).get("pipetteId", "") + logger.info( + "Loaded pipette %s on %s mount -> ID: %s", + pipette_name, + mount, + pipette_id, + ) + return pipette_id + + # --- Homing --- + + async def home(self) -> Dict[str, Any]: + """Home all axes. The gantry moves to the rear-left-top.""" + return await self._execute_command("home", {}) + + async def _discover_pipette(self) -> PipetteInfo: + data = await self._get_instruments() + pipettes = self._parse_pipettes(data) + if not pipettes: + raise OpentronsError("No pipette detected", f"{self.host}:{self.port}") + pip = pipettes[0] + pip.pipette_id = await self._load_pipette(pip.pipette_name, pip.mount) + return pip diff --git a/pylabrobot/resources/opentrons/__init__.py b/pylabrobot/resources/opentrons/__init__.py index 38d1dc2e55c..bf2cff7fb50 100644 --- a/pylabrobot/resources/opentrons/__init__.py +++ b/pylabrobot/resources/opentrons/__init__.py @@ -1,4 +1,6 @@ from .deck import OTDeck +from .flex_deck import FlexDeck +from .flex_tip_racks import * from .load import load_ot_tip_rack from .module import OTModule from .ot2_geometry import OT2RobotGeometry diff --git a/pylabrobot/resources/opentrons/flex_deck.py b/pylabrobot/resources/opentrons/flex_deck.py new file mode 100644 index 00000000000..4cd847bc2a7 --- /dev/null +++ b/pylabrobot/resources/opentrons/flex_deck.py @@ -0,0 +1,362 @@ +"""FlexDeck — Opentrons Flex deck with A1–D3 grid layout plus staging area. + +The Flex has 12 standard slots in a 4-row x 3-column grid (rows A–D +from rear to front, columns 1–3 from left to right), plus 4 staging +area slots in column 4. + +Coordinates sourced from Opentrons ot3_standard deck definition v5. +Slot bounding box: 128.0 x 86.0 mm. + +Provides collision detection for single-nozzle tip pickup: when +the 8-channel pipette uses only 1 nozzle, the other 7 extend into +the adjacent slot's airspace and could hit tall labware. +""" + +from __future__ import annotations + +import re +from typing import Dict, Optional + +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.deck import Deck +from pylabrobot.resources.resource import Resource +from pylabrobot.resources.resource_holder import ResourceHolder +from pylabrobot.resources.trash import Trash + +# OT-2 slot number → Flex slot identifier mapping +_OT2_TO_FLEX = { + 1: "D1", + 2: "D2", + 3: "D3", + 4: "C1", + 5: "C2", + 6: "C3", + 7: "B1", + 8: "B2", + 9: "B3", + 10: "A1", + 11: "A2", + 12: "A3", +} + +# Valid slot pattern: A-D followed by 1-4 +_SLOT_PATTERN = re.compile(r"^[A-D][1-4]$") + +# Row ordering from front (D, index 0) to rear (A, index 3) +_ROW_ORDER = ["D", "C", "B", "A"] + +# Slot coordinates (mm) from ot3_standard.json v5 cutout positions. +# Origin is front-left corner of slot D1. +SLOT_LOCATIONS: Dict[str, Dict[str, float]] = { + "D1": {"x": 0.0, "y": 0.0, "z": 0.0}, + "D2": {"x": 164.0, "y": 0.0, "z": 0.0}, + "D3": {"x": 328.0, "y": 0.0, "z": 0.0}, + "C1": {"x": 0.0, "y": 107.0, "z": 0.0}, + "C2": {"x": 164.0, "y": 107.0, "z": 0.0}, + "C3": {"x": 328.0, "y": 107.0, "z": 0.0}, + "B1": {"x": 0.0, "y": 214.0, "z": 0.0}, + "B2": {"x": 164.0, "y": 214.0, "z": 0.0}, + "B3": {"x": 328.0, "y": 214.0, "z": 0.0}, + "A1": {"x": 0.0, "y": 321.0, "z": 0.0}, + "A2": {"x": 164.0, "y": 321.0, "z": 0.0}, + "A3": {"x": 328.0, "y": 321.0, "z": 0.0}, +} + +# Staging area coordinates (column 4). +STAGING_LOCATIONS: Dict[str, Dict[str, float]] = { + "D4": {"x": 492.0, "y": 0.0, "z": 14.5}, + "C4": {"x": 492.0, "y": 107.0, "z": 14.5}, + "B4": {"x": 492.0, "y": 214.0, "z": 14.5}, + "A4": {"x": 492.0, "y": 321.0, "z": 14.5}, +} + +# Slot bounding box (mm) +SLOT_WIDTH = 128.0 # x dimension +SLOT_DEPTH = 86.0 # y dimension + +# Overall deck footprint (mm), including the frame around the slot grid. +_DECK_SIZE_X = 855.0 +_DECK_SIZE_Y = 582.0 + +# Default clearance Z when no operation height is provided. +# Conservative estimate — measured at 51.3mm on real hardware +# (tip at bottom of flat well plate, A1 nozzle to deck surface). +_DEFAULT_CLEARANCE_Z = 50.0 + + +class FlexDeck(Deck): + """Opentrons Flex deck — 16 slots with placement and collision detection. + + Each slot (the 12 standard A1–D3 slots plus the 4 staging slots A4–D4) + is modeled as a :class:`ResourceHolder` child assigned into this deck's + resource tree, so labware placed at a slot is properly parented — + ``resource.parent`` walks up through the slot holder to this deck, and + ``get_absolute_location()`` works through the standard PLR mechanism. + Labware is placed into a slot's holder with :meth:`assign_child_at_slot`. + + Example:: + + deck = FlexDeck() + deck.assign_child_at_slot(tip_rack, slot="C1") + print(deck.summary()) + deck.check_single_nozzle_clearance("C3", primary_nozzle="H1") + """ + + def __init__( + self, + with_trash_bin: bool = True, + name: str = "flex_deck", + ) -> None: + super().__init__(size_x=_DECK_SIZE_X, size_y=_DECK_SIZE_Y, size_z=0.0, name=name) + + self._slot_holders: Dict[str, ResourceHolder] = {} + for slot_id, loc in {**SLOT_LOCATIONS, **STAGING_LOCATIONS}.items(): + holder = ResourceHolder( + name=f"{self.name}_slot_{slot_id}", + size_x=SLOT_WIDTH, + size_y=SLOT_DEPTH, + size_z=0, + ) + self._slot_holders[slot_id] = holder + super().assign_child_resource(holder, location=Coordinate(x=loc["x"], y=loc["y"], z=loc["z"])) + + if with_trash_bin: + trash = Trash(name="trash", size_x=SLOT_WIDTH, size_y=SLOT_DEPTH, size_z=82.0) + self.assign_child_at_slot(trash, "A3") + + # --- Slot Validation --- + + @staticmethod + def _validate_slot(slot: str) -> str: + """Validate and normalize a slot identifier. Returns uppercase slot.""" + slot = slot.upper() + if _SLOT_PATTERN.match(slot): + return slot + + # Check if user passed an OT-2 integer slot + try: + ot2_slot = int(slot) + if 1 <= ot2_slot <= 12: + flex_slot = _OT2_TO_FLEX[ot2_slot] + raise ValueError( + f"'{slot}' looks like an OT-2 slot number. " + f"The Flex uses letter-number identifiers: " + f"slot {ot2_slot} on OT-2 is '{flex_slot}' on the Flex. " + f"Use deck.assign_child_at_slot(resource, slot='{flex_slot}')." + ) + except ValueError as e: + if "OT-2" in str(e): + raise + + raise ValueError( + f"Invalid slot identifier '{slot}'. " + f"Must be A1–D3 (standard) or A4–D4 (staging). " + f"Examples: 'C1', 'A3', 'B4'." + ) + + # --- Slot Access --- + + def get_slot_location(self, slot: str) -> Dict[str, float]: + """Get the XYZ coordinate for a slot.""" + slot = self._validate_slot(slot) + if slot in SLOT_LOCATIONS: + return SLOT_LOCATIONS[slot] + if slot in STAGING_LOCATIONS: + return STAGING_LOCATIONS[slot] + raise ValueError(f"Unknown slot '{slot}'.") + + def assign_child_at_slot(self, resource: Resource, slot: str) -> None: + """Place a resource at a named slot. + + Args: + resource: The resource (tip rack, plate, etc.) to place. + slot: Slot identifier, e.g., "C1", "A4". + + Raises: + ValueError: If slot is invalid or already occupied. + """ + slot = self._validate_slot(slot) + holder = self._slot_holders[slot] + if holder.resource is not None: + name = getattr(holder.resource, "name", str(holder.resource)) + raise ValueError(f"Slot {slot} is already occupied by '{name}'.") + holder.assign_child_resource(resource) + + def unassign_child_at_slot(self, slot: str) -> None: + """Remove a resource from a slot.""" + slot = self._validate_slot(slot) + holder = self._slot_holders[slot] + if holder.resource is not None: + holder.unassign_child_resource(holder.resource) + + def get_slot(self, resource: Resource) -> Optional[str]: + """Get the slot identifier for a placed resource, or None.""" + for slot_id, holder in self._slot_holders.items(): + if holder.resource is resource: + return slot_id + return None + + def get_resource_at_slot(self, slot: str) -> Optional[Resource]: + """Return the resource placed at a slot, or None.""" + slot = self._validate_slot(slot) + return self._slot_holders[slot].resource + + def get_trash_area(self) -> Trash: + """Return the trash resource (default at A3).""" + for holder in self._slot_holders.values(): + if isinstance(holder.resource, Trash): + return holder.resource + raise ValueError("No trash area configured on this deck.") + + # --- OT-2 Conversion --- + + @staticmethod + def ot2_slot_to_flex(ot2_slot: int) -> str: + """Convert an OT-2 slot number to the Flex equivalent. + + Useful for migrating protocols. E.g., 5 → "C2". + """ + if ot2_slot not in _OT2_TO_FLEX: + mapping = ", ".join(f"{k}→{v}" for k, v in sorted(_OT2_TO_FLEX.items())) + raise ValueError(f"OT-2 slot must be 1–12, got {ot2_slot}. Full mapping: {mapping}") + return _OT2_TO_FLEX[ot2_slot] + + # --- Collision Detection --- + + def check_single_nozzle_clearance( + self, + slot: str, + primary_nozzle: str = "H1", + operation_z: Optional[float] = None, + ) -> None: + """Check that adjacent slots are clear for single-nozzle operations. + + When an 8-channel pipette uses a single nozzle, the 7 inactive + nozzles extend ~63mm into the adjacent slot's airspace. Two rules: + + 1. TipRack in adjacent slot → always blocked (inactive nozzles + would physically engage tips). + 2. Other labware → blocked if taller than the operation Z + (the height the nozzle descends to). + + Args: + slot: Deck slot where the operation happens. + primary_nozzle: "H1" (front) or "A1" (rear). + operation_z: The Z height the nozzle descends to (mm). + If None, uses the default conservative threshold. + + Raises: + ValueError: If a collision risk is detected. + """ + from pylabrobot.resources.tip_rack import TipRack + + slot = self._validate_slot(slot) + row = slot[0] + col = slot[1] + row_idx = _ROW_ORDER.index(row) + + if primary_nozzle == "H1": + # Front nozzle → inactive extend toward rear + if row_idx + 1 < len(_ROW_ORDER): + danger_slot = f"{_ROW_ORDER[row_idx + 1]}{col}" + else: + return # Rearmost row (A), nothing behind + elif primary_nozzle == "A1": + # Rear nozzle → inactive extend toward front + if row_idx - 1 >= 0: + danger_slot = f"{_ROW_ORDER[row_idx - 1]}{col}" + else: + return # Frontmost row (D), nothing in front + else: + return # Other nozzle configs — skip for now + + resource = self._slot_holders[danger_slot].resource + if resource is None: + return # Slot empty, safe + + direction = "behind" if primary_nozzle == "H1" else "in front of" + name = getattr(resource, "name", str(resource)) + + # Rule 1: TipRack always blocked — nozzles would grab tips + if isinstance(resource, TipRack): + raise ValueError( + f"Collision risk: single-nozzle operation at {slot} " + f"with nozzle {primary_nozzle} — the 7 inactive nozzles " + f"extend into slot {danger_slot}, which contains tip rack " + f"'{name}'. Inactive nozzles would engage tips. " + f"Move the tip rack or use a different nozzle direction." + ) + + # Rule 2: Other labware — check against operation Z + if hasattr(resource, "get_size_z"): + resource_z = resource.get_size_z() + else: + resource_z = getattr(resource, "_size_z", 0) or getattr(resource, "size_z", 0) + + clearance_z = operation_z if operation_z is not None else _DEFAULT_CLEARANCE_Z + + if resource_z > clearance_z: + raise ValueError( + f"Collision risk: single-nozzle operation at {slot} " + f"with nozzle {primary_nozzle} — the 7 inactive nozzles " + f"extend into slot {danger_slot} at Z={clearance_z:.0f}mm, " + f"which contains '{name}' (height {resource_z:.0f}mm). " + f"Move '{name}' to a different slot, or use a slot with " + f"no tall labware {direction} it." + ) + + def check_deck_clearance(self, slot: str, operation: str = "move") -> None: + """Verify a slot has labware for an operation that requires it.""" + slot = self._validate_slot(slot) + resource = self._slot_holders[slot].resource + if resource is None and operation in ("pick_up_tips", "aspirate", "dispense"): + raise ValueError( + f"Cannot {operation} at slot {slot}: no labware assigned. " + f"Use deck.assign_child_at_slot(resource, slot='{slot}') first." + ) + + # --- Summary --- + + def summary(self) -> str: + """ASCII representation of the Flex deck. + + Example:: + + Flex Deck (855mm x 582mm) + + +----------+----------+----------+----------+ + | A1 | A2 | A3 | A4 | + | Empty | Empty | trash | (staging)| + +----------+----------+----------+----------+ + | B1 | B2 | B3 | B4 | + | Empty | Empty | Empty | (staging)| + +----------+----------+----------+----------+ + ... + """ + + def _slot_label(slot_id: str) -> str: + resource = self._slot_holders[slot_id].resource + if resource is None: + if slot_id.endswith("4"): + return "(staging)" + return "Empty" + name = getattr(resource, "name", str(resource)) + if len(name) > 8: + name = name[:6] + ".." + return name + + sep = "+----------+----------+----------+----------+" + lines = [ + f"Flex Deck ({self.get_absolute_size_x():g}mm x {self.get_absolute_size_y():g}mm)", + "", + sep, + ] + + for row_letter in "ABCD": + row_ids = [f"| {row_letter}{col} " for col in "1234"] + row_names = [f"| {_slot_label(f'{row_letter}{col}'):8s} " for col in "1234"] + lines.append("".join(row_ids) + "|") + lines.append("".join(row_names) + "|") + lines.append(sep) + + return "\n".join(lines) diff --git a/pylabrobot/resources/opentrons/flex_tip_racks.py b/pylabrobot/resources/opentrons/flex_tip_racks.py new file mode 100644 index 00000000000..73ef3c346f4 --- /dev/null +++ b/pylabrobot/resources/opentrons/flex_tip_racks.py @@ -0,0 +1,148 @@ +"""Flex tip rack definitions — uses PLR's TipRack/TipSpot/Tip classes. + +Tip positions are hardcoded from the Opentrons labware definition JSON +(downloaded from the Flex robot on 2026-04-01). Standard 96-well layout +with 9mm pitch. + +Each factory function returns a PLR TipRack with: +- Standard TipSpots with TipTrackers for tip tracking and management +- Tips with VolumeTrackers for liquid volume tracking +- ``ot_load_name`` attribute for loading into the Flex robot's labware system +""" + +from __future__ import annotations + +from typing import Dict + +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.tip import Tip +from pylabrobot.resources.tip_rack import TipRack, TipSpot + +# --- Standard 96-well positions (9mm pitch, from Opentrons JSON) --- + +# A1 at (14.38, 74.38), rows go down in Y, columns go right in X. +# Z = 1.5mm (base of tip pocket). +_WELL_DX = 14.38 # left margin to A1 center +_WELL_DY = 74.38 # bottom margin to A1 center (measured from front) +_WELL_DZ = 1.5 # z offset +_PITCH = 9.0 # center-to-center spacing + +# Tip spot size (diameter of the tip pocket) +_SPOT_SIZE = 5.58 + + +def _make_flex_tip_rack( + name: str, + ot_load_name: str, + tip_volume: float, + total_tip_length: float, + fitting_depth: float, + has_filter: bool = False, +) -> TipRack: + """Create a PLR TipRack with 96 positions in Flex geometry. + + Returns a standard PLR TipRack with an extra ``ot_load_name`` + attribute identifying the Opentrons labware definition for the Flex robot. + """ + + def make_tip(name: str) -> Tip: + return Tip( + name=name, + maximal_volume=tip_volume, + total_tip_length=total_tip_length, + fitting_depth=fitting_depth, + has_filter=has_filter, + ) + + # Create ordered_items dict: {"A1": TipSpot, "B1": TipSpot, ...} + # Column-major order (A1, B1, ..., H1, A2, B2, ..., H12) + ordered_items: Dict[str, TipSpot] = {} + for col_idx in range(12): + for row_idx, row_letter in enumerate("ABCDEFGH"): + identifier = f"{row_letter}{col_idx + 1}" + spot = TipSpot( + name=identifier, + size_x=_SPOT_SIZE, + size_y=_SPOT_SIZE, + make_tip=make_tip, + ) + spot.location = Coordinate( + x=_WELL_DX + col_idx * _PITCH, + y=_WELL_DY - row_idx * _PITCH, + z=_WELL_DZ, + ) + ordered_items[identifier] = spot + + rack = TipRack( + name=name, + size_x=127.75, + size_y=85.75, + size_z=99.0, + ordered_items=ordered_items, + model=ot_load_name, + ) + + # Flex-specific: Opentrons labware load name for JIT loading + rack.ot_load_name = ot_load_name # type: ignore[attr-defined] + + return rack + + +# --- Tip Rack Factory Functions --- + + +def flex_96_tiprack_50ul(name: str = "flex_96_tiprack_50ul") -> TipRack: + """Opentrons Flex 96 Tip Rack 50 µL. + + Tip length 57.9mm, fitting depth 10.5mm (from Opentrons specs). + """ + return _make_flex_tip_rack( + name=name, + ot_load_name="opentrons_flex_96_tiprack_50ul", + tip_volume=50.0, + total_tip_length=57.9, + fitting_depth=10.5, + ) + + +def flex_96_filtertiprack_50ul( + name: str = "flex_96_filtertiprack_50ul", +) -> TipRack: + """Opentrons Flex 96 Filter Tip Rack 50 µL. + + Physically identical geometry to ``flex_96_tiprack_50ul`` (same 96-well + layout, tip length 57.9mm, fitting depth 10.5mm) — the only difference is the + aerosol filter, so it is the same rack with ``has_filter=True`` and the + Opentrons filter load name. Lets a protocol that uses filter tips resolve + against the resource model. + """ + return _make_flex_tip_rack( + name=name, + ot_load_name="opentrons_flex_96_filtertiprack_50ul", + tip_volume=50.0, + total_tip_length=57.9, + fitting_depth=10.5, + has_filter=True, + ) + + +def flex_96_tiprack_200ul(name: str = "flex_96_tiprack_200ul") -> TipRack: + """Opentrons Flex 96 Tip Rack 200 µL.""" + return _make_flex_tip_rack( + name=name, + ot_load_name="opentrons_flex_96_tiprack_200ul", + tip_volume=200.0, + total_tip_length=58.35, + fitting_depth=10.5, + ) + + +def flex_96_tiprack_1000ul(name: str = "flex_96_tiprack_1000ul") -> TipRack: + """Opentrons Flex 96 Tip Rack 1000 µL.""" + return _make_flex_tip_rack( + name=name, + ot_load_name="opentrons_flex_96_tiprack_1000ul", + tip_volume=1000.0, + total_tip_length=95.6, + fitting_depth=10.5, + )