feat(api): add WellVolumeOffset to WellLocation

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

@@ -86,6 +86,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
         pipette_id = params.pipetteId
         labware_id = params.labwareId
         well_name = params.wellName
+        volume = params.volume
 
         ready_to_aspirate = self._pipetting.get_is_ready_to_aspirate(
             pipette_id=pipette_id
@@ -118,6 +119,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
             well_name=well_name,
             well_location=params.wellLocation,
             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 +132,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(
@@ -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,21 @@ 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_ending_height(labware_id, well_name, starting_liquid_height, volume)
+                    height_after_operation = starting_liquid_height  # delete, use above method once implemented
+                    if volume:  # delete
+                        pass  # delete
+                    offset = offset.copy(
+                        update={"z": offset.z + height_after_operation}
+                    )
                 else:
                     raise errors.LiquidHeightUnknownError(
                         "Must liquid probe before specifying WellOrigin.MENISCUS."

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,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.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):

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 z-offset to account for volume in an operation.",
+      "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"
         }
       }
     },