feat(api): add new user-facing helper functions for static liquid tracking (#17360)

Author: caila-marashaj

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

@@ -1060,6 +1060,27 @@ def detect_liquid_presence(self, well_core: WellCore, loc: Location) -> bool:
 
         return result.z_position is not None
 
+    def get_minimum_liquid_sense_height(self) -> float:
+        attached_tip = self._engine_client.state.pipettes.get_attached_tip(
+            self._pipette_id
+        )
+        if attached_tip:
+            tip_volume = attached_tip.volume
+        else:
+            raise TipNotAttachedError(
+                "Need to have a tip attached for liquid-sense operations."
+            )
+        lld_settings = self._engine_client.state.pipettes.get_pipette_lld_settings(
+            pipette_id=self.pipette_id
+        )
+        if lld_settings:
+            lld_min_height_for_tip_attached = lld_settings[f"t{tip_volume}"][
+                "minHeight"
+            ]
+            return lld_min_height_for_tip_attached
+        else:
+            raise ValueError("liquid-level detection settings not found.")
+
     def liquid_probe_with_recovery(self, well_core: WellCore, loc: Location) -> None:
         labware_id = well_core.labware_id
         well_name = well_core.get_name()

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

@@ -155,3 +155,35 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
             y_ratio=y,
             z_ratio=z,
         )
+
+    def estimate_liquid_height_after_pipetting(
+        self,
+        operation_volume: float,
+    ) -> float:
+        """Return an estimate of liquid height after pipetting without raising an error."""
+        labware_id = self.labware_id
+        well_name = self._name
+        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,
+            initial_height=starting_liquid_height,
+            volume=operation_volume,
+        )
+        return projected_final_height
+
+    def current_liquid_height(self) -> float:
+        """Return the current liquid height within a well."""
+        labware_id = self.labware_id
+        well_name = self._name
+        return self._engine_client.state.geometry.get_meniscus_height(
+            labware_id=labware_id, well_name=well_name
+        )
+
+    def get_liquid_volume(self) -> float:
+        """Return the current volume in a well."""
+        labware_id = self.labware_id
+        well_name = self._name
+        return self._engine_client.state.geometry.get_current_well_volume(
+            labware_id=labware_id, well_name=well_name
+        )

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

@@ -243,6 +243,10 @@ def get_current_volume(self) -> float:
     def get_available_volume(self) -> float:
         ...
 
+    @abstractmethod
+    def get_minimum_liquid_sense_height(self) -> float:
+        ...
+
     @abstractmethod
     def get_hardware_state(self) -> PipetteDict:
         """Get the current state of the pipette hardware as a dictionary."""

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -667,3 +667,14 @@ def _pressure_supported_by_pipette(self) -> bool:
     def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""
         return False
+
+    def get_minimum_liquid_sense_height(self) -> float:
+        return 0.0
+
+    def estimate_liquid_height(
+        self,
+        well_core: LegacyWellCore,
+        starting_liquid_height: float,
+        operation_volume: float,
+    ) -> float:
+        return 0.0

api/src/opentrons/protocol_api/core/legacy/legacy_well_core.py

@@ -118,6 +118,21 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """Gets point in deck coordinates based on percentage of the radius of each axis."""
         return self._geometry.from_center_cartesian(x, y, z)
 
+    def estimate_liquid_height_after_pipetting(
+        self,
+        operation_volume: float,
+    ) -> float:
+        """Estimate what the liquid height will be after pipetting, without raising an error."""
+        return 0.0
+
+    def current_liquid_height(self) -> float:
+        """Get the current liquid height."""
+        return 0.0
+
+    def get_liquid_volume(self) -> float:
+        """Get the current well volume."""
+        return 0.0
+
     # TODO(mc, 2022-10-28): is this used and/or necessary?
     def __repr__(self) -> str:
         """Use the well's display name as its repr."""

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -559,3 +559,15 @@ def _pressure_supported_by_pipette(self) -> bool:
     def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""
         return False
+
+    def get_minimum_liquid_sense_height(self) -> float:
+        return 0.0
+
+    def estimate_liquid_height(
+        self,
+        well_core: LegacyWellCore,
+        starting_liquid_height: float,
+        operation_volume: float,
+    ) -> float:
+        """This will never be called because it was added in API 2.21."""
+        assert False, "estimate_liquid_height only supported in API 2.21 & later"

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

@@ -83,5 +83,20 @@ def load_liquid(
     def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """Gets point in deck coordinates based on percentage of the radius of each axis."""
 
+    @abstractmethod
+    def estimate_liquid_height_after_pipetting(
+        self,
+        operation_volume: float,
+    ) -> float:
+        """Estimate what the liquid height will be after pipetting, without raising an error."""
+
+    @abstractmethod
+    def current_liquid_height(self) -> float:
+        """Get the current liquid height."""
+
+    @abstractmethod
+    def get_liquid_volume(self) -> float:
+        """Get the current volume within a well."""
+
 
 WellCoreType = TypeVar("WellCoreType", bound=AbstractWellCore)

api/src/opentrons/protocol_api/instrument_context.py

@@ -166,6 +166,11 @@ def default_speed(self) -> float:
     def default_speed(self, speed: float) -> None:
         self._core.set_default_speed(speed)
 
+    @requires_version(2, 21)
+    def get_minimum_liquid_sense_height(self) -> float:
+        """Get the minimum allowed height for liquid-level detection."""
+        return self._core.get_minimum_liquid_sense_height()
+
     @requires_version(2, 0)
     def aspirate(
         self,

api/src/opentrons/protocol_api/labware.py

@@ -36,7 +36,6 @@
     UnsupportedAPIError,
 )
 
-
 # TODO(mc, 2022-09-02): re-exports provided for backwards compatibility
 # remove when their usage is no longer needed
 from opentrons.protocols.labware import (  # noqa: F401
@@ -49,7 +48,10 @@
 from ._liquid import Liquid
 from ._types import OffDeckType
 from .core import well_grid
-from .core.engine import ENGINE_CORE_API_VERSION, SET_OFFSET_RESTORED_API_VERSION
+from .core.engine import (
+    ENGINE_CORE_API_VERSION,
+    SET_OFFSET_RESTORED_API_VERSION,
+)
 from .core.labware import AbstractLabware
 from .core.module import AbstractModuleCore
 from .core.core_map import LoadedCoreMap
@@ -301,6 +303,32 @@ def load_liquid(self, liquid: Liquid, volume: float) -> None:
             volume=volume,
         )
 
+    @requires_version(2, 21)
+    def current_liquid_height(self) -> float:
+        """Get the current liquid height in a well."""
+        return self._core.current_liquid_height()
+
+    @requires_version(2, 21)
+    def current_liquid_volume(self) -> float:
+        """Get the current liquid volume in a well."""
+        return self._core.get_liquid_volume()
+
+    @requires_version(2, 21)
+    def estimate_liquid_height_after_pipetting(self, operation_volume: float) -> float:
+        """Check the height of the liquid within a well.
+
+        :returns: The height, in mm, of the liquid from the deck.
+
+        :meta private:
+
+        This is intended for Opentrons internal use only and is not a guaranteed API.
+        """
+
+        projected_final_height = self._core.estimate_liquid_height_after_pipetting(
+            operation_volume=operation_volume,
+        )
+        return projected_final_height
+
     def _from_center_cartesian(self, x: float, y: float, z: float) -> Point:
         """
         Private version of from_center_cartesian. Present only for backward

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

@@ -148,7 +148,6 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 labware_id=labware_id,
                 well_name=well_name,
             )
-
         move_result = await move_to_well(
             movement=self._movement,
             model_utils=self._model_utils,

api/src/opentrons/protocol_engine/errors/__init__.py

@@ -76,6 +76,7 @@
     CommandNotAllowedError,
     InvalidLiquidHeightFound,
     LiquidHeightUnknownError,
+    LiquidVolumeUnknownError,
     IncompleteLabwareDefinitionError,
     IncompleteWellDefinitionError,
     OperationLocationNotInWellError,
@@ -169,6 +170,7 @@
     "CommandNotAllowedError",
     "InvalidLiquidHeightFound",
     "LiquidHeightUnknownError",
+    "LiquidVolumeUnknownError",
     "IncompleteLabwareDefinitionError",
     "IncompleteWellDefinitionError",
     "OperationLocationNotInWellError",

api/src/opentrons/protocol_engine/errors/exceptions.py

@@ -1114,6 +1114,19 @@ def __init__(
         super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
 
 
+class LiquidVolumeUnknownError(ProtocolEngineError):
+    """Raised when attempting to report an unknown liquid volume."""
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+        wrapping: Optional[Sequence[EnumeratedError]] = None,
+    ) -> None:
+        """Build a LiquidVolumeUnknownError."""
+        super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
+
+
 class EStopActivatedError(ProtocolEngineError):
     """Represents an E-stop event."""
 

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

@@ -241,9 +241,8 @@ def _get_segment_capacity(segment: WellSegment) -> float:
 def get_well_volumetric_capacity(
     well_geometry: InnerWellGeometry,
 ) -> List[Tuple[float, float]]:
-    """Return the total volumetric capacity of a well as a map of height borders to volume."""
-    # dictionary map of heights to volumetric capacities within their respective segment
-    # {top_height_0: volume_0, top_height_1: volume_1, top_height_2: volume_2}
+    """Return the volumetric capacity of a well as a list of pairs relating segment heights to volumes."""
+    #  [(top_height_0, section_0_volume), (top_height_1, section_1_volume), ...]
     well_volume = []
 
     # get the well segments sorted in ascending order
@@ -417,13 +416,16 @@ def _find_height_in_partial_frustum(
 
 
 def find_height_at_well_volume(
-    target_volume: float, well_geometry: InnerWellGeometry
+    target_volume: float,
+    well_geometry: InnerWellGeometry,
+    raise_error_if_result_invalid: bool = True,
 ) -> float:
     """Find the height within a well, at a known volume."""
     volumetric_capacity = get_well_volumetric_capacity(well_geometry)
     max_volume = sum(row[1] for row in volumetric_capacity)
-    if target_volume < 0 or target_volume > max_volume:
-        raise InvalidLiquidHeightFound("Invalid target volume.")
+    if raise_error_if_result_invalid:
+        if target_volume < 0 or target_volume > max_volume:
+            raise InvalidLiquidHeightFound("Invalid target volume.")
 
     sorted_well = sorted(well_geometry.sections, key=lambda section: section.topHeight)
     # find the section the target volume is in and compute the height

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

@@ -1431,7 +1431,7 @@ def get_well_offset_adjustment(
             volume = operation_volume or 0.0
 
         if volume:
-            return self.get_well_height_after_volume(
+            return self.get_well_height_after_liquid_handling(
                 labware_id=labware_id,
                 well_name=well_name,
                 initial_height=initial_handling_height,
@@ -1440,6 +1440,49 @@ def get_well_offset_adjustment(
         else:
             return initial_handling_height
 
+    def get_current_well_volume(
+        self,
+        labware_id: str,
+        well_name: str,
+    ) -> float:
+        """Returns most recently updated volume in specified well."""
+        last_updated = self._wells.get_last_liquid_update(labware_id, well_name)
+        if last_updated is None:
+            raise errors.LiquidHeightUnknownError(
+                "Must LiquidProbe or LoadLiquid before specifying WellOrigin.MENISCUS."
+            )
+
+        well_liquid = self._wells.get_well_liquid_info(
+            labware_id=labware_id, well_name=well_name
+        )
+        if (
+            well_liquid.probed_height is not None
+            and well_liquid.probed_height.height is not None
+            and well_liquid.probed_height.last_probed == last_updated
+        ):
+            return self.get_well_volume_at_height(
+                labware_id=labware_id,
+                well_name=well_name,
+                height=well_liquid.probed_height.height,
+            )
+        elif (
+            well_liquid.loaded_volume is not None
+            and well_liquid.loaded_volume.volume is not None
+            and well_liquid.loaded_volume.last_loaded == last_updated
+        ):
+            return well_liquid.loaded_volume.volume
+        elif (
+            well_liquid.probed_volume is not None
+            and well_liquid.probed_volume.volume is not None
+            and well_liquid.probed_volume.last_probed == last_updated
+        ):
+            return well_liquid.probed_volume.volume
+        else:
+            # This should not happen if there was an update but who knows
+            raise errors.LiquidVolumeUnknownError(
+                f"Unable to find liquid volume despite an update at {last_updated}."
+            )
+
     def get_meniscus_height(
         self,
         labware_id: str,
@@ -1496,7 +1539,7 @@ def get_well_handling_height(
             )
         return float(handling_height)
 
-    def get_well_height_after_volume(
+    def get_well_height_after_liquid_handling(
         self, labware_id: str, well_name: str, initial_height: float, volume: float
     ) -> float:
         """Return the height of liquid in a labware well after a given volume has been handled.
@@ -1514,6 +1557,28 @@ def get_well_height_after_volume(
             target_volume=final_volume, well_geometry=well_geometry
         )
 
+    def get_well_height_after_liquid_handling_no_error(
+        self, labware_id: str, well_name: str, initial_height: float, volume: float
+    ) -> float:
+        """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
+        )
+        initial_volume = find_volume_at_well_height(
+            target_height=initial_height, well_geometry=well_geometry
+        )
+        final_volume = initial_volume + volume
+        well_volume = find_height_at_well_volume(
+            target_volume=final_volume,
+            well_geometry=well_geometry,
+            raise_error_if_result_invalid=False,
+        )
+        return well_volume
+
     def get_well_height_at_volume(
         self, labware_id: str, well_name: str, volume: float
     ) -> float: