refactor(api): Port touchTip, liquidProbe, and tryLiquidProbe l…

…ocation updates to `StateUpdate` (#16261)

## Overview

More incremental work towards EXEC-652.

## Changelog

This continues the pattern started in #16160. The following commands now
use the new `StateUpdate` mechanism to update the pipette's logical
state for the purposes of path planning:

* `touchTip`
* `liquidProbe`
* `tryLiquidProbe`

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

@@ -11,7 +11,6 @@
 from .pipetting_common import (
     OverpressureError,
     LiquidNotFoundError,
-    LiquidNotFoundErrorInternalData,
 )
 
 from . import absorbance_reader
@@ -706,7 +705,7 @@
 CommandDefinedErrorData = Union[
     DefinedErrorData[TipPhysicallyMissingError, None],
     DefinedErrorData[OverpressureError, None],
-    DefinedErrorData[LiquidNotFoundError, LiquidNotFoundErrorInternalData],
+    DefinedErrorData[LiquidNotFoundError, None],
 ]
 
 

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

@@ -1,20 +1,21 @@
 """The liquidProbe and tryLiquidProbe commands."""
 
 from __future__ import annotations
-from typing import TYPE_CHECKING, Optional, Type, Union
+from typing import TYPE_CHECKING, NamedTuple, Optional, Type, Union
+from typing_extensions import Literal
+
+from pydantic import Field
+
 from opentrons.protocol_engine.errors.exceptions import MustHomeError, TipNotEmptyError
+from opentrons.protocol_engine.state import update_types
 from opentrons.types import MountType
 from opentrons_shared_data.errors.exceptions import (
     PipetteLiquidNotFoundError,
 )
-from typing_extensions import Literal
-
-from pydantic import Field
 
 from ..types import DeckPoint
 from .pipetting_common import (
     LiquidNotFoundError,
-    LiquidNotFoundErrorInternalData,
     PipetteIdMixin,
     WellLocationMixin,
     DestinationPositionResult,
@@ -80,11 +81,76 @@ class TryLiquidProbeResult(DestinationPositionResult):
 
 _LiquidProbeExecuteReturn = Union[
     SuccessData[LiquidProbeResult, None],
-    DefinedErrorData[LiquidNotFoundError, LiquidNotFoundErrorInternalData],
+    DefinedErrorData[LiquidNotFoundError, None],
 ]
 _TryLiquidProbeExecuteReturn = SuccessData[TryLiquidProbeResult, None]
 
 
+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
+
+    state_update: update_types.StateUpdate
+    deck_point: DeckPoint
+
+
+async def _execute_common(
+    movement: MovementHandler, pipetting: PipettingHandler, params: _CommonParams
+) -> _ExecuteCommonResult:
+    pipette_id = params.pipetteId
+    labware_id = params.labwareId
+    well_name = params.wellName
+
+    state_update = update_types.StateUpdate()
+
+    # _validate_tip_attached in pipetting.py is a private method so we're using
+    # get_is_ready_to_aspirate as an indirect way to throw a TipNotAttachedError if appropriate
+    pipetting.get_is_ready_to_aspirate(pipette_id=pipette_id)
+
+    if pipetting.get_is_empty(pipette_id=pipette_id) is False:
+        raise TipNotEmptyError(
+            message="This operation requires a tip with no liquid in it."
+        )
+
+    if await movement.check_for_valid_position(mount=MountType.LEFT) is False:
+        raise MustHomeError(
+            message="Current position of pipette is invalid. Please home."
+        )
+
+    # liquid_probe process start position
+    position = await movement.move_to_well(
+        pipette_id=pipette_id,
+        labware_id=labware_id,
+        well_name=well_name,
+        well_location=params.wellLocation,
+    )
+    deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z)
+    state_update.set_pipette_location(
+        pipette_id=pipette_id,
+        new_labware_id=labware_id,
+        new_well_name=well_name,
+        new_deck_point=deck_point,
+    )
+
+    try:
+        z_pos = await pipetting.liquid_probe_in_place(
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=params.wellLocation,
+        )
+    except PipetteLiquidNotFoundError as exception:
+        return _ExecuteCommonResult(
+            z_pos_or_error=exception, state_update=state_update, deck_point=deck_point
+        )
+    else:
+        return _ExecuteCommonResult(
+            z_pos_or_error=z_pos, state_update=state_update, deck_point=deck_point
+        )
+
+
 class LiquidProbeImplementation(
     AbstractCommandImpl[LiquidProbeParams, _LiquidProbeExecuteReturn]
 ):
@@ -115,40 +181,10 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
             MustHomeError: as an undefined error, if the plunger is not in a valid
                 position.
         """
-        pipette_id = params.pipetteId
-        labware_id = params.labwareId
-        well_name = params.wellName
-
-        # _validate_tip_attached in pipetting.py is a private method so we're using
-        # get_is_ready_to_aspirate as an indirect way to throw a TipNotAttachedError if appropriate
-        self._pipetting.get_is_ready_to_aspirate(pipette_id=pipette_id)
-
-        if self._pipetting.get_is_empty(pipette_id=pipette_id) is False:
-            raise TipNotEmptyError(
-                message="This operation requires a tip with no liquid in it."
-            )
-
-        if await self._movement.check_for_valid_position(mount=MountType.LEFT) is False:
-            raise MustHomeError(
-                message="Current position of pipette is invalid. Please home."
-            )
-
-        # liquid_probe process start position
-        position = await self._movement.move_to_well(
-            pipette_id=pipette_id,
-            labware_id=labware_id,
-            well_name=well_name,
-            well_location=params.wellLocation,
+        z_pos_or_error, state_update, deck_point = await _execute_common(
+            self._movement, self._pipetting, params
         )
-
-        try:
-            z_pos = await self._pipetting.liquid_probe_in_place(
-                pipette_id=pipette_id,
-                labware_id=labware_id,
-                well_name=well_name,
-                well_location=params.wellLocation,
-            )
-        except PipetteLiquidNotFoundError as e:
+        if isinstance(z_pos_or_error, PipetteLiquidNotFoundError):
             return DefinedErrorData(
                 public=LiquidNotFoundError(
                     id=self._model_utils.generate_id(),
@@ -157,21 +193,20 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
                         ErrorOccurrence.from_failed(
                             id=self._model_utils.generate_id(),
                             createdAt=self._model_utils.get_timestamp(),
-                            error=e,
+                            error=z_pos_or_error,
                         )
                     ],
                 ),
-                private=LiquidNotFoundErrorInternalData(
-                    position=DeckPoint(x=position.x, y=position.y, z=position.z)
-                ),
+                private=None,
+                state_update=state_update,
             )
         else:
             return SuccessData(
                 public=LiquidProbeResult(
-                    z_position=z_pos,
-                    position=DeckPoint(x=position.x, y=position.y, z=position.z),
+                    z_position=z_pos_or_error, position=deck_point
                 ),
                 private=None,
+                state_update=state_update,
             )
 
 
@@ -184,12 +219,10 @@ 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.
@@ -198,39 +231,23 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
         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,
+        z_pos_or_error, state_update, deck_point = await _execute_common(
+            self._movement, self._pipetting, params
+        )
+
+        z_pos = (
+            None
+            if isinstance(z_pos_or_error, PipetteLiquidNotFoundError)
+            else z_pos_or_error
+        )
+        return SuccessData(
+            public=TryLiquidProbeResult(
+                z_position=z_pos,
+                position=deck_point,
+            ),
+            private=None,
+            state_update=state_update,
         )
-        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(

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

@@ -1,5 +1,4 @@
 """Common pipetting command base models."""
-from dataclasses import dataclass
 from opentrons_shared_data.errors import ErrorCodes
 from pydantic import BaseModel, Field
 from typing import Literal, Optional, Tuple, TypedDict
@@ -169,11 +168,3 @@ class LiquidNotFoundError(ErrorOccurrence):
 
     errorCode: str = ErrorCodes.PIPETTE_LIQUID_NOT_FOUND.value.code
     detail: str = ErrorCodes.PIPETTE_LIQUID_NOT_FOUND.value.detail
-
-
-@dataclass(frozen=True)
-class LiquidNotFoundErrorInternalData:
-    """Internal-to-ProtocolEngine data about a LiquidNotFoundError."""
-
-    position: DeckPoint
-    """Same meaning as DestinationPositionResult.position."""

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

@@ -4,6 +4,8 @@
 from typing import TYPE_CHECKING, Optional, Type
 from typing_extensions import Literal
 
+from opentrons.protocol_engine.state import update_types
+
 from ..errors import TouchTipDisabledError, LabwareIsTipRackError
 from ..types import DeckPoint
 from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
@@ -71,6 +73,8 @@ async def execute(
         labware_id = params.labwareId
         well_name = params.wellName
 
+        state_update = update_types.StateUpdate()
+
         if self._state_view.labware.get_has_quirk(labware_id, "touchTipDisabled"):
             raise TouchTipDisabledError(
                 f"Touch tip not allowed on labware {labware_id}"
@@ -98,14 +102,25 @@ async def execute(
             center_point=center_point,
         )
 
-        x, y, z = await self._gantry_mover.move_to(
+        final_point = await self._gantry_mover.move_to(
             pipette_id=pipette_id,
             waypoints=touch_waypoints,
             speed=touch_speed,
         )
+        final_deck_point = DeckPoint.construct(
+            x=final_point.x, y=final_point.y, z=final_point.z
+        )
+        state_update.set_pipette_location(
+            pipette_id=pipette_id,
+            new_labware_id=labware_id,
+            new_well_name=well_name,
+            new_deck_point=final_deck_point,
+        )
 
         return SuccessData(
-            public=TouchTipResult(position=DeckPoint(x=x, y=y, z=z)), private=None
+            public=TouchTipResult(position=final_deck_point),
+            private=None,
+            state_update=state_update,
         )
 
 

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

@@ -22,7 +22,6 @@
 )
 from opentrons.protocol_engine.actions.actions import FailCommandAction
 from opentrons.protocol_engine.commands.command import DefinedErrorData
-from opentrons.protocol_engine.commands.pipetting_common import LiquidNotFoundError
 from opentrons.types import MountType, Mount as HwMount, Point
 
 from . import update_types
@@ -320,33 +319,6 @@ def _update_current_location(  # noqa: C901
         # These commands leave the pipette in a new location.
         # Update current_location to reflect that.
         if isinstance(action, SucceedCommandAction) and isinstance(
-            action.command.result,
-            (
-                commands.TouchTipResult,
-                commands.LiquidProbeResult,
-                commands.TryLiquidProbeResult,
-            ),
-        ):
-            self._state.current_location = CurrentWell(
-                pipette_id=action.command.params.pipetteId,
-                labware_id=action.command.params.labwareId,
-                well_name=action.command.params.wellName,
-            )
-        elif isinstance(action, FailCommandAction) and (
-            isinstance(action.error, DefinedErrorData)
-            and (
-                (
-                    isinstance(action.running_command, commands.LiquidProbe)
-                    and isinstance(action.error.public, LiquidNotFoundError)
-                )
-            )
-        ):
-            self._state.current_location = CurrentWell(
-                pipette_id=action.running_command.params.pipetteId,
-                labware_id=action.running_command.params.labwareId,
-                well_name=action.running_command.params.wellName,
-            )
-        elif isinstance(action, SucceedCommandAction) and isinstance(
             action.command.result,
             (
                 commands.MoveToAddressableAreaResult,
@@ -440,7 +412,6 @@ def _update_deck_point(  # noqa: C901
                 commands.MoveRelativeResult,
                 commands.MoveToAddressableAreaResult,
                 commands.MoveToAddressableAreaForDropTipResult,
-                commands.TouchTipResult,
             ),
         ):
             pipette_id = action.command.params.pipetteId