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,13 +147,13 @@ 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,
             )
+            if is_meniscus:
+                well_location.volumeOffset = "operationVolume"
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
                 pipette_id=self._pipette_id,
@@ -237,12 +238,10 @@ 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,
             )
             pipette_movement_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
@@ -416,10 +415,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/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:

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:

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

api/src/opentrons/protocol_api/instrument_context.py

@@ -217,6 +217,7 @@ def aspirate(
             )
         )
 
+        is_meniscus: Optional[bool] = None
         well: Optional[labware.Well] = None
         move_to_location: types.Location
         last_location = self._get_last_location_by_api_version()
@@ -237,6 +238,7 @@ def aspirate(
                 z=self._well_bottom_clearances.aspirate
             )
             well = target.well
+            is_meniscus = target.is_meniscus
         if isinstance(target, validation.PointTarget):
             move_to_location = target.location
         if isinstance(target, (TrashBin, WasteChute)):
@@ -282,6 +284,7 @@ def aspirate(
                 rate=rate,
                 flow_rate=flow_rate,
                 in_place=target.in_place,
+                is_meniscus=is_meniscus if is_meniscus is not None else None,
             )
 
         return self

api/src/opentrons/protocol_api/labware.py

@@ -232,7 +232,7 @@ def meniscus(self, z: float = 0.0) -> Location:
 
         :meta private:
         """
-        return Location(self._core.get_meniscus(z_offset=z), self)
+        return Location(self._core.get_meniscus(z_offset=z), self, True)
 
     @requires_version(2, 8)
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:

api/src/opentrons/protocol_api/validation.py

@@ -418,6 +418,7 @@ class WellTarget(NamedTuple):
     well: Well
     location: Optional[Location]
     in_place: bool
+    is_meniscus: Optional[bool] = None
 
 
 class PointTarget(NamedTuple):
@@ -476,7 +477,12 @@ def validate_location(
     _, well = target_location.labware.get_parent_labware_and_well()
 
     return (
-        WellTarget(well=well, location=target_location, in_place=in_place)
+        WellTarget(
+            well=well,
+            location=target_location,
+            in_place=in_place,
+            is_meniscus=target_location.is_meniscus,
+        )
         if well is not None
         else PointTarget(location=target_location, in_place=in_place)
     )

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(

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

@@ -13,7 +13,7 @@
     PipetteIdMixin,
     DispenseVolumeMixin,
     FlowRateMixin,
-    WellLocationMixin,
+    LiquidHandlingWellLocationMixin,
     BaseLiquidHandlingResult,
     DestinationPositionResult,
     OverpressureError,
@@ -30,13 +30,14 @@
 if TYPE_CHECKING:
     from ..execution import MovementHandler, PipettingHandler
     from ..resources import ModelUtils
+    from ..state.state import StateView
 
 
 DispenseCommandType = Literal["dispense"]
 
 
 class DispenseParams(
-    PipetteIdMixin, DispenseVolumeMixin, FlowRateMixin, WellLocationMixin
+    PipetteIdMixin, DispenseVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
 ):
     """Payload required to dispense to a specific well."""
 
@@ -63,37 +64,50 @@ class DispenseImplementation(AbstractCommandImpl[DispenseParams, _ExecuteReturn]
 
     def __init__(
         self,
+        state_view: StateView,
         movement: MovementHandler,
         pipetting: PipettingHandler,
         model_utils: ModelUtils,
         **kwargs: object,
     ) -> None:
+        self._state_view = state_view
         self._movement = movement
         self._pipetting = pipetting
         self._model_utils = model_utils
 
     async def execute(self, params: DispenseParams) -> _ExecuteReturn:
         """Move to and dispense to the requested well."""
         state_update = StateUpdate()
+        well_location = params.wellLocation
+        labware_id = params.labwareId
+        well_name = params.wellName
+        volume = params.volume
+
+        self._state_view.geometry.validate_dispense_volume_into_well(
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
+            volume=volume,
+        )
 
         position = await self._movement.move_to_well(
             pipette_id=params.pipetteId,
-            labware_id=params.labwareId,
-            well_name=params.wellName,
-            well_location=params.wellLocation,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
         )
         deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
         state_update.set_pipette_location(
             pipette_id=params.pipetteId,
-            new_labware_id=params.labwareId,
-            new_well_name=params.wellName,
+            new_labware_id=labware_id,
+            new_well_name=well_name,
             new_deck_point=deck_point,
         )
 
         try:
             volume = await self._pipetting.dispense_in_place(
                 pipette_id=params.pipetteId,
-                volume=params.volume,
+                volume=volume,
                 flow_rate=params.flowRate,
                 push_out=params.pushOut,
             )

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

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING, Optional, Type
 from typing_extensions import Literal
 
-from ..types import DeckPoint
+from ..types import DeckPoint, WellOrigin
 from .pipetting_common import (
     PipetteIdMixin,
     WellLocationMixin,
@@ -13,9 +13,11 @@
 from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ..errors.error_occurrence import ErrorOccurrence
 from ..state import update_types
+from ..errors import LabwareIsTipRackError
 
 if TYPE_CHECKING:
     from ..execution import MovementHandler
+    from ..state.state import StateView
 
 MoveToWellCommandType = Literal["moveToWell"]
 
@@ -37,29 +39,44 @@ class MoveToWellImplementation(
 ):
     """Move to well command implementation."""
 
-    def __init__(self, movement: MovementHandler, **kwargs: object) -> None:
+    def __init__(
+        self, state_view: StateView, movement: MovementHandler, **kwargs: object
+    ) -> None:
+        self._state_view = state_view
         self._movement = movement
 
     async def execute(
         self, params: MoveToWellParams
     ) -> SuccessData[MoveToWellResult, None]:
         """Move the requested pipette to the requested well."""
+        pipette_id = params.pipetteId
+        labware_id = params.labwareId
+        well_name = params.wellName
+        well_location = params.wellLocation
+
         state_update = update_types.StateUpdate()
 
+        if (self._state_view.labware.is_tiprack(labware_id)) and (
+            well_location.origin == WellOrigin.MENISCUS or well_location.volumeOffset
+        ):
+            raise LabwareIsTipRackError(
+                "Cannot specify a WellLocation with an origin of MENISCUS or any volumeOffset with movement to a tip rack"
+            )
+
         x, y, z = await self._movement.move_to_well(
-            pipette_id=params.pipetteId,
-            labware_id=params.labwareId,
-            well_name=params.wellName,
-            well_location=params.wellLocation,
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
             force_direct=params.forceDirect,
             minimum_z_height=params.minimumZHeight,
             speed=params.speed,
         )
         deck_point = DeckPoint.construct(x=x, y=y, z=z)
         state_update.set_pipette_location(
-            pipette_id=params.pipetteId,
-            new_labware_id=params.labwareId,
-            new_well_name=params.wellName,
+            pipette_id=pipette_id,
+            new_labware_id=labware_id,
+            new_well_name=well_name,
             new_deck_point=deck_point,
         )