feat(api): add WellVolumeOffset to WellLocation

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

@@ -118,6 +118,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
             well_name=well_name,
             well_location=params.wellLocation,
             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/liquid_probe.py

@@ -13,11 +13,10 @@
     PipetteLiquidNotFoundError,
 )
 
-from ..types import DeckPoint
+from ..types import LiquidProbeWellLocation, DeckPoint
 from .pipetting_common import (
     LiquidNotFoundError,
     PipetteIdMixin,
-    WellLocationMixin,
     DestinationPositionResult,
 )
 from .command import (
@@ -42,8 +41,13 @@
 # Both command variants should have identical parameters.
 # But we need two separate parameter model classes because
 # `command_unions.CREATE_TYPES_BY_PARAMS_TYPE` needs to be a 1:1 mapping.
-class _CommonParams(PipetteIdMixin, WellLocationMixin):
-    pass
+class _CommonParams(PipetteIdMixin):
+    labwareId: str = Field(..., description="Identifier of labware to use.")
+    wellName: str = Field(..., description="Name of well to use in labware.")
+    wellLocation: LiquidProbeWellLocation = Field(
+        default_factory=LiquidProbeWellLocation,
+        description="Relative well location at which to liquid probe.",
+    )
 
 
 class LiquidProbeParams(_CommonParams):

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

@@ -9,10 +9,9 @@
 from ..errors import ErrorOccurrence, TipNotAttachedError
 from ..resources import ModelUtils
 from ..state import update_types
-from ..types import DeckPoint
+from ..types import PickUpTipWellLocation, DeckPoint
 from .pipetting_common import (
     PipetteIdMixin,
-    WellLocationMixin,
     DestinationPositionResult,
 )
 from .command import (
@@ -31,10 +30,15 @@
 PickUpTipCommandType = Literal["pickUpTip"]
 
 
-class PickUpTipParams(PipetteIdMixin, WellLocationMixin):
+class PickUpTipParams(PipetteIdMixin):
     """Payload needed to move a pipette to a specific well."""
 
-    pass
+    labwareId: str = Field(..., description="Identifier of labware to use.")
+    wellName: str = Field(..., description="Name of well to use in labware.")
+    wellLocation: PickUpTipWellLocation = Field(
+        default_factory=PickUpTipWellLocation,
+        description="Relative well location at which to pick up the tip.",
+    )
 
 
 class PickUpTipResult(DestinationPositionResult):

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(
@@ -129,6 +130,7 @@ async def move_to_well(
             current_well=current_well,
             force_direct=force_direct,
             minimum_z_height=minimum_z_height,
+            operation_volume=operation_volume,
         )
 
         speed = self._state_store.pipettes.get_movement_speed(

api/src/opentrons/protocol_engine/state/geometry.py

@@ -420,6 +420,7 @@ def get_well_position(
         labware_id: str,
         well_name: str,
         well_location: Optional[WellLocation] = None,
+        operation_volume: Optional[float] = None,
     ) -> Point:
         """Given relative well location in a labware, get absolute position."""
         labware_pos = self.get_labware_position(labware_id)
@@ -434,11 +435,18 @@ def get_well_position(
             elif well_location.origin == WellOrigin.CENTER:
                 offset = offset.copy(update={"z": offset.z + well_depth / 2.0})
             elif well_location.origin == WellOrigin.MENISCUS:
-                liquid_height = self._wells.get_last_measured_liquid_height(
+                starting_liquid_height = self._wells.get_last_measured_liquid_height(
                     labware_id, well_name
                 )
-                if liquid_height is not None:
-                    offset = offset.copy(update={"z": offset.z + liquid_height})
+                if starting_liquid_height is not None:
+                    if well_location.volumeOffset.volumeOffset == "operationVolume":
+                        volume = operation_volume or 0.0
+                    else:
+                        volume = well_location.volumeOffset.volumeOffset
+                    height_after_operation = self.get_height_after_volume(labware_id=labware_id, well_name=well_name, starting_height=starting_liquid_height, volume=volume)
+                    offset = offset.copy(
+                        update={"z": offset.z + height_after_operation}
+                    )
                 else:
                     raise errors.LiquidHeightUnknownError(
                         "Must liquid probe before specifying WellOrigin.MENISCUS."
@@ -1218,3 +1226,24 @@ def get_well_volumetric_capacity(
                 message=f"No InnerWellGeometry found for well id: {well_id}"
             )
         return get_well_volumetric_capacity(well_geometry)
+
+    def get_volume_at_height(
+        self, labware_id: str, well_id: str, target_height: float
+    ) -> float:
+        pass
+
+    def get_height_at_volume(
+        self, labware_id: str, well_id: str, target_volume: float
+    ) -> float:
+        pass
+
+    def get_height_after_volume(
+        self, labware_id: str, well_name: str, starting_height: float, volume: float
+    ) -> float:
+        well_def = self._labware.get_well_definition(labware_id, well_name)
+        well_id = well_def.geometryDefinitionId
+        starting_volume = self.get_volume_at_height(labware_id=labware_id, well_id=well_id, target_height=starting_height)
+        ending_volume = starting_volume + volume
+        ending_height = self.get_height_at_volume(labware_id=labware_id, well_id=well_id, target_volume=ending_volume)
+        # return ending_height
+        return starting_height # delete, use line above once sub-methods implemented

api/src/opentrons/protocol_engine/state/motion.py

@@ -96,6 +96,7 @@ def get_movement_waypoints_to_well(
         current_well: Optional[CurrentWell] = None,
         force_direct: bool = False,
         minimum_z_height: Optional[float] = None,
+        operation_volume: Optional[float] = None,
     ) -> List[motion_planning.Waypoint]:
         """Calculate waypoints to a destination that's specified as a well."""
         location = current_well or self._pipettes.get_current_location()
@@ -107,9 +108,10 @@ def get_movement_waypoints_to_well(
             destination_cp = CriticalPoint.XY_CENTER
 
         destination = self._geometry.get_well_position(
-            labware_id,
-            well_name,
-            well_location,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
+            operation_volume=operation_volume,
         )
 
         move_type = _move_types.get_move_type_to_well(

api/src/opentrons/protocol_engine/types.py

@@ -239,8 +239,31 @@ class WellOffset(BaseModel):
     z: float = 0
 
 
+# get rid of operationVolume below and operation_volume parameter in methods?
+# TODO(pm, 2024-09-23): PE should raise error if volumeOffset specified with a tip rack location
 class WellLocation(BaseModel):
-    """A relative location in reference to a well's location."""
+    """A relative location in reference to a well's location.""" # update
+
+    origin: WellOrigin = WellOrigin.TOP
+    offset: WellOffset = Field(default_factory=WellOffset)
+    volumeOffset: Union[float, Literal["operationVolume"]] = Field(default=0.0, description="""A volume of liquid to account for when 
+                                executing commands with an origin of WellOrigin.MENISCUS. Specifying 
+                                `operationVolume` results in this class acting as a sentinel and should 
+                                be used when volume can be determined from the command parameters, for 
+                                example commanding Aspirate. A volume should be specified when it cannot 
+                                be determined from the command parameters, for example commanding 
+                                MoveToWell prior to AspirateInPlace.""") # update comment
+
+
+class LiquidProbeWellLocation(BaseModel):
+    """A relative location in reference to a well's location.""" # update
+
+    origin: WellOrigin = WellOrigin.TOP
+    offset: WellOffset = Field(default_factory=WellOffset)
+
+
+class PickUpTipWellLocation(BaseModel):
+    """A relative location in reference to a well's location.""" # update
 
     origin: WellOrigin = WellOrigin.TOP
     offset: WellOffset = Field(default_factory=WellOffset)

api/tests/opentrons/protocol_engine/commands/test_aspirate.py

@@ -79,6 +79,7 @@ async def test_aspirate_implementation_no_prep(
             well_name="A3",
             well_location=location,
             current_well=None,
+            operation_volume=-50,
         ),
     ).then_return(Point(x=1, y=2, z=3))
 
@@ -145,6 +146,7 @@ async def test_aspirate_implementation_with_prep(
                 labware_id="123",
                 well_name="A3",
             ),
+            operation_volume=-50,
         ),
     ).then_return(Point(x=1, y=2, z=3))
 
@@ -210,6 +212,7 @@ async def test_aspirate_raises_volume_error(
             well_name="A3",
             well_location=location,
             current_well=None,
+            operation_volume=-50,
         ),
     ).then_return(Point(1, 2, 3))
 
@@ -267,6 +270,7 @@ async def test_overpressure_error(
             well_name=well_name,
             well_location=well_location,
             current_well=None,
+            operation_volume=-50,
         ),
     ).then_return(position)
 

api/tests/opentrons/protocol_engine/execution/test_movement_handler.py

@@ -149,6 +149,7 @@ async def test_move_to_well(
             current_well=None,
             force_direct=True,
             minimum_z_height=12.3,
+            operation_volume=None,
         )
     ).then_return(
         [Waypoint(Point(1, 2, 3), CriticalPoint.XY_CENTER), Waypoint(Point(4, 5, 6))]
@@ -257,6 +258,7 @@ async def test_move_to_well_from_starting_location(
             well_location=well_location,
             force_direct=False,
             minimum_z_height=None,
+            operation_volume=None,
         )
     ).then_return([Waypoint(Point(1, 2, 3), CriticalPoint.XY_CENTER)])
 

api/tests/opentrons/protocol_engine/state/test_geometry_view.py

@@ -1543,6 +1543,7 @@ def test_get_well_position_with_meniscus_offset(
             origin=WellOrigin.MENISCUS,
             offset=WellOffset(x=2, y=3, z=4),
         ),
+        operation_volume=0.0,
     )
 
     assert result == Point(

api/tests/opentrons/protocol_engine/state/test_motion_view.py

@@ -309,7 +309,7 @@ def test_get_movement_waypoints_to_well_for_y_center(
     ).then_return(False)
 
     decoy.when(
-        geometry_view.get_well_position("labware-id", "well-name", WellLocation())
+        geometry_view.get_well_position("labware-id", "well-name", WellLocation(), None)
     ).then_return(Point(x=4, y=5, z=6))
 
     decoy.when(
@@ -391,7 +391,7 @@ def test_get_movement_waypoints_to_well_for_xy_center(
     ).then_return(True)
 
     decoy.when(
-        geometry_view.get_well_position("labware-id", "well-name", WellLocation())
+        geometry_view.get_well_position("labware-id", "well-name", WellLocation(), None)
     ).then_return(Point(x=4, y=5, z=6))
 
     decoy.when(
@@ -460,6 +460,7 @@ def test_get_movement_waypoints_to_well_raises(
             labware_id="labware-id",
             well_name="A1",
             well_location=None,
+            operation_volume=None,
         )
     ).then_return(Point(x=4, y=5, z=6))
     decoy.when(pipette_view.get_current_location()).then_return(None)

shared-data/command/schemas/9.json

@@ -318,6 +318,26 @@
         }
       }
     },
+    "WellVolumeOffset": {
+      "title": "WellVolumeOffset",
+      "description": "A volume of liquid to account for when executing commands with an origin of WellOrigin.MENISCUS.\n\nSpecifying `operationVolume` results in this class acting as a sentinel and should be used when\nvolume can be determined from the command parameters, for example commanding Aspirate. A volume\nshould be specified when it cannot be determined from the command parameters, for example commanding\nMoveToWell prior to AspirateInPlace.",
+      "type": "object",
+      "properties": {
+        "volumeOffset": {
+          "title": "Volumeoffset",
+          "default": 0.0,
+          "anyOf": [
+            {
+              "type": "number"
+            },
+            {
+              "enum": ["operationVolume"],
+              "type": "string"
+            }
+          ]
+        }
+      }
+    },
     "WellLocation": {
       "title": "WellLocation",
       "description": "A relative location in reference to a well's location.",
@@ -333,6 +353,9 @@
         },
         "offset": {
           "$ref": "#/definitions/WellOffset"
+        },
+        "volumeOffset": {
+          "$ref": "#/definitions/WellVolumeOffset"
         }
       }
     },