instrument ctx test data function

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

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Optional, TYPE_CHECKING, cast, Union, List
+from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple
 from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -976,6 +976,43 @@ def liquid_probe_with_recovery(self, well_core: WellCore, loc: Location) -> None
 
         self._protocol_core.set_last_location(location=loc, mount=self.get_mount())
 
+    def liquid_probe_testing_data(
+        self,
+        well_core: WellCore,
+        loc: Location,
+        operation_volume: float,
+    ) -> Tuple[float, float, float]:
+        labware_id = well_core.labware_id
+        well_name = well_core.get_name()
+        well_location = WellLocation(
+            origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=2)
+        )
+        result = self._engine_client.execute_command_without_recovery(
+            cmd.LiquidProbeParams(
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+                pipetteId=self.pipette_id,
+            )
+        )
+
+        self._protocol_core.set_last_location(location=loc, mount=self.get_mount())
+        projected_current_volume = (
+            self._engine_client.state.geometry.get_well_volume_at_height(
+                labware_id=labware_id, well_name=well_name, height=result.z_position
+            )
+        )
+        projected_final_height = (
+            self._engine_client.state.geometry.get_well_height_after_volume(
+                labware_id=labware_id,
+                well_name=well_name,
+                initial_height=result.z_position,
+                volume=operation_volume,
+            )
+        )
+
+        return result.z_position, projected_current_volume, projected_final_height
+
     def liquid_probe_without_recovery(
         self, well_core: WellCore, loc: Location
     ) -> float:

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

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod, ABC
-from typing import Any, Generic, Optional, TypeVar, Union, List
+from typing import Any, Generic, Optional, TypeVar, Union, List, Tuple
 
 from opentrons import types
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -371,6 +371,15 @@ def liquid_probe_without_recovery(
         """Do a liquid probe to find the level of the liquid in the well."""
         ...
 
+    @abstractmethod
+    def liquid_probe_testing_data(
+        self,
+        well_core: types.WellCore,
+        loc: types.Location,
+        operation_volume: float,
+    ) -> Tuple[float, float, float]:
+        ...
+
     @abstractmethod
     def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""

api/src/opentrons/protocol_api/instrument_context.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 import logging
 from contextlib import ExitStack
-from typing import Any, List, Optional, Sequence, Union, cast, Dict
+from typing import Any, List, Optional, Sequence, Union, cast, Dict, Tuple
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
@@ -2538,8 +2538,32 @@ def measure_liquid_height(self, well: labware.Well) -> float:
         self._raise_if_pressure_not_supported_by_pipette()
         loc = well.top()
         height = self._core.liquid_probe_without_recovery(well._core, loc)
+        # return current projected volume, height of liquid found, projected height after aspirate
         return height
 
+    @requires_version(2, 20)
+    def hw_testing_liquid_probe(
+        self, well: labware.Well, operation_volume: float
+    ) -> Tuple[float, 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.
+        """
+        self._raise_if_pressure_not_supported_by_pipette()
+        loc = well.top()
+        (
+            current_height,
+            estimated_current_volume,
+            projected_final_height,
+        ) = self._core.liquid_probe_testing_data(
+            well_core=well._core, loc=loc, operation_volume=operation_volume
+        )
+        return current_height, estimated_current_volume, projected_final_height
+
     def _raise_if_configuration_not_supported_by_pipette(
         self, style: NozzleLayout
     ) -> None:

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

@@ -69,7 +69,9 @@ class AspirateResult(BaseLiquidHandlingResult, DestinationPositionResult):
 ]
 
 
-class AspirateImplementation(AbstractCommandImpl[AspirateWhileTrackingParams, _ExecuteReturn]):
+class AspirateImplementation(
+    AbstractCommandImpl[AspirateWhileTrackingParams, _ExecuteReturn]
+):
     """Aspirate command implementation."""
 
     def __init__(
@@ -225,7 +227,9 @@ async def execute(self, params: AspirateWhileTrackingParams) -> _ExecuteReturn:
 
 class Aspirate(
     BaseCommand[
-        AspirateWhileTrackingParams, AspirateResult, OverpressureError | StallOrCollisionError
+        AspirateWhileTrackingParams,
+        AspirateResult,
+        OverpressureError | StallOrCollisionError,
     ]
 ):
     """Aspirate command model."""

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

@@ -71,7 +71,9 @@ class DispenseWhileTrackingResult(BaseLiquidHandlingResult, DestinationPositionR
 ]
 
 
-class DispenseWhileTrackingImplementation(AbstractCommandImpl[DispenseWhileTrackingParams, _ExecuteReturn]):
+class DispenseWhileTrackingImplementation(
+    AbstractCommandImpl[DispenseWhileTrackingParams, _ExecuteReturn]
+):
     """Dispense command implementation."""
 
     def __init__(
@@ -176,7 +178,9 @@ async def execute(self, params: DispenseWhileTrackingParams) -> _ExecuteReturn:
 
 class DispenseWhileTracking(
     BaseCommand[
-        DispenseWhileTrackingParams, DispenseWhileTrackingResult, OverpressureError | StallOrCollisionError
+        DispenseWhileTrackingParams,
+        DispenseWhileTrackingResult,
+        OverpressureError | StallOrCollisionError,
     ]
 ):
     """Dispense command model."""
@@ -185,7 +189,9 @@ class DispenseWhileTracking(
     params: DispenseWhileTrackingParams
     result: Optional[DispenseWhileTrackingResult] = None
 
-    _ImplementationCls: Type[DispenseWhileTrackingImplementation] = DispenseWhileTrackingImplementation
+    _ImplementationCls: Type[
+        DispenseWhileTrackingImplementation
+    ] = DispenseWhileTrackingImplementation
 
 
 class DispenseWhileTrackingCreate(BaseCommandCreate[DispenseWhileTrackingParams]):

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

@@ -362,6 +362,8 @@ def find_volume_at_well_height(
     volumetric_capacity = get_well_volumetric_capacity(well_geometry)
     max_height = volumetric_capacity[-1][0]
     if target_height < 0 or target_height > max_height:
+        # find where we fail bc height < lld minimum and kick off error recovery
+        # to allow the pipette to go to well.bottom() rather than probe in the future
         raise InvalidLiquidHeightFound(f"Invalid target height {target_height}.")
     # volumes in volumetric_capacity are relative to each frustum,
     # so we have to find the volume of all the full sections enclosed