update

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

@@ -111,7 +111,7 @@ def set_default_speed(self, speed: float) -> None:
 
     def aspirate(
         self,
-        location: Location,
+        location: Union[Location, WellLocation],
         well_core: Optional[WellCore],
         volume: float,
         rate: float,
@@ -129,6 +129,7 @@ def aspirate(
         """
         if well_core is None:
             if not in_place:
+                assert isinstance(location, Location)
                 self._engine_client.execute_command(
                     cmd.MoveToCoordinatesParams(
                         pipetteId=self._pipette_id,
@@ -150,14 +151,17 @@ def aspirate(
         else:
             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,
+            if isinstance(location, WellLocation):
+                well_location = location
+                location = Location(well_core.get_meniscus(z_offset=location.offset.z))
+            else:
+                well_location = (
+                    self._engine_client.state.geometry.get_relative_well_location(
+                        labware_id=labware_id,
+                        well_name=well_name,
+                        absolute_point=location.point,
+                    )
                 )
-            )
             deck_conflict.check_safe_for_pipette_movement(
                 engine_state=self._engine_client.state,
                 pipette_id=self._pipette_id,
@@ -176,6 +180,7 @@ def aspirate(
                 )
             )
 
+        assert isinstance(location, Location)
         self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
     def dispense(

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

@@ -24,6 +24,7 @@ class WellCore(AbstractWellCore):
         engine_client: Synchronous ProtocolEngine client.
     """
 
+    # get and store parent like WellCore? For InstrumentCore.aspirate work
     def __init__(self, name: str, labware_id: str, engine_client: EngineClient) -> None:
         self._labware_id = labware_id
         self._engine_client = engine_client

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,
@@ -160,9 +161,6 @@ def default_speed(self) -> float:
     def default_speed(self, speed: float) -> None:
         self._core.set_default_speed(speed)
 
-    # we do volume tracking already
-    # update old_well/new_well heights/volumes after (successful) liquid handling actions (using handled volume)
-    # (initially) populate well heights/volumes via user input?! Error (and recovery) if not enough volume to liquid probe?
     @requires_version(2, 0)
     def aspirate(
         self,
@@ -2143,7 +2141,10 @@ def _determine_aspirate_move_to_location(
                     offset_from_meniscus_mm=offset_from_meniscus_mm,
                     volume=volume,
                 ):
-                    move_to_location = target.well.meniscus(z=offset_from_meniscus_mm)
+                    move_to_location = WellLocation(
+                        origin=WellOrigin.MENISCUS,
+                        offset=WellOffset(x=0, y=0, z=offset_from_meniscus_mm),
+                    )
         if isinstance(target, validation.PointTarget):
             move_to_location = target.location
         if self.api_version >= APIVersion(2, 11):
@@ -2185,10 +2186,6 @@ def _liquid_probe_before_aspirate(
             height = self._core.get_last_measured_liquid_height(well_core=well._core)
             if height is None:
                 self.measure_liquid_height(well=well)
-            # new parameters fit into location param, sent to Protocol Engine command (via well.meniscus symbol)
-            # need below? well.meniscus
-            # convert height to volume, subtract `volume`, convert volume to height, use as offset below
-            # raise error if not enough liquid present to aspirate desired volume. Error Recovery option
             return True
         elif (
             self.api_version >= APIVersion(2, 20)

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

@@ -86,6 +86,8 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
         pipette_id = params.pipetteId
         labware_id = params.labwareId
         well_name = params.wellName
+        well_location = params.wellLocation
+        volume = params.volume
 
         ready_to_aspirate = self._pipetting.get_is_ready_to_aspirate(
             pipette_id=pipette_id
@@ -112,12 +114,19 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
             )
 
+        # PE stuff first, make it's own PR (WellVolumeOffset, tests). Below not needed. Test on robot before Protocol API work
+        if well_location.origin == WellOrigin.MENISCUS:
+            # update well_location based on geometry maths. Create WellOrigin.MENISCUS_AT_VOLUME?
+            # well_location.offset = self._state_view._geometry.get_well_position_at_volume(well_location, volume)
+            pass
+
         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=-volume,
         )
         deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
         state_update.set_pipette_location(
@@ -130,7 +139,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
         try:
             volume_aspirated = await self._pipetting.aspirate_in_place(
                 pipette_id=pipette_id,
-                volume=params.volume,
+                volume=volume,
                 flow_rate=params.flowRate,
                 command_note_adder=self._command_note_adder,
             )

api/src/opentrons/protocol_engine/execution/movement.py

@@ -71,6 +71,7 @@ async def move_to_well(
         force_direct: bool = False,
         minimum_z_height: Optional[float] = None,
         speed: Optional[float] = None,
+        operation_volume: Optional[float] = None,
     ) -> Point:
         """Move to a specific well."""
         self._state_store.labware.raise_if_labware_inaccessible_by_pipette(

api/src/opentrons/protocol_engine/types.py

@@ -239,11 +239,18 @@ class WellOffset(BaseModel):
     z: float = 0
 
 
+class WellVolumeOffset(BaseModel):
+    """A z-offset to account for volume in an operation."""
+
+    volumeOffset: Union[float, Literal["operationVolume"]] = 0
+
+
 class WellLocation(BaseModel):
     """A relative location in reference to a well's location."""
 
     origin: WellOrigin = WellOrigin.TOP
     offset: WellOffset = Field(default_factory=WellOffset)
+    volumeOffset: WellVolumeOffset = Field(default_factory=WellVolumeOffset)
 
 
 class DropTipWellLocation(BaseModel):