feat(api): simulate liquid probe results

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

@@ -2,8 +2,22 @@
 
 from __future__ import annotations
 
-from typing import Optional, TYPE_CHECKING, cast, Union, List, Tuple, NamedTuple
-from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
+from typing import (
+    Optional,
+    TYPE_CHECKING,
+    cast,
+    Union,
+    List,
+    Tuple,
+    NamedTuple,
+    Literal,
+)
+from opentrons.types import (
+    Location,
+    Mount,
+    NozzleConfigurationType,
+    NozzleMapInterface,
+)
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version
@@ -1681,7 +1695,7 @@ def liquid_probe_with_recovery(self, well_core: WellCore, loc: Location) -> None
 
     def liquid_probe_without_recovery(
         self, well_core: WellCore, loc: Location
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         labware_id = well_core.labware_id
         well_name = well_core.get_name()
         well_location = WellLocation(
@@ -1697,7 +1711,8 @@ def liquid_probe_without_recovery(
         )
 
         self._protocol_core.set_last_location(location=loc, mount=self.get_mount())
-
+        if not isinstance(result.z_position, float):
+            return "SimulatedProbeResult"
         return result.z_position
 
     def nozzle_configuration_valid_for_lld(self) -> bool:

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

@@ -587,7 +587,11 @@ def absolute_point_from_position_reference_and_offset(
         case PositionReference.WELL_CENTER:
             reference_point = well.get_center()
         case PositionReference.LIQUID_MENISCUS:
-            reference_point = well.get_meniscus()
+            meniscus_point = well.get_meniscus()
+            if not isinstance(meniscus_point, Point):
+                reference_point = well.get_center()
+            else:
+                reference_point = meniscus_point
         case _:
             raise ValueError(f"Unknown position reference {position_reference}")
     return reference_point + Point(offset.x, offset.y, offset.z)

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

@@ -1,12 +1,13 @@
 """ProtocolEngine-based Well core implementations."""
-from typing import Optional
+from typing import Optional, Union, Literal
 
 from opentrons_shared_data.labware.constants import WELL_NAME_PATTERN
 
 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.protocols.api_support.util import UnsupportedAPIError
+
 from opentrons.types import Point
 
 from . import point_calculations
@@ -135,9 +136,13 @@ def get_center(self) -> Point:
             well_location=WellLocation(origin=WellOrigin.CENTER),
         )
 
-    def get_meniscus(self) -> Point:
+    def get_meniscus(self) -> Union[Point, Literal["SimulatedProbeResult"]]:
         """Get the coordinate of the well's meniscus."""
-        return self.get_bottom(self.current_liquid_height())
+        current_liquid_height = self.current_liquid_height()
+        if isinstance(current_liquid_height, float):
+            return self.get_bottom(z_offset=current_liquid_height)
+        else:
+            return current_liquid_height
 
     def load_liquid(
         self,
@@ -173,7 +178,7 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
     def estimate_liquid_height_after_pipetting(
         self,
         operation_volume: float,
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Return an estimate of liquid height after pipetting without raising an error."""
         labware_id = self.labware_id
         well_name = self._name
@@ -186,15 +191,15 @@ def estimate_liquid_height_after_pipetting(
         )
         return projected_final_height
 
-    def current_liquid_height(self) -> float:
+    def current_liquid_height(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """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:
+    def get_liquid_volume(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Return the current volume in a well."""
         labware_id = self.labware_id
         well_name = self._name

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

@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 from abc import abstractmethod, ABC
-from typing import Any, Generic, Optional, TypeVar, Union, List, Tuple
+from typing import Any, Generic, Optional, TypeVar, Union, List, Tuple, Literal
 
 from opentrons import types
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.protocols.api_support.util import FlowRates
 from opentrons.protocols.advanced_control.transfers.common import TransferTipPolicyV2
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.protocol_api._liquid import LiquidClass
+
 from ..disposal_locations import TrashBin, WasteChute
 from .well import WellCoreType
 from .labware import LabwareCoreType
@@ -426,7 +427,7 @@ def liquid_probe_with_recovery(
     @abstractmethod
     def liquid_probe_without_recovery(
         self, well_core: WellCoreType, loc: types.Location
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Do a liquid probe to find the level of the liquid in the well."""
         ...
 

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

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Optional, Union, List, Tuple
+from typing import TYPE_CHECKING, Optional, Union, List, Tuple, Literal
 
 from opentrons import types
 from opentrons.hardware_control import CriticalPoint
@@ -686,7 +686,7 @@ def liquid_probe_with_recovery(
 
     def liquid_probe_without_recovery(
         self, well_core: WellCore, loc: types.Location
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
 

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

@@ -1,5 +1,5 @@
 """Legacy Well core implementation."""
-from typing import Optional
+from typing import Optional, Union, Literal
 
 from opentrons_shared_data.labware.constants import WELL_NAME_PATTERN
 
@@ -106,7 +106,7 @@ def get_center(self) -> Point:
         """Get the coordinate of the well's center."""
         return self._geometry.center()
 
-    def get_meniscus(self) -> Point:
+    def get_meniscus(self) -> Union[Point, Literal["SimulatedProbeResult"]]:
         """Get the coordinate of the well's center."""
         raise APIVersionError(api_element="Getting a meniscus")
 
@@ -125,15 +125,15 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
     def estimate_liquid_height_after_pipetting(
         self,
         operation_volume: float,
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Estimate what the liquid height will be after pipetting, without raising an error."""
         return 0.0
 
-    def current_liquid_height(self) -> float:
+    def current_liquid_height(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Get the current liquid height."""
         return 0.0
 
-    def get_liquid_volume(self) -> float:
+    def get_liquid_volume(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Get the current well volume."""
         return 0.0
 

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

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Optional, Union, List, Tuple
+from typing import TYPE_CHECKING, Optional, Union, List, Tuple, Literal
 
 from opentrons import types
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -578,7 +578,7 @@ def liquid_probe_with_recovery(
 
     def liquid_probe_without_recovery(
         self, well_core: WellCore, loc: types.Location
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
 

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

@@ -1,7 +1,7 @@
 """Abstract interface for Well core implementations."""
 
 from abc import ABC, abstractmethod
-from typing import TypeVar, Optional
+from typing import TypeVar, Optional, Union, Literal
 
 from opentrons.types import Point
 
@@ -72,7 +72,7 @@ def get_center(self) -> Point:
         """Get the coordinate of the well's center."""
 
     @abstractmethod
-    def get_meniscus(self) -> Point:
+    def get_meniscus(self) -> Union[Point, Literal["SimulatedProbeResult"]]:
         """Get the coordinate of the well's meniscus."""
 
     @abstractmethod
@@ -91,15 +91,15 @@ def from_center_cartesian(self, x: float, y: float, z: float) -> Point:
     def estimate_liquid_height_after_pipetting(
         self,
         operation_volume: float,
-    ) -> float:
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Estimate what the liquid height will be after pipetting, without raising an error."""
 
     @abstractmethod
-    def current_liquid_height(self) -> float:
+    def current_liquid_height(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Get the current liquid height."""
 
     @abstractmethod
-    def get_liquid_volume(self) -> float:
+    def get_liquid_volume(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Get the current volume within a well."""
 
 

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, Literal
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
@@ -2577,7 +2577,9 @@ def require_liquid_presence(self, well: labware.Well) -> None:
         self._core.liquid_probe_with_recovery(well._core, loc)
 
     @requires_version(2, 20)
-    def measure_liquid_height(self, well: labware.Well) -> float:
+    def measure_liquid_height(
+        self, well: labware.Well
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Check the height of the liquid within a well.
 
         :returns: The height, in mm, of the liquid from the deck.
@@ -2588,8 +2590,7 @@ 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 height
+        return self._core.liquid_probe_without_recovery(well._core, loc)
 
     def _raise_if_configuration_not_supported_by_pipette(
         self, style: NozzleLayout

api/src/opentrons/protocol_api/labware.py

@@ -19,11 +19,12 @@
     List,
     Dict,
     Optional,
-    Union,
     Tuple,
     cast,
     Sequence,
     Mapping,
+    Union,
+    Literal,
 )
 
 from opentrons_shared_data.labware.types import (
@@ -324,17 +325,20 @@ def load_liquid(self, liquid: Liquid, volume: float) -> None:
         )
 
     @requires_version(2, 21)
-    def current_liquid_height(self) -> float:
+    def current_liquid_height(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Get the current liquid height in a well."""
         return self._core.current_liquid_height()
 
     @requires_version(2, 21)
-    def current_liquid_volume(self) -> float:
+    def current_liquid_volume(self) -> Union[float, Literal["SimulatedProbeResult"]]:
         """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:
+    def estimate_liquid_height_after_pipetting(
+        self,
+        operation_volume: float,
+    ) -> Union[float, Literal["SimulatedProbeResult"]]:
         """Check the height of the liquid within a well.
 
         :returns: The height, in mm, of the liquid from the deck.

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

@@ -80,7 +80,7 @@ class TryLiquidProbeParams(_CommonParams):
 class LiquidProbeResult(DestinationPositionResult):
     """Result data from the execution of a `liquidProbe` command."""
 
-    z_position: float = Field(
+    z_position: Union[float, Literal["SimulatedProbeResult"]] = Field(
         ..., description="The Z coordinate, in mm, of the found liquid in deck space."
     )
     # New fields should use camelCase. z_position is snake_case for historical reasons.
@@ -89,7 +89,9 @@ class LiquidProbeResult(DestinationPositionResult):
 class TryLiquidProbeResult(DestinationPositionResult):
     """Result data from the execution of a `tryLiquidProbe` command."""
 
-    z_position: float | SkipJsonSchema[None] = Field(
+    z_position: Union[
+        float, SkipJsonSchema[None], Literal["SimulatedProbeResult"]
+    ] = Field(
         ...,
         description=(
             "The Z coordinate, in mm, of the found liquid in deck space."
@@ -116,8 +118,9 @@ class _ExecuteCommonResult(NamedTuple):
     # If the probe succeeded, the z_pos that it returned.
     # Or, if the probe found no liquid, the error representing that,
     # so calling code can propagate those details up.
-    z_pos_or_error: float | PipetteLiquidNotFoundError | PipetteOverpressureError
-
+    z_pos_or_error: float | PipetteLiquidNotFoundError | PipetteOverpressureError | Literal[
+        "SimulatedProbeResult"
+    ]
     state_update: update_types.StateUpdate
     deck_point: DeckPoint
 
@@ -303,12 +306,12 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
             )
         else:
             try:
-                well_volume: float | update_types.ClearType = (
-                    self._state_view.geometry.get_well_volume_at_height(
-                        labware_id=params.labwareId,
-                        well_name=params.wellName,
-                        height=z_pos_or_error,
-                    )
+                well_volume: Union[
+                    float, update_types.ClearType, Literal["SimulatedProbeResult"]
+                ] = self._state_view.geometry.get_well_volume_at_height(
+                    labware_id=params.labwareId,
+                    well_name=params.wellName,
+                    height=z_pos_or_error,
                 )
             except IncompleteLabwareDefinitionError:
                 well_volume = update_types.CLEAR
@@ -370,7 +373,9 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
             z_pos_or_error, (PipetteLiquidNotFoundError, PipetteOverpressureError)
         ):
             z_pos = None
-            well_volume: float | update_types.ClearType = update_types.CLEAR
+            well_volume: Union[
+                float, update_types.ClearType, Literal["SimulatedProbeResult"]
+            ] = update_types.CLEAR
         else:
             z_pos = z_pos_or_error
             try: