rm starting_height arg, add get_current_volume

Author: caila-marashaj

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

@@ -158,12 +158,12 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
 
     def estimate_liquid_height_after_pipetting(
         self,
-        starting_liquid_height: float,
         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,
@@ -179,3 +179,11 @@ def current_liquid_height(self) -> float:
         return self._engine_client.state.geometry.get_meniscus_height(
             labware_id=labware_id, well_name=well_name
         )
+
+    def get_well_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/legacy/legacy_well_core.py

@@ -120,7 +120,6 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
 
     def estimate_liquid_height_after_pipetting(
         self,
-        starting_liquid_height: float,
         operation_volume: float,
     ) -> float:
         """Estimate what the liquid height will be after pipetting, without raising an error."""
@@ -130,6 +129,10 @@ def current_liquid_height(self) -> float:
         """Get the current liquid height."""
         return 0.0
 
+    def get_well_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/well.py

@@ -86,7 +86,6 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
     @abstractmethod
     def estimate_liquid_height_after_pipetting(
         self,
-        starting_liquid_height: float,
         operation_volume: float,
     ) -> float:
         """Estimate what the liquid height will be after pipetting, without raising an error."""
@@ -95,5 +94,9 @@ def estimate_liquid_height_after_pipetting(
     def current_liquid_height(self) -> float:
         """Get the current liquid height."""
 
+    @abstractmethod
+    def get_well_volume(self) -> float:
+        """Get the current volume within a well."""
+
 
 WellCoreType = TypeVar("WellCoreType", bound=AbstractWellCore)

api/src/opentrons/protocol_api/labware.py

@@ -305,12 +305,16 @@ def load_liquid(self, liquid: Liquid, volume: float) -> None:
 
     @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 estimate_liquid_height_after_pipetting(
-        self, starting_liquid_height: float, operation_volume: float
-    ) -> float:
+    def current_well_volume(self) -> float:
+        """Get the current liquid volume in a well."""
+        return self._core.get_well_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.
@@ -321,7 +325,6 @@ def estimate_liquid_height_after_pipetting(
         """
 
         projected_final_height = self._core.estimate_liquid_height_after_pipetting(
-            starting_liquid_height=starting_liquid_height,
             operation_volume=operation_volume,
         )
         return projected_final_height

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/geometry.py

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

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

@@ -11,7 +11,10 @@
 from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset
 from opentrons.protocol_engine import commands as cmd
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
-from opentrons.protocol_engine.errors.exceptions import LiquidHeightUnknownError
+from opentrons.protocol_engine.errors.exceptions import (
+    LiquidHeightUnknownError,
+    LiquidVolumeUnknownError,
+)
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.api_support.util import UnsupportedAPIError
 from opentrons.types import Point
@@ -256,6 +259,29 @@ def test_current_liquid_height(
         subject.current_liquid_height()
 
 
+def test_current_liquid_volume(
+    decoy: Decoy, subject: WellCore, mock_engine_client: EngineClient
+) -> None:
+    """Make sure current_liquid_volume returns the correct value or raises an error."""
+    fake_volume = 2222.2
+    decoy.when(
+        mock_engine_client.state.geometry.get_current_well_volume(
+            labware_id="labware-id", well_name="well-name"
+        )
+    ).then_return(fake_volume)
+    assert subject.get_well_volume() == fake_volume
+
+    # make sure that WellCore propagates a LiquidVolumeUnknownError
+    decoy.when(
+        mock_engine_client.state.geometry.get_current_well_volume(
+            labware_id="labware-id", well_name="well-name"
+        )
+    ).then_raise(LiquidVolumeUnknownError())
+
+    with pytest.raises(LiquidVolumeUnknownError):
+        subject.get_well_volume()
+
+
 @pytest.mark.parametrize("operation_volume", [0.0, 100, -100, 2, -4, 5])
 def test_estimate_liquid_height_after_pipetting(
     decoy: Decoy,
@@ -295,6 +321,7 @@ def test_estimate_liquid_height_after_pipetting(
     ).then_return(fake_well_geometry)
     initial_liquid_height = 5.6
     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(
             labware_id="labware-id",
@@ -306,7 +333,6 @@ def test_estimate_liquid_height_after_pipetting(
 
     # make sure that no error was raised
     final_height = subject.estimate_liquid_height_after_pipetting(
-        starting_liquid_height=initial_liquid_height,
         operation_volume=operation_volume,
     )
     assert final_height == fake_final_height