refactor(api): silently recover from invalid liquid heights after aspirating/dispensing (#18292)

Author: caila-marashaj

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

@@ -197,12 +197,14 @@ def estimate_liquid_height_after_pipetting(
             raise PipetteNotAttachedError(f"No pipette present on mount {mount}")
         pipette_id = pipette_from_mount.id
         starting_liquid_height = self.current_liquid_height()
-        projected_final_height = self._engine_client.state.geometry.get_well_height_after_liquid_handling_no_error(
-            labware_id=labware_id,
-            well_name=well_name,
-            pipette_id=pipette_id,
-            initial_height=starting_liquid_height,
-            volume=operation_volume,
+        projected_final_height = (
+            self._engine_client.state.geometry.get_well_height_after_liquid_handling(
+                labware_id=labware_id,
+                well_name=well_name,
+                pipette_id=pipette_id,
+                initial_height=starting_liquid_height,
+                volume=operation_volume,
+            )
         )
         return projected_final_height
 

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

@@ -423,7 +423,8 @@ def _find_height_in_partial_frustum(
     # if we finish looping through the whole well, bottom_section will be the well's volume
     total_well_volume = bottom_section_volume
     # if we've looked through all sections and can't find the target volume, raise an error
-    # also this code should never be reached bc an error should be raised by find_height_at_well_volume
+    # also this code should never be reached bc an invalid target volume should be changed
+    # by find_height_at_well_volume
     raise InvalidLiquidHeightFound(
         f"Target volume {target_volume} uL exceeds the well volume {total_well_volume} uL."
     )
@@ -432,7 +433,6 @@ def _find_height_in_partial_frustum(
 def find_height_at_well_volume(
     target_volume: LiquidTrackingType,
     well_geometry: InnerWellGeometry,
-    raise_error_if_result_invalid: bool = True,
 ) -> LiquidTrackingType:
     """Find the height within a well, at a known volume."""
     # comparisons with SimulatedProbeResult objects aren't meaningful, just
@@ -443,12 +443,10 @@ def find_height_at_well_volume(
     volumetric_capacity = get_well_volumetric_capacity(well_geometry)
     max_volume = sum(row[1] for row in volumetric_capacity)
 
-    if raise_error_if_result_invalid:
-        if target_volume < 0 or target_volume > max_volume:
-            raise InvalidLiquidHeightFound(
-                f"Invalid target volume {target_volume} uL; max volume is {max_volume} uL"
-            )
-
+    if target_volume < 0:
+        target_volume = 0
+    elif target_volume > max_volume:
+        target_volume = max_volume
     sorted_well = sorted(well_geometry.sections, key=lambda section: section.topHeight)
     # find the section the target volume is in and compute the height
     return _find_height_in_partial_frustum(

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

@@ -7,7 +7,6 @@
 from typing import Optional, List, Tuple, Union, cast, TypeVar, Dict, Set
 from dataclasses import dataclass
 from functools import cached_property
-from math import isclose
 
 from opentrons.types import (
     Point,
@@ -33,7 +32,6 @@
     LabwareNotLoadedOnLabwareError,
     LabwareNotLoadedOnModuleError,
     LabwareMovementNotAllowedError,
-    OperationLocationNotInWellError,
     InvalidLabwarePositionError,
     LabwareNotOnDeckError,
 )
@@ -522,39 +520,23 @@ def get_labware_position(self, labware_id: str) -> Point:
             z=origin_pos.z + cal_offset.z,
         )
 
-    def validate_well_position(
+    def _validate_well_position(
         self,
-        well_location: WellLocationType,
-        z_offset: float,
-        pipette_id: Optional[str] = None,
-    ) -> None:
-        """Raise exception if operation location is not within well.
-
-        Primarily this checks if there is not enough liquid in a well to do meniscus-relative static aspiration.
-        """
-        if well_location.origin == WellOrigin.MENISCUS:
-            assert pipette_id is not None, "pipette id is None"
-            lld_min_height = self._pipettes.get_current_tip_lld_settings(
-                pipette_id=pipette_id
-            )
-            if z_offset < lld_min_height:
-                if isinstance(well_location, LiquidHandlingWellLocation):
-                    raise OperationLocationNotInWellError(
-                        f"Specifying {well_location.origin} with a height offset of {well_location.offset.z} results in a height of {z_offset} mm; the minimum allowed height for liquid tracking is {lld_min_height} mm"
-                    )
-                else:
-                    raise OperationLocationNotInWellError(
-                        f"Specifying {well_location.origin} with an offset of {well_location.offset} results in an operation location that could be below the bottom of the well"
-                    )
-        elif z_offset < 0 and not isclose(z_offset, 0, abs_tol=0.0000001):
-            if isinstance(well_location, LiquidHandlingWellLocation):
-                raise OperationLocationNotInWellError(
-                    f"Specifying {well_location.origin} with an offset of {well_location.offset} and a volume offset of {well_location.volumeOffset} results in an operation location below the bottom of the well"
-                )
-            else:
-                raise OperationLocationNotInWellError(
-                    f"Specifying {well_location.origin} with an offset of {well_location.offset} results in an operation location below the bottom of the well"
-                )
+        target_height: LiquidTrackingType,  # height in mm inside a well relative to the bottom
+        well_max_height: float,
+        pipette_id: str,
+    ) -> LiquidTrackingType:
+        """If well offset would be outside the bounds of a well, silently bring it back to the boundary."""
+        if isinstance(target_height, SimulatedProbeResult):
+            return target_height
+        lld_min_height = self._pipettes.get_current_tip_lld_settings(
+            pipette_id=pipette_id
+        )
+        if target_height < lld_min_height:
+            target_height = lld_min_height
+        elif target_height > well_max_height:
+            target_height = well_max_height
+        return target_height
 
     def validate_probed_height(
         self,
@@ -595,7 +577,7 @@ def get_well_position(
 
         offset = WellOffset(x=0, y=0, z=well_depth)
         if well_location is not None:
-            offset = well_location.offset
+            offset = well_location.offset  # location of the bottom of the well
             offset_adjustment = self.get_well_offset_adjustment(
                 labware_id=labware_id,
                 well_name=well_name,
@@ -606,11 +588,6 @@ def get_well_position(
             )
             if not isinstance(offset_adjustment, SimulatedProbeResult):
                 offset = offset.model_copy(update={"z": offset.z + offset_adjustment})
-                self.validate_well_position(
-                    well_location=well_location,
-                    z_offset=offset.z,
-                    pipette_id=pipette_id,
-                )
         return Point(
             x=labware_pos.x + offset.x + well_def.x,
             y=labware_pos.y + offset.y + well_def.y,
@@ -2107,6 +2084,8 @@ def get_well_height_after_liquid_handling(
 
         This is given an initial handling height, with reference to the well bottom.
         """
+        well_def = self._labware.get_well_definition(labware_id, well_name)
+        well_depth = well_def.depth
         well_geometry = self._labware.get_well_geometry(
             labware_id=labware_id, well_name=well_name
         )
@@ -2122,49 +2101,17 @@ def get_well_height_after_liquid_handling(
                     pipette_id=pipette_id,
                 )
             )
-            return find_height_at_well_volume(
+            # NOTE(cm): if final_volume is outside the bounds of the well, it will get
+            # adjusted inside find_height_at_well_volume to accomodate well the height
+            # calculation.
+            height_inside_well = find_height_at_well_volume(
                 target_volume=final_volume, well_geometry=well_geometry
             )
-        except InvalidLiquidHeightFound as _exception:
-            raise InvalidLiquidHeightFound(
-                message=_exception.message
-                + f"for well {well_name} of {self._labware.get_display_name(labware_id)} on slot {self.get_ancestor_slot_name(labware_id)}"
-            )
-
-    def get_well_height_after_liquid_handling_no_error(
-        self,
-        labware_id: str,
-        well_name: str,
-        pipette_id: str,
-        initial_height: LiquidTrackingType,
-        volume: float,
-    ) -> LiquidTrackingType:
-        """Return what the height of liquid in a labware well after liquid handling will be.
-
-        This raises no error if the value returned is an invalid physical location, so it should never be
-        used for navigation, only for a pre-emptive estimate.
-        """
-        well_geometry = self._labware.get_well_geometry(
-            labware_id=labware_id, well_name=well_name
-        )
-        try:
-            initial_volume = find_volume_at_well_height(
-                target_height=initial_height, well_geometry=well_geometry
-            )
-            final_volume = initial_volume + (
-                volume
-                * self.get_nozzles_per_well(
-                    labware_id=labware_id,
-                    target_well_name=well_name,
-                    pipette_id=pipette_id,
-                )
-            )
-            well_volume = find_height_at_well_volume(
-                target_volume=final_volume,
-                well_geometry=well_geometry,
-                raise_error_if_result_invalid=False,
+            return self._validate_well_position(
+                target_height=height_inside_well,
+                well_max_height=well_depth,
+                pipette_id=pipette_id,
             )
-            return well_volume
         except InvalidLiquidHeightFound as _exception:
             raise InvalidLiquidHeightFound(
                 message=_exception.message

api/tests/opentrons/protocol_api/core/engine/test_well_core.py

@@ -360,7 +360,7 @@ def test_estimate_liquid_height_after_pipetting(
     fake_final_height = 10000000
     decoy.when(subject.current_liquid_height()).then_return(initial_liquid_height)
     decoy.when(
-        mock_engine_client.state.geometry.get_well_height_after_liquid_handling_no_error(
+        mock_engine_client.state.geometry.get_well_height_after_liquid_handling(
             labware_id="labware-id",
             well_name="well-name",
             pipette_id="pipette-id",