feat(api): static meniscus-relative aspiration

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -31,6 +31,7 @@
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
 from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
 from opentrons_shared_data.pipette.types import PipetteNameType
+from opentrons_shared_data.robot.types import RobotType
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 from opentrons.hardware_control.nozzle_manager import NozzleMap
@@ -92,6 +93,10 @@ def pipette_id(self) -> str:
         """The instrument's unique ProtocolEngine ID."""
         return self._pipette_id
 
+    @property
+    def robot_type(self) -> RobotType:
+        return self._engine_client.state.config.robot_type
+
     def get_default_speed(self) -> float:
         speed = self._engine_client.state.pipettes.get_movement_speed(
             pipette_id=self._pipette_id
@@ -843,6 +848,13 @@ def retract(self) -> None:
         z_axis = self._engine_client.state.pipettes.get_z_axis(self._pipette_id)
         self._engine_client.execute_command(cmd.HomeParams(axes=[z_axis]))
 
+    def get_last_measured_liquid_height(self, well_core: WellCore) -> Optional[float]:
+        well_name = well_core.get_name()
+        labware_id = well_core.labware_id
+        return self._engine_client.state.wells.get_last_measured_liquid_height(
+            labware_id=labware_id, well_name=well_name
+        )
+
     def detect_liquid_presence(self, well_core: WellCore, loc: Location) -> bool:
         labware_id = well_core.labware_id
         well_name = well_core.get_name()

api/src/opentrons/protocol_api/core/engine/well.py

@@ -3,7 +3,12 @@
 
 from opentrons_shared_data.labware.constants import WELL_NAME_PATTERN
 
-from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset
+from opentrons.protocol_engine import (
+    WellLocation,
+    WellOrigin,
+    WellOffset,
+    WellVolumeOffset,
+)
 from opentrons.protocol_engine import commands as cmd
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
 from opentrons.protocols.api_support.util import UnsupportedAPIError
@@ -125,15 +130,19 @@ def get_center(self) -> Point:
             well_location=WellLocation(origin=WellOrigin.CENTER),
         )
 
-    def get_meniscus(self, z_offset: float) -> Point:
+    def get_meniscus(
+        self, z_offset: float, operation_volume: Optional[float] = None
+    ) -> Point:
         """Get the coordinate of the well's meniscus, with a z-offset."""
         return self._engine_client.state.geometry.get_well_position(
             well_name=self._name,
             labware_id=self._labware_id,
             well_location=WellLocation(
                 origin=WellOrigin.MENISCUS,
                 offset=WellOffset(x=0, y=0, z=z_offset),
+                volumeOffset=WellVolumeOffset(volumeOffset="operationVolume"),
             ),
+            operational_volume=operation_volume,
         )
 
     def load_liquid(

api/src/opentrons/protocol_api/core/instrument.py

@@ -10,12 +10,18 @@
 from opentrons.protocols.api_support.util import FlowRates
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.hardware_control.nozzle_manager import NozzleMap
+from opentrons_shared_data.robot.types import RobotType
 
 from ..disposal_locations import TrashBin, WasteChute
 from .well import WellCoreType
 
 
 class AbstractInstrument(ABC, Generic[WellCoreType]):
+    @property
+    @abstractmethod
+    def robot_type(self) -> RobotType:
+        ...
+
     @abstractmethod
     def get_default_speed(self) -> float:
         ...
@@ -305,6 +311,12 @@ def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         ...
 
+    @abstractmethod
+    def get_last_measured_liquid_height(
+        self, well_core: WellCoreType
+    ) -> Optional[float]:
+        ...
+
     @abstractmethod
     def detect_liquid_presence(
         self, well_core: WellCoreType, loc: types.Location

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -20,6 +20,7 @@
 from opentrons.protocols.geometry import planning
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.hardware_control.nozzle_manager import NozzleMap
+from opentrons_shared_data.robot.types import RobotType
 
 from ...disposal_locations import TrashBin, WasteChute
 from ..instrument import AbstractInstrument
@@ -64,6 +65,10 @@ def __init__(
         )
         self._liquid_presence_detection = False
 
+    @property
+    def robot_type(self) -> RobotType:
+        return "OT-2 Standard"
+
     def get_default_speed(self) -> float:
         """Gets the speed at which the robot's gantry moves."""
         return self._default_speed
@@ -566,6 +571,12 @@ def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]
 
+    def get_last_measured_liquid_height(self, well_core: WellCore) -> Optional[float]:
+        """This will never be called because it was added in API 2.21."""
+        assert (
+            False
+        ), "get_last_measured_liquid_height only supported in API 2.21 & later"
+
     def detect_liquid_presence(self, well_core: WellCore, loc: types.Location) -> bool:
         """This will never be called because it was added in API 2.20."""
         assert False, "detect_liquid_presence only supported in API 2.20 & later"

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -21,6 +21,7 @@
     UnexpectedTipRemovalError,
     UnexpectedTipAttachError,
 )
+from opentrons_shared_data.robot.types import RobotType
 
 from ...disposal_locations import TrashBin, WasteChute
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
@@ -77,6 +78,10 @@ def __init__(
         )
         self._liquid_presence_detection = False
 
+    @property
+    def robot_type(self) -> RobotType:
+        return "OT-2 Standard"
+
     def get_default_speed(self) -> float:
         return self._default_speed
 
@@ -484,6 +489,12 @@ def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]
 
+    def get_last_measured_liquid_height(self, well_core: WellCore) -> Optional[float]:
+        """This will never be called because it was added in API 2.21."""
+        assert (
+            False
+        ), "get_last_measured_liquid_height only supported in API 2.21 & later"
+
     def detect_liquid_presence(self, well_core: WellCore, loc: types.Location) -> bool:
         """This will never be called because it was added in API 2.20."""
         assert False, "detect_liquid_presence only supported in API 2.20 & later"

api/src/opentrons/protocol_api/instrument_context.py

@@ -3,6 +3,7 @@
 from contextlib import ExitStack
 from typing import Any, List, Optional, Sequence, Union, cast, Dict
 from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError
+from opentrons.protocol_engine.types import WellLocation, WellOrigin, WellOffset
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
@@ -27,6 +28,7 @@
     requires_version,
     APIVersionError,
     UnsupportedAPIError,
+    RobotTypeError,
 )
 from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 
@@ -165,6 +167,8 @@ def aspirate(
         volume: Optional[float] = None,
         location: Optional[Union[types.Location, labware.Well]] = None,
         rate: float = 1.0,
+        meniscus_relative: bool = False,
+        offset_from_meniscus_mm: float = -2.0,
     ) -> InstrumentContext:
         """
         Draw liquid into a pipette tip.
@@ -217,52 +221,22 @@ def aspirate(
             )
         )
 
-        well: Optional[labware.Well] = None
-        move_to_location: types.Location
-        last_location = self._get_last_location_by_api_version()
-        try:
-            target = validation.validate_location(
-                location=location, last_location=last_location
-            )
-        except validation.NoLocationError as e:
-            raise RuntimeError(
-                "If aspirate is called without an explicit location, another"
-                " method that moves to a location (such as move_to or "
-                "dispense) must previously have been called so the robot "
-                "knows where it is."
-            ) from e
-
-        if isinstance(target, validation.WellTarget):
-            move_to_location = target.location or target.well.bottom(
-                z=self._well_bottom_clearances.aspirate
-            )
-            well = target.well
-        if isinstance(target, validation.PointTarget):
-            move_to_location = target.location
-        if isinstance(target, (TrashBin, WasteChute)):
-            raise ValueError(
-                "Trash Bin and Waste Chute are not acceptable location parameters for Aspirate commands."
-            )
-        if self.api_version >= APIVersion(2, 11):
-            instrument.validate_takes_liquid(
-                location=move_to_location,
-                reject_module=self.api_version >= APIVersion(2, 13),
-                reject_adapter=self.api_version >= APIVersion(2, 15),
-            )
-
         if self.api_version >= APIVersion(2, 16):
             c_vol = self._core.get_available_volume() if volume is None else volume
         else:
             c_vol = self._core.get_available_volume() if not volume else volume
         flow_rate = self._core.get_aspirate_flow_rate(rate)
 
-        if (
-            self.api_version >= APIVersion(2, 20)
-            and well is not None
-            and self.liquid_presence_detection
-            and self._96_tip_config_valid()
-        ):
-            self.require_liquid_presence(well=well)
+        target, move_to_location, well = self._determine_aspirate_move_to_location(
+            location=location,
+            volume=c_vol,
+            meniscus_relative=meniscus_relative,
+            offset_from_meniscus_mm=offset_from_meniscus_mm,
+        )
+        if isinstance(target, (TrashBin, WasteChute)):
+            raise ValueError(
+                "Trash Bin and Waste Chute are not acceptable location parameters for Aspirate commands."
+            )
 
         with publisher.publish_context(
             broker=self.broker,
@@ -1689,6 +1663,7 @@ def liquid_presence_detection(self) -> bool:
         """
         return self._core.get_liquid_presence_detection()
 
+    # Is this the `configure_liquid_presence_detection` wrapper? See Syntax and Semantics doc
     @liquid_presence_detection.setter
     @requires_version(2, 20)
     def liquid_presence_detection(self, enable: bool) -> None:
@@ -2132,6 +2107,97 @@ def configure_nozzle_layout(
         )
         self._tip_racks = tip_racks or []
 
+    def _determine_aspirate_move_to_location(
+        self,
+        location: Optional[Union[types.Location, labware.Well]],
+        volume: float,
+        meniscus_relative: bool,
+        offset_from_meniscus_mm: float,
+    ) -> tuple[validation.ValidTarget, types.Location, Optional[labware.Well]]:
+        well: Optional[labware.Well] = None
+        move_to_location: types.Location
+        last_location = self._get_last_location_by_api_version()
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
+            )
+        except validation.NoLocationError as e:
+            raise RuntimeError(
+                "If aspirate is called without an explicit location, another"
+                " method that moves to a location (such as move_to or "
+                "dispense) must previously have been called so the robot "
+                "knows where it is."
+            ) from e
+
+        if isinstance(target, validation.WellTarget):
+            move_to_location = target.location or target.well.bottom(
+                z=self._well_bottom_clearances.aspirate
+            )
+            well = target.well
+            if self.api_version >= APIVersion(2, 20):
+                if self._liquid_probe_before_aspirate(
+                    well=well,
+                    meniscus_relative=meniscus_relative,
+                    offset_from_meniscus_mm=offset_from_meniscus_mm,
+                    volume=volume,
+                ):
+                    move_to_location = target.well.meniscus(
+                        z=offset_from_meniscus_mm,
+                        operation_volume=volume,
+                    )
+        if isinstance(target, validation.PointTarget):
+            move_to_location = target.location
+        if self.api_version >= APIVersion(2, 11):
+            instrument.validate_takes_liquid(
+                location=move_to_location,
+                reject_module=self.api_version >= APIVersion(2, 13),
+                reject_adapter=self.api_version >= APIVersion(2, 15),
+            )
+        return target, move_to_location, well
+
+    def _liquid_probe_before_aspirate(
+        self,
+        well: labware.Well,
+        meniscus_relative: bool,
+        offset_from_meniscus_mm: float,
+        volume: float,
+    ) -> bool:
+        if self.api_version < APIVersion(2, 21) and meniscus_relative:
+            raise APIVersionError(
+                api_element="Meniscus-relative aspiration",
+                until_version="2.21",
+                current_version=f"{self.api_version}",
+            )
+        elif (
+            self._core.robot_type != "OT-3 Standard"
+            and self.liquid_presence_detection
+            or meniscus_relative
+        ):
+            raise RobotTypeError(
+                "Liquid presence detection only available on Flex robot."
+            )
+        if (
+            self.api_version >= APIVersion(2, 21)
+            and well is not None
+            and self.liquid_presence_detection
+            and meniscus_relative
+            and self._96_tip_config_valid()
+        ):
+            height = self._core.get_last_measured_liquid_height(well_core=well._core)
+            if height is None:
+                self.measure_liquid_height(well=well)
+            return True
+        elif (
+            self.api_version >= APIVersion(2, 20)
+            and well is not None
+            and self.liquid_presence_detection
+            and self._96_tip_config_valid()
+        ):
+            self.require_liquid_presence(well=well)
+            return False
+        else:
+            return False
+
     @requires_version(2, 20)
     def detect_liquid_presence(self, well: labware.Well) -> bool:
         """Checks for liquid in a well.
@@ -2171,6 +2237,7 @@ def measure_liquid_height(self, well: labware.Well) -> float:
 
         loc = well.top()
         self._96_tip_config_valid()
+        # ensure this raises LiquidPresenceNotDetectedError
         height = self._core.liquid_probe_without_recovery(well._core, loc)
         return height
 

api/src/opentrons/protocol_api/labware.py

@@ -222,15 +222,19 @@ def center(self) -> Location:
         return Location(self._core.get_center(), self)
 
     @requires_version(2, 21)
-    def meniscus(self, z: float = 0.0) -> Location:
+    def meniscus(
+        self, z: float = 0.0, operation_volume: Optional[float] = None
+    ) -> Location:
         """
         :param z: An offset on the z-axis, in mm. Positive offsets are higher and
             negative offsets are lower.
         :return: A :py:class:`~opentrons.types.Location` corresponding to the
             absolute position of the meniscus-center of the well, plus the ``z`` offset
             (if specified).
         """
-        return Location(self._core.get_meniscus(z_offset=z), self)
+        return Location(
+            self._core.get_meniscus(z_offset=z, operation_volume=operation_volume), self
+        )
 
     @requires_version(2, 8)
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point: