feat(protocol-engine): Add tryLiquidProbe to complement liquidProbe

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

@@ -331,6 +331,11 @@
     LiquidProbeCreate,
     LiquidProbeResult,
     LiquidProbeCommandType,
+    TryLiquidProbe,
+    TryLiquidProbeParams,
+    TryLiquidProbeCreate,
+    TryLiquidProbeResult,
+    TryLiquidProbeCommandType,
 )
 
 __all__ = [
@@ -580,4 +585,9 @@
     "LiquidProbeCreate",
     "LiquidProbeResult",
     "LiquidProbeCommandType",
+    "TryLiquidProbe",
+    "TryLiquidProbeParams",
+    "TryLiquidProbeCreate",
+    "TryLiquidProbeResult",
+    "TryLiquidProbeCommandType",
 ]

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

@@ -1,6 +1,7 @@
 """Union types of concrete command definitions."""
 
-from typing import Annotated, Iterable, Type, Union, get_type_hints
+from collections.abc import Collection
+from typing import Annotated, Type, Union, get_type_hints
 
 from pydantic import Field
 
@@ -313,6 +314,11 @@
     LiquidProbeCreate,
     LiquidProbeResult,
     LiquidProbeCommandType,
+    TryLiquidProbe,
+    TryLiquidProbeParams,
+    TryLiquidProbeCreate,
+    TryLiquidProbeResult,
+    TryLiquidProbeCommandType,
 )
 
 Command = Annotated[
@@ -353,6 +359,7 @@
         VerifyTipPresence,
         GetTipPresence,
         LiquidProbe,
+        TryLiquidProbe,
         heater_shaker.WaitForTemperature,
         heater_shaker.SetTargetTemperature,
         heater_shaker.DeactivateHeater,
@@ -421,6 +428,7 @@
     VerifyTipPresenceParams,
     GetTipPresenceParams,
     LiquidProbeParams,
+    TryLiquidProbeParams,
     heater_shaker.WaitForTemperatureParams,
     heater_shaker.SetTargetTemperatureParams,
     heater_shaker.DeactivateHeaterParams,
@@ -487,6 +495,7 @@
     VerifyTipPresenceCommandType,
     GetTipPresenceCommandType,
     LiquidProbeCommandType,
+    TryLiquidProbeCommandType,
     heater_shaker.WaitForTemperatureCommandType,
     heater_shaker.SetTargetTemperatureCommandType,
     heater_shaker.DeactivateHeaterCommandType,
@@ -554,6 +563,7 @@
         VerifyTipPresenceCreate,
         GetTipPresenceCreate,
         LiquidProbeCreate,
+        TryLiquidProbeCreate,
         heater_shaker.WaitForTemperatureCreate,
         heater_shaker.SetTargetTemperatureCreate,
         heater_shaker.DeactivateHeaterCreate,
@@ -622,6 +632,7 @@
     VerifyTipPresenceResult,
     GetTipPresenceResult,
     LiquidProbeResult,
+    TryLiquidProbeResult,
     heater_shaker.WaitForTemperatureResult,
     heater_shaker.SetTargetTemperatureResult,
     heater_shaker.DeactivateHeaterResult,
@@ -671,12 +682,20 @@
 
 
 def _map_create_types_by_params_type(
-    create_types: Iterable[Type[CommandCreate]],
+    create_types: Collection[Type[CommandCreate]],
 ) -> dict[Type[CommandParams], Type[CommandCreate]]:
     def get_params_type(create_type: Type[CommandCreate]) -> Type[CommandParams]:
         return get_type_hints(create_type)["params"]  # type: ignore[no-any-return]
 
-    return {get_params_type(create_type): create_type for create_type in create_types}
+    result = {get_params_type(create_type): create_type for create_type in create_types}
+
+    # This isn't an inherent requirement of opentrons.protocol_engine,
+    # but this mapping is only useful to higher-level code if this holds true.
+    assert len(result) == len(
+        create_types
+    ), "Param models should map to create models 1:1."
+
+    return result
 
 
 CREATE_TYPES_BY_PARAMS_TYPE = _map_create_types_by_params_type(

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

@@ -1,4 +1,5 @@
-"""Liquid-probe command for OT3 hardware. request, result, and implementation models."""
+"""The liquidProbe and tryLiquidProbe commands."""
+
 from __future__ import annotations
 from typing import TYPE_CHECKING, Optional, Type, Union
 from opentrons.protocol_engine.errors.exceptions import MustHomeError, TipNotEmptyError
@@ -34,30 +35,59 @@
 
 
 LiquidProbeCommandType = Literal["liquidProbe"]
+TryLiquidProbeCommandType = Literal["tryLiquidProbe"]
+
 
+# Both command variants should have identical parameters.
+# But we need two separate parameter model classes because
+# `command_unions.CREATE_TYPES_BY_PARAMS_TYPE` needs to be a 1:1 mapping.
+class _CommonParams(PipetteIdMixin, WellLocationMixin):
+    pass
 
-class LiquidProbeParams(PipetteIdMixin, WellLocationMixin):
-    """Parameters required to liquid probe a specific well."""
+
+class LiquidProbeParams(_CommonParams):
+    """Parameters required for a `liquidProbe` command."""
+
+    pass
+
+
+class TryLiquidProbeParams(_CommonParams):
+    """Parameters required for a `tryLiquidProbe` command."""
 
     pass
 
 
 class LiquidProbeResult(DestinationPositionResult):
-    """Result data from the execution of a liquid-probe command."""
+    """Result data from the execution of a `liquidProbe` command."""
 
     z_position: float = 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.
 
 
-_ExecuteReturn = Union[
+class TryLiquidProbeResult(DestinationPositionResult):
+    """Result data from the execution of a `tryLiquidProbe` command."""
+
+    z_position: Optional[float] = Field(
+        ...,
+        description=(
+            "The Z coordinate, in mm, of the found liquid in deck space."
+            " If no liquid was found, `null` or omitted."
+        ),
+    )
+
+
+_LiquidProbeExecuteReturn = Union[
     SuccessData[LiquidProbeResult, None],
     DefinedErrorData[LiquidNotFoundError, LiquidNotFoundErrorInternalData],
 ]
+_TryLiquidProbeExecuteReturn = SuccessData[TryLiquidProbeResult, None]
 
 
-class LiquidProbeImplementation(AbstractCommandImpl[LiquidProbeParams, _ExecuteReturn]):
+class LiquidProbeImplementation(
+    AbstractCommandImpl[LiquidProbeParams, _LiquidProbeExecuteReturn]
+):
     """The implementation of a `liquidProbe` command."""
 
     def __init__(
@@ -71,16 +101,19 @@ def __init__(
         self._pipetting = pipetting
         self._model_utils = model_utils
 
-    async def execute(self, params: LiquidProbeParams) -> _ExecuteReturn:
+    async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
         """Move to and liquid probe the requested well.
 
         Return the z-position of the found liquid.
+        If no liquid is found, return a LiquidNotFoundError as a defined error.
 
         Raises:
-            TipNotAttachedError: if there is no tip attached to the pipette
-            MustHomeError: if the plunger is not in a valid position
-            TipNotEmptyError: if the tip starts with liquid in it
-            LiquidNotFoundError: if liquid is not found during the probe process.
+            TipNotAttachedError: as an undefined error, if there is not tip attached to
+                the pipette.
+            TipNotEmptyError: as an undefined error, if the tip starts with liquid
+                in it.
+            MustHomeError: as an undefined error, if the plunger is not in a valid
+                position.
         """
         pipette_id = params.pipetteId
         labware_id = params.labwareId
@@ -139,8 +172,68 @@ async def execute(self, params: LiquidProbeParams) -> _ExecuteReturn:
             )
 
 
-class LiquidProbe(BaseCommand[LiquidProbeParams, LiquidProbeResult, ErrorOccurrence]):
-    """LiquidProbe command model."""
+class TryLiquidProbeImplementation(
+    AbstractCommandImpl[TryLiquidProbeParams, _TryLiquidProbeExecuteReturn]
+):
+    """The implementation of a `tryLiquidProbe` command."""
+
+    def __init__(
+        self,
+        movement: MovementHandler,
+        pipetting: PipettingHandler,
+        model_utils: ModelUtils,
+        **kwargs: object,
+    ) -> None:
+        self._movement = movement
+        self._pipetting = pipetting
+        self._model_utils = model_utils
+
+    async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
+        """Execute a `tryLiquidProbe` command.
+
+        `tryLiquidProbe` is identical to `liquidProbe`, except that if no liquid is
+        found, `tryLiquidProbe` returns a success result with `z_position=null` instead
+        of a defined error.
+        """
+        # We defer to the `liquidProbe` implementation. If it returns a defined
+        # `liquidNotFound` error, we remap that to a success result.
+        # Otherwise, we return the result or propagate the exception unchanged.
+
+        original_impl = LiquidProbeImplementation(
+            movement=self._movement,
+            pipetting=self._pipetting,
+            model_utils=self._model_utils,
+        )
+        original_result = await original_impl.execute(params)
+
+        match original_result:
+            case DefinedErrorData(
+                public=LiquidNotFoundError(),
+                private=LiquidNotFoundErrorInternalData() as original_private,
+            ):
+                return SuccessData(
+                    public=TryLiquidProbeResult(
+                        z_position=None,
+                        position=original_private.position,
+                    ),
+                    private=None,
+                )
+            case SuccessData(
+                public=LiquidProbeResult() as original_public, private=None
+            ):
+                return SuccessData(
+                    public=TryLiquidProbeResult(
+                        position=original_public.position,
+                        z_position=original_public.z_position,
+                    ),
+                    private=None,
+                )
+
+
+class LiquidProbe(
+    BaseCommand[LiquidProbeParams, LiquidProbeResult, LiquidNotFoundError]
+):
+    """The model for a full `liquidProbe` command."""
 
     commandType: LiquidProbeCommandType = "liquidProbe"
     params: LiquidProbeParams
@@ -149,10 +242,33 @@ class LiquidProbe(BaseCommand[LiquidProbeParams, LiquidProbeResult, ErrorOccurre
     _ImplementationCls: Type[LiquidProbeImplementation] = LiquidProbeImplementation
 
 
+class TryLiquidProbe(
+    BaseCommand[TryLiquidProbeParams, TryLiquidProbeResult, ErrorOccurrence]
+):
+    """The model for a full `tryLiquidProbe` command."""
+
+    commandType: TryLiquidProbeCommandType = "tryLiquidProbe"
+    params: TryLiquidProbeParams
+    result: Optional[TryLiquidProbeResult]
+
+    _ImplementationCls: Type[
+        TryLiquidProbeImplementation
+    ] = TryLiquidProbeImplementation
+
+
 class LiquidProbeCreate(BaseCommandCreate[LiquidProbeParams]):
-    """Create LiquidProbe command request model."""
+    """The request model for a `liquidProbe` command."""
 
     commandType: LiquidProbeCommandType = "liquidProbe"
     params: LiquidProbeParams
 
     _CommandCls: Type[LiquidProbe] = LiquidProbe
+
+
+class TryLiquidProbeCreate(BaseCommandCreate[TryLiquidProbeParams]):
+    """The request model for a `tryLiquidProbe` command."""
+
+    commandType: TryLiquidProbeCommandType = "tryLiquidProbe"
+    params: TryLiquidProbeParams
+
+    _CommandCls: Type[TryLiquidProbe] = TryLiquidProbe