feat(api): add WellVolumeOffset to WellLocation

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

@@ -112,6 +112,7 @@ def aspirate(
         rate: float,
         flow_rate: float,
         in_place: bool,
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -146,12 +147,11 @@ def aspirate(
             well_name = well_core.get_name()
             labware_id = well_core.labware_id
 
-            well_location = (
-                self._engine_client.state.geometry.get_relative_well_location(
-                    labware_id=labware_id,
-                    well_name=well_name,
-                    absolute_point=location.point,
-                )
+            well_location = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
+                labware_id=labware_id,
+                well_name=well_name,
+                absolute_point=location.point,
+                is_meniscus=is_meniscus,
             )
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
@@ -182,6 +182,7 @@ def dispense(
         flow_rate: float,
         in_place: bool,
         push_out: Optional[float],
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:
@@ -237,12 +238,11 @@ def dispense(
             well_name = well_core.get_name()
             labware_id = well_core.labware_id
 
-            well_location = (
-                self._engine_client.state.geometry.get_relative_well_location(
-                    labware_id=labware_id,
-                    well_name=well_name,
-                    absolute_point=location.point,
-                )
+            well_location = self._engine_client.state.geometry.get_relative_liquid_handling_well_location(
+                labware_id=labware_id,
+                well_name=well_name,
+                absolute_point=location.point,
+                is_meniscus=is_meniscus,
             )
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
@@ -416,10 +416,12 @@ def pick_up_tip(
         well_name = well_core.get_name()
         labware_id = well_core.labware_id
 
-        well_location = self._engine_client.state.geometry.get_relative_well_location(
-            labware_id=labware_id,
-            well_name=well_name,
-            absolute_point=location.point,
+        well_location = (
+            self._engine_client.state.geometry.get_relative_pick_up_tip_well_location(
+                labware_id=labware_id,
+                well_name=well_name,
+                absolute_point=location.point,
+            )
         )
         pipette_movement_conflict.check_safe_for_tip_pickup_and_return(
             engine_state=self._engine_client.state,

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

@@ -19,6 +19,8 @@
     DeckSlotLocation,
     OnLabwareLocation,
     WellLocation,
+    LiquidHandlingWellLocation,
+    PickUpTipWellLocation,
     DropTipWellLocation,
 )
 from opentrons.protocol_engine.types import (
@@ -66,7 +68,12 @@ def check_safe_for_pipette_movement(
     pipette_id: str,
     labware_id: str,
     well_name: str,
-    well_location: Union[WellLocation, DropTipWellLocation],
+    well_location: Union[
+        WellLocation,
+        LiquidHandlingWellLocation,
+        PickUpTipWellLocation,
+        DropTipWellLocation,
+    ],
 ) -> None:
     """Check if the labware is safe to move to with a pipette in partial tip configuration.
 

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

@@ -125,17 +125,6 @@ def get_center(self) -> Point:
             well_location=WellLocation(origin=WellOrigin.CENTER),
         )
 
-    def get_meniscus(self, z_offset: float) -> 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),
-            ),
-        )
-
     def load_liquid(
         self,
         liquid: Liquid,

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

@@ -33,6 +33,7 @@ def aspirate(
         rate: float,
         flow_rate: float,
         in_place: bool,
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -55,6 +56,7 @@ def dispense(
         flow_rate: float,
         in_place: bool,
         push_out: Optional[float],
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:

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

@@ -80,6 +80,7 @@ def aspirate(
         rate: float,
         flow_rate: float,
         in_place: bool,
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -122,6 +123,7 @@ def dispense(
         flow_rate: float,
         in_place: bool,
         push_out: Optional[float],
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         """Dispense a given volume of liquid into the specified location.
         Args:

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

@@ -106,10 +106,6 @@ def get_center(self) -> Point:
         """Get the coordinate of the well's center."""
         return self._geometry.center()
 
-    def get_meniscus(self, z_offset: float) -> Point:
-        """This will never be called because it was added in API 2.21."""
-        assert False, "get_meniscus only supported in API 2.21 & later"
-
     def load_liquid(
         self,
         liquid: Liquid,

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

@@ -91,6 +91,7 @@ def aspirate(
         rate: float,
         flow_rate: float,
         in_place: bool,
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any
@@ -132,6 +133,7 @@ def dispense(
         flow_rate: float,
         in_place: bool,
         push_out: Optional[float],
+        is_meniscus: Optional[bool] = None,
     ) -> None:
         if isinstance(location, (TrashBin, WasteChute)):
             raise APIVersionError(

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

@@ -71,10 +71,6 @@ def get_bottom(self, z_offset: float) -> Point:
     def get_center(self) -> Point:
         """Get the coordinate of the well's center."""
 
-    @abstractmethod
-    def get_meniscus(self, z_offset: float) -> Point:
-        """Get the coordinate of the well's meniscus, with an z-offset."""
-
     @abstractmethod
     def load_liquid(
         self,

api/src/opentrons/protocol_api/instrument_context.py

@@ -217,8 +217,9 @@ def aspirate(
             )
         )
 
-        well: Optional[labware.Well] = None
         move_to_location: types.Location
+        well: Optional[labware.Well] = None
+        is_meniscus: Optional[bool] = None
         last_location = self._get_last_location_by_api_version()
         try:
             target = validation.validate_location(
@@ -232,17 +233,13 @@ def aspirate(
                 "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."
             )
+        move_to_location, well, is_meniscus = self._handle_aspirate_target(
+            target=target
+        )
         if self.api_version >= APIVersion(2, 11):
             instrument.validate_takes_liquid(
                 location=move_to_location,
@@ -282,6 +279,7 @@ def aspirate(
                 rate=rate,
                 flow_rate=flow_rate,
                 in_place=target.in_place,
+                is_meniscus=is_meniscus,
             )
 
         return self
@@ -384,6 +382,7 @@ def dispense(  # noqa: C901
             )
         )
         well: Optional[labware.Well] = None
+        is_meniscus: Optional[bool] = None
         last_location = self._get_last_location_by_api_version()
 
         try:
@@ -402,6 +401,7 @@ def dispense(  # noqa: C901
             well = target.well
             if target.location:
                 move_to_location = target.location
+                is_meniscus = target.location.is_meniscus
             elif well.parent._core.is_fixed_trash():
                 move_to_location = target.well.top()
             else:
@@ -467,6 +467,7 @@ def dispense(  # noqa: C901
                 flow_rate=flow_rate,
                 in_place=target.in_place,
                 push_out=push_out,
+                is_meniscus=is_meniscus,
             )
 
         return self
@@ -2191,6 +2192,26 @@ def _raise_if_configuration_not_supported_by_pipette(
                     )
             # SINGLE, QUADRANT and ALL are supported by all pipettes
 
+    def _handle_aspirate_target(
+        self, target: validation.ValidTarget
+    ) -> tuple[types.Location, Optional[labware.Well], Optional[bool]]:
+        move_to_location: types.Location
+        well: Optional[labware.Well] = None
+        is_meniscus: Optional[bool] = None
+        if isinstance(target, validation.WellTarget):
+            well = target.well
+            if target.location:
+                move_to_location = target.location
+                is_meniscus = target.location.is_meniscus
+
+            else:
+                move_to_location = target.well.bottom(
+                    z=self._well_bottom_clearances.aspirate
+                )
+        if isinstance(target, validation.PointTarget):
+            move_to_location = target.location
+        return (move_to_location, well, is_meniscus)
+
 
 class AutoProbeDisable:
     """Use this class to temporarily disable automatic liquid presence detection."""

api/src/opentrons/protocol_api/labware.py

@@ -226,13 +226,13 @@ def meniscus(self, z: float = 0.0) -> 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: A :py:class:`~opentrons.types.Location` that indicates location is meniscus and that holds the ``z`` offset in its point.z field.
 
         :meta private:
         """
-        return Location(self._core.get_meniscus(z_offset=z), self)
+        return Location(
+            point=Point(x=0, y=0, z=z), labware=self, _ot_internal_is_meniscus=True
+        )
 
     @requires_version(2, 8)
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:

api/src/opentrons/protocol_api/validation.py

@@ -435,10 +435,13 @@ class LocationTypeError(TypeError):
     """Error representing that the location supplied is of different expected type."""
 
 
+ValidTarget = Union[WellTarget, PointTarget, TrashBin, WasteChute]
+
+
 def validate_location(
     location: Union[Location, Well, TrashBin, WasteChute, None],
     last_location: Optional[Location],
-) -> Union[WellTarget, PointTarget, TrashBin, WasteChute]:
+) -> ValidTarget:
     """Validate a given location for a liquid handling command.
 
     Args:

api/src/opentrons/protocol_engine/__init__.py

@@ -47,9 +47,12 @@
     LoadedPipette,
     MotorAxis,
     WellLocation,
+    LiquidHandlingWellLocation,
+    PickUpTipWellLocation,
     DropTipWellLocation,
     WellOrigin,
     DropTipWellOrigin,
+    PickUpTipWellOrigin,
     WellOffset,
     ModuleModel,
     ModuleDefinition,
@@ -109,9 +112,12 @@
     "LoadedPipette",
     "MotorAxis",
     "WellLocation",
+    "LiquidHandlingWellLocation",
+    "PickUpTipWellLocation",
     "DropTipWellLocation",
     "WellOrigin",
     "DropTipWellOrigin",
+    "PickUpTipWellOrigin",
     "WellOffset",
     "ModuleModel",
     "ModuleDefinition",

api/src/opentrons/protocol_engine/commands/aspirate.py

@@ -9,7 +9,7 @@
     PipetteIdMixin,
     AspirateVolumeMixin,
     FlowRateMixin,
-    WellLocationMixin,
+    LiquidHandlingWellLocationMixin,
     BaseLiquidHandlingResult,
     DestinationPositionResult,
 )
@@ -38,7 +38,7 @@
 
 
 class AspirateParams(
-    PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, WellLocationMixin
+    PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
 ):
     """Parameters required to aspirate from a specific well."""
 
@@ -112,12 +112,17 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
             )
 
+        well_location = params.wellLocation
+        if well_location.origin == WellOrigin.MENISCUS:
+            well_location.volumeOffset = "operationVolume"
+
         position = await self._movement.move_to_well(
             pipette_id=pipette_id,
             labware_id=labware_id,
             well_name=well_name,
-            well_location=params.wellLocation,
+            well_location=well_location,
             current_well=current_well,
+            operation_volume=-params.volume,
         )
         deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
         state_update.set_pipette_location(