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..7c12310623c 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, @@ -23,17 +28,16 @@ ) from pylabrobot.resources import ( Coordinate, + Resource, Tip, + does_tip_tracking, ) -from pylabrobot.resources.opentrons import OTDeck -from pylabrobot.resources.tip_rack import TipRack +from pylabrobot.resources.opentrons import OT2RobotGeometry, OTDeck +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 +48,44 @@ # https://labautomation.io/t/connect-pylabrobot-to-ot2/2862/18 _OT_DECK_IS_ADDRESSABLE_AREA_VERSION = "7.1.0" +# head8 surrounding-resource collision guard: pad the head footprint by this in x and y, and treat a +# resource as in the way when its top is within this z of the tip-pickup height. +_HEAD8_XY_TOLERANCE = 5.0 # mm +_HEAD8_Z_TOLERANCE = 10.0 # mm +_HEAD8_NOZZLE_PITCH = 9.0 # mm between adjacent nozzles on the 8-channel head + +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 +107,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 +119,19 @@ 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 + + self.geometry = OT2RobotGeometry() + + # 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 +150,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 +199,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,18 +306,20 @@ 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 labware_uuid = self.get_ot_name(tip_rack.name) deck = tip_rack.parent + while deck is not None and not isinstance(deck, OTDeck): + deck = deck.parent # labware sits in a slot holder, whose parent is the deck assert isinstance(deck, OTDeck) 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 +330,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 +407,223 @@ 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 = _HEAD8_NOZZLE_PITCH + 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) + + def _deck_resources(self) -> List[Resource]: + """Labware assigned to deck slots (the children of the slot holders).""" + resources: List[Resource] = [] + for holder in self.deck.children: + resources.extend(holder.children) + return resources + + def _check_head8_surrounding_resources( + self, ops: List[Pickup], use_channels: List[int] + ) -> List[Resource]: + """Other deck resources the head8's footprint would collide with (surrounding-resource guard). + + Builds a bounding box around the head - the full eight-nozzle column span, padded by + ``_HEAD8_XY_TOLERANCE`` in x and y. Any other deck resource whose footprint overlaps that box, + and whose top sits within ``_HEAD8_Z_TOLERANCE`` of the tip-pickup height (so it is tall enough + to be in the head's path), is a collision. Unlike _check_head8_pickup (which only inspects the + target rack and ignores z), this is z-aware: a short resource under the box is cleared. + """ + pitch = _HEAD8_NOZZLE_PITCH + + primary = next(op for op, ch in zip(ops, use_channels) if ch == 0) + spot = primary.resource.get_absolute_location("c", "c", "b") + col_x = spot.x + # the head's lowest point over an adjacent slot is the bare nozzle bottoms, which during pickup + # sit where the nozzle engages the tip: the tip top minus the fitting depth. + pickup_z = spot.z + primary.tip.total_tip_length - primary.tip.fitting_depth + # the head physically spans all eight nozzles: channel 0 at the primary (back), channel 7 in front + y_back, y_front = spot.y, spot.y - 7 * pitch + box_x0, box_x1 = col_x - _HEAD8_XY_TOLERANCE, col_x + _HEAD8_XY_TOLERANCE + box_y0, box_y1 = y_front - _HEAD8_XY_TOLERANCE, y_back + _HEAD8_XY_TOLERANCE + + target_rack = ops[0].resource.parent + collisions: List[Resource] = [] + for res in self._deck_resources(): + if res is target_rack: + continue + corner = res.get_absolute_location() + rx0, ry0 = corner.x, corner.y + rx1, ry1 = rx0 + res.get_absolute_size_x(), ry0 + res.get_absolute_size_y() + if rx1 < box_x0 or rx0 > box_x1 or ry1 < box_y0 or ry0 > box_y1: + continue # no x-y overlap with the head box + res_top_z = corner.z + res.get_absolute_size_z() + if res_top_z >= pickup_z - _HEAD8_Z_TOLERANCE: # tall enough to be in the head's path + collisions.append(res) + return collisions + + def can_reach_position(self, channel_idx: int, position: Coordinate) -> bool: + """Whether ``channel_idx`` can reach ``position`` (deck frame), centre-based. + + Maps the channel to its mount and per-nozzle y offset - a single pipette resolves to its + centre (channel_offset 0), a multi to each nozzle's offset - converts the target into the + robot frame, and tests it against the OT-2 gantry envelope. Works for single and multi alike. + """ + pipette, nozzle = self._channel_map()[channel_idx] + mount = "left" if pipette is self.left_pipette else "right" + n = self._pipette_channel_count(pipette) + channel_offset = self.geometry.channel_y_offsets(n)[nozzle] if n > 1 else 0.0 + return self.geometry.can_reach_position( + mount, self._deck_to_robot_frame(position), channel_offset + ) + + def ensure_can_reach_position( + self, use_channels: List[int], ops: Sequence[PipettingOp], op_name: str + ) -> None: + """Raise if any declared channel cannot reach its target position (the STAR pattern).""" + cant = [ + ch + for ch, op in zip(use_channels, ops) + if not self.can_reach_position( + ch, op.resource.get_location_wrt(self.deck, "c", "c", "b") + op.offset + ) + ] + if cant: + raise ValueError( + f"Channels {cant} cannot reach their target positions in '{op_name}'. Near the front/back " + f"deck limits only a subset of a multi's nozzles reach; try different channels or placement." + ) + + 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) + self.ensure_can_reach_position(use_channels, ops, "pick_up_tips") + + 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) + + collisions = self._check_head8_surrounding_resources(ops, use_channels) + if collisions: + where = ", ".join(r.name for r in collisions) + raise ValueError( + f"OT-2 8-channel head would collide with an adjacent resource ({where}): the head's " + f"footprint overhangs into that slot and the resource is tall enough to be in its path. " + f"Use a slot without a tall resource in the overhang direction, or clear it." + ) offset_x, offset_y, offset_z = ( op.offset.x, @@ -321,7 +639,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,15 +651,21 @@ 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 and op.resource.name == "trash" ) + + if not use_fixed_trash: + self.ensure_can_reach_position(use_channels, ops, "drop_tips") if use_fixed_trash: labware_id = "fixedTrash" else: @@ -361,15 +685,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, @@ -437,17 +761,31 @@ def _get_default_aspiration_flow_rate(self, pipette_name: str) -> float: "p20_multi_gen2": 7.6, }[pipette_name] + def _deck_to_robot_frame(self, location: Coordinate) -> Coordinate: + """Convert a deck-frame coordinate to the OT-2 robot frame. + + pylabrobot positions OT deck slots from the deck plate corner, whereas the OT-2 motion API + expects coordinates in the robot frame whose origin is slot 1's corner. The two frames differ by + slot 1's position in the deck frame, so subtract it. + """ + return location - cast(OTDeck, self.deck).slot_locations[0] + 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) + self.ensure_can_reach_position(use_channels, ops, "aspirate") + op = cast(SingleChannelAspiration, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) flow_rate = op.flow_rate or self._get_default_aspiration_flow_rate(pipette_name) - location = ( + location = self._deck_to_robot_frame( op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset + Coordinate(z=op.liquid_height or 0) @@ -461,24 +799,24 @@ 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, ) - traversal_location = ( + traversal_location = self._deck_to_robot_frame( op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset ) traversal_location.z = self.traversal_height @@ -513,16 +851,21 @@ 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) + self.ensure_can_reach_position(use_channels, ops, "dispense") + op = cast(SingleChannelDispense, primary) volume = op.volume pipette_name = self.get_pipette_name(pipette_id) flow_rate = op.flow_rate or self._get_default_dispense_flow_rate(pipette_name) - location = ( + location = self._deck_to_robot_frame( op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset + Coordinate(z=op.liquid_height or 0) @@ -533,7 +876,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,18 +884,18 @@ 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, ) - traversal_location = ( + traversal_location = self._deck_to_robot_frame( op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset ) traversal_location.z = self.traversal_height @@ -563,7 +906,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 +933,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 +1016,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 +1036,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..b96e7da77c0 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).""" @@ -239,6 +286,19 @@ def test_get_pickup_pipette_raises_when_tip_already_mounted(self): with self.assertRaises(NoChannelError): self.backend._get_pickup_pipette(ops) + # -- _deck_to_robot_frame -- + + def test_deck_to_robot_frame_maps_slot1_corner_to_robot_origin(self): + """The deck->robot transform subtracts slot 1's corner, so a deck-frame point at slot 1's + corner becomes the robot origin and a point offset from it keeps that offset.""" + self.backend.set_deck(self.deck) + corner = self.deck.slot_locations[0] + self.assertEqual(self.backend._deck_to_robot_frame(corner), Coordinate(0, 0, 0)) + self.assertEqual( + self.backend._deck_to_robot_frame(corner + Coordinate(10, 20, 3)), + Coordinate(10, 20, 3), + ) + # -- _get_drop_pipette -- def test_get_drop_pipette_selects_right_for_20ul(self): @@ -315,3 +375,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..492062407b4 --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox.py @@ -0,0 +1,184 @@ +"""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, +) +from pylabrobot.resources.opentrons import OT2RobotGeometry + +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 + self.geometry = OT2RobotGeometry() + + 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..7244d607064 --- /dev/null +++ b/pylabrobot/liquid_handling/backends/opentrons_chatterbox_tests.py @@ -0,0 +1,414 @@ +"""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 Coordinate, set_tip_tracking, set_volume_tracking +from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb +from pylabrobot.resources.opentrons import ( + OTDeck, + opentrons_96_filtertiprack_20ul, + opentrons_96_tiprack_300ul, +) + + +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)) + + +class _FarTarget: + """Stub resource that always reports a location far beyond the gantry envelope.""" + + def get_location_wrt(self, *args, **kwargs): + return Coordinate(150, 800, 5) + + +class _FarOp: + def __init__(self): + self.resource = _FarTarget() + self.offset = Coordinate(0, 0, 0) + + +class OpentronsChatterboxReachTests(unittest.IsolatedAsyncioTestCase): + """Center-based reachability check (OT2RobotGeometry wired into the operations).""" + + 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=5) + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + async def test_can_reach_position_central_yes_far_no(self): + """A central deck coordinate is reachable; one far past the back envelope is not - + for both a multi nozzle (channel 0) and the single (channel 8).""" + central, far = Coordinate(150, 150, 5), Coordinate(150, 800, 5) + self.assertTrue(self.backend.can_reach_position(0, central)) + self.assertTrue(self.backend.can_reach_position(8, central)) + self.assertFalse(self.backend.can_reach_position(0, far)) + self.assertFalse(self.backend.can_reach_position(8, far)) + + async def test_out_of_envelope_target_raises(self): + """ensure_can_reach_position rejects a target outside the gantry envelope.""" + with self.assertRaises(ValueError): + self.backend.ensure_can_reach_position([0], [_FarOp()], "pick_up_tips") # type: ignore[list-item] + + async def test_reachable_head8_column_pickup_passes(self): + """A normal on-deck head8 column pickup passes the reach guard and runs.""" + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEFGH"], use_channels=list(range(8)) + ) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(8))) + + async def test_trash_discard_skips_reach_check(self): + """Discarding to the fixed trash is exempt from the reach check (it routes through the + addressable area, not a reach-bounded move), so it is not falsely rejected.""" + await self.lh.pick_up_tips( + [self.tips.get_item(f"{r}1") for r in "ABCDEFGH"], use_channels=list(range(8)) + ) + await self.lh.discard_tips() + self.assertFalse(any(self.lh.head[c].has_tip for c in range(8))) + + +class OpentronsChatterboxPartialPickupTests(unittest.IsolatedAsyncioTestCase): + """Careful coverage of partial head8 pickup: a channel-0-anchored block of k < 8 channels. + + Allowed only when the tipspots the unused nozzles cover are empty; the unused nozzles may + overhang past the last row into empty space. Exercises the deck-accessibility guard at both + the front (slot 1) and back (slot 10) edges. (Surrounding-resource collision - e.g. a tall + resource in an adjacent slot the head overhangs into - is NOT modelled and not tested here.) + """ + + 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() + + async def asyncTearDown(self): + set_tip_tracking(False) + set_volume_tracking(False) + + def _rack(self, slot, present_rows): + """A 20 uL tiprack on ``slot`` whose column-1 tips exist only on ``present_rows``.""" + rack = opentrons_96_filtertiprack_20ul(name=f"tips_{slot}") + self.deck.assign_child_at_slot(rack, slot=slot) + for r in "ABCDEFGH": + if r not in present_rows: + rack.get_item(f"{r}1").tracker.remove_tip(commit=True) + return rack + + async def test_partial_pickup_picks_exactly_k(self): + """For k = 1..7 a channel-0-anchored block picks exactly k tips (channels 0..k-1) in one + command, leaving the unused channels empty - when the rows below the selection are empty.""" + rows = "ABCDEFGH" + for k in range(1, 8): + with self.subTest(k=k): + rack = self._rack(slot=k, present_rows=rows[:k]) + mark = len(self.backend.commands) + await self.lh.pick_up_tips( + [rack.get_item(f"{r}1") for r in rows[:k]], use_channels=list(range(k)) + ) + names = [n for n, _, _ in self.backend.commands[mark:]] + self.assertEqual(names.count("lh.pick_up_tip"), 1) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(k))) + self.assertFalse(any(self.lh.head[c].has_tip for c in range(k, 8))) + await self.lh.return_tips() + + async def test_partial_pickup_of_bottom_rows_overhangs_into_empty_space(self): + """Picking the bottom rows E-H with channels 0-3 (anchored at E1) is allowed at both the + front (slot 1) and back (slot 10) edge - the unused nozzles overhang past row H, over no tips.""" + for slot in (1, 10): + with self.subTest(slot=slot): + rack = self._rack(slot=slot, present_rows="EFGH") + await self.lh.pick_up_tips( + [rack.get_item(f"{r}1") for r in "EFGH"], use_channels=[0, 1, 2, 3] + ) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(4))) + self.assertFalse(any(self.lh.head[c].has_tip for c in range(4, 8))) + await self.lh.return_tips() + + async def test_partial_pickup_rejected_when_a_covered_tipspot_is_occupied(self): + """Picking A-C (channels 0-2) is rejected when row D - covered by unused nozzle 3 - still + holds a tip: the head would grab it too.""" + rack = self._rack(slot=2, present_rows="ABCD") + with self.assertRaises(ValueError): + await self.lh.pick_up_tips([rack.get_item(f"{r}1") for r in "ABC"], use_channels=[0, 1, 2]) + + async def test_overhang_into_tall_adjacent_resource_is_rejected(self): + """Guard 2 (surrounding-resource): a partial pickup at slot 10 overhangs into slot 7. Allowed + when slot 7 is empty; rejected once slot 7 holds a tall tiprack the head footprint would hit.""" + rack = self._rack(slot=10, present_rows="EFGH") + await self.lh.pick_up_tips( + [rack.get_item(f"{r}1") for r in "EFGH"], use_channels=[0, 1, 2, 3] + ) # slot 7 empty -> allowed + await self.lh.return_tips() + + self.deck.assign_child_at_slot(opentrons_96_filtertiprack_20ul(name="tall_neighbour"), slot=7) + with self.assertRaises(ValueError): + await self.lh.pick_up_tips( + [rack.get_item(f"{r}1") for r in "EFGH"], use_channels=[0, 1, 2, 3] + ) + + async def test_overhang_clears_short_adjacent_resource(self): + """The collision check is z-aware: the same slot-10 overhang is allowed when slot 7 holds only + a short resource (a 96-well plate) whose top sits below the head's path.""" + rack = self._rack(slot=10, present_rows="EFGH") + self.deck.assign_child_at_slot(CellTreat_96_wellplate_350ul_Fb(name="short_neighbour"), slot=7) + await self.lh.pick_up_tips([rack.get_item(f"{r}1") for r in "EFGH"], use_channels=[0, 1, 2, 3]) + self.assertTrue(all(self.lh.head[c].has_tip for c in range(4))) + + async def test_collision_z_reference_is_the_nozzle_engagement_height(self): + """Pins the collision z-reference to where the nozzles engage (tip top - fitting depth), not + the tipspot anchor (the tip bottom). For a 300 uL tip the tip bottom sits near the deck, so the + buggy reference would drive the z threshold negative and wrongly reject even a short neighbour. + So a 300 uL partial pickup over a short 96-well plate must be ALLOWED - cleared only with the + correct reference. (This is the test that would fail if the z-fix were reverted.)""" + backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p300_multi_gen2", right_pipette_name="p20_single_gen2", verbose=False + ) + deck = OTDeck() + lh = LiquidHandler(backend=backend, deck=deck) + await lh.setup() + rack = opentrons_96_tiprack_300ul(name="t300") + deck.assign_child_at_slot(rack, slot=10) + for r in "ABCD": + rack.get_item(f"{r}1").tracker.remove_tip(commit=True) + deck.assign_child_at_slot(CellTreat_96_wellplate_350ul_Fb(name="short"), slot=7) + await lh.pick_up_tips([rack.get_item(f"{r}1") for r in "EFGH"], use_channels=[0, 1, 2, 3]) + self.assertTrue(all(lh.head[c].has_tip for c in range(4))) + + +class _StubResource: + """A deck resource with a fixed footprint and height, for the collision-math unit test.""" + + children: list = [] + + def __init__(self, name, x, y, z, sx, sy, sz, parent=None): + self.name = name + self.parent = parent + self._loc = Coordinate(x, y, z) + self._sx, self._sy, self._sz = sx, sy, sz + + def get_absolute_location(self, *args, **kwargs): + return self._loc + + def get_absolute_size_x(self): + return self._sx + + def get_absolute_size_y(self): + return self._sy + + def get_absolute_size_z(self): + return self._sz + + +class _StubTip: + total_tip_length = 39.2 + fitting_depth = 8.25 + + +class _StubPickup: + def __init__(self, x, y, z, parent): + self.resource = _StubResource("primary_spot", x, y, z, 0, 0, 0, parent=parent) + self.tip = _StubTip() + + +class OpentronsCollisionMathTests(unittest.TestCase): + """Unit test of _check_head8_surrounding_resources with synthetic coordinates, decoupled from the + real deck geometry: verifies the x-y bounding-box overlap and the z-height threshold directly.""" + + def test_box_overlap_and_z_threshold(self): + backend = OpentronsOT2ChatterboxBackend( + left_pipette_name="p20_multi_gen2", right_pipette_name="p20_single_gen2", verbose=False + ) + target_rack = object() + # primary tip at (130, 380, 25.5): col_x=130, pickup_z = 25.5 + 39.2 - 8.25 = 56.45; + # head box x [125, 135], y [380 - 7*9 - 5, 385] = [312, 385]; collide iff top z >= 46.45. + primary = _StubPickup(130, 380, 25.5, parent=target_rack) + + tall_overlap = _StubResource("tall_overlap", 120, 340, 0, 20, 20, 64.7) # overlaps + tall + short_overlap = _StubResource("short_overlap", 120, 340, 0, 20, 20, 14.3) # overlaps but short + tall_far = _StubResource("tall_far", 200, 340, 0, 20, 20, 64.7) # tall but no x overlap + + class _Holder: + def __init__(self, res): + self.children = [res] + + class _Deck: + children = [_Holder(tall_overlap), _Holder(short_overlap), _Holder(tall_far)] + + backend.set_deck(_Deck()) # type: ignore[arg-type] + names = [r.name for r in backend._check_head8_surrounding_resources([primary], [0])] # type: ignore[list-item] + self.assertIn("tall_overlap", names) # overlap + tall enough -> collision + self.assertNotIn("short_overlap", names) # overlap but below the head's path -> cleared + self.assertNotIn("tall_far", names) # tall but outside the x band -> no collision + + +if __name__ == "__main__": + unittest.main() diff --git a/pylabrobot/resources/opentrons/__init__.py b/pylabrobot/resources/opentrons/__init__.py index 47225260c04..38d1dc2e55c 100644 --- a/pylabrobot/resources/opentrons/__init__.py +++ b/pylabrobot/resources/opentrons/__init__.py @@ -1,4 +1,5 @@ from .deck import OTDeck from .load import load_ot_tip_rack from .module import OTModule +from .ot2_geometry import OT2RobotGeometry from .tip_racks import * diff --git a/pylabrobot/resources/opentrons/deck.py b/pylabrobot/resources/opentrons/deck.py index db63bd2a2f0..096ab6a9525 100644 --- a/pylabrobot/resources/opentrons/deck.py +++ b/pylabrobot/resources/opentrons/deck.py @@ -1,50 +1,90 @@ import textwrap -from typing import List, Optional +from typing import List, Optional, cast 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 +_SLOT_SIZE_X = 128.0 +_SLOT_SIZE_Y = 86.0 + +# The fixed trash (opentrons_1_trash_1100ml_fixed) is larger than a standard slot and overhangs it, +# so slot 12's holder takes the trash footprint instead of the standard slot size. +_TRASH_SIZE_X = 172.86 +_TRASH_SIZE_Y = 165.86 +_TRASH_SIZE_Z = 82.0 + +# Base OT-2 slot grid in the robot frame, where slot 1's corner is the origin. +_BASE_SLOT_LOCATIONS = [ + Coordinate(x=0.0, y=0.0, z=0.0), + Coordinate(x=132.5, y=0.0, z=0.0), + Coordinate(x=265.0, y=0.0, z=0.0), + Coordinate(x=0.0, y=90.5, z=0.0), + Coordinate(x=132.5, y=90.5, z=0.0), + Coordinate(x=265.0, y=90.5, z=0.0), + Coordinate(x=0.0, y=181.0, z=0.0), + Coordinate(x=132.5, y=181.0, z=0.0), + Coordinate(x=265.0, y=181.0, z=0.0), + Coordinate(x=0.0, y=271.5, z=0.0), + Coordinate(x=132.5, y=271.5, z=0.0), + Coordinate(x=265.0, y=271.5, z=0.0), +] +# The deck plate corner sits at (-115.65, -68.03) in the robot frame, which is this resource's +# local origin, so each slot is re-based onto the plate corner by adding the offset. +_SLOT_CORNER_OFFSET = Coordinate(x=115.65, y=68.03, z=0.0) + class OTDeck(Deck): - """The OpenTron deck.""" + """The Opentrons OT-2 deck. + + The 12 slots are modeled as :class:`ResourceHolder` children, one per slot, so the deck geometry + has a single source of truth and renders directly from the serialized resource tree. Labware is + placed into a slot's holder with :meth:`assign_child_at_slot`. + """ def __init__( self, size_x: float = 624.3, size_y: float = 565.2, - size_z: float = 900, + size_z: float = 0, origin: Coordinate = Coordinate(0, 0, 0), with_trash: bool = True, - name: str = "deck", + name: str = "ot2_deck", + category: str = "deck", ): - # size_z is probably wrong - - super().__init__(size_x=size_x, size_y=size_y, size_z=size_z, origin=origin) - - self.slots: List[Optional[Resource]] = [None] * 12 - - self.slot_locations = [ - Coordinate(x=0.0, y=0.0, z=0.0), - Coordinate(x=132.5, y=0.0, z=0.0), - Coordinate(x=265.0, y=0.0, z=0.0), - Coordinate(x=0.0, y=90.5, z=0.0), - Coordinate(x=132.5, y=90.5, z=0.0), - Coordinate(x=265.0, y=90.5, z=0.0), - Coordinate(x=0.0, y=181.0, z=0.0), - Coordinate(x=132.5, y=181.0, z=0.0), - Coordinate(x=265.0, y=181.0, z=0.0), - Coordinate(x=0.0, y=271.5, z=0.0), - Coordinate(x=132.5, y=271.5, z=0.0), - Coordinate(x=265.0, y=271.5, z=0.0), - ] + super().__init__( + size_x=size_x, size_y=size_y, size_z=size_z, name=name, origin=origin, category=category + ) + + self._slot_holders: List[ResourceHolder] = [] + for i, base in enumerate(_BASE_SLOT_LOCATIONS): + is_trash_slot = i == 11 and with_trash + holder = ResourceHolder( + name=f"{self.name}_slot_{i + 1}", + size_x=_TRASH_SIZE_X if is_trash_slot else _SLOT_SIZE_X, + size_y=_TRASH_SIZE_Y if is_trash_slot else _SLOT_SIZE_Y, + size_z=_TRASH_SIZE_Z if is_trash_slot else 0, + ) + self._slot_holders.append(holder) + super().assign_child_resource(holder, location=base + _SLOT_CORNER_OFFSET) if with_trash: self._assign_trash() + @property + def slots(self) -> List[Optional[Resource]]: + """The labware in each slot, or ``None`` for empty slots (slot 1 first).""" + return [holder.resource for holder in self._slot_holders] + + @property + def slot_locations(self) -> List[Coordinate]: + """The location of each slot, re-based onto the deck plate corner (slot 1 first).""" + return [cast(Coordinate, holder.location) for holder in self._slot_holders] + def _assign_trash(self): - """Assign the trash area to the deck. + """Assign the trash area to slot 12. Because all opentrons operations require that the resource passed references a parent, we need to create a dummy resource to represent the container of the actual trash area. @@ -52,16 +92,16 @@ def _assign_trash(self): trash_container = Trash( name="trash_container", - size_x=172.86, - size_y=165.86, - size_z=82, + size_x=_TRASH_SIZE_X, + size_y=_TRASH_SIZE_Y, + size_z=_TRASH_SIZE_Z, ) actual_trash = Trash( name="trash", - size_x=172.86, - size_y=165.86, - size_z=82, + size_x=_TRASH_SIZE_X, + size_y=_TRASH_SIZE_Y, + size_z=_TRASH_SIZE_Z, ) # Trash location used to be Coordinate(x=86.43, y=82.93, z=0), @@ -75,47 +115,57 @@ def _assign_trash(self): def assign_child_resource( self, resource: Resource, - location: Optional[Coordinate], + location: Optional[Coordinate] = None, reassign: bool = True, ): - """Assign a resource to a slot. + """Assign a slot holder to the deck. - ..warning:: This method exists only for deserialization. You should use - :meth:`assign_child_at_slot` instead. + The deck's direct children are the 12 slot holders created in ``__init__``. Deserialization + re-assigns those holders by name, replacing the placeholder with the loaded one (which carries + its labware). Labware itself is placed with :meth:`assign_child_at_slot`, not here. """ - if location is None: - raise ValueError("location must be provided for resources on the deck") - - if location not in self.slot_locations: - super().assign_child_resource(resource, location=location) - else: - slot = self.slot_locations.index(location) + 1 - self.assign_child_at_slot(resource, slot) + existing = next((child for child in self.children if child.name == resource.name), None) + if existing is not None: + if not reassign: + raise ValueError(f"Resource '{resource.name}' already assigned to deck") + super().unassign_child_resource(existing) + for i, holder in enumerate(self._slot_holders): + if holder is existing: + self._slot_holders[i] = cast(ResourceHolder, resource) + break + elif not isinstance(resource, ResourceHolder): + raise ValueError( + f"Cannot assign '{resource.name}' directly to the deck. Use assign_child_at_slot to place " + "labware into a slot. A deck serialized before slots became resource holders stores labware " + "as direct children and will not load." + ) + + super().assign_child_resource(resource, location=location, reassign=reassign) def assign_child_at_slot(self, resource: Resource, slot: int): if slot not in range(1, 13): raise ValueError("slot must be between 1 and 12") - if self.slots[slot - 1] is not None: + holder = self._slot_holders[slot - 1] + if holder.resource is not None: raise ValueError(f"Spot {slot} is already occupied") - self.slots[slot - 1] = resource - super().assign_child_resource(resource, location=self.slot_locations[slot - 1]) + holder.assign_child_resource(resource) def unassign_child_resource(self, resource: Resource): - if resource not in self.slots: - raise ValueError(f"Resource {resource.name} is not assigned to this deck") - - slot = self.slots.index(resource) - self.slots[slot] = None + for holder in self._slot_holders: + if holder.resource is resource: + holder.unassign_child_resource(resource) + return super().unassign_child_resource(resource) def get_slot(self, resource: Resource) -> Optional[int]: """Get the slot number of a resource.""" - if resource not in self.slots: - return None - return self.slots.index(resource) + 1 + for i, holder in enumerate(self._slot_holders): + if holder.resource is resource: + return i + 1 + return None def summary(self) -> str: """Get a summary of the deck. diff --git a/pylabrobot/resources/opentrons/deck_tests.py b/pylabrobot/resources/opentrons/deck_tests.py index 475903ad0ae..dfa457bba3a 100644 --- a/pylabrobot/resources/opentrons/deck_tests.py +++ b/pylabrobot/resources/opentrons/deck_tests.py @@ -1,6 +1,7 @@ import textwrap import unittest +from pylabrobot.resources.coordinate import Coordinate from pylabrobot.resources.corning.plates import ( Cor_96_wellplate_360ul_Fb, ) @@ -8,6 +9,8 @@ from pylabrobot.resources.opentrons.tip_racks import ( opentrons_96_tiprack_300ul, ) +from pylabrobot.resources.resource import Resource +from pylabrobot.resources.resource_holder import ResourceHolder class TestOTDeck(unittest.TestCase): @@ -23,6 +26,41 @@ def setUp(self) -> None: self.deck.assign_child_at_slot(Cor_96_wellplate_360ul_Fb("my_plate"), 4) self.deck.assign_child_at_slot(Cor_96_wellplate_360ul_Fb("my_other_plate"), 5) + def test_slot_locations_inset_from_plate_corner(self): + """Slots are re-based onto the deck plate corner, so slot 1 sits at the corner offset + (115.65, 68.03) rather than the deck origin, and every other slot shifts by the same amount.""" + self.assertEqual(self.deck.slot_locations[0], Coordinate(115.65, 68.03, 0)) + self.assertEqual(self.deck.slot_locations[11], Coordinate(380.65, 339.53, 0)) + + def test_slots_are_resource_holders_at_inset_positions(self): + """The 12 slots are ResourceHolder children at the inset slot locations, and a slot's labware + is that holder's child.""" + holders = self.deck._slot_holders + self.assertEqual(len(holders), 12) + self.assertTrue(all(isinstance(h, ResourceHolder) for h in holders)) + self.assertEqual(holders[0].location, Coordinate(115.65, 68.03, 0)) + self.assertIs(holders[6].resource, self.deck.slots[6]) # tip_rack_1 lives in slot 7's holder + + def test_assign_child_resource_rejects_direct_labware(self): + """Labware must enter a slot via assign_child_at_slot; assigning it directly to the deck (as a + deck serialized before slots became holders would) is rejected rather than misplaced.""" + deck = OTDeck() + plate = Cor_96_wellplate_360ul_Fb("p") + with self.assertRaises(ValueError): + deck.assign_child_resource(plate, location=Coordinate(115.65, 68.03, 0)) + + def test_serialize_deserialize_round_trip(self): + """A deck with labware survives serialize -> deserialize, with slots resolving to the same + labware (validates replacing the placeholder holders with the loaded ones).""" + loaded = Resource.deserialize(self.deck.serialize()) + assert isinstance(loaded, OTDeck) + slot_6, slot_3 = loaded.slots[6], loaded.slots[3] + assert slot_6 is not None and slot_3 is not None + self.assertEqual(slot_6.name, "tip_rack_1") + self.assertEqual(loaded.get_slot(slot_6), 7) + self.assertEqual(slot_3.name, "my_plate") + self.assertEqual(loaded.slot_locations[0], Coordinate(115.65, 68.03, 0)) + def test_summary(self): self.assertEqual( self.deck.summary(), diff --git a/pylabrobot/resources/opentrons/ot2_geometry.py b/pylabrobot/resources/opentrons/ot2_geometry.py new file mode 100644 index 00000000000..00c783a93ae --- /dev/null +++ b/pylabrobot/resources/opentrons/ot2_geometry.py @@ -0,0 +1,87 @@ +import dataclasses +from typing import List, Tuple + +from pylabrobot.resources.coordinate import Coordinate + + +@dataclasses.dataclass(frozen=True) +class OT2RobotGeometry: + """Static geometry of an Opentrons OT-2, expressed in the robot frame. + + Every OT-2 Standard shares this geometry, so it is a constant rather than something probed from + the device. This is the key difference from Hamilton STAR hardware, whose installed configuration + varies per machine and must be read from firmware; the only OT-2 data worth discovering at runtime + is which pipettes are mounted, not where the deck is. + + The robot frame has its origin at the front-left corner of slot 1, with +x to the right, +y to + the back, and +z up. All distances are in mm. + """ + + # Gantry working area reachable by the reference (right) mount. + extents: Coordinate = dataclasses.field(default_factory=lambda: Coordinate(446.75, 347.5, 0.0)) + + # Partial-tip body clearance: how far a pipette's bounding box may sit beyond the deck extents at + # each edge. These bound the pipette body in partial nozzle configurations only; they do not + # constrain a fully-configured single or multi-channel pipette and are not used by the + # single-channel reach check below. + padding_front: float = 31.89 + padding_rear: float = -35.91 + padding_left: float = 0.0 + padding_right: float = 0.0 + + # Nozzle offset of each mount from the gantry carriage; the right mount is the reference. + left_mount_offset: Coordinate = dataclasses.field( + default_factory=lambda: Coordinate(-34.0, 0.0, 0.0) + ) + right_mount_offset: Coordinate = dataclasses.field( + default_factory=lambda: Coordinate(0.0, 0.0, 0.0) + ) + + def mount_offset(self, mount: str) -> Coordinate: + """Nozzle offset of ``mount`` ("left" or "right") from the gantry carriage.""" + if mount == "left": + return self.left_mount_offset + if mount == "right": + return self.right_mount_offset + raise ValueError(f"mount must be 'left' or 'right', got {mount!r}") + + def single_channel_reach(self, mount: str) -> Tuple[float, float, float, float]: + """Reachable nozzle region for a fully-configured pipette on ``mount``. + + Returns ``(x_min, x_max, y_min, y_max)`` in the robot frame: the gantry working area + (``extents``) translated by the mount offset. This mirrors how the OT-2 motion system bounds a + fully-configured pipette - the deck extents per mount, running from the front-left corner to + ``extents`` plus the mount offset. It is independent of ``padding_*``, which govern partial-tip + nozzle configurations only. The mount offset has no y component, so both mounts share the same + front/back reach and differ only in x. + """ + off = self.mount_offset(mount) + return (off.x, self.extents.x + off.x, 0.0, self.extents.y) + + @staticmethod + def channel_y_offsets(num_channels: int = 8, channel_pitch: float = 9.0) -> List[float]: + """Y offset of each channel from the head center, in mm, for a linear multi-channel head. + + Indexed back-to-front to match the Opentrons nozzle map: index 0 is the back-most nozzle (row + A, the primary/critical point) at the largest +y, descending to the front-most nozzle at -y. + Defaults describe the head8: 8 channels at 9 mm pitch spanning the head center by +-31.5 mm. + """ + half = (num_channels - 1) / 2 + return [(half - i) * channel_pitch for i in range(num_channels)] + + def can_reach_position( + self, mount: str, position: Coordinate, channel_offset: float = 0.0 + ) -> bool: + """Whether a nozzle displaced ``channel_offset`` mm in y from the head center can reach + ``position`` in x and y on ``mount``. + + Mirrors ``STARBackend.can_reach_position``: a pure predicate over the reachable region, here the + single-channel deck extents for the mount. For a single channel use the default + ``channel_offset=0``. For a head8, pass each channel's offset from :meth:`channel_y_offsets`; the + center must land within the extents, so near the front/back limits only a subset of channels can + reach, which is why an edge column is picked up partially rather than all 8. Z is not bounded by + deck extents and is checked separately by the motion layer. + """ + x_min, x_max, y_min, y_max = self.single_channel_reach(mount) + center_y = position.y - channel_offset + return x_min <= position.x <= x_max and y_min <= center_y <= y_max diff --git a/pylabrobot/resources/opentrons/ot2_geometry_tests.py b/pylabrobot/resources/opentrons/ot2_geometry_tests.py new file mode 100644 index 00000000000..6b233fe31da --- /dev/null +++ b/pylabrobot/resources/opentrons/ot2_geometry_tests.py @@ -0,0 +1,54 @@ +import unittest + +from pylabrobot.resources.coordinate import Coordinate +from pylabrobot.resources.opentrons.ot2_geometry import OT2RobotGeometry + + +class TestOT2RobotGeometry(unittest.TestCase): + """Tests for the static OT-2 robot geometry.""" + + def assert_bounds_almost_equal(self, actual, expected): + for a, e in zip(actual, expected): + self.assertAlmostEqual(a, e, places=4) + + def test_single_channel_reach_right_mount_is_full_extents(self): + """Right mount (reference): reach is the full gantry extents anchored at the origin.""" + self.assert_bounds_almost_equal( + OT2RobotGeometry().single_channel_reach("right"), (0.0, 446.75, 0.0, 347.5) + ) + + def test_single_channel_reach_left_mount_shifted_by_mount_offset(self): + """Left mount sits 34 mm left of the reference, so its x window shifts by -34 while y is + unchanged (the mount offset has no y component).""" + self.assert_bounds_almost_equal( + OT2RobotGeometry().single_channel_reach("left"), (-34.0, 412.75, 0.0, 347.5) + ) + + def test_can_reach_position_inside_and_outside(self): + """A point inside the mount's extents reaches; one past the front edge (y<0) does not.""" + geo = OT2RobotGeometry() + self.assertTrue(geo.can_reach_position("right", Coordinate(265.0, 271.5, 0))) # slot 12 corner + self.assertFalse( + geo.can_reach_position("right", Coordinate(265.0, -5.0, 0)) + ) # in front of deck + + def test_channel_y_offsets_head8_matches_opentrons_nozzle_map(self): + """8 channels at 9 mm pitch, indexed back-to-front: index 0 is A1 at +31.5 (the back-most, + primary nozzle), descending to the front-most at -31.5, matching the Opentrons nozzle map.""" + offsets = OT2RobotGeometry().channel_y_offsets() + self.assertEqual(len(offsets), 8) + self.assertAlmostEqual(offsets[0], 31.5) + self.assertAlmostEqual(offsets[-1], -31.5) + + def test_can_reach_position_channel_offset_partial_at_back_edge(self): + """Near the back limit a back-displaced channel reaches but a front-displaced one does not, so + a head8 engages only a subset of channels there.""" + geo = OT2RobotGeometry() + target = Coordinate(265.0, 340.0, 0) + self.assertTrue(geo.can_reach_position("right", target, channel_offset=31.5)) + self.assertFalse(geo.can_reach_position("right", target, channel_offset=-31.5)) + + def test_mount_offset_rejects_unknown_mount(self): + """Only 'left' and 'right' are valid mounts.""" + with self.assertRaises(ValueError): + OT2RobotGeometry().mount_offset("middle") diff --git a/pylabrobot/visualizer/lib.js b/pylabrobot/visualizer/lib.js index 4eda8f639d4..094fbd264ba 100644 --- a/pylabrobot/visualizer/lib.js +++ b/pylabrobot/visualizer/lib.js @@ -640,6 +640,11 @@ var frameImages = []; let frameInterval = 8; var _recordingTimer = null; +// The deck resource, found by type rather than a hardcoded name (decks may be named anything). +function getDeck() { + return Object.values(resources).find((r) => r instanceof Deck); +} + function getSnappingResourceAndLocationAndSnappingBox(resourceToSnap, x, y) { // Return the snapping resource that the given point is within, or undefined if there is no such resource. // A snapping resource is a spot within a plate/tip carrier or the OT deck. @@ -670,7 +675,7 @@ function getSnappingResourceAndLocationAndSnappingBox(resourceToSnap, x, y) { } // Check if the resource is in a ResourceHolder. - let deck = resources["deck"]; + let deck = getDeck(); for (let resource_name in deck.children) { const resource = deck.children[resource_name]; @@ -753,7 +758,7 @@ function getSnappingGrid(x, y, width, height) { let snappingLines = {}; - const deck = resources["deck"]; + const deck = getDeck(); if ( deck.constructor.name === "HamiltonSTARDeck" || deck.constructor.name === "VantageDeck" @@ -1350,72 +1355,78 @@ class VantageDeck extends Deck { } } +// Slot positions re-based onto the deck plate corner (matching the Python ResourceHolder slots). +// Used only for the slot-number labels and drag-snapping; the slot rectangles themselves are +// rendered by the ResourceHolder children. const otDeckSiteLocations = [ - { x: 0.0, y: 0.0 }, - { x: 132.5, y: 0.0 }, - { x: 265.0, y: 0.0 }, - { x: 0.0, y: 90.5 }, - { x: 132.5, y: 90.5 }, - { x: 265.0, y: 90.5 }, - { x: 0.0, y: 181.0 }, - { x: 132.5, y: 181.0 }, - { x: 265.0, y: 181.0 }, - { x: 0.0, y: 271.5 }, - { x: 132.5, y: 271.5 }, - { x: 265.0, y: 271.5 }, + { x: 115.65, y: 68.03 }, + { x: 248.15, y: 68.03 }, + { x: 380.65, y: 68.03 }, + { x: 115.65, y: 158.53 }, + { x: 248.15, y: 158.53 }, + { x: 380.65, y: 158.53 }, + { x: 115.65, y: 249.03 }, + { x: 248.15, y: 249.03 }, + { x: 380.65, y: 249.03 }, + { x: 115.65, y: 339.53 }, + { x: 248.15, y: 339.53 }, + { x: 380.65, y: 339.53 }, ]; class OTDeck extends Deck { constructor(resourceData) { - resourceData.location = { x: 115.65, y: 68.03 }; super(resourceData, undefined); } drawMainShape() { + // The slot rectangles are drawn by the ResourceHolder children; the main shape is just the deck + // border. The slot-number labels are added in draw() so they sit on top of the holders. let group = new Konva.Group({}); - const width = 128.0; - const height = 86.0; - // Draw the sites - for (let i = 0; i < otDeckSiteLocations.length; i++) { - const siteLocation = otDeckSiteLocations[i]; - const site = new Konva.Rect({ - x: siteLocation.x, - y: siteLocation.y, - width: width, - height: height, - fill: "white", + group.add( + new Konva.Rect({ + width: this.size_x, + height: this.size_y, stroke: "black", strokeWidth: 1, - }); - group.add(site); + }) + ); + return group; + } - // Add a text label in the site + draw(layer) { + super.draw(layer); + // Holder children are drawn after the main shape, so a slot number placed in the main shape is + // hidden. Add the number to the deck group last (on top) but only for empty slots, so the + // labware in an occupied slot is shown instead of being covered by the number. + const width = 128.0; + const height = 86.0; + for (const holder of this.children) { + if (!holder.location || (holder.children && holder.children.length > 0)) { + continue; // skip non-slot children and occupied slots + } + const match = holder.name ? holder.name.match(/slot_(\d+)$/) : null; + if (match === null) { + continue; + } const siteLabel = new Konva.Text({ - x: siteLocation.x, - y: siteLocation.y + height, - text: i + 1, + x: holder.location.x, + y: holder.location.y + height, + text: match[1], width: width, height: height, fontSize: 16, - fill: "black", + fill: "white", + stroke: "black", + strokeWidth: 0.5, + fillAfterStrokeEnabled: true, align: "center", verticalAlign: "middle", scaleY: -1, // Flip the text vertically + listening: false, }); - group.add(siteLabel); + this.group.add(siteLabel); + siteLabel.moveToTop(); } - - // draw border around the deck - group.add( - new Konva.Rect({ - width: this.size_x, - height: this.size_y, - stroke: "black", - strokeWidth: 1, - }) - ); - - return group; } serialize() { @@ -1729,7 +1740,7 @@ class Tube extends Container { // Nothing special. class Trash extends Resource { drawMainShape() { - if (resources["deck"].constructor.name) { + if (getDeck()) { return undefined; } return super.drawMainShape();