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

@@ -2,15 +2,7 @@
 
 from __future__ import annotations
 
-from typing import (
-    Optional,
-    TYPE_CHECKING,
-    cast,
-    Union,
-    List,
-    Tuple,
-    NamedTuple,
-)
+from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple, NamedTuple
 from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -1369,6 +1361,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

@@ -225,6 +225,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

@@ -643,3 +643,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

@@ -561,3 +561,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

@@ -165,6 +165,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

@@ -152,7 +152,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

@@ -74,6 +74,7 @@
     CommandNotAllowedError,
     InvalidLiquidHeightFound,
     LiquidHeightUnknownError,
+    LiquidVolumeUnknownError,
     IncompleteLabwareDefinitionError,
     IncompleteWellDefinitionError,
     OperationLocationNotInWellError,
@@ -167,6 +168,7 @@
     "CommandNotAllowedError",
     "InvalidLiquidHeightFound",
     "LiquidHeightUnknownError",
+    "LiquidVolumeUnknownError",
     "IncompleteLabwareDefinitionError",
     "IncompleteWellDefinitionError",
     "OperationLocationNotInWellError",

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

@@ -1101,6 +1101,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