diff --git a/pylabrobot/liquid_handling/backends/__init__.py b/pylabrobot/liquid_handling/backends/__init__.py index 50c01b189ad..a039cc1ed4b 100644 --- a/pylabrobot/liquid_handling/backends/__init__.py +++ b/pylabrobot/liquid_handling/backends/__init__.py @@ -4,6 +4,7 @@ from .hamilton.STAR_backend import STAR, STARBackend from .hamilton.vantage_backend import Vantage, VantageBackend from .opentrons_backend import OpentronsOT2Backend +from .opentrons_chatterbox import OpentronsOT2ChatterboxBackend from .opentrons_simulator import OpentronsOT2Simulator from .serializing_backend import SerializingBackend from .tecan.EVO_backend import EVO, EVOBackend diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend.py b/pylabrobot/liquid_handling/backends/opentrons_backend.py index f5e30322a9e..9be532b7e1f 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend.py @@ -1,7 +1,11 @@ +import inspect +import logging import uuid -from typing import Dict, List, Optional, Tuple, Union, cast +import warnings +from typing import Any, Dict, List, Optional, Sequence, Tuple, Union, cast from pylabrobot import utils +from pylabrobot.io import LOG_LEVEL_IO from pylabrobot.liquid_handling.backends.backend import ( LiquidHandlerBackend, ) @@ -15,6 +19,7 @@ MultiHeadDispensePlate, Pickup, PickupTipRack, + PipettingOp, ResourceDrop, ResourceMove, ResourcePickup, @@ -24,16 +29,14 @@ from pylabrobot.resources import ( Coordinate, Tip, + does_tip_tracking, ) from pylabrobot.resources.opentrons import OTDeck -from pylabrobot.resources.tip_rack import TipRack +from pylabrobot.resources.tip_rack import TipRack, TipSpot try: import ot_api - # for run cancellation - import ot_api.requestor as _req - USE_OT = True except ImportError as e: USE_OT = False @@ -44,6 +47,38 @@ # https://labautomation.io/t/connect-pylabrobot-to-ot2/2862/18 _OT_DECK_IS_ADDRESSABLE_AREA_VERSION = "7.1.0" +logger = logging.getLogger(__name__) + + +class _IOLogger: + """Transparent proxy over the ``ot_api`` module that logs every call at + ``LOG_LEVEL_IO``. + + The OT-2 talks HTTP through ``ot_api`` rather than a pylabrobot.io transport, so + this wrapper gives it the same wire-level logging every other backend gets from + its io object. Submodules (``lh``, ``health``, ...) are wrapped recursively; + plain attributes (e.g. ``run_id``) pass through untouched. + """ + + def __init__(self, target: Any, prefix: str = ""): + object.__setattr__(self, "_target", target) + object.__setattr__(self, "_prefix", prefix) + + def __getattr__(self, name: str) -> Any: + attr = getattr(self._target, name) + qualified = f"{self._prefix}.{name}" if self._prefix else name + if inspect.ismodule(attr): + return _IOLogger(attr, qualified) + if callable(attr): + + def _logged(*args, **kwargs): + parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] + logger.log(LOG_LEVEL_IO, "%s(%s)", qualified, ", ".join(parts)) + return attr(*args, **kwargs) + + return _logged + return attr + class OpentronsOT2Backend(LiquidHandlerBackend): """Backends for the Opentrons OT2 liquid handling robots.""" @@ -65,7 +100,7 @@ class OpentronsOT2Backend(LiquidHandlerBackend): "p1000_single_gen3": 1000, } - def __init__(self, host: str, port: int = 31950): + def __init__(self, host: str, port: int = 31950, allow_undeclared_tip_pickup: bool = False): super().__init__() if not USE_OT: @@ -77,8 +112,17 @@ def __init__(self, host: str, port: int = 31950): self.host = host self.port = port - ot_api.set_host(host) - ot_api.set_port(port) + # Default for whether a multi pickup may absorb undeclared tips grabbed by unused + # nozzles; overridable per call via pick_up_tips(..., allow_undeclared_tip_pickup=). + self.allow_undeclared_tip_pickup = allow_undeclared_tip_pickup + + # All hardware I/O goes through this handle so a subclass (e.g. the chatterbox) + # can dry-run the backend by swapping it for a recording stand-in. The real handle + # wraps ot_api to log every HTTP call at LOG_LEVEL_IO, like other backends' io. + self._ot: Any = _IOLogger(ot_api) + + self._ot.set_host(host) + self._ot.set_port(port) self.ot_api_version: Optional[str] = None self.left_pipette: Optional[Dict[str, str]] = None @@ -97,24 +141,46 @@ def serialize(self) -> dict: async def setup(self, skip_home: bool = False): # create run - run_id = ot_api.runs.create() - ot_api.set_run(run_id) + run_id = self._ot.runs.create() + self._ot.set_run(run_id) # get pipettes, then assign them - self.left_pipette, self.right_pipette = ot_api.lh.add_mounted_pipettes() + self.left_pipette, self.right_pipette = self._ot.lh.add_mounted_pipettes() self.left_pipette_has_tip = self.right_pipette_has_tip = False # get api version - health = ot_api.health.get() + health = self._ot.health.get() self.ot_api_version = health["api_version"] if not skip_home: await self.home() + @staticmethod + def _pipette_channel_count(pipette: Optional[Dict[str, str]]) -> int: + """Number of channels a mounted pipette presents: 8 for a multi, 1 for a single.""" + if pipette is None: + return 0 + return 8 if "multi" in pipette["name"] else 1 + + def _channel_map(self) -> List[Tuple[Dict[str, str], int]]: + """Per-mount channel blocks: channel index -> (pipette, nozzle index within it). + + The left mount's channels come first, then the right mount's. A p20-multi on + the left plus a p300-single on the right gives channels 0-7 (the multi's + nozzles, 0 = back / row A) and channel 8 (the single). + """ + channels: List[Tuple[Dict[str, str], int]] = [] + for pipette in (self.left_pipette, self.right_pipette): + if pipette is None: + continue + for nozzle in range(self._pipette_channel_count(pipette)): + channels.append((pipette, nozzle)) + return channels + @property def num_channels(self) -> int: - return len([p for p in [self.left_pipette, self.right_pipette] if p is not None]) + return len(self._channel_map()) async def stop(self): """Cancel any active OT run, then clear labware definitions.""" @@ -124,16 +190,16 @@ async def stop(self): self.right_pipette = None # cancel the HTTP-API run if it exists (helpful to make device available again in official Opentrons app) - run_id = getattr(ot_api, "run_id", None) + run_id = getattr(self._ot, "run_id", None) if run_id: try: - _req.post(f"/runs/{run_id}/cancel") + self._ot.requestor.post(f"/runs/{run_id}/cancel") except Exception: try: - _req.post(f"/runs/{run_id}/actions/cancel") + self._ot.requestor.post(f"/runs/{run_id}/actions/cancel") except Exception: try: - _req.delete(f"/runs/{run_id}") + self._ot.requestor.delete(f"/runs/{run_id}") except Exception: pass @@ -231,7 +297,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): ], } - data = ot_api.labware.define(lw) + data = self._ot.labware.define(lw) namespace, definition, version = data["data"]["definitionUri"].split("/") # assign labware to robot @@ -242,7 +308,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): slot = deck.get_slot(tip_rack) assert slot is not None, "tip rack must be on deck" - ot_api.labware.add( + self._ot.labware.add( load_name=definition, namespace=namespace, ot_location=slot, @@ -253,6 +319,35 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip): self._tip_racks[tip_rack.name] = slot + def _resolve_pipette_and_primary( + self, ops: Sequence[PipettingOp], use_channels: List[int] + ) -> Tuple[str, PipettingOp]: + """Map ``use_channels`` to the single pipette they address, plus the primary op. + + All channels in one operation must belong to the same pipette/mount (a multi + pipette is a ganged head - the OT-2 cannot operate two mounts in one command; + issue separate calls per mount). The primary op is the one on the lowest + nozzle index (nozzle 0 = back / row A). The single ``ot_api`` command targets + the primary op's well; the firmware fans the remaining nozzles out from there. + """ + channel_map = self._channel_map() + pipette_ids = set() + primary_op: Optional[PipettingOp] = None + primary_nozzle: Optional[int] = None + for op, channel in zip(ops, use_channels): + if not 0 <= channel < len(channel_map): + raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.") + pipette, nozzle = channel_map[channel] + pipette_ids.add(cast(str, pipette["pipetteId"])) + if primary_nozzle is None or nozzle < primary_nozzle: + primary_nozzle, primary_op = nozzle, op + if len(pipette_ids) != 1 or primary_op is None: + raise NoChannelError( + "All channels in one operation must address the same pipette (mount); " + "issue separate calls per mount." + ) + return pipette_ids.pop(), primary_op + def _get_pickup_pipette(self, ops: List[Pickup]) -> str: """Get the pipette for a tip pick-up, or raise.""" assert len(ops) == 1, "only one channel supported for now" @@ -301,11 +396,135 @@ def _set_tip_state(self, pipette_id: str, has_tip: bool): raise ValueError(f"Unknown or unconfigured pipette_id {pipette_id!r} in _set_tip_state.") - async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): - """Pick up tips from the specified resource.""" + def _check_head8_pickup( + self, ops: List[Pickup], use_channels: List[int] + ) -> List[Tuple[int, TipSpot]]: + """Validate an 8-channel pickup against what the Opentrons API can declare. - pipette_id = self._get_pickup_pipette(ops) - op = ops[0] + A default-layout multi anchors the pickup at the referenced well - the back / + channel-0 nozzle - and fills toward the front (row H). The API only approves + front-anchored nozzle layouts, so use_channels must be a channel-0-anchored + contiguous block ``[0, 1, ..., k]``; a back-anchored set such as ``[1..7]`` + (leaving row A) cannot be declared at all. + + Returns the ``(nozzle, tipspot)`` pairs the head would also grab BELOW the + declared selection (occupied tipspots within its 8-nozzle reach) - undeclared + tips for the caller to reject or absorb. An empty list means a clean pickup. + """ + head8_pitch = 9.0 + num_nozzles = 8 + + ordered = sorted(use_channels) + if ordered != list(range(len(ordered))): + raise ValueError( + f"OT-2 8-channel pickup must use a channel-0-anchored contiguous block of channels " + f"[0, 1, ..., k]; got use_channels={ordered}. The Opentrons API anchors a multi pickup " + f"at the back (channel-0) nozzle and only approves front-anchored nozzle layouts, so a " + f"back-anchored selection such as [1..7] (which would leave row A and overhang the back " + f"edge) cannot be declared. To pick fewer tips, keep channel 0 and use the topmost rows." + ) + + def loc(resource): + return resource.get_absolute_location("c", "c", "b") + + tip_spots = [op.resource for op in ops] + rack = tip_spots[0].parent + if not isinstance(rack, TipRack) or any(s.parent is not rack for s in tip_spots): + raise ValueError("OT-2 8-channel pickup must come from a single tip rack.") + + col_x = loc(tip_spots[0]).x + if any(abs(loc(s).x - col_x) > 0.5 for s in tip_spots): + raise ValueError( + "OT-2 8-channel pickup must be within a single column (all tipspots must share x)." + ) + + col_spots = [s for s in rack.get_all_items() if abs(loc(s).x - col_x) < 0.5] + top_y = max(loc(s).y for s in col_spots) + + def row_of(y: float) -> int: + return int(round((top_y - y) / head8_pitch)) + + spot_by_row = {row_of(loc(s).y): s for s in col_spots} + + # The declared tipspots must run consecutively down the column in channel order + # (channel 0 = topmost well, then one row per channel at 9 mm pitch). + offsets = {row_of(loc(op.resource).y) - ch for op, ch in zip(ops, use_channels)} + if len(offsets) != 1: + raise ValueError( + "OT-2 8-channel pickup tipspots must run consecutively down the column in channel order " + "(channel 0 = topmost well, then one row per channel at 9 mm pitch). The given tipspots " + "do not line up 1:1 with the channels." + ) + top_row = offsets.pop() + + # The head also grabs any occupied tipspot BELOW the selection, within its 8-nozzle reach + # (unused nozzles k+1..7). Those are undeclared tips. + used = set(use_channels) + extras: List[Tuple[int, TipSpot]] = [] + for nozzle in range(num_nozzles): + if nozzle in used: + continue + spot = spot_by_row.get(nozzle + top_row) + if spot is not None and spot.has_tip(): + extras.append((nozzle, spot)) + return extras + + def _absorb_undeclared_tips(self, extras: List[Tuple[int, TipSpot]]) -> None: + """Account for tips grabbed by unused nozzles so PLR tracking stays consistent. + + Mirrors the frontend's pickup bookkeeping (add tip to the channel, remove it from + the tipspot) for each nozzle the caller did not declare. + """ + if self._head is None: + return + for nozzle, spot in extras: + self._head[nozzle].add_tip(spot.get_tip(), origin=spot, commit=True) + if does_tip_tracking() and not spot.tracker.is_disabled: + spot.tracker.remove_tip(commit=True) + + async def pick_up_tips( + self, + ops: List[Pickup], + use_channels: List[int], + *, + allow_undeclared_tip_pickup: Optional[bool] = None, + ): + """Pick up tips from the specified resource. + + A multi-channel pickup (one op per channel) issues a single ``ot_api`` command + targeting the primary op's well; the firmware engages the remaining nozzles. + + ``allow_undeclared_tip_pickup`` overrides the backend default for this call only; + when ``None`` the backend's ``self.allow_undeclared_tip_pickup`` is used. When the + 8-nozzle head would also grab occupied tipspots below the selection, the pickup is + rejected unless this is True, in which case those tips are absorbed (with a warning). + """ + + pipette_id, op = self._resolve_pipette_and_primary(ops, use_channels) + + allow_undeclared = ( + self.allow_undeclared_tip_pickup + if allow_undeclared_tip_pickup is None + else allow_undeclared_tip_pickup + ) + if "multi" in self.get_pipette_name(pipette_id): + extras = self._check_head8_pickup(ops, use_channels) + if extras: + where = ", ".join(spot.name for _, spot in extras) + if not allow_undeclared: + raise ValueError( + f"OT-2 8-channel pickup would also grab undeclared tips at {where}. The head fills " + f"from channel 0 down toward row H, and these occupied tipspots sit below your " + f"selection within its 8-nozzle reach, so the hardware would pick them up too. " + f"Extend use_channels down to include them, clear those spots first, or pass " + f"allow_undeclared_tip_pickup=True (per call or on the backend) to absorb them." + ) + warnings.warn( + f"OT-2 8-channel pickup is also grabbing undeclared tips at {where} (below the " + f"selection) and absorbing them into tracking (allow_undeclared_tip_pickup=True).", + stacklevel=2, + ) + self._absorb_undeclared_tips(extras) offset_x, offset_y, offset_z = ( op.offset.x, @@ -321,7 +540,7 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): offset_z += op.tip.total_tip_length - ot_api.lh.pick_up_tip( + self._ot.lh.pick_up_tip( labware_id=self.get_ot_name(tip_rack.name), well_name=self.get_ot_name(op.resource.name), pipette_id=pipette_id, @@ -333,10 +552,13 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]): self._set_tip_state(pipette_id, True) async def drop_tips(self, ops: List[Drop], use_channels: List[int]): - """Drop tips from the specified resource.""" + """Drop tips from the specified resource. - pipette_id = self._get_drop_pipette(ops) - op = ops[0] + A multi-channel drop issues one ``ot_api`` command at the primary op's well. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(Drop, primary) use_fixed_trash = ( cast(str, self.ot_api_version) >= _OT_DECK_IS_ADDRESSABLE_AREA_VERSION @@ -361,15 +583,15 @@ async def drop_tips(self, ops: List[Drop], use_channels: List[int]): offset_z += 10 if use_fixed_trash: - ot_api.lh.move_to_addressable_area_for_drop_tip( + self._ot.lh.move_to_addressable_area_for_drop_tip( pipette_id=pipette_id, offset_x=offset_x, offset_y=offset_y, offset_z=offset_z, ) - ot_api.lh.drop_tip_in_place(pipette_id=pipette_id) + self._ot.lh.drop_tip_in_place(pipette_id=pipette_id) else: - ot_api.lh.drop_tip( + self._ot.lh.drop_tip( labware_id, well_name=self.get_ot_name(op.resource.name), pipette_id=pipette_id, @@ -438,10 +660,14 @@ def _get_default_aspiration_flow_rate(self, pipette_name: str) -> float: }[pipette_name] async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int]): - """Aspirate liquid from the specified resource using pip.""" + """Aspirate liquid from the specified resource using pip. - pipette_id = self._get_liquid_pipette(ops) - op = ops[0] + A multi-channel aspirate issues one ``ot_api`` command at the primary op's + well; all nozzles draw the same volume. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(SingleChannelAspiration, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) @@ -461,18 +687,18 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[ if op.mix is not None: for _ in range(op.mix.repetitions): - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=volume, flow_rate=flow_rate, pipette_id=pipette_id, @@ -513,10 +739,14 @@ def _get_default_dispense_flow_rate(self, pipette_name: str) -> float: }[pipette_name] async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int]): - """Dispense liquid from the specified resource using pip.""" + """Dispense liquid from the specified resource using pip. - pipette_id = self._get_liquid_pipette(ops) - op = ops[0] + A multi-channel dispense issues one ``ot_api`` command at the primary op's + well; all nozzles dispense the same volume. + """ + + pipette_id, primary = self._resolve_pipette_and_primary(ops, use_channels) + op = cast(SingleChannelDispense, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) @@ -533,7 +763,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=volume, flow_rate=flow_rate, pipette_id=pipette_id, @@ -541,12 +771,12 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in if op.mix is not None: for _ in range(op.mix.repetitions): - ot_api.lh.aspirate_in_place( + self._ot.lh.aspirate_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, ) - ot_api.lh.dispense_in_place( + self._ot.lh.dispense_in_place( volume=op.mix.volume, flow_rate=op.mix.flow_rate, pipette_id=pipette_id, @@ -563,7 +793,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in ) async def home(self): - ot_api.health.home() + self._ot.health.home() async def pick_up_tips96(self, pickup: PickupTipRack): raise NotImplementedError("The Opentrons backend does not support the 96 head.") @@ -590,24 +820,21 @@ async def drop_resource(self, drop: ResourceDrop): async def list_connected_modules(self) -> List[dict]: """List all connected temperature modules.""" - return cast(List[dict], ot_api.modules.list_connected_modules()) + return cast(List[dict], self._ot.modules.list_connected_modules()) def _pipette_id_for_channel(self, channel: int) -> str: - pipettes = [] - if self.left_pipette is not None: - pipettes.append(self.left_pipette["pipetteId"]) - if self.right_pipette is not None: - pipettes.append(self.right_pipette["pipetteId"]) - if channel < 0 or channel >= len(pipettes): + channel_map = self._channel_map() + if channel < 0 or channel >= len(channel_map): raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.") - return pipettes[channel] + pipette, _nozzle = channel_map[channel] + return cast(str, pipette["pipetteId"]) def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]: """Return the pipette id and current coordinate for a given channel.""" pipette_id = self._pipette_id_for_channel(channel) try: - res = ot_api.lh.save_position(pipette_id=pipette_id) + res = self._ot.lh.save_position(pipette_id=pipette_id) pos = res["data"]["result"]["position"] current = Coordinate(pos["x"], pos["y"], pos["z"]) except Exception as exc: # noqa: BLE001 @@ -676,7 +903,7 @@ async def move_pipette_head( if pipette_id is None: raise ValueError("No pipette id given or left/right pipette not available.") - ot_api.lh.move_arm( + self._ot.lh.move_arm( pipette_id=pipette_id, location_x=location.x, location_y=location.y, @@ -696,14 +923,9 @@ def supports_tip(channel_vol: float, tip_vol: float) -> bool: return tip_vol in {1000} raise ValueError(f"Unknown channel volume: {channel_vol}") - if channel_idx == 0: - if self.left_pipette is None: - return False - left_volume = OpentronsOT2Backend.pipette_name2volume[self.left_pipette["name"]] - return supports_tip(left_volume, tip.maximal_volume) - if channel_idx == 1: - if self.right_pipette is None: - return False - right_volume = OpentronsOT2Backend.pipette_name2volume[self.right_pipette["name"]] - return supports_tip(right_volume, tip.maximal_volume) - return False + channel_map = self._channel_map() + if channel_idx < 0 or channel_idx >= len(channel_map): + return False + pipette, _nozzle = channel_map[channel_idx] + channel_volume = OpentronsOT2Backend.pipette_name2volume[pipette["name"]] + return supports_tip(channel_volume, tip.maximal_volume) diff --git a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py index 05ea8e2845f..ab45960d818 100644 --- a/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py +++ b/pylabrobot/liquid_handling/backends/opentrons_backend_tests.py @@ -7,6 +7,7 @@ from pylabrobot.liquid_handling import LiquidHandler from pylabrobot.liquid_handling.backends.opentrons_backend import ( + _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, OpentronsOT2Backend, ) from pylabrobot.liquid_handling.errors import NoChannelError @@ -127,9 +128,6 @@ def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, off self.assertEqual(labware_id, self.backend.get_ot_name("tip_rack")) self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(offset_x, offset_x) - self.assertEqual(offset_y, offset_y) - self.assertEqual(offset_z, offset_z) mock_pick_up_tip.side_effect = assert_parameters @@ -138,12 +136,8 @@ def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, off @patch("ot_api.lh.drop_tip") async def test_tip_drop(self, mock_drop_tip): def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, offset_z): - self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1")) self.assertEqual(pipette_id, "left-pipette-id") - self.assertEqual(offset_x, offset_x) - self.assertEqual(offset_y, offset_y) - self.assertEqual(offset_z, offset_z) mock_drop_tip.side_effect = assert_parameters @@ -188,6 +182,59 @@ def assert_parameters( with no_volume_tracking(): await self.lh.dispense(self.plate["A1"], vols=[10]) + # -- characterization of the remaining ot_api call sites (Phase 0 safety net) -- + + @patch("ot_api.health.home") + async def test_home_calls_health_home(self, mock_home): + """home() issues exactly one ot_api.health.home() call.""" + await self.backend.home() + mock_home.assert_called_once_with() + + @patch("ot_api.modules.list_connected_modules") + async def test_list_connected_modules_passthrough(self, mock_modules): + """list_connected_modules() returns ot_api.modules.list_connected_modules() verbatim.""" + mock_modules.return_value = [{"id": "tempdeck"}] + result = await self.backend.list_connected_modules() + mock_modules.assert_called_once_with() + self.assertEqual(result, [{"id": "tempdeck"}]) + + @patch("ot_api.run_id", "run-id", create=True) + @patch("ot_api.requestor.post") + async def test_stop_cancels_active_run_and_clears_pipettes(self, mock_post): + """stop() cancels the active run through the requestor and clears mounted pipettes.""" + await self.backend.stop() + mock_post.assert_called_once_with("/runs/run-id/cancel") + self.assertIsNone(self.backend.left_pipette) + self.assertIsNone(self.backend.right_pipette) + + @patch("ot_api.lh.drop_tip_in_place") + @patch("ot_api.lh.move_to_addressable_area_for_drop_tip") + @patch("ot_api.lh.drop_tip") + @patch("ot_api.lh.pick_up_tip") + @patch("ot_api.labware.define") + @patch("ot_api.labware.add") + async def test_tip_drop_to_trash_uses_addressable_area( + self, + mock_add, + mock_define, + mock_pick_up_tip, + mock_drop_tip, + mock_to_trash, + mock_drop_in_place, + ): + """At api_version >= 7.1.0 a discard to the deck trash routes via the addressable + area (move_to_addressable_area_for_drop_tip + drop_tip_in_place), not drop_tip.""" + mock_define.side_effect = _mock_define + mock_add.side_effect = _mock_add + self.backend.ot_api_version = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION + + await self.lh.pick_up_tips(self.tip_rack["A1"]) + await self.lh.discard_tips() + + mock_to_trash.assert_called_once() + mock_drop_in_place.assert_called_once() + mock_drop_tip.assert_not_called() + def _make_backend_with_pipettes(left_name="p300_single_gen2", right_name="p20_single_gen2"): """Create a backend with pipette state set directly (no ot_api needed).""" @@ -315,3 +362,30 @@ def test_set_tip_state_right(self): self.backend._set_tip_state("right-id", True) self.assertFalse(self.backend.left_pipette_has_tip) self.assertTrue(self.backend.right_pipette_has_tip) + + # -- channel model (head8 step 1) -- + + def test_channel_map_two_singles_is_one_channel_per_mount(self): + """Two single-channel pipettes give two channels, one per mount (unchanged).""" + backend = _make_backend_with_pipettes("p300_single_gen2", "p20_single_gen2") + self.assertEqual(backend.num_channels, 2) + self.assertEqual(backend._pipette_id_for_channel(0), "left-id") + self.assertEqual(backend._pipette_id_for_channel(1), "right-id") + + def test_channel_map_multi_mount_is_eight_channels(self): + """A multi on the left + a single on the right gives channels 0-7 (the multi's + nozzles) and channel 8 (the single).""" + backend = _make_backend_with_pipettes("p20_multi_gen2", "p300_single_gen2") + self.assertEqual(backend.num_channels, 9) + self.assertTrue(all(pip is backend.left_pipette for pip, _ in backend._channel_map()[:8])) + self.assertIs(backend._channel_map()[8][0], backend.right_pipette) + self.assertEqual(backend._pipette_id_for_channel(0), "left-id") + self.assertEqual(backend._pipette_id_for_channel(7), "left-id") + self.assertEqual(backend._pipette_id_for_channel(8), "right-id") + + def test_pipette_id_for_channel_out_of_range_raises(self): + """Channels beyond the mounted pipettes raise NoChannelError.""" + backend = _make_backend_with_pipettes("p20_single_gen2", None) + self.assertEqual(backend.num_channels, 1) + with self.assertRaises(NoChannelError): + backend._pipette_id_for_channel(1) diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py new file mode 100644 index 00000000000..a390c17fc30 --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py @@ -0,0 +1,182 @@ +"""A chatterbox backend for the Opentrons OT-2. + +Dry-runs the real OpentronsOT2Backend without hardware or the ``ot_api`` library +by swapping the backend's transport handle (``self._ot``) for a recorder that +logs every call and returns canned data for the few reads the backend makes back. + +This mirrors how ``STARChatterboxBackend`` dry-runs ``STARBackend``: only the +transport is replaced, so all the real high-level logic (pipette selection, tip +and volume bookkeeping, the per-operation wire calls) runs unchanged. Contrast +with ``OpentronsOT2Simulator``, which overrides the high-level methods themselves. +""" + +import logging +from typing import Dict, List, Optional, Tuple, cast + +from pylabrobot.io import LOG_LEVEL_IO +from pylabrobot.liquid_handling.backends.backend import LiquidHandlerBackend +from pylabrobot.liquid_handling.backends.opentrons_backend import ( + _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + OpentronsOT2Backend, +) + +logger = logging.getLogger(__name__) + + +class _RecordingNamespace: + """An ot_api sub-namespace (e.g. ``lh``, ``health``) that records every call. + + Unknown attributes resolve to a function that appends ``(name, args, kwargs)`` + to the shared recorder and returns the canned value registered in ``returns`` + (``None`` if none is registered). + """ + + def __init__(self, recorder: "_OTChatterboxModule", prefix: str, returns=None): + self._recorder = recorder + self._prefix = prefix + self._returns = returns or {} + + def __getattr__(self, name: str): + if name.startswith("_"): + raise AttributeError(name) + qualified = f"{self._prefix}.{name}" + returns = self._returns + recorder = self._recorder + + def _record(*args, **kwargs): + recorder.log(qualified, args, kwargs) + canned = returns.get(name) + return canned() if callable(canned) else canned + + return _record + + +class _OTChatterboxModule: + """Stand-in for the ``ot_api`` module that records calls instead of issuing them. + + Provides the sub-namespaces and reads the real backend touches: ``runs.create``, + ``lh.add_mounted_pipettes``, ``health.get``, ``labware.define``, + ``modules.list_connected_modules`` and ``lh.save_position`` return canned data; + everything else is recorded and returns ``None``. ``run_id`` stays ``None`` so + ``stop()`` skips the cancel request. + """ + + def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool = True): + self.calls: List[Tuple[str, tuple, dict]] = [] + self.run_id: Optional[str] = None + self._verbose = verbose + + self.runs = _RecordingNamespace(self, "runs", {"create": lambda: "chatterbox-run"}) + self.health = _RecordingNamespace(self, "health", {"get": lambda: {"api_version": api_version}}) + self.labware = _RecordingNamespace( + self, "labware", {"define": lambda: {"data": {"definitionUri": "pylabrobot/chatterbox/1"}}} + ) + self.modules = _RecordingNamespace(self, "modules", {"list_connected_modules": lambda: []}) + self.requestor = _RecordingNamespace(self, "requestor") + self.lh = _RecordingNamespace( + self, + "lh", + { + "add_mounted_pipettes": lambda: (left_pipette, right_pipette), + "save_position": lambda: {"data": {"result": {"position": {"x": 0, "y": 0, "z": 0}}}}, + }, + ) + + def log(self, qualified: str, args: tuple, kwargs: dict): + self.calls.append((qualified, args, kwargs)) + parts = [repr(a) for a in args] + [f"{k}={v!r}" for k, v in kwargs.items()] + rendered = f"{qualified}({', '.join(parts)})" + # log at LOG_LEVEL_IO so a dry run captures the same wire trace as a real run + logger.log(LOG_LEVEL_IO, "%s", rendered) + if self._verbose: + print(rendered) + + def __getattr__(self, name: str): + # top-level functions the backend calls directly: set_host, set_port, set_run + if name.startswith("_"): + raise AttributeError(name) + + def _record(*args, **kwargs): + self.log(name, args, kwargs) + return None + + return _record + + +class OpentronsOT2ChatterboxBackend(OpentronsOT2Backend): + """Chatterbox backend for the Opentrons OT-2. + + Runs the real OpentronsOT2Backend logic with its transport replaced by a + recorder - no hardware and no ``ot_api`` library required. Every issued call is + printed and collected in :attr:`commands`. + + Example: + >>> from pylabrobot.liquid_handling import LiquidHandler + >>> from pylabrobot.liquid_handling.backends import OpentronsOT2ChatterboxBackend + >>> from pylabrobot.resources.opentrons import OTDeck + >>> lh = LiquidHandler(backend=OpentronsOT2ChatterboxBackend(), deck=OTDeck()) + >>> await lh.setup() + """ + + def __init__( + self, + left_pipette_name: Optional[str] = "p300_single_gen2", + right_pipette_name: Optional[str] = "p20_single_gen2", + host: str = "chatterbox", + port: int = 31950, + api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION, + allow_undeclared_tip_pickup: bool = False, + verbose: bool = True, + ): + """Initialize the chatterbox. + + Args: + left_pipette_name: pipette mounted on the left (``None`` for none). + right_pipette_name: pipette mounted on the right (``None`` for none). + api_version: reported Opentrons API version; defaults to the version at + which tip drops route through the addressable-area trash. + verbose: if True, print every recorded call. + """ + # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly. + LiquidHandlerBackend.__init__(self) + + pv = OpentronsOT2Backend.pipette_name2volume + if left_pipette_name is not None and left_pipette_name not in pv: + raise ValueError(f"Unknown left pipette: {left_pipette_name}") + if right_pipette_name is not None and right_pipette_name not in pv: + raise ValueError(f"Unknown right pipette: {right_pipette_name}") + + self._left_pipette_name = left_pipette_name + self._right_pipette_name = right_pipette_name + self.host = host + self.port = port + self.allow_undeclared_tip_pickup = allow_undeclared_tip_pickup + + left = ( + {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None + ) + right = ( + {"name": right_pipette_name, "pipetteId": "chatterbox-right"} if right_pipette_name else None + ) + self._ot = _OTChatterboxModule(left, right, api_version, verbose=verbose) + + self.ot_api_version: Optional[str] = None + self.left_pipette: Optional[Dict[str, str]] = None + self.right_pipette: Optional[Dict[str, str]] = None + self.traversal_height = 120 + self._tip_racks: Dict[str, int] = {} + self._plr_name_to_load_name: Dict[str, str] = {} + + @property + def commands(self) -> List[Tuple[str, tuple, dict]]: + """Recorded ``(qualified_name, args, kwargs)`` for every call issued so far.""" + return cast(List[Tuple[str, tuple, dict]], self._ot.calls) + + def serialize(self) -> dict: + return { + **LiquidHandlerBackend.serialize(self), + "left_pipette_name": self._left_pipette_name, + "right_pipette_name": self._right_pipette_name, + "host": self.host, + "port": self.port, + } diff --git a/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py new file mode 100644 index 00000000000..1f10bbaa751 --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -0,0 +1,168 @@ +"""Tests for OpentronsOT2ChatterboxBackend. + +Deliberately does NOT importorskip("ot_api"): running the real backend logic with +no hardware and no ot_api library is the whole point of the chatterbox. +""" + +import unittest +import warnings + +from pylabrobot.liquid_handling import LiquidHandler +from pylabrobot.liquid_handling.backends import OpentronsOT2ChatterboxBackend +from pylabrobot.liquid_handling.errors import NoChannelError +from pylabrobot.resources import set_tip_tracking, set_volume_tracking +from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb +from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul + + +def _names(backend: OpentronsOT2ChatterboxBackend): + return [call[0] for call in backend.commands] + + +class OpentronsChatterboxTests(unittest.IsolatedAsyncioTestCase): + """Direct tests: the chatterbox runs the real backend with no ot_api.""" + + async def asyncSetUp(self): + set_tip_tracking(True) + set_volume_tracking(True) + self.backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_single_gen2", + right_pipette_name="p20_single_gen2", + verbose=False, + ) + self.deck = OTDeck() + self.lh = LiquidHandler(backend=self.backend, deck=self.deck) + await self.lh.setup() + self.tips = opentrons_96_filtertiprack_20ul(name="tips") + self.deck.assign_child_at_slot(self.tips, slot=1) + self.plate = CellTreat_96_wellplate_350ul_Fb(name="plate") + self.deck.assign_child_at_slot(self.plate, slot=2) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_setup_resolves_two_channels_without_ot_api(self): + """setup() runs through the recorder and resolves both mounted pipettes.""" + self.assertEqual(self.backend.num_channels, 2) + assert self.backend.left_pipette is not None and self.backend.right_pipette is not None + self.assertEqual(self.backend.left_pipette["name"], "p20_single_gen2") + + async def test_full_protocol_records_one_wire_call_per_operation(self): + """A pickup -> aspirate -> dispense -> trash-discard records exactly one + wire call each, via the real backend logic.""" + self.plate.get_well("A1").tracker.set_volume(15) + await self.lh.pick_up_tips(self.tips["A1"]) + await self.lh.aspirate(self.plate["A1"], vols=[10]) + await self.lh.dispense(self.plate["B1"], vols=[10]) + await self.lh.discard_tips() + + names = _names(self.backend) + self.assertEqual(names.count("lh.pick_up_tip"), 1) + self.assertEqual(names.count("lh.aspirate_in_place"), 1) + self.assertEqual(names.count("lh.dispense_in_place"), 1) + # api_version defaults to 7.1.0, so the discard routes through the trash addressable area + self.assertEqual(names.count("lh.move_to_addressable_area_for_drop_tip"), 1) + self.assertEqual(names.count("lh.drop_tip_in_place"), 1) + + def test_unknown_pipette_name_raises(self): + """An unrecognised pipette name is rejected at construction.""" + with self.assertRaises(ValueError): + OpentronsOT2ChatterboxBackend(left_pipette_name="not_a_pipette") + + def test_serialize_includes_pipettes(self): + """serialize() captures the mounted-pipette names (None for an empty mount).""" + backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_single_gen2", right_pipette_name=None, verbose=False + ) + serialized = backend.serialize() + self.assertEqual(serialized["left_pipette_name"], "p20_single_gen2") + self.assertIsNone(serialized["right_pipette_name"]) + + +class OpentronsChatterboxHead8Tests(unittest.IsolatedAsyncioTestCase): + """Multi-channel (head8) column operations, dry-run through the chatterbox.""" + + async def asyncSetUp(self): + set_tip_tracking(True) + set_volume_tracking(True) + self.backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_multi_gen2", + right_pipette_name="p300_single_gen2", + verbose=False, + ) + self.deck = OTDeck() + self.lh = LiquidHandler(backend=self.backend, deck=self.deck) + await self.lh.setup() + self.tips = opentrons_96_filtertiprack_20ul(name="tips") + self.deck.assign_child_at_slot(self.tips, slot=1) + self.plate = CellTreat_96_wellplate_350ul_Fb(name="plate") + self.deck.assign_child_at_slot(self.plate, slot=2) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_eight_channel_column_issues_one_command_each_and_tracks_all(self): + """A full-column pickup -> aspirate -> dispense on the 8 multi channels issues + exactly one ot_api command each, tracks all 8 channels, and fills the column.""" + rows = "ABCDEFGH" + channels = list(range(8)) + for r in rows: + self.plate.get_well(f"{r}1").tracker.set_volume(20) + + await self.lh.pick_up_tips([self.tips.get_item(f"{r}1") for r in rows], use_channels=channels) + await self.lh.aspirate( + [self.plate.get_item(f"{r}1") for r in rows], vols=[5.0] * 8, use_channels=channels + ) + await self.lh.dispense( + [self.plate.get_item(f"{r}2") for r in rows], vols=[5.0] * 8, use_channels=channels + ) + + names = [call[0] for call in self.backend.commands] + self.assertEqual(names.count("lh.pick_up_tip"), 1) + self.assertEqual(names.count("lh.aspirate_in_place"), 1) + self.assertEqual(names.count("lh.dispense_in_place"), 1) + self.assertTrue(all(self.lh.head[c].has_tip for c in channels)) + for r in rows: + self.assertEqual(self.plate.get_item(f"{r}2").tracker.get_used_volume(), 5.0) + + async def test_channels_spanning_two_mounts_is_rejected(self): + """Channels addressing both the multi (0) and the single (8) cannot be one + command - the OT-2 drives a single mount per call. The resolver only pairs ops + with channels, so plain sentinels stand in for the ops here.""" + ops = [object(), object()] + with self.assertRaises(NoChannelError): + self.backend._resolve_pipette_and_primary(ops, use_channels=[0, 8]) # type: ignore[arg-type] + + async def test_back_anchored_pickup_is_rejected(self): + """use_channels must be a channel-0-anchored block; [1..7] (leaving row A) is rejected.""" + with self.assertRaises(ValueError): + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "BCDEFGH"], use_channels=list(range(1, 8)) + ) + + async def test_pickup_grabbing_undeclared_tips_below_is_rejected(self): + """Picking A1:F1 from a full column would also grab the occupied G1/H1 below it within + the head's 8-nozzle reach; rejected by default.""" + with self.assertRaises(ValueError): + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEF"], use_channels=list(range(6)) + ) + + async def test_allow_undeclared_absorbs_grabbed_tips(self): + """allow_undeclared_tip_pickup=True absorbs the extra grabbed tips into tracking (with a + warning) so all eight nozzles are accounted for.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEF"], + use_channels=list(range(6)), + allow_undeclared_tip_pickup=True, + ) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(8))) + self.assertTrue(any("undeclared tips" in str(w.message) for w in caught)) + + +if __name__ == "__main__": + unittest.main()