From f4e8a4d9c5c23cb54a889aa64b46282abd578981 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Wed, 22 Jul 2026 09:52:24 +0200
Subject: [PATCH 01/13] feat(resources): Opentrons Flex deck + tip racks

---
 pylabrobot/resources/opentrons/__init__.py    |   2 +
 pylabrobot/resources/opentrons/flex_deck.py   | 357 ++++++++++++++++++
 .../resources/opentrons/flex_tip_racks.py     | 150 ++++++++
 3 files changed, 509 insertions(+)
 create mode 100644 pylabrobot/resources/opentrons/flex_deck.py
 create mode 100644 pylabrobot/resources/opentrons/flex_tip_racks.py

diff --git a/pylabrobot/resources/opentrons/__init__.py b/pylabrobot/resources/opentrons/__init__.py
index 38d1dc2e55c..a9e895898e2 100644
--- a/pylabrobot/resources/opentrons/__init__.py
+++ b/pylabrobot/resources/opentrons/__init__.py
@@ -1,5 +1,7 @@
 from .deck import OTDeck
+from .flex_deck import FlexDeck
 from .load import load_ot_tip_rack
 from .module import OTModule
 from .ot2_geometry import OT2RobotGeometry
 from .tip_racks import *
+from .flex_tip_racks import *
diff --git a/pylabrobot/resources/opentrons/flex_deck.py b/pylabrobot/resources/opentrons/flex_deck.py
new file mode 100644
index 00000000000..184ad95e4d4
--- /dev/null
+++ b/pylabrobot/resources/opentrons/flex_deck.py
@@ -0,0 +1,357 @@
+"""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.
+
+[Capability Branch]
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Optional
+
+from pylabrobot.resources.coordinate import Coordinate
+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
+
+# 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:
+    """Opentrons Flex deck — 16 slots with placement and collision detection.
+
+    Standalone class (no PLR Deck dependency). Manages which resources
+    are placed at which slots, provides coordinate lookups, and checks
+    collision clearance for single-nozzle operations.
+
+    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:
+        self.name = name
+        self._slots: Dict[str, Optional[Any]] = {}
+        for slot_id in list(SLOT_LOCATIONS) + list(STAGING_LOCATIONS):
+            self._slots[slot_id] = None
+
+        if with_trash_bin:
+            trash = Trash(
+                name="trash",
+                size_x=SLOT_WIDTH,
+                size_y=SLOT_DEPTH,
+                size_z=82.0,
+            )
+            trash.location = Coordinate(
+                x=SLOT_LOCATIONS["A3"]["x"],
+                y=SLOT_LOCATIONS["A3"]["y"],
+                z=SLOT_LOCATIONS["A3"]["z"],
+            )
+            self._slots["A3"] = trash
+
+    # --- 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: Any, 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)
+        current = self._slots.get(slot)
+        if current is not None:
+            name = getattr(current, "name", str(current))
+            raise ValueError(
+                f"Slot {slot} is already occupied by '{name}'."
+            )
+        self._slots[slot] = resource
+
+        # Set the resource's location so PLR's get_absolute_location()
+        # works through the standard resource tree
+        loc = self.get_slot_location(slot)
+        resource.location = Coordinate(x=loc["x"], y=loc["y"], z=loc["z"])
+
+    def unassign_child_at_slot(self, slot: str) -> None:
+        """Remove a resource from a slot."""
+        slot = self._validate_slot(slot)
+        resource = self._slots.get(slot)
+        if resource is not None:
+            resource.location = None
+        self._slots[slot] = None
+
+    def get_slot(self, resource: Any) -> Optional[str]:
+        """Get the slot identifier for a placed resource, or None."""
+        for slot_id, slot_resource in self._slots.items():
+            if slot_resource is resource:
+                return slot_id
+        return None
+
+    def get_resource_at_slot(self, slot: str) -> Optional[Any]:
+        """Return the resource placed at a slot, or None."""
+        slot = self._validate_slot(slot)
+        return self._slots.get(slot)
+
+    def get_trash_area(self) -> Trash:
+        """Return the trash resource (default at A3)."""
+        for slot_id, resource in self._slots.items():
+            if isinstance(resource, Trash):
+                return 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._slots.get(danger_slot)
+        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._slots.get(slot)
+        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._slots.get(slot_id)
+            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 = ["Flex Deck (855mm x 582mm)", "", 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..521f6f602fc
--- /dev/null
+++ b/pylabrobot/resources/opentrons/flex_tip_racks.py
@@ -0,0 +1,150 @@
+"""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 (compatible with PIP capability)
+- Tips with VolumeTrackers (compatible with PIP's volume tracking)
+- ``ot_load_name`` attribute for JIT loading into the Flex run
+
+[Capability Branch]
+"""
+
+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 for FlexPIPBackend's JIT labware loading.
+    """
+
+    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,
+    )

From 8f94a26190b882cb0172084f3b8ddf93a71560e5 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 08:37:47 +0200
Subject: [PATCH 02/13] feat(opentrons): OpentronsRobot plain-class base (httpx
 transport + protocol + discovery)

---
 pylabrobot/opentrons/robot.py | 324 ++++++++++++++++++++++++++++++++++
 1 file changed, 324 insertions(+)
 create mode 100644 pylabrobot/opentrons/robot.py

diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py
new file mode 100644
index 00000000000..7031b746d61
--- /dev/null
+++ b/pylabrobot/opentrons/robot.py
@@ -0,0 +1,324 @@
+import abc
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+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.
+  """
+
+  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 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 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 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": {}})
+    self.run_id = result["data"]["id"]
+    logger.info("Created run %s", self.run_id)
+    return self.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 = 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: "
+          f"{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
+
+  def check_gripper(self, instruments_data: Dict[str, Any]) -> bool:
+    """Check if a gripper is attached."""
+    for instrument in instruments_data.get("data", []):
+      if instrument.get("instrumentType") == "gripper":
+        return True
+    return False
+
+  async def get_modules(self) -> List[Dict[str, Any]]:
+    """Query connected modules via GET /modules.
+
+    Returns the module data dict for each USB-connected module. Passive
+    hardware (magnetic block, waste chute) is not discoverable.
+    """
+    data = await self._get("/modules")
+    return data.get("data", [])
+
+  # --- 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 = 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", {})
+
+  # --- Movement Commands ---
+
+  async def move_to_coordinates(
+    self,
+    pipette_id: str,
+    x: float,
+    y: float,
+    z: float,
+    minimum_z_height: Optional[float] = None,
+    speed: Optional[float] = None,
+  ) -> Dict[str, Any]:
+    """Move a pipette to absolute deck coordinates.
+
+    The robot automatically arcs to a safe Z height before lateral
+    movement (unless forceDirect=True, which we never set).
+    The minimumZHeight parameter raises the arc if needed.
+    """
+    params: Dict[str, Any] = {
+      "pipetteId": pipette_id,
+      "coordinates": {"x": x, "y": y, "z": z},
+    }
+    if minimum_z_height is not None:
+      params["minimumZHeight"] = minimum_z_height
+    if speed is not None:
+      params["speed"] = speed
+    return await self.execute_command("moveToCoordinates", params)
+
+  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

From c12a83ddd56ebd3ae60fe18f48076d8e1295841b Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 08:45:08 +0200
Subject: [PATCH 03/13] =?UTF-8?q?feat(opentrons):=20OpentronsFlex=20plain?=
 =?UTF-8?q?=20class=20=E2=80=94=20setup=20+=20tip=20ops=20commit=20to=20re?=
 =?UTF-8?q?source=20trackers?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds OpentronsFlex(OpentronsRobot): _model_setup homes the gantry;
pick_up_tips/drop_tips commit tip state to TipSpot.tracker (remove_tip/
add_tip + commit) rather than a private per-device copy. Only which
physical channel holds which tip is device-local state
(self._channel_tips). Trash drops use the addressable-area path
(moveToAddressableAreaForDropTip + dropTipInPlace) and skip the tracker
add. JIT labware loading and load-name resolution ported verbatim from
titronic plr_v4/flex/head8_backend.py.

wellName is derived via TipRack.get_child_identifier(tip_spot) — the
canonical PLR API for item->identifier — not string-splitting spot.name
(which is prefixed with the rack name, e.g. "tips_A1").
---
 pylabrobot/opentrons/__init__.py |   2 +
 pylabrobot/opentrons/flex.py     | 153 +++++++++++++++++++++++++++++++
 2 files changed, 155 insertions(+)
 create mode 100644 pylabrobot/opentrons/flex.py

diff --git a/pylabrobot/opentrons/__init__.py b/pylabrobot/opentrons/__init__.py
index 45ef90311f4..8b914f0012b 100644
--- a/pylabrobot/opentrons/__init__.py
+++ b/pylabrobot/opentrons/__init__.py
@@ -5,3 +5,5 @@
   OpentronsTemperatureModuleUSBTemperatureBackend,
   OpentronsTemperatureModuleV2,
 )
+from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot, PipetteInfo
+from pylabrobot.opentrons.flex import OpentronsFlex
diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
new file mode 100644
index 00000000000..ff21be68468
--- /dev/null
+++ b/pylabrobot/opentrons/flex.py
@@ -0,0 +1,153 @@
+import logging
+import uuid
+from typing import List, Optional
+
+from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot
+from pylabrobot.resources import Coordinate, TipSpot, Trash
+from pylabrobot.resources.opentrons.flex_deck import FlexDeck
+
+logger = logging.getLogger(__name__)
+
+_OT_NAMESPACE = "opentrons"
+_OT_VERSION = 1
+
+_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 = {}
+    self._channel_tips: List[Optional[object]] = [None] * 8
+
+  async def _model_setup(self) -> None:
+    await self.home()
+
+  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)))
+    labware_id = await self._ensure_labware_loaded(tip_spots[0].parent)
+    well_name = tip_spots[0].parent.get_child_identifier(tip_spots[0])
+    params = {
+      "pipetteId": self.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()
+      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)))
+    target = tip_spots[0]
+    if isinstance(target, Trash) or isinstance(target.parent, Trash):
+      await self.execute_command("moveToAddressableAreaForDropTip", {
+        "pipetteId": self.pipette.pipette_id,
+        "addressableAreaName": "movableTrashA3",
+        "alternateDropLocation": True,
+      })
+      await self.execute_command("dropTipInPlace", {"pipetteId": self.pipette.pipette_id})
+    else:
+      labware_id = await self._ensure_labware_loaded(target.parent)
+      well_name = target.parent.get_child_identifier(target)
+      params = {
+        "pipetteId": self.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 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 _ensure_labware_loaded(self, 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 = 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) -> str:
+    """Resolve a PLR resource to its Opentrons labware load name."""
+    if hasattr(resource, "ot_load_name"):
+      return 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.",
+    )

From 18421499c1a304c9d0dcf2ef68cc60bb0e37973c Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 08:54:35 +0200
Subject: [PATCH 04/13] =?UTF-8?q?feat(opentrons):=20OpentronsFlex=20aspira?=
 =?UTF-8?q?te/dispense=20=E2=80=94=20established=20signature,=20volume=20t?=
 =?UTF-8?q?rackers?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pylabrobot/opentrons/flex.py | 89 +++++++++++++++++++++++++++++++++++-
 1 file changed, 87 insertions(+), 2 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index ff21be68468..f1abd0f6578 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -1,9 +1,9 @@
 import logging
 import uuid
-from typing import List, Optional
+from typing import List, Literal, Optional, Sequence
 
 from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot
-from pylabrobot.resources import Coordinate, TipSpot, Trash
+from pylabrobot.resources import Container, Coordinate, TipSpot, Trash
 from pylabrobot.resources.opentrons.flex_deck import FlexDeck
 
 logger = logging.getLogger(__name__)
@@ -11,6 +11,11 @@
 _OT_NAMESPACE = "opentrons"
 _OT_VERSION = 1
 
+# Flex-managed positioning flow-rate defaults (uL/s), copied verbatim from
+# titronic plr_v4/flex/head8_backend.py.
+_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",
@@ -98,6 +103,86 @@ async def drop_tips(
         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:
+    labware_id = await self._ensure_labware_loaded(resources[0].parent)
+    well_name = resources[0].parent.get_child_identifier(resources[0])
+    flow_rate = (flow_rates[0] if flow_rates else None) or _DEFAULT_ASPIRATE_FLOW_RATE
+    params = {
+      "pipetteId": self.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)
+    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:
+    labware_id = await self._ensure_labware_loaded(resources[0].parent)
+    well_name = resources[0].parent.get_child_identifier(resources[0])
+    flow_rate = (flow_rates[0] if flow_rates else None) or _DEFAULT_DISPENSE_FLOW_RATE
+    params = {
+      "pipetteId": self.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)
+    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.
+
+    Mirrors titronic ``head8_backend._aspirate_flex``/``_dispense_flex``, which sets
+    ``offset.z`` from ``liquid_height`` (origin "bottom" per ``FlexDriver.aspirate``/
+    ``dispense``); extended here to also honor an explicit x/y/z offset.
+    """
+    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) -> str:
     """Load labware into the Flex run if not already loaded."""
     name = getattr(resource, "name", str(resource))

From f03a601c8e96314be13d5b0b060eadf19a459394 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 09:00:05 +0200
Subject: [PATCH 05/13] style(opentrons): drop provenance comments from flex.py

Removes narrative comments referencing source origins (titronic, plr_v4)
in favor of pure behavioral descriptions. Fixes lines 14-15 (flow-rate
comment) and 169-173 (wellLocation docstring).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 pylabrobot/opentrons/flex.py | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index f1abd0f6578..95b5e98a0d0 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -11,8 +11,7 @@
 _OT_NAMESPACE = "opentrons"
 _OT_VERSION = 1
 
-# Flex-managed positioning flow-rate defaults (uL/s), copied verbatim from
-# titronic plr_v4/flex/head8_backend.py.
+# Flex-managed positioning flow-rate defaults (uL/s).
 _DEFAULT_ASPIRATE_FLOW_RATE = 35.0
 _DEFAULT_DISPENSE_FLOW_RATE = 57.0
 
@@ -168,9 +167,8 @@ def _well_location(
   ) -> Optional[dict]:
     """Build the Flex ``wellLocation`` param from an offset and/or liquid height.
 
-    Mirrors titronic ``head8_backend._aspirate_flex``/``_dispense_flex``, which sets
-    ``offset.z`` from ``liquid_height`` (origin "bottom" per ``FlexDriver.aspirate``/
-    ``dispense``); extended here to also honor an explicit x/y/z offset.
+    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:

From 3e8a4304ae94e7f73da691c69a38a263a792d004 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 09:12:03 +0200
Subject: [PATCH 06/13] =?UTF-8?q?fix(opentrons):=20mypy-clean=20robot.py/f?=
 =?UTF-8?q?lex.py=20=E2=80=94=20narrow=20Optionals,=20type=20dict=20return?=
 =?UTF-8?q?s?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Casts httpx JSON returns instead of leaking Any; narrows self.pipette and
TipSpot/Container.parent with explicit asserts instead of unchecked attribute
access; types _loaded_labware and _channel_tips precisely. Behavior unchanged
(harness still passes all sections).
---
 pylabrobot/opentrons/flex.py  | 81 +++++++++++++++++++++++------------
 pylabrobot/opentrons/robot.py | 19 ++++----
 2 files changed, 64 insertions(+), 36 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index 95b5e98a0d0..9131b6a22db 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -1,10 +1,12 @@
 import logging
 import uuid
-from typing import List, Literal, Optional, Sequence
+from typing import Any, Dict, List, Literal, Optional, Sequence, cast
 
-from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot
-from pylabrobot.resources import Container, Coordinate, TipSpot, Trash
+from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot, PipetteInfo
+from pylabrobot.resources import Container, Coordinate, Resource, TipSpot, Trash
+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__)
 
@@ -38,12 +40,29 @@ class OpentronsFlex(OpentronsRobot):
   def __init__(self, deck: FlexDeck, host: str, port: int = 31950) -> None:
     super().__init__(host=host, port=port)
     self.deck = deck
-    self._loaded_labware: dict = {}
-    self._channel_tips: List[Optional[object]] = [None] * 8
+    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],
@@ -51,10 +70,12 @@ async def pick_up_tips(
     offsets: Optional[List[Coordinate]] = None,
   ) -> None:
     use_channels = use_channels if use_channels is not None else list(range(len(tip_spots)))
-    labware_id = await self._ensure_labware_loaded(tip_spots[0].parent)
-    well_name = tip_spots[0].parent.get_child_identifier(tip_spots[0])
-    params = {
-      "pipetteId": self.pipette.pipette_id,
+    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,
     }
@@ -75,19 +96,21 @@ async def drop_tips(
     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": self.pipette.pipette_id,
+        "pipetteId": pipette.pipette_id,
         "addressableAreaName": "movableTrashA3",
         "alternateDropLocation": True,
       })
-      await self.execute_command("dropTipInPlace", {"pipetteId": self.pipette.pipette_id})
+      await self.execute_command("dropTipInPlace", {"pipetteId": pipette.pipette_id})
     else:
-      labware_id = await self._ensure_labware_loaded(target.parent)
-      well_name = target.parent.get_child_identifier(target)
-      params = {
-        "pipetteId": self.pipette.pipette_id,
+      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,
       }
@@ -113,11 +136,13 @@ async def aspirate(
     blow_out_air_volume: Optional[List[Optional[float]]] = None,
     spread: Literal["wide", "tight", "custom"] = "wide",
   ) -> None:
-    labware_id = await self._ensure_labware_loaded(resources[0].parent)
-    well_name = resources[0].parent.get_child_identifier(resources[0])
+    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 = {
-      "pipetteId": self.pipette.pipette_id,
+    params: Dict[str, Any] = {
+      "pipetteId": pipette.pipette_id,
       "labwareId": labware_id,
       "wellName": well_name,
       "volume": vols[0],
@@ -142,11 +167,13 @@ async def dispense(
     blow_out_air_volume: Optional[List[Optional[float]]] = None,
     spread: Literal["wide", "tight", "custom"] = "wide",
   ) -> None:
-    labware_id = await self._ensure_labware_loaded(resources[0].parent)
-    well_name = resources[0].parent.get_child_identifier(resources[0])
+    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 = {
-      "pipetteId": self.pipette.pipette_id,
+    params: Dict[str, Any] = {
+      "pipetteId": pipette.pipette_id,
       "labwareId": labware_id,
       "wellName": well_name,
       "volume": vols[0],
@@ -181,7 +208,7 @@ def _well_location(
       return None
     return {"origin": "bottom", "offset": offset}
 
-  async def _ensure_labware_loaded(self, resource) -> str:
+  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:
@@ -205,7 +232,7 @@ async def _ensure_labware_loaded(self, resource) -> str:
       "labwareId": labware_id,
       "displayName": name,
     })
-    labware_id = result.get("result", {}).get("labwareId", labware_id)
+    labware_id = cast(str, result.get("result", {}).get("labwareId", labware_id))
 
     self._loaded_labware[name] = labware_id
     logger.info(
@@ -215,10 +242,10 @@ async def _ensure_labware_loaded(self, resource) -> str:
     return labware_id
 
   @staticmethod
-  def _ot_load_name(resource) -> str:
+  def _ot_load_name(resource: Resource) -> str:
     """Resolve a PLR resource to its Opentrons labware load name."""
     if hasattr(resource, "ot_load_name"):
-      return resource.ot_load_name
+      return cast(str, resource.ot_load_name)
 
     name_lower = getattr(resource, "name", "").lower()
 
diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py
index 7031b746d61..2d9024c54d7 100644
--- a/pylabrobot/opentrons/robot.py
+++ b/pylabrobot/opentrons/robot.py
@@ -3,7 +3,7 @@
 import logging
 import time
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, cast
 
 try:
   import httpx
@@ -103,21 +103,21 @@ async def _get(self, path: str) -> Dict[str, Any]:
     assert self._client is not None, "Not connected. Call connect() first."
     response = await self._client.get(path)
     response.raise_for_status()
-    return response.json()
+    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 response.json()
+    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 response.json()
+    return cast(Dict[str, Any], response.json())
 
   # --- Run Management ---
 
@@ -128,9 +128,10 @@ async def _create_run(self) -> str:
     interactively, which is how PLR controls the robot.
     """
     result = await self._post("/runs", {"data": {}})
-    self.run_id = result["data"]["id"]
+    run_id = cast(str, result["data"]["id"])
+    self.run_id = run_id
     logger.info("Created run %s", self.run_id)
-    return self.run_id
+    return run_id
 
   async def _cancel_run(self) -> None:
     """Cancel the current run. Safe to call if no run is active."""
@@ -185,7 +186,7 @@ async def execute_command(
       }
     }
     result = await self._post(f"/runs/{self.run_id}/commands", payload)
-    cmd_data = result.get("data", {})
+    cmd_data: Dict[str, Any] = result.get("data", {})
 
     if not wait:
       return cmd_data
@@ -258,7 +259,7 @@ async def get_modules(self) -> List[Dict[str, Any]]:
     hardware (magnetic block, waste chute) is not discoverable.
     """
     data = await self._get("/modules")
-    return data.get("data", [])
+    return cast(List[Dict[str, Any]], data.get("data", []))
 
   # --- Pipette Loading ---
 
@@ -274,7 +275,7 @@ async def _load_pipette(self, pipette_name: str, mount: str) -> str:
       {"pipetteName": pipette_name, "mount": mount},
       wait=True,
     )
-    pipette_id = result.get("result", {}).get("pipetteId", "")
+    pipette_id: str = result.get("result", {}).get("pipetteId", "")
     logger.info(
       "Loaded pipette %s on %s mount -> ID: %s",
       pipette_name, mount, pipette_id,

From 17e1b3649d356d4a48f6656935e3f3fb32ece338 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 09:12:12 +0200
Subject: [PATCH 07/13] docs(opentrons): Flex hello-world notebook + API docs

Adds docs/user_guide/opentrons/flex/hello-world.ipynb covering setup(),
pick_up_tips, aspirate, dispense, and drop_tips on the plain-class
OpentronsFlex with a FlexDeck, a flex_96_tiprack_50ul, and a Corning 96-well
plate. Wires it into the opentrons/index toctree (opentrons/index was
already listed in the Manufacturers toctree in user_guide/index.md).
Documents OpentronsRobot/OpentronsFlex/OpentronsError/PipetteInfo in
docs/api/pylabrobot.opentrons.rst (already referenced from pylabrobot.rst).
---
 docs/api/pylabrobot.opentrons.rst             |  13 +
 .../opentrons/flex/hello-world.ipynb          | 229 ++++++++++++++++++
 docs/user_guide/opentrons/index.md            |   1 +
 3 files changed, 243 insertions(+)
 create mode 100644 docs/user_guide/opentrons/flex/hello-world.ipynb

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
 ```

From e434b0a71fca9b105716884b75e2ddc7789413c0 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 09:26:00 +0200
Subject: [PATCH 08/13] style(opentrons): ruff format + drop
 deprecated-architecture references from resources

Apply ruff format to standardize code layout. Remove [Capability Branch] markers
and reword docstrings that referenced PIP capability/FlexPIPBackend, replacing
with factual descriptions of what the racks and deck are (Opentrons Flex device
definitions with ot_load_name for the robot's labware system).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 pylabrobot/opentrons/flex.py                  |  53 +-
 pylabrobot/opentrons/robot.py                 |  19 +-
 pylabrobot/resources/opentrons/flex_deck.py   | 602 +++++++++---------
 .../resources/opentrons/flex_tip_racks.py     | 203 +++---
 4 files changed, 447 insertions(+), 430 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index 9131b6a22db..80116380738 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -58,9 +58,9 @@ def _require_pipette(self) -> PipetteInfo:
   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)."
-    )
+    assert isinstance(
+      parent, ItemizedResource
+    ), f"'{item.name}' has no itemized parent resource (rack/plate)."
     return parent
 
   async def pick_up_tips(
@@ -81,7 +81,10 @@ async def pick_up_tips(
     }
     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}}
+      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()
@@ -99,11 +102,14 @@ async def drop_tips(
     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(
+        "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)
@@ -116,7 +122,10 @@ async def drop_tips(
       }
       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}}
+        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]
@@ -224,20 +233,26 @@ async def _ensure_labware_loaded(self, resource: Resource) -> str:
     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,
-    })
+    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,
+      name,
+      slot,
+      labware_id,
+      load_name,
     )
     return labware_id
 
diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py
index 2d9024c54d7..cea7e0419ec 100644
--- a/pylabrobot/opentrons/robot.py
+++ b/pylabrobot/opentrons/robot.py
@@ -7,6 +7,7 @@
 
 try:
   import httpx
+
   _HAS_HTTPX = True
 except ImportError:
   _HAS_HTTPX = False
@@ -86,8 +87,11 @@ async def _connect(self) -> None:
     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,
+      robot_name,
+      self.host,
+      self.port,
+      self.api_version,
+      self.robot_model,
     )
 
   async def _disconnect(self) -> None:
@@ -206,14 +210,11 @@ async def execute_command(
       elif status == "failed":
         error = cmd_data.get("error", {})
         raise RuntimeError(
-          f"Opentrons command '{command_type}' failed: "
-          f"{error.get('detail', error)}"
+          f"Opentrons command '{command_type}' failed: " f"{error.get('detail', error)}"
         )
       await asyncio.sleep(0.2)
 
-    raise RuntimeError(
-      f"Opentrons command '{command_type}' timed out after {timeout}s"
-    )
+    raise RuntimeError(f"Opentrons command '{command_type}' timed out after {timeout}s")
 
   # --- Instrument Discovery ---
 
@@ -278,7 +279,9 @@ async def _load_pipette(self, pipette_name: str, mount: str) -> str:
     pipette_id: str = result.get("result", {}).get("pipetteId", "")
     logger.info(
       "Loaded pipette %s on %s mount -> ID: %s",
-      pipette_name, mount, pipette_id,
+      pipette_name,
+      mount,
+      pipette_id,
     )
     return pipette_id
 
diff --git a/pylabrobot/resources/opentrons/flex_deck.py b/pylabrobot/resources/opentrons/flex_deck.py
index 184ad95e4d4..22ba3382be8 100644
--- a/pylabrobot/resources/opentrons/flex_deck.py
+++ b/pylabrobot/resources/opentrons/flex_deck.py
@@ -10,8 +10,6 @@
 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.
-
-[Capability Branch]
 """
 
 from __future__ import annotations
@@ -25,10 +23,18 @@
 
 # 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",
+  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
@@ -40,31 +46,31 @@
 # 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},
+  "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},
+  "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
+SLOT_WIDTH = 128.0  # x dimension
+SLOT_DEPTH = 86.0  # y dimension
 
 # Default clearance Z when no operation height is provided.
 # Conservative estimate — measured at 51.3mm on real hardware
@@ -73,285 +79,279 @@
 
 
 class FlexDeck:
-    """Opentrons Flex deck — 16 slots with placement and collision detection.
-
-    Standalone class (no PLR Deck dependency). Manages which resources
-    are placed at which slots, provides coordinate lookups, and checks
-    collision clearance for single-nozzle operations.
-
-    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:
-        self.name = name
-        self._slots: Dict[str, Optional[Any]] = {}
-        for slot_id in list(SLOT_LOCATIONS) + list(STAGING_LOCATIONS):
-            self._slots[slot_id] = None
-
-        if with_trash_bin:
-            trash = Trash(
-                name="trash",
-                size_x=SLOT_WIDTH,
-                size_y=SLOT_DEPTH,
-                size_z=82.0,
-            )
-            trash.location = Coordinate(
-                x=SLOT_LOCATIONS["A3"]["x"],
-                y=SLOT_LOCATIONS["A3"]["y"],
-                z=SLOT_LOCATIONS["A3"]["z"],
-            )
-            self._slots["A3"] = trash
-
-    # --- 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
-
+  """Opentrons Flex deck — 16 slots with placement and collision detection.
+
+  Standalone class (no PLR Deck dependency). Manages which resources
+  are placed at which slots, provides coordinate lookups, and checks
+  collision clearance for single-nozzle operations.
+
+  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:
+    self.name = name
+    self._slots: Dict[str, Optional[Any]] = {}
+    for slot_id in list(SLOT_LOCATIONS) + list(STAGING_LOCATIONS):
+      self._slots[slot_id] = None
+
+    if with_trash_bin:
+      trash = Trash(
+        name="trash",
+        size_x=SLOT_WIDTH,
+        size_y=SLOT_DEPTH,
+        size_z=82.0,
+      )
+      trash.location = Coordinate(
+        x=SLOT_LOCATIONS["A3"]["x"],
+        y=SLOT_LOCATIONS["A3"]["y"],
+        z=SLOT_LOCATIONS["A3"]["z"],
+      )
+      self._slots["A3"] = trash
+
+  # --- 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"Invalid slot identifier '{slot}'. "
-            f"Must be A1–D3 (standard) or A4–D4 (staging). "
-            f"Examples: 'C1', 'A3', 'B4'."
+          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: Any, 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)
+    current = self._slots.get(slot)
+    if current is not None:
+      name = getattr(current, "name", str(current))
+      raise ValueError(f"Slot {slot} is already occupied by '{name}'.")
+    self._slots[slot] = resource
+
+    # Set the resource's location so PLR's get_absolute_location()
+    # works through the standard resource tree
+    loc = self.get_slot_location(slot)
+    resource.location = Coordinate(x=loc["x"], y=loc["y"], z=loc["z"])
+
+  def unassign_child_at_slot(self, slot: str) -> None:
+    """Remove a resource from a slot."""
+    slot = self._validate_slot(slot)
+    resource = self._slots.get(slot)
+    if resource is not None:
+      resource.location = None
+    self._slots[slot] = None
+
+  def get_slot(self, resource: Any) -> Optional[str]:
+    """Get the slot identifier for a placed resource, or None."""
+    for slot_id, slot_resource in self._slots.items():
+      if slot_resource is resource:
+        return slot_id
+    return None
+
+  def get_resource_at_slot(self, slot: str) -> Optional[Any]:
+    """Return the resource placed at a slot, or None."""
+    slot = self._validate_slot(slot)
+    return self._slots.get(slot)
+
+  def get_trash_area(self) -> Trash:
+    """Return the trash resource (default at A3)."""
+    for slot_id, resource in self._slots.items():
+      if isinstance(resource, Trash):
+        return 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._slots.get(danger_slot)
+    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._slots.get(slot)
+    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.
 
-    # --- 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: Any, 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)
-        current = self._slots.get(slot)
-        if current is not None:
-            name = getattr(current, "name", str(current))
-            raise ValueError(
-                f"Slot {slot} is already occupied by '{name}'."
-            )
-        self._slots[slot] = resource
-
-        # Set the resource's location so PLR's get_absolute_location()
-        # works through the standard resource tree
-        loc = self.get_slot_location(slot)
-        resource.location = Coordinate(x=loc["x"], y=loc["y"], z=loc["z"])
-
-    def unassign_child_at_slot(self, slot: str) -> None:
-        """Remove a resource from a slot."""
-        slot = self._validate_slot(slot)
-        resource = self._slots.get(slot)
-        if resource is not None:
-            resource.location = None
-        self._slots[slot] = None
-
-    def get_slot(self, resource: Any) -> Optional[str]:
-        """Get the slot identifier for a placed resource, or None."""
-        for slot_id, slot_resource in self._slots.items():
-            if slot_resource is resource:
-                return slot_id
-        return None
-
-    def get_resource_at_slot(self, slot: str) -> Optional[Any]:
-        """Return the resource placed at a slot, or None."""
-        slot = self._validate_slot(slot)
-        return self._slots.get(slot)
-
-    def get_trash_area(self) -> Trash:
-        """Return the trash resource (default at A3)."""
-        for slot_id, resource in self._slots.items():
-            if isinstance(resource, Trash):
-                return 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._slots.get(danger_slot)
-        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._slots.get(slot)
-        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._slots.get(slot_id)
-            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 = ["Flex Deck (855mm x 582mm)", "", 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)
+    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._slots.get(slot_id)
+      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 = ["Flex Deck (855mm x 582mm)", "", 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
index 521f6f602fc..c329de10b02 100644
--- a/pylabrobot/resources/opentrons/flex_tip_racks.py
+++ b/pylabrobot/resources/opentrons/flex_tip_racks.py
@@ -5,11 +5,9 @@
 with 9mm pitch.
 
 Each factory function returns a PLR TipRack with:
-- Standard TipSpots with TipTrackers (compatible with PIP capability)
-- Tips with VolumeTrackers (compatible with PIP's volume tracking)
-- ``ot_load_name`` attribute for JIT loading into the Flex run
-
-[Capability Branch]
+- 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
@@ -27,124 +25,125 @@
 # 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
+_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,
+  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 for FlexPIPBackend's JIT labware loading.
-    """
-
-    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,
+  """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,
     )
 
-    # Flex-specific: Opentrons labware load name for JIT loading
-    rack.ot_load_name = ot_load_name  # type: ignore[attr-defined]
-
-    return rack
+  # 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,
-    )
+  """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",
+  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,
-    )
+  """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,
-    )
+  """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,
-    )
+  """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,
+  )

From d9b50821e53976f87476012eead00759c5225c43 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 09:35:34 +0200
Subject: [PATCH 09/13] feat(opentrons): honor global tip/volume tracking
 switches

---
 pylabrobot/opentrons/flex.py | 36 ++++++++++++++++++++++++++----------
 1 file changed, 26 insertions(+), 10 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index 80116380738..4dcfb18022a 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -3,7 +3,15 @@
 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
+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
@@ -88,8 +96,9 @@ async def pick_up_tips(
     await self.execute_command("pickUpTip", params)
     for ch, spot in zip(use_channels, tip_spots):
       tip = spot.get_tip()
-      spot.tracker.remove_tip()
-      spot.tracker.commit()
+      if does_tip_tracking():
+        spot.tracker.remove_tip()
+        spot.tracker.commit()
       self._channel_tips[ch] = tip
 
   async def drop_tips(
@@ -129,7 +138,12 @@ async def drop_tips(
       await self.execute_command("dropTip", params)
     for ch, spot in zip(use_channels, tip_spots):
       tip = self._channel_tips[ch]
-      if tip is not None and not isinstance(spot, Trash) and not isinstance(spot.parent, Trash):
+      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
@@ -161,9 +175,10 @@ async def aspirate(
     if well_location is not None:
       params["wellLocation"] = well_location
     await self.execute_command("aspirate", params)
-    for well, vol in zip(resources, vols):
-      well.tracker.remove_liquid(vol)
-      well.tracker.commit()
+    if does_volume_tracking():
+      for well, vol in zip(resources, vols):
+        well.tracker.remove_liquid(vol)
+        well.tracker.commit()
 
   async def dispense(
     self,
@@ -192,9 +207,10 @@ async def dispense(
     if well_location is not None:
       params["wellLocation"] = well_location
     await self.execute_command("dispense", params)
-    for well, vol in zip(resources, vols):
-      well.tracker.add_liquid(vol)
-      well.tracker.commit()
+    if does_volume_tracking():
+      for well, vol in zip(resources, vols):
+        well.tracker.add_liquid(vol)
+        well.tracker.commit()
 
   @staticmethod
   def _well_location(

From ac9fe8764381fb32fb5f3ae1c0264102cc0ba048 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 10:07:12 +0200
Subject: [PATCH 10/13] =?UTF-8?q?refactor(resources):=20FlexDeck=20is=20a?=
 =?UTF-8?q?=20real=20Deck=20=E2=80=94=20parent=20labware=20into=20the=20re?=
 =?UTF-8?q?source=20tree?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pylabrobot/resources/opentrons/flex_deck.py | 97 +++++++++++----------
 1 file changed, 51 insertions(+), 46 deletions(-)

diff --git a/pylabrobot/resources/opentrons/flex_deck.py b/pylabrobot/resources/opentrons/flex_deck.py
index 22ba3382be8..4cd847bc2a7 100644
--- a/pylabrobot/resources/opentrons/flex_deck.py
+++ b/pylabrobot/resources/opentrons/flex_deck.py
@@ -15,12 +15,14 @@
 from __future__ import annotations
 
 import re
-from typing import Any, Dict, Optional
+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",
@@ -72,18 +74,25 @@
 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:
+class FlexDeck(Deck):
   """Opentrons Flex deck — 16 slots with placement and collision detection.
 
-  Standalone class (no PLR Deck dependency). Manages which resources
-  are placed at which slots, provides coordinate lookups, and checks
-  collision clearance for single-nozzle operations.
+  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::
 
@@ -98,24 +107,22 @@ def __init__(
     with_trash_bin: bool = True,
     name: str = "flex_deck",
   ) -> None:
-    self.name = name
-    self._slots: Dict[str, Optional[Any]] = {}
-    for slot_id in list(SLOT_LOCATIONS) + list(STAGING_LOCATIONS):
-      self._slots[slot_id] = None
+    super().__init__(size_x=_DECK_SIZE_X, size_y=_DECK_SIZE_Y, size_z=0.0, name=name)
 
-    if with_trash_bin:
-      trash = Trash(
-        name="trash",
+    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=82.0,
-      )
-      trash.location = Coordinate(
-        x=SLOT_LOCATIONS["A3"]["x"],
-        y=SLOT_LOCATIONS["A3"]["y"],
-        z=SLOT_LOCATIONS["A3"]["z"],
+        size_z=0,
       )
-      self._slots["A3"] = trash
+      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 ---
 
@@ -158,7 +165,7 @@ def get_slot_location(self, slot: str) -> Dict[str, float]:
       return STAGING_LOCATIONS[slot]
     raise ValueError(f"Unknown slot '{slot}'.")
 
-  def assign_child_at_slot(self, resource: Any, slot: str) -> None:
+  def assign_child_at_slot(self, resource: Resource, slot: str) -> None:
     """Place a resource at a named slot.
 
     Args:
@@ -169,42 +176,36 @@ def assign_child_at_slot(self, resource: Any, slot: str) -> None:
         ValueError: If slot is invalid or already occupied.
     """
     slot = self._validate_slot(slot)
-    current = self._slots.get(slot)
-    if current is not None:
-      name = getattr(current, "name", str(current))
+    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}'.")
-    self._slots[slot] = resource
-
-    # Set the resource's location so PLR's get_absolute_location()
-    # works through the standard resource tree
-    loc = self.get_slot_location(slot)
-    resource.location = Coordinate(x=loc["x"], y=loc["y"], z=loc["z"])
+    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)
-    resource = self._slots.get(slot)
-    if resource is not None:
-      resource.location = None
-    self._slots[slot] = None
+    holder = self._slot_holders[slot]
+    if holder.resource is not None:
+      holder.unassign_child_resource(holder.resource)
 
-  def get_slot(self, resource: Any) -> Optional[str]:
+  def get_slot(self, resource: Resource) -> Optional[str]:
     """Get the slot identifier for a placed resource, or None."""
-    for slot_id, slot_resource in self._slots.items():
-      if slot_resource is resource:
+    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[Any]:
+  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._slots.get(slot)
+    return self._slot_holders[slot].resource
 
   def get_trash_area(self) -> Trash:
     """Return the trash resource (default at A3)."""
-    for slot_id, resource in self._slots.items():
-      if isinstance(resource, Trash):
-        return resource
+    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 ---
@@ -269,7 +270,7 @@ def check_single_nozzle_clearance(
     else:
       return  # Other nozzle configs — skip for now
 
-    resource = self._slots.get(danger_slot)
+    resource = self._slot_holders[danger_slot].resource
     if resource is None:
       return  # Slot empty, safe
 
@@ -307,7 +308,7 @@ def check_single_nozzle_clearance(
   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._slots.get(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. "
@@ -334,7 +335,7 @@ def summary(self) -> str:
     """
 
     def _slot_label(slot_id: str) -> str:
-      resource = self._slots.get(slot_id)
+      resource = self._slot_holders[slot_id].resource
       if resource is None:
         if slot_id.endswith("4"):
           return "(staging)"
@@ -345,7 +346,11 @@ def _slot_label(slot_id: str) -> str:
       return name
 
     sep = "+----------+----------+----------+----------+"
-    lines = ["Flex Deck (855mm x 582mm)", "", 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"]

From ead3fb6cafd03f4e4546a394fcd8d461016bae3d Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 10:26:07 +0200
Subject: [PATCH 11/13] refactor(opentrons): drop dead base methods, privatize
 the wire primitive, note httpx transport

---
 pylabrobot/opentrons/flex.py  | 14 ++++-----
 pylabrobot/opentrons/robot.py | 53 +++++------------------------------
 2 files changed, 14 insertions(+), 53 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index 4dcfb18022a..9613f835415 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -93,7 +93,7 @@ async def pick_up_tips(
         "origin": "top",
         "offset": {"x": offset.x, "y": offset.y, "z": offset.z},
       }
-    await self.execute_command("pickUpTip", params)
+    await self._execute_command("pickUpTip", params)
     for ch, spot in zip(use_channels, tip_spots):
       tip = spot.get_tip()
       if does_tip_tracking():
@@ -111,7 +111,7 @@ async def drop_tips(
     pipette = self._require_pipette()
     target = tip_spots[0]
     if isinstance(target, Trash) or isinstance(target.parent, Trash):
-      await self.execute_command(
+      await self._execute_command(
         "moveToAddressableAreaForDropTip",
         {
           "pipetteId": pipette.pipette_id,
@@ -119,7 +119,7 @@ async def drop_tips(
           "alternateDropLocation": True,
         },
       )
-      await self.execute_command("dropTipInPlace", {"pipetteId": pipette.pipette_id})
+      await self._execute_command("dropTipInPlace", {"pipetteId": pipette.pipette_id})
     else:
       rack = self._require_itemized_parent(target)
       labware_id = await self._ensure_labware_loaded(rack)
@@ -135,7 +135,7 @@ async def drop_tips(
           "origin": "top",
           "offset": {"x": offset.x, "y": offset.y, "z": offset.z},
         }
-      await self.execute_command("dropTip", params)
+      await self._execute_command("dropTip", params)
     for ch, spot in zip(use_channels, tip_spots):
       tip = self._channel_tips[ch]
       if (
@@ -174,7 +174,7 @@ async def aspirate(
     well_location = self._well_location(offsets, liquid_height)
     if well_location is not None:
       params["wellLocation"] = well_location
-    await self.execute_command("aspirate", params)
+    await self._execute_command("aspirate", params)
     if does_volume_tracking():
       for well, vol in zip(resources, vols):
         well.tracker.remove_liquid(vol)
@@ -206,7 +206,7 @@ async def dispense(
     well_location = self._well_location(offsets, liquid_height)
     if well_location is not None:
       params["wellLocation"] = well_location
-    await self.execute_command("dispense", params)
+    await self._execute_command("dispense", params)
     if does_volume_tracking():
       for well, vol in zip(resources, vols):
         well.tracker.add_liquid(vol)
@@ -249,7 +249,7 @@ async def _ensure_labware_loaded(self, resource: Resource) -> str:
     load_name = self._ot_load_name(resource)
     labware_id = uuid.uuid4().hex[:12]
 
-    result = await self.execute_command(
+    result = await self._execute_command(
       "loadLabware",
       {
         "loadName": load_name,
diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py
index cea7e0419ec..584aa8aef18 100644
--- a/pylabrobot/opentrons/robot.py
+++ b/pylabrobot/opentrons/robot.py
@@ -37,6 +37,10 @@ class OpentronsRobot(abc.ABC):
 
   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:
@@ -155,7 +159,7 @@ async def _cancel_run(self) -> None:
 
   # --- Command Execution ---
 
-  async def execute_command(
+  async def _execute_command(
     self,
     command_type: str,
     params: Dict[str, Any],
@@ -246,22 +250,6 @@ def _parse_pipettes(self, instruments_data: Dict[str, Any]) -> List[PipetteInfo]
       )
     return pipettes
 
-  def check_gripper(self, instruments_data: Dict[str, Any]) -> bool:
-    """Check if a gripper is attached."""
-    for instrument in instruments_data.get("data", []):
-      if instrument.get("instrumentType") == "gripper":
-        return True
-    return False
-
-  async def get_modules(self) -> List[Dict[str, Any]]:
-    """Query connected modules via GET /modules.
-
-    Returns the module data dict for each USB-connected module. Passive
-    hardware (magnetic block, waste chute) is not discoverable.
-    """
-    data = await self._get("/modules")
-    return cast(List[Dict[str, Any]], data.get("data", []))
-
   # --- Pipette Loading ---
 
   async def _load_pipette(self, pipette_name: str, mount: str) -> str:
@@ -271,7 +259,7 @@ async def _load_pipette(self, pipette_name: str, mount: str) -> str:
     commands (pickUpTip, aspirateInPlace, moveToCoordinates, etc.).
     Must be called after _create_run().
     """
-    result = await self.execute_command(
+    result = await self._execute_command(
       "loadPipette",
       {"pipetteName": pipette_name, "mount": mount},
       wait=True,
@@ -289,34 +277,7 @@ async def _load_pipette(self, pipette_name: str, mount: str) -> str:
 
   async def home(self) -> Dict[str, Any]:
     """Home all axes. The gantry moves to the rear-left-top."""
-    return await self.execute_command("home", {})
-
-  # --- Movement Commands ---
-
-  async def move_to_coordinates(
-    self,
-    pipette_id: str,
-    x: float,
-    y: float,
-    z: float,
-    minimum_z_height: Optional[float] = None,
-    speed: Optional[float] = None,
-  ) -> Dict[str, Any]:
-    """Move a pipette to absolute deck coordinates.
-
-    The robot automatically arcs to a safe Z height before lateral
-    movement (unless forceDirect=True, which we never set).
-    The minimumZHeight parameter raises the arc if needed.
-    """
-    params: Dict[str, Any] = {
-      "pipetteId": pipette_id,
-      "coordinates": {"x": x, "y": y, "z": z},
-    }
-    if minimum_z_height is not None:
-      params["minimumZHeight"] = minimum_z_height
-    if speed is not None:
-      params["speed"] = speed
-    return await self.execute_command("moveToCoordinates", params)
+    return await self._execute_command("home", {})
 
   async def _discover_pipette(self) -> PipetteInfo:
     data = await self._get_instruments()

From f2889e938fb40f8a5506ea07e28435f83cfca9b4 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 10:28:30 +0200
Subject: [PATCH 12/13] =?UTF-8?q?style(opentrons):=20format=20with=20the?=
 =?UTF-8?q?=20repo's=20ruff=20(0.15.4)=20=E2=80=94=20CI-gate=20clean?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pylabrobot/opentrons/flex.py  | 6 +++---
 pylabrobot/opentrons/robot.py | 2 +-
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/pylabrobot/opentrons/flex.py b/pylabrobot/opentrons/flex.py
index 9613f835415..7dd625be3ad 100644
--- a/pylabrobot/opentrons/flex.py
+++ b/pylabrobot/opentrons/flex.py
@@ -66,9 +66,9 @@ def _require_pipette(self) -> PipetteInfo:
   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)."
+    assert isinstance(parent, ItemizedResource), (
+      f"'{item.name}' has no itemized parent resource (rack/plate)."
+    )
     return parent
 
   async def pick_up_tips(
diff --git a/pylabrobot/opentrons/robot.py b/pylabrobot/opentrons/robot.py
index 584aa8aef18..ad0081bfc3e 100644
--- a/pylabrobot/opentrons/robot.py
+++ b/pylabrobot/opentrons/robot.py
@@ -214,7 +214,7 @@ async def _execute_command(
       elif status == "failed":
         error = cmd_data.get("error", {})
         raise RuntimeError(
-          f"Opentrons command '{command_type}' failed: " f"{error.get('detail', error)}"
+          f"Opentrons command '{command_type}' failed: {error.get('detail', error)}"
         )
       await asyncio.sleep(0.2)
 

From 6ee378e5af672c92b59d53f2a0e33d9b68783613 Mon Sep 17 00:00:00 2001
From: vcjdeboer <vcjdeboer@gmail.com>
Date: Thu, 23 Jul 2026 10:53:51 +0200
Subject: [PATCH 13/13] style(opentrons): sort imports (ruff check --select I)
 for CI lint gate

---
 pylabrobot/opentrons/__init__.py                 | 5 +++--
 pylabrobot/resources/opentrons/__init__.py       | 2 +-
 pylabrobot/resources/opentrons/flex_tip_racks.py | 1 -
 3 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/pylabrobot/opentrons/__init__.py b/pylabrobot/opentrons/__init__.py
index 8b914f0012b..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,
@@ -5,5 +8,3 @@
   OpentronsTemperatureModuleUSBTemperatureBackend,
   OpentronsTemperatureModuleV2,
 )
-from pylabrobot.opentrons.robot import OpentronsError, OpentronsRobot, PipetteInfo
-from pylabrobot.opentrons.flex import OpentronsFlex
diff --git a/pylabrobot/resources/opentrons/__init__.py b/pylabrobot/resources/opentrons/__init__.py
index a9e895898e2..bf2cff7fb50 100644
--- a/pylabrobot/resources/opentrons/__init__.py
+++ b/pylabrobot/resources/opentrons/__init__.py
@@ -1,7 +1,7 @@
 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
 from .tip_racks import *
-from .flex_tip_racks import *
diff --git a/pylabrobot/resources/opentrons/flex_tip_racks.py b/pylabrobot/resources/opentrons/flex_tip_racks.py
index c329de10b02..73ef3c346f4 100644
--- a/pylabrobot/resources/opentrons/flex_tip_racks.py
+++ b/pylabrobot/resources/opentrons/flex_tip_racks.py
@@ -18,7 +18,6 @@
 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.