feat(api): implement liquid probe protocol engine command

api/src/opentrons/hardware_control/ot3api.py

@@ -2594,7 +2594,8 @@ def _get_probe_distances(
 
     async def liquid_probe(
         self,
-        mount: OT3Mount,
+        mount: Union[top_types.Mount, OT3Mount],
+        max_z_dist: float, # use!
         probe_settings: Optional[LiquidProbeSettings] = None,
         probe: Optional[InstrumentProbeType] = None,
     ) -> float:
@@ -2622,9 +2623,9 @@ async def liquid_probe(
         if not probe_settings:
             probe_settings = self.config.liquid_sense
 
-        pos = await self.gantry_position(mount, refresh=True)
+        pos = await self.gantry_position(checked_mount, refresh=True)
         probe_start_pos = pos._replace(z=probe_settings.starting_mount_height)
-        await self.move_to(mount, probe_start_pos)
+        await self.move_to(checked_mount, probe_start_pos)
         total_z_travel = probe_settings.max_z_distance
         z_travels = self._get_probe_distances(
             checked_mount,
@@ -2636,7 +2637,7 @@ async def liquid_probe(
         for z_travel in z_travels:
 
             if probe_settings.aspirate_while_sensing:
-                await self._move_to_plunger_bottom(mount, rate=1.0)
+                await self._move_to_plunger_bottom(checked_mount, rate=1.0)
             else:
                 # find the ideal travel distance by multiplying the plunger speed
                 # by the time it will take to complete the z move.
@@ -2656,7 +2657,7 @@ async def liquid_probe(
                 await self._move(target_pos, speed=speed, acquire_lock=True)
             try:
                 height = await self._liquid_probe_pass(
-                    mount,
+                    checked_mount,
                     probe_settings,
                     probe if probe else InstrumentProbeType.PRIMARY,
                     z_travel,
@@ -2666,7 +2667,7 @@ async def liquid_probe(
                 break
             except LiquidNotFoundError as lnfe:
                 error = lnfe
-        await self.move_to(mount, probe_start_pos)
+        await self.move_to(checked_mount, probe_start_pos)
         if error is not None:
             # if we never found an liquid raise an error
             raise error

api/src/opentrons/hardware_control/protocols/liquid_handler.py

@@ -171,3 +171,15 @@ async def drop_tip(
                                 the ejector shroud after a drop.
         """
         ...
+
+    async def liquid_probe(
+        self,
+        mount: MountArgType,
+        max_z_dist: float,
+    ) -> float:
+        """Search for and return liquid level height using this pipette
+        at the current location.
+
+        mount : Mount.LEFT or Mount.RIGHT
+        """
+        ...

api/src/opentrons/protocol_engine/clients/sync_client.py

@@ -900,3 +900,22 @@ def load_liquid(
         )
         result = self._transport.execute_command(request=request)
         return cast(commands.LoadLiquidResult, result)
+
+    def liquid_probe(
+        self,
+        pipette_id: str,
+        labware_id: str,
+        well_name: str,
+        well_location: WellLocation,
+    ) -> commands.LiquidProbeResult:
+        """Execute the liquid_probe command and return the result."""
+        request = commands.LiquidProbeCreate(
+            params=commands.LiquidProbeParams(
+                pipetteId=pipette_id,
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+            )
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.LiquidProbeResult, result)

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

@@ -324,6 +324,14 @@
     VerifyTipPresenceCommandType,
 )
 
+from .liquid_probe import (
+    LiquidProbe,
+    LiquidProbeParams,
+    LiquidProbeCreate,
+    LiquidProbeResult,
+    LiquidProbeCommandType,
+)
+
 __all__ = [
     # command type unions
     "Command",
@@ -564,4 +572,10 @@
     "VerifyTipPresenceParams",
     "VerifyTipPresenceResult",
     "VerifyTipPresenceCommandType",
+    # liquid probe command bundle
+    "LiquidProbe",
+    "LiquidProbeParams",
+    "LiquidProbeCreate",
+    "LiquidProbeResult",
+    "LiquidProbeCommandType",
 ]

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

@@ -300,6 +300,14 @@
     GetTipPresenceCommandType,
 )
 
+from .liquid_probe import (
+    LiquidProbe,
+    LiquidProbeParams,
+    LiquidProbeCreate,
+    LiquidProbeResult,
+    LiquidProbeCommandType,
+)
+
 Command = Annotated[
     Union[
         Aspirate,
@@ -337,6 +345,7 @@
         SetStatusBar,
         VerifyTipPresence,
         GetTipPresence,
+        LiquidProbe,
         heater_shaker.WaitForTemperature,
         heater_shaker.SetTargetTemperature,
         heater_shaker.DeactivateHeater,
@@ -402,6 +411,7 @@
     SetStatusBarParams,
     VerifyTipPresenceParams,
     GetTipPresenceParams,
+    LiquidProbeParams,
     heater_shaker.WaitForTemperatureParams,
     heater_shaker.SetTargetTemperatureParams,
     heater_shaker.DeactivateHeaterParams,
@@ -466,6 +476,7 @@
     SetStatusBarCommandType,
     VerifyTipPresenceCommandType,
     GetTipPresenceCommandType,
+    LiquidProbeCommandType,
     heater_shaker.WaitForTemperatureCommandType,
     heater_shaker.SetTargetTemperatureCommandType,
     heater_shaker.DeactivateHeaterCommandType,
@@ -530,6 +541,7 @@
         SetStatusBarCreate,
         VerifyTipPresenceCreate,
         GetTipPresenceCreate,
+        LiquidProbeCreate,
         heater_shaker.WaitForTemperatureCreate,
         heater_shaker.SetTargetTemperatureCreate,
         heater_shaker.DeactivateHeaterCreate,
@@ -595,6 +607,7 @@
     SetStatusBarResult,
     VerifyTipPresenceResult,
     GetTipPresenceResult,
+    LiquidProbeResult,
     heater_shaker.WaitForTemperatureResult,
     heater_shaker.SetTargetTemperatureResult,
     heater_shaker.DeactivateHeaterResult,

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

@@ -0,0 +1,116 @@
+"""Liquid-probe command for OT3 hardware. request, result, and implementation models."""
+from __future__ import annotations
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+
+from pydantic import Field
+
+from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint
+from .pipetting_common import (
+    PipetteIdMixin,
+    WellLocationMixin,
+    DestinationPositionResult,
+)
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
+from ..errors.error_occurrence import ErrorOccurrence
+
+if TYPE_CHECKING:
+    from ..execution import MovementHandler, PipettingHandler
+
+
+LiquidProbeCommandType = Literal["liquidProbe"]
+
+
+class LiquidProbeParams(PipetteIdMixin, WellLocationMixin):
+    """Payload required to liquid probe."""
+
+    pass
+
+
+class LiquidProbeResult(DestinationPositionResult):
+    """Result data from the execution of a liquid-probe command."""
+
+    z_position: float = Field(..., description="Z position of the found liquid.")
+
+
+class LiquidProbeImplementation(
+    AbstractCommandImpl[LiquidProbeParams, SuccessData[LiquidProbeResult, None]]
+):
+    """The implementation of a `liquidProbe` command."""
+
+    def __init__(
+        self, movement: MovementHandler, pipetting: PipettingHandler, **kwargs: object
+    ) -> None:
+        self._movement = movement
+        self._pipetting = pipetting
+
+    async def execute(
+        self, params: LiquidProbeParams
+    ) -> SuccessData[LiquidProbeResult, None]:
+        """Execute a `liquidProbe` command.
+
+        Return the z-position of the found liquid.
+        """
+        pipette_id = params.pipetteId
+        labware_id = params.labwareId
+        well_name = params.wellName
+
+        ready_to_probe = self._pipetting.get_is_ready_to_aspirate(pipette_id=pipette_id)
+
+        current_well = None
+
+        if not ready_to_probe:
+            await self._movement.move_to_well(
+                pipette_id=pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+                well_location=WellLocation(origin=WellOrigin.TOP),
+            )
+
+            current_well = CurrentWell(
+                pipette_id=pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+            )
+
+        position = await self._movement.move_to_well(
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=params.wellLocation,
+            current_well=current_well,
+        )
+        well_def = self._movement._state_store.labware.get_well_definition(
+            labware_id, well_name
+        )
+        well_depth = well_def.depth
+        z_pos = await self._pipetting.liquid_probe_in_place(
+            pipette_id=pipette_id, max_z_dist=well_depth
+        )
+
+        return SuccessData(
+            public=LiquidProbeResult(
+                z_position=z_pos,
+                position=DeckPoint(x=position.x, y=position.y, z=position.z),
+            ),
+            private=None,
+        )
+
+
+class LiquidProbe(BaseCommand[LiquidProbeParams, LiquidProbeResult, ErrorOccurrence]):
+    """A `liquidProbe` command."""
+
+    commandType: LiquidProbeCommandType = "liquidProbe"
+    params: LiquidProbeParams
+    result: Optional[LiquidProbeResult]
+
+    _ImplementationCls: Type[LiquidProbeImplementation] = LiquidProbeImplementation
+
+
+class LiquidProbeCreate(BaseCommandCreate[LiquidProbeParams]):
+    """A request to create a `liquidProbe` command."""
+
+    commandType: LiquidProbeCommandType = "liquidProbe"
+    params: LiquidProbeParams
+
+    _CommandCls: Type[LiquidProbe] = LiquidProbe

api/src/opentrons/protocol_engine/execution/pipetting.py

@@ -60,6 +60,13 @@ async def blow_out_in_place(
     ) -> None:
         """Set flow rate and blow-out."""
 
+    async def liquid_probe_in_place(
+        self,
+        pipette_id: str,
+        max_z_dist: float,
+    ) -> float:
+        """Detect liquid level."""
+
 
 class HardwarePipettingHandler(PipettingHandler):
     """Liquid handling, using the Hardware API.""" ""
@@ -156,6 +163,21 @@ async def blow_out_in_place(
         with self._set_flow_rate(pipette=hw_pipette, blow_out_flow_rate=flow_rate):
             await self._hardware_api.blow_out(mount=hw_pipette.mount)
 
+    async def liquid_probe_in_place(
+        self,
+        pipette_id: str,
+        max_z_dist: float,
+    ) -> float:
+        """Detect liquid level."""
+        hw_pipette = self._state_view.pipettes.get_hardware_pipette(
+            pipette_id=pipette_id,
+            attached_pipettes=self._hardware_api.attached_instruments,
+        )
+        z_pos = await self._hardware_api.liquid_probe(
+            mount=hw_pipette.mount, max_z_dist=max_z_dist
+        )
+        return float(z_pos)
+
     @contextmanager
     def _set_flow_rate(
         self,
@@ -245,6 +267,14 @@ async def blow_out_in_place(
     ) -> None:
         """Virtually blow out (no-op)."""
 
+    async def liquid_probe_in_place(
+        self,
+        pipette_id: str,
+        max_z_dist: float,
+    ) -> float:
+        """Detect liquid level."""
+        return 0.0  # fix
+
     def _validate_tip_attached(self, pipette_id: str, command_name: str) -> None:
         """Validate if there is a tip attached."""
         tip_geometry = self._state_view.pipettes.get_attached_tip(pipette_id)