feat(api): express stalls in a recoverable way (#16861)

Commands that use `move_to_well`, `move_to_coordinates`,
`move_to_addressable_area`, and `move_relative` now will return stalls
as DefinedErrors, which means they can be hooked into error recovery en
masse.

This just leaves move labware.

Closes EXEC-831

## Reviews
- Feel like it paid off?

## Testing
- Do some stalls and make sure they end up with defined errors

## Further work and questions
- I think we should probably handle errors during `aspirate`'s automatic
prepare for aspirate at the beginning and probably the same with drop
tip but we didn't handle it at all so far so I don't know

---------

Co-authored-by: Max Marrone <[email protected]>

Author: sfoster1

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

@@ -15,6 +15,7 @@
 from .movement_common import (
     LiquidHandlingWellLocationMixin,
     DestinationPositionResult,
+    StallOrCollisionError,
     move_to_well,
 )
 from .command import (
@@ -60,7 +61,7 @@ class AspirateResult(BaseLiquidHandlingResult, DestinationPositionResult):
 
 _ExecuteReturn = Union[
     SuccessData[AspirateResult],
-    DefinedErrorData[OverpressureError],
+    DefinedErrorData[OverpressureError] | DefinedErrorData[StallOrCollisionError],
 ]
 
 
@@ -120,13 +121,16 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
 
         move_result = await move_to_well(
             movement=self._movement,
+            model_utils=self._model_utils,
             pipette_id=pipette_id,
             labware_id=labware_id,
             well_name=well_name,
             well_location=params.wellLocation,
             current_well=current_well,
             operation_volume=-params.volume,
         )
+        if isinstance(move_result, DefinedErrorData):
+            return move_result
 
         aspirate_result = await aspirate_in_place(
             pipette_id=pipette_id,
@@ -185,7 +189,11 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
             )
 
 
-class Aspirate(BaseCommand[AspirateParams, AspirateResult, OverpressureError]):
+class Aspirate(
+    BaseCommand[
+        AspirateParams, AspirateResult, OverpressureError | StallOrCollisionError
+    ]
+):
     """Aspirate command model."""
 
     commandType: AspirateCommandType = "aspirate"

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

@@ -11,15 +11,19 @@
     FlowRateMixin,
     blow_out_in_place,
 )
-from .movement_common import WellLocationMixin, DestinationPositionResult, move_to_well
+from .movement_common import (
+    WellLocationMixin,
+    DestinationPositionResult,
+    move_to_well,
+    StallOrCollisionError,
+)
 from .command import (
     AbstractCommandImpl,
     BaseCommand,
     BaseCommandCreate,
     DefinedErrorData,
     SuccessData,
 )
-from ..errors.error_occurrence import ErrorOccurrence
 from ..state.update_types import StateUpdate
 
 from opentrons.hardware_control import HardwareControlAPI
@@ -48,7 +52,7 @@ class BlowOutResult(DestinationPositionResult):
 
 _ExecuteReturn = Union[
     SuccessData[BlowOutResult],
-    DefinedErrorData[OverpressureError],
+    DefinedErrorData[OverpressureError] | DefinedErrorData[StallOrCollisionError],
 ]
 
 
@@ -74,11 +78,14 @@ async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
         """Move to and blow-out the requested well."""
         move_result = await move_to_well(
             movement=self._movement,
+            model_utils=self._model_utils,
             pipette_id=params.pipetteId,
             labware_id=params.labwareId,
             well_name=params.wellName,
             well_location=params.wellLocation,
         )
+        if isinstance(move_result, DefinedErrorData):
+            return move_result
         blow_out_result = await blow_out_in_place(
             pipette_id=params.pipetteId,
             flow_rate=params.flowRate,
@@ -112,7 +119,13 @@ async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
             )
 
 
-class BlowOut(BaseCommand[BlowOutParams, BlowOutResult, ErrorOccurrence]):
+class BlowOut(
+    BaseCommand[
+        BlowOutParams,
+        BlowOutResult,
+        OverpressureError | StallOrCollisionError,
+    ]
+):
     """Blow-out command model."""
 
     commandType: BlowOutCommandType = "blowout"

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

@@ -13,6 +13,7 @@
     LiquidNotFoundError,
     TipPhysicallyAttachedError,
 )
+from .movement_common import StallOrCollisionError
 
 from . import absorbance_reader
 from . import heater_shaker
@@ -754,6 +755,7 @@
     DefinedErrorData[OverpressureError],
     DefinedErrorData[LiquidNotFoundError],
     DefinedErrorData[GripperMovementError],
+    DefinedErrorData[StallOrCollisionError],
 ]
 
 

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

@@ -19,6 +19,7 @@
 from .movement_common import (
     LiquidHandlingWellLocationMixin,
     DestinationPositionResult,
+    StallOrCollisionError,
     move_to_well,
 )
 from .command import (
@@ -57,7 +58,7 @@ class DispenseResult(BaseLiquidHandlingResult, DestinationPositionResult):
 
 _ExecuteReturn = Union[
     SuccessData[DispenseResult],
-    DefinedErrorData[OverpressureError],
+    DefinedErrorData[OverpressureError] | DefinedErrorData[StallOrCollisionError],
 ]
 
 
@@ -88,11 +89,14 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
 
         move_result = await move_to_well(
             movement=self._movement,
+            model_utils=self._model_utils,
             pipette_id=params.pipetteId,
             labware_id=labware_id,
             well_name=well_name,
             well_location=well_location,
         )
+        if isinstance(move_result, DefinedErrorData):
+            return move_result
         dispense_result = await dispense_in_place(
             pipette_id=params.pipetteId,
             volume=volume,
@@ -159,7 +163,11 @@ async def execute(self, params: DispenseParams) -> _ExecuteReturn:
             )
 
 
-class Dispense(BaseCommand[DispenseParams, DispenseResult, OverpressureError]):
+class Dispense(
+    BaseCommand[
+        DispenseParams, DispenseResult, OverpressureError | StallOrCollisionError
+    ]
+):
     """Dispense command model."""
 
     commandType: DispenseCommandType = "dispense"

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

@@ -15,7 +15,11 @@
     PipetteIdMixin,
     TipPhysicallyAttachedError,
 )
-from .movement_common import DestinationPositionResult, move_to_well
+from .movement_common import (
+    DestinationPositionResult,
+    move_to_well,
+    StallOrCollisionError,
+)
 from .command import (
     AbstractCommandImpl,
     BaseCommand,
@@ -69,7 +73,9 @@ class DropTipResult(DestinationPositionResult):
 
 
 _ExecuteReturn = (
-    SuccessData[DropTipResult] | DefinedErrorData[TipPhysicallyAttachedError]
+    SuccessData[DropTipResult]
+    | DefinedErrorData[TipPhysicallyAttachedError]
+    | DefinedErrorData[StallOrCollisionError]
 )
 
 
@@ -117,11 +123,14 @@ async def execute(self, params: DropTipParams) -> _ExecuteReturn:
 
         move_result = await move_to_well(
             movement=self._movement_handler,
+            model_utils=self._model_utils,
             pipette_id=pipette_id,
             labware_id=labware_id,
             well_name=well_name,
             well_location=tip_drop_location,
         )
+        if isinstance(move_result, DefinedErrorData):
+            return move_result
 
         try:
             await self._tip_handler.drop_tip(

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

@@ -27,6 +27,7 @@
 from .movement_common import (
     WellLocationMixin,
     DestinationPositionResult,
+    StallOrCollisionError,
     move_to_well,
 )
 from .command import (
@@ -91,9 +92,11 @@ class TryLiquidProbeResult(DestinationPositionResult):
 
 _LiquidProbeExecuteReturn = Union[
     SuccessData[LiquidProbeResult],
-    DefinedErrorData[LiquidNotFoundError],
+    DefinedErrorData[LiquidNotFoundError] | DefinedErrorData[StallOrCollisionError],
 ]
-_TryLiquidProbeExecuteReturn = SuccessData[TryLiquidProbeResult]
+_TryLiquidProbeExecuteReturn = (
+    SuccessData[TryLiquidProbeResult] | DefinedErrorData[StallOrCollisionError]
+)
 
 
 class _ExecuteCommonResult(NamedTuple):
@@ -110,8 +113,9 @@ async def _execute_common(
     state_view: StateView,
     movement: MovementHandler,
     pipetting: PipettingHandler,
+    model_utils: ModelUtils,
     params: _CommonParams,
-) -> _ExecuteCommonResult:
+) -> _ExecuteCommonResult | DefinedErrorData[StallOrCollisionError]:
     pipette_id = params.pipetteId
     labware_id = params.labwareId
     well_name = params.wellName
@@ -145,12 +149,14 @@ async def _execute_common(
     # liquid_probe process start position
     move_result = await move_to_well(
         movement=movement,
+        model_utils=model_utils,
         pipette_id=pipette_id,
         labware_id=labware_id,
         well_name=well_name,
         well_location=params.wellLocation,
     )
-
+    if isinstance(move_result, DefinedErrorData):
+        return move_result
     try:
         z_pos = await pipetting.liquid_probe_in_place(
             pipette_id=pipette_id,
@@ -206,9 +212,16 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
             MustHomeError: as an undefined error, if the plunger is not in a valid
                 position.
         """
-        z_pos_or_error, state_update, deck_point = await _execute_common(
-            self._state_view, self._movement, self._pipetting, params
+        result = await _execute_common(
+            state_view=self._state_view,
+            movement=self._movement,
+            pipetting=self._pipetting,
+            model_utils=self._model_utils,
+            params=params,
         )
+        if isinstance(result, DefinedErrorData):
+            return result
+        z_pos_or_error, state_update, deck_point = result
         if isinstance(z_pos_or_error, PipetteLiquidNotFoundError):
             state_update.set_liquid_probed(
                 labware_id=params.labwareId,
@@ -282,9 +295,16 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
         found, `tryLiquidProbe` returns a success result with `z_position=null` instead
         of a defined error.
         """
-        z_pos_or_error, state_update, deck_point = await _execute_common(
-            self._state_view, self._movement, self._pipetting, params
+        result = await _execute_common(
+            state_view=self._state_view,
+            movement=self._movement,
+            pipetting=self._pipetting,
+            model_utils=self._model_utils,
+            params=params,
         )
+        if isinstance(result, DefinedErrorData):
+            return result
+        z_pos_or_error, state_update, deck_point = result
 
         if isinstance(z_pos_or_error, PipetteLiquidNotFoundError):
             z_pos = None
@@ -316,7 +336,11 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
 
 
 class LiquidProbe(
-    BaseCommand[LiquidProbeParams, LiquidProbeResult, LiquidNotFoundError]
+    BaseCommand[
+        LiquidProbeParams,
+        LiquidProbeResult,
+        LiquidNotFoundError | StallOrCollisionError,
+    ]
 ):
     """The model for a full `liquidProbe` command."""
 
@@ -328,7 +352,7 @@ class LiquidProbe(
 
 
 class TryLiquidProbe(
-    BaseCommand[TryLiquidProbeParams, TryLiquidProbeResult, ErrorOccurrence]
+    BaseCommand[TryLiquidProbeParams, TryLiquidProbeResult, StallOrCollisionError]
 ):
     """The model for a full `tryLiquidProbe` command."""