blow out

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

@@ -1,18 +1,26 @@
 """Blow-out command request, result, and implementation models."""
 from __future__ import annotations
-from typing import TYPE_CHECKING, Optional, Type
+from typing import TYPE_CHECKING, Optional, Type, Union
+from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
 from typing_extensions import Literal
 
 
 from ..state.update_types import StateUpdate
 from ..types import DeckPoint
 from .pipetting_common import (
+    OverpressureError,
     PipetteIdMixin,
     FlowRateMixin,
     WellLocationMixin,
     DestinationPositionResult,
 )
-from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    DefinedErrorData,
+    SuccessData,
+)
 from ..errors.error_occurrence import ErrorOccurrence
 
 from opentrons.hardware_control import HardwareControlAPI
@@ -21,6 +29,8 @@
 if TYPE_CHECKING:
     from ..execution import MovementHandler, PipettingHandler
     from ..state.state import StateView
+    from ..resources import ModelUtils
+
 
 BlowOutCommandType = Literal["blowout"]
 
@@ -37,9 +47,13 @@ class BlowOutResult(DestinationPositionResult):
     pass
 
 
-class BlowOutImplementation(
-    AbstractCommandImpl[BlowOutParams, SuccessData[BlowOutResult, None]]
-):
+_ExecuteReturn = Union[
+    SuccessData[BlowOutResult, None],
+    DefinedErrorData[OverpressureError],
+]
+
+
+class BlowOutImplementation(AbstractCommandImpl[BlowOutParams, _ExecuteReturn]):
     """BlowOut command implementation."""
 
     def __init__(
@@ -48,14 +62,16 @@ def __init__(
         pipetting: PipettingHandler,
         state_view: StateView,
         hardware_api: HardwareControlAPI,
+        model_utils: ModelUtils,
         **kwargs: object,
     ) -> None:
         self._movement = movement
         self._pipetting = pipetting
         self._state_view = state_view
         self._hardware_api = hardware_api
+        self._model_utils = model_utils
 
-    async def execute(self, params: BlowOutParams) -> SuccessData[BlowOutResult, None]:
+    async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
         """Move to and blow-out the requested well."""
         state_update = StateUpdate()
 
@@ -72,16 +88,37 @@ async def execute(self, params: BlowOutParams) -> SuccessData[BlowOutResult, Non
             new_well_name=params.wellName,
             new_deck_point=deck_point,
         )
-
-        await self._pipetting.blow_out_in_place(
-            pipette_id=params.pipetteId, flow_rate=params.flowRate
-        )
-
-        return SuccessData(
-            public=BlowOutResult(position=deck_point),
-            private=None,
-            state_update=state_update,
-        )
+        try:
+            await self._pipetting.blow_out_in_place(
+                pipette_id=params.pipetteId, flow_rate=params.flowRate
+            )
+        except PipetteOverpressureError as e:
+            return DefinedErrorData(
+                public=OverpressureError(
+                    id=self._model_utils.generate_id(),
+                    createdAt=self._model_utils.get_timestamp(),
+                    wrappedErrors=[
+                        ErrorOccurrence.from_failed(
+                            id=self._model_utils.generate_id(),
+                            createdAt=self._model_utils.get_timestamp(),
+                            error=e,
+                        )
+                    ],
+                    errorInfo={
+                        "retryLocation": (
+                            x,
+                            y,
+                            z,
+                        )
+                    },
+                ),
+            )
+        else:
+            return SuccessData(
+                public=BlowOutResult(position=deck_point),
+                private=None,
+                state_update=state_update,
+            )
 
 
 class BlowOut(BaseCommand[BlowOutParams, BlowOutResult, ErrorOccurrence]):

api/tests/opentrons/protocol_engine/commands/test_blow_out.py

@@ -1,6 +1,9 @@
 """Test blow-out command."""
-from decoy import Decoy
+from datetime import datetime
+from decoy import Decoy, matchers
 
+from opentrons.protocol_engine.commands.pipetting_common import OverpressureError
+from opentrons.protocol_engine.resources.model_utils import ModelUtils
 from opentrons.types import Point
 from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset, DeckPoint
 from opentrons.protocol_engine.state import update_types
@@ -10,29 +13,41 @@
     BlowOutImplementation,
     BlowOutParams,
 )
-from opentrons.protocol_engine.commands.command import SuccessData
+from opentrons.protocol_engine.commands.command import DefinedErrorData, SuccessData
 from opentrons.protocol_engine.execution import (
     MovementHandler,
     PipettingHandler,
 )
 from opentrons.hardware_control import HardwareControlAPI
+from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
+import pytest
 
 
-async def test_blow_out_implementation(
-    decoy: Decoy,
+@pytest.fixture
+def subject(
     state_view: StateView,
     hardware_api: HardwareControlAPI,
     movement: MovementHandler,
+    model_utils: ModelUtils,
     pipetting: PipettingHandler,
-) -> None:
-    """Test BlowOut command execution."""
-    subject = BlowOutImplementation(
+) -> BlowOutImplementation:
+    return BlowOutImplementation(
         state_view=state_view,
         movement=movement,
         hardware_api=hardware_api,
         pipetting=pipetting,
+        model_utils=model_utils,
     )
 
+
+async def test_blow_out_implementation(
+    decoy: Decoy,
+    movement: MovementHandler,
+    pipetting: PipettingHandler,
+    subject: BlowOutImplementation,
+) -> None:
+    """Test BlowOut command execution."""
+
     location = WellLocation(origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1))
 
     data = BlowOutParams(
@@ -73,3 +88,57 @@ async def test_blow_out_implementation(
         await pipetting.blow_out_in_place(pipette_id="pipette-id", flow_rate=1.234),
         times=1,
     )
+
+
+async def test_overpressure_error(
+    decoy: Decoy,
+    pipetting: PipettingHandler,
+    subject: BlowOutImplementation,
+    model_utils: ModelUtils,
+    movement: MovementHandler,
+) -> None:
+    """It should return an overpressure error if the hardware API indicates that."""
+    pipette_id = "pipette-id"
+
+    error_id = "error-id"
+    error_timestamp = datetime(year=2020, month=1, day=2)
+
+    location = WellLocation(origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1))
+
+    data = BlowOutParams(
+        pipetteId="pipette-id",
+        labwareId="labware-id",
+        wellName="C6",
+        wellLocation=location,
+        flowRate=1.234,
+    )
+
+    decoy.when(pipetting.get_is_ready_to_aspirate(pipette_id=pipette_id)).then_return(
+        True
+    )
+
+    decoy.when(
+        await pipetting.blow_out_in_place(pipette_id="pipette-id", flow_rate=1.234)
+    ).then_raise(PipetteOverpressureError())
+
+    decoy.when(model_utils.generate_id()).then_return(error_id)
+    decoy.when(model_utils.get_timestamp()).then_return(error_timestamp)
+    decoy.when(
+        await movement.move_to_well(
+            pipette_id="pipette-id",
+            labware_id="labware-id",
+            well_name="C6",
+            well_location=location,
+        )
+    ).then_return(Point(x=1, y=2, z=3))
+
+    result = await subject.execute(data)
+
+    assert result == DefinedErrorData(
+        public=OverpressureError.construct(
+            id=error_id,
+            createdAt=error_timestamp,
+            wrappedErrors=[matchers.Anything()],
+            errorInfo={"retryLocation": (1, 2, 3)},
+        ),
+    )