From 90972252f4c0836f55520643954de4df78e3d226 Mon Sep 17 00:00:00 2001 From: Max Marrone Date: Thu, 6 Jun 2024 17:40:06 -0400 Subject: [PATCH] feat(api): Prepare PipetteStore to handle overpressure errors from aspirate commands (#15321) --- .../commands/command_unions.py | 7 +- .../protocol_engine/commands/pick_up_tip.py | 2 +- .../commands/pipetting_common.py | 30 +- .../protocol_engine/state/pipettes.py | 197 +++++++----- .../state/test_pipette_store.py | 292 ++++++++++++------ 5 files changed, 364 insertions(+), 164 deletions(-) diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py index 9196cdf89cd..362b2eb0a02 100644 --- a/api/src/opentrons/protocol_engine/commands/command_unions.py +++ b/api/src/opentrons/protocol_engine/commands/command_unions.py @@ -6,6 +6,7 @@ from pydantic import Field from .command import DefinedErrorData +from .pipetting_common import OverpressureError, OverpressureErrorInternalData from . import heater_shaker from . import magnetic_module @@ -629,7 +630,7 @@ ] # All `DefinedErrorData`s that implementations will actually return in practice. -# There's just one right now, but this will eventually be a Union. -CommandDefinedErrorData = DefinedErrorData[ - TipPhysicallyMissingError, TipPhysicallyMissingErrorInternalData +CommandDefinedErrorData = Union[ + DefinedErrorData[TipPhysicallyMissingError, TipPhysicallyMissingErrorInternalData], + DefinedErrorData[OverpressureError, OverpressureErrorInternalData], ] diff --git a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py index 684a8fff476..0022c517eeb 100644 --- a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py +++ b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py @@ -78,7 +78,7 @@ class TipPhysicallyMissingError(ErrorOccurrence): detail: str = "No tip detected." -@dataclass +@dataclass(frozen=True) class TipPhysicallyMissingErrorInternalData: """Internal-to-ProtocolEngine data about a TipPhysicallyMissingError.""" diff --git a/api/src/opentrons/protocol_engine/commands/pipetting_common.py b/api/src/opentrons/protocol_engine/commands/pipetting_common.py index 2d165e4894a..9b080275898 100644 --- a/api/src/opentrons/protocol_engine/commands/pipetting_common.py +++ b/api/src/opentrons/protocol_engine/commands/pipetting_common.py @@ -1,6 +1,10 @@ """Common pipetting command base models.""" +from dataclasses import dataclass +from opentrons_shared_data.errors import ErrorCodes from pydantic import BaseModel, Field -from typing import Optional +from typing import Literal, Optional + +from opentrons.protocol_engine.errors.error_occurrence import ErrorOccurrence from ..types import WellLocation, DeckPoint @@ -117,3 +121,27 @@ class DestinationPositionResult(BaseModel): " after the move was completed." ), ) + + +class OverpressureError(ErrorOccurrence): + """Returned when sensors detect an overpressure error while moving liquid. + + The pipette plunger motion is stopped at the point of the error. The next thing to + move the plunger must be a `home` or `blowout` command; commands like `aspirate` + will return an error. + """ + + isDefined: bool = True + + errorType: Literal["overpressure"] = "overpressure" + + errorCode: str = ErrorCodes.PIPETTE_OVERPRESSURE.value.code + detail: str = ErrorCodes.PIPETTE_OVERPRESSURE.value.detail + + +@dataclass(frozen=True) +class OverpressureErrorInternalData: + """Internal-to-ProtocolEngine data about an OverpressureError.""" + + position: DeckPoint + """Same meaning as DestinationPositionResult.position.""" diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index 6803d19272b..d1153e9a51e 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -1,7 +1,8 @@ """Basic pipette data state and store.""" from __future__ import annotations from dataclasses import dataclass -from typing import Dict, List, Mapping, Optional, Tuple +from typing import Dict, List, Mapping, Optional, Tuple, Union +from typing_extensions import assert_type from opentrons_shared_data.pipette import pipette_definition from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE @@ -10,6 +11,13 @@ NozzleConfigurationType, NozzleMap, ) +from opentrons.protocol_engine.actions.actions import FailCommandAction +from opentrons.protocol_engine.commands.aspirate import Aspirate +from opentrons.protocol_engine.commands.command import DefinedErrorData +from opentrons.protocol_engine.commands.pipetting_common import ( + OverpressureError, + OverpressureErrorInternalData, +) from opentrons.types import MountType, Mount as HwMount, Point from .. import errors @@ -24,7 +32,6 @@ TipGeometry, ) from ..commands import ( - Command, LoadPipetteResult, AspirateResult, AspirateInPlaceResult, @@ -46,7 +53,6 @@ TouchTipResult, thermocycler, heater_shaker, - CommandPrivateResult, PrepareToAspirateResult, ) from ..commands.configuring_common import ( @@ -150,16 +156,22 @@ def __init__(self) -> None: def handle_action(self, action: Action) -> None: """Modify state in reaction to an action.""" - if isinstance(action, SucceedCommandAction): - self._handle_command(action.command, action.private_result) + if isinstance(action, (SucceedCommandAction, FailCommandAction)): + self._handle_command(action) elif isinstance(action, SetPipetteMovementSpeedAction): self._state.movement_speed_by_id[action.pipette_id] = action.speed def _handle_command( # noqa: C901 - self, command: Command, private_result: CommandPrivateResult + self, action: Union[SucceedCommandAction, FailCommandAction] ) -> None: - self._update_current_location(command) - self._update_deck_point(command) + self._update_current_location(action) + self._update_deck_point(action) + self._update_volumes(action) + + if not isinstance(action, SucceedCommandAction): + return + + command, private_result = action.command, action.private_result if isinstance(private_result, PipetteConfigUpdateResultMixin): config = private_result.config @@ -212,23 +224,6 @@ def _handle_command( # noqa: C901 pipette_id ] = static_config.default_nozzle_map - elif isinstance(command.result, (AspirateResult, AspirateInPlaceResult)): - pipette_id = command.params.pipetteId - previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0 - # PipetteHandler will have clamped command.result.volume for us, so - # next_volume should always be in bounds. - next_volume = previous_volume + command.result.volume - - self._state.aspirated_volume_by_id[pipette_id] = next_volume - - elif isinstance(command.result, (DispenseResult, DispenseInPlaceResult)): - pipette_id = command.params.pipetteId - previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0 - # PipetteHandler will have clamped command.result.volume for us, so - # next_volume should always be in bounds. - next_volume = previous_volume - command.result.volume - self._state.aspirated_volume_by_id[pipette_id] = next_volume - elif isinstance(command.result, PickUpTipResult): pipette_id = command.params.pipetteId attached_tip = TipGeometry( @@ -277,19 +272,14 @@ def _handle_command( # noqa: C901 default_aspirate=tip_configuration.default_aspirate_flowrate.values_by_api_level, default_dispense=tip_configuration.default_dispense_flowrate.values_by_api_level, ) - elif isinstance(command.result, (BlowOutResult, BlowOutInPlaceResult)): - pipette_id = command.params.pipetteId - self._state.aspirated_volume_by_id[pipette_id] = None - elif isinstance(command.result, PrepareToAspirateResult): - pipette_id = command.params.pipetteId - self._state.aspirated_volume_by_id[pipette_id] = 0 - - def _update_current_location(self, command: Command) -> None: + def _update_current_location( # noqa: C901 + self, action: Union[SucceedCommandAction, FailCommandAction] + ) -> None: # These commands leave the pipette in a new location. # Update current_location to reflect that. - if isinstance( - command.result, + if isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( MoveToWellResult, PickUpTipResult, @@ -301,18 +291,28 @@ def _update_current_location(self, command: Command) -> None: ), ): self._state.current_location = CurrentWell( - pipette_id=command.params.pipetteId, - labware_id=command.params.labwareId, - well_name=command.params.wellName, + pipette_id=action.command.params.pipetteId, + labware_id=action.command.params.labwareId, + well_name=action.command.params.wellName, ) - - elif isinstance( - command.result, + elif ( + isinstance(action, FailCommandAction) + and isinstance(action.running_command, Aspirate) + and isinstance(action.error, DefinedErrorData) + and isinstance(action.error.public, OverpressureError) + ): + 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, (MoveToAddressableAreaResult, MoveToAddressableAreaForDropTipResult), ): self._state.current_location = CurrentAddressableArea( - pipette_id=command.params.pipetteId, - addressable_area_name=command.params.addressableAreaName, + pipette_id=action.command.params.pipetteId, + addressable_area_name=action.command.params.addressableAreaName, ) # These commands leave the pipette in a place that we can't logically associate @@ -320,8 +320,8 @@ def _update_current_location(self, command: Command) -> None: # # TODO(mc, 2021-11-12): Wipe out current_location on movement failures, too. # TODO(jbl 2023-02-14): Need to investigate whether move relative should clear current location - elif isinstance( - command.result, + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( HomeResult, RetractAxisResult, @@ -334,14 +334,14 @@ def _update_current_location(self, command: Command) -> None: # Heater-Shaker commands may have left the pipette in a place that we can't # associate with a logical location, depending on their result. - elif isinstance( - command.result, + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( heater_shaker.SetAndWaitForShakeSpeedResult, heater_shaker.OpenLabwareLatchResult, ), ): - if command.result.pipetteRetracted: + if action.command.result.pipetteRetracted: self._state.current_location = None # A moveLabware command may have moved the labware that contains the current @@ -350,9 +350,11 @@ def _update_current_location(self, command: Command) -> None: # # This is necessary for safe motion planning in case the next movement # goes to the same labware (now in a new place). - elif isinstance(command.result, MoveLabwareResult): - moved_labware_id = command.params.labwareId - if command.params.strategy == "usingGripper": + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, MoveLabwareResult + ): + moved_labware_id = action.command.params.labwareId + if action.command.params.strategy == "usingGripper": # All mounts will have been retracted. self._state.current_location = None elif ( @@ -361,9 +363,14 @@ def _update_current_location(self, command: Command) -> None: ): self._state.current_location = None - def _update_deck_point(self, command: Command) -> None: - if isinstance( - command.result, + def _update_deck_point( + self, action: Union[SucceedCommandAction, FailCommandAction] + ) -> None: + # This function mostly mirrors self._update_current_location(). + # See there for explanations. + + if isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( MoveToWellResult, MoveToCoordinatesResult, @@ -378,20 +385,28 @@ def _update_deck_point(self, command: Command) -> None: TouchTipResult, ), ): - pipette_id = command.params.pipetteId - deck_point = command.result.position - - try: - loaded_pipette = self._state.pipettes_by_id[pipette_id] - except KeyError: - self._clear_deck_point() - else: - self._state.current_deck_point = CurrentDeckPoint( - mount=loaded_pipette.mount, deck_point=deck_point - ) + pipette_id = action.command.params.pipetteId + deck_point = action.command.result.position + loaded_pipette = self._state.pipettes_by_id[pipette_id] + self._state.current_deck_point = CurrentDeckPoint( + mount=loaded_pipette.mount, deck_point=deck_point + ) + elif ( + isinstance(action, FailCommandAction) + and isinstance(action.running_command, Aspirate) + and isinstance(action.error, DefinedErrorData) + and isinstance(action.error.public, OverpressureError) + ): + assert_type(action.error.private, OverpressureErrorInternalData) + pipette_id = action.running_command.params.pipetteId + deck_point = action.error.private.position + loaded_pipette = self._state.pipettes_by_id[pipette_id] + self._state.current_deck_point = CurrentDeckPoint( + mount=loaded_pipette.mount, deck_point=deck_point + ) - elif isinstance( - command.result, + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( HomeResult, RetractAxisResult, @@ -401,21 +416,59 @@ def _update_deck_point(self, command: Command) -> None: ): self._clear_deck_point() - elif isinstance( - command.result, + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, ( heater_shaker.SetAndWaitForShakeSpeedResult, heater_shaker.OpenLabwareLatchResult, ), ): - if command.result.pipetteRetracted: + if action.command.result.pipetteRetracted: self._clear_deck_point() - elif isinstance(command.result, MoveLabwareResult): - if command.params.strategy == "usingGripper": + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, MoveLabwareResult + ): + if action.command.params.strategy == "usingGripper": # All mounts will have been retracted. self._clear_deck_point() + def _update_volumes( + self, action: Union[SucceedCommandAction, FailCommandAction] + ) -> None: + if isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, (AspirateResult, AspirateInPlaceResult) + ): + pipette_id = action.command.params.pipetteId + previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0 + # PipetteHandler will have clamped action.command.result.volume for us, so + # next_volume should always be in bounds. + next_volume = previous_volume + action.command.result.volume + + self._state.aspirated_volume_by_id[pipette_id] = next_volume + + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, (DispenseResult, DispenseInPlaceResult) + ): + pipette_id = action.command.params.pipetteId + previous_volume = self._state.aspirated_volume_by_id[pipette_id] or 0 + # PipetteHandler will have clamped action.command.result.volume for us, so + # next_volume should always be in bounds. + next_volume = previous_volume - action.command.result.volume + self._state.aspirated_volume_by_id[pipette_id] = next_volume + + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, (BlowOutResult, BlowOutInPlaceResult) + ): + pipette_id = action.command.params.pipetteId + self._state.aspirated_volume_by_id[pipette_id] = None + + elif isinstance(action, SucceedCommandAction) and isinstance( + action.command.result, PrepareToAspirateResult + ): + pipette_id = action.command.params.pipetteId + self._state.aspirated_volume_by_id[pipette_id] = 0 + def _clear_deck_point(self) -> None: """Reset last deck point to default None value for mount and point.""" self._state.current_deck_point = CurrentDeckPoint(mount=None, deck_point=None) diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py index d2479a55bc8..d1b4ac29f7c 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py @@ -1,13 +1,19 @@ """Tests for pipette state changes in the protocol_engine state store.""" import pytest from datetime import datetime -from typing import Optional +from typing import Optional, Union from opentrons_shared_data.pipette.dev_types import PipetteNameType from opentrons_shared_data.pipette import pipette_definition from opentrons.types import DeckSlotName, MountType, Point from opentrons.protocol_engine import commands as cmd +from opentrons.protocol_engine.commands.command import DefinedErrorData +from opentrons.protocol_engine.commands.pipetting_common import ( + OverpressureError, + OverpressureErrorInternalData, +) +from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryType from opentrons.protocol_engine.types import ( DeckPoint, DeckSlotLocation, @@ -19,6 +25,7 @@ TipGeometry, ) from opentrons.protocol_engine.actions import ( + FailCommandAction, SetPipetteMovementSpeedAction, SucceedCommandAction, ) @@ -284,81 +291,135 @@ def test_blow_out_clears_volume( @pytest.mark.parametrize( - ("command", "expected_location"), + ("action", "expected_location"), ( ( - create_aspirate_command( - pipette_id="aspirate-pipette-id", + SucceedCommandAction( + command=create_aspirate_command( + pipette_id="pipette-id", + labware_id="aspirate-labware-id", + well_name="aspirate-well-name", + volume=1337, + flow_rate=1.23, + ), + private_result=None, + ), + CurrentWell( + pipette_id="pipette-id", labware_id="aspirate-labware-id", well_name="aspirate-well-name", - volume=1337, - flow_rate=1.23, + ), + ), + ( + FailCommandAction( + running_command=cmd.Aspirate( + params=cmd.AspirateParams( + pipetteId="pipette-id", + labwareId="aspirate-labware-id", + wellName="aspirate-well-name", + volume=99999, + flowRate=1.23, + ), + id="command-id", + key="command-key", + createdAt=datetime.now(), + status=cmd.CommandStatus.RUNNING, + ), + error=DefinedErrorData( + public=OverpressureError( + id="error-id", + createdAt=datetime.now(), + ), + private=OverpressureErrorInternalData( + position=DeckPoint(x=0, y=0, z=0) + ), + ), + command_id="command-id", + error_id="error-id", + failed_at=datetime.now(), + notes=[], + type=ErrorRecoveryType.WAIT_FOR_RECOVERY, ), CurrentWell( - pipette_id="aspirate-pipette-id", + pipette_id="pipette-id", labware_id="aspirate-labware-id", well_name="aspirate-well-name", ), ), ( - create_dispense_command( - pipette_id="dispense-pipette-id", - labware_id="dispense-labware-id", - well_name="dispense-well-name", - volume=1337, - flow_rate=1.23, + SucceedCommandAction( + command=create_dispense_command( + pipette_id="pipette-id", + labware_id="dispense-labware-id", + well_name="dispense-well-name", + volume=1337, + flow_rate=1.23, + ), + private_result=None, ), CurrentWell( - pipette_id="dispense-pipette-id", + pipette_id="pipette-id", labware_id="dispense-labware-id", well_name="dispense-well-name", ), ), ( - create_pick_up_tip_command( - pipette_id="pick-up-tip-pipette-id", - labware_id="pick-up-tip-labware-id", - well_name="pick-up-tip-well-name", + SucceedCommandAction( + command=create_pick_up_tip_command( + pipette_id="pipette-id", + labware_id="pick-up-tip-labware-id", + well_name="pick-up-tip-well-name", + ), + private_result=None, ), CurrentWell( - pipette_id="pick-up-tip-pipette-id", + pipette_id="pipette-id", labware_id="pick-up-tip-labware-id", well_name="pick-up-tip-well-name", ), ), ( - create_drop_tip_command( - pipette_id="drop-tip-pipette-id", - labware_id="drop-tip-labware-id", - well_name="drop-tip-well-name", + SucceedCommandAction( + command=create_drop_tip_command( + pipette_id="pipette-id", + labware_id="drop-tip-labware-id", + well_name="drop-tip-well-name", + ), + private_result=None, ), CurrentWell( - pipette_id="drop-tip-pipette-id", + pipette_id="pipette-id", labware_id="drop-tip-labware-id", well_name="drop-tip-well-name", ), ), ( - create_move_to_well_command( - pipette_id="move-to-well-pipette-id", - labware_id="move-to-well-labware-id", - well_name="move-to-well-well-name", + SucceedCommandAction( + command=create_move_to_well_command( + pipette_id="pipette-id", + labware_id="move-to-well-labware-id", + well_name="move-to-well-well-name", + ), + private_result=None, ), CurrentWell( - pipette_id="move-to-well-pipette-id", + pipette_id="pipette-id", labware_id="move-to-well-labware-id", well_name="move-to-well-well-name", ), ), ( - create_blow_out_command( - pipette_id="move-to-well-pipette-id", - labware_id="move-to-well-labware-id", - well_name="move-to-well-well-name", - flow_rate=1.23, + SucceedCommandAction( + command=create_blow_out_command( + pipette_id="pipette-id", + labware_id="move-to-well-labware-id", + well_name="move-to-well-well-name", + flow_rate=1.23, + ), + private_result=None, ), CurrentWell( - pipette_id="move-to-well-pipette-id", + pipette_id="pipette-id", labware_id="move-to-well-labware-id", well_name="move-to-well-well-name", ), @@ -366,13 +427,13 @@ def test_blow_out_clears_volume( ), ) def test_movement_commands_update_current_well( - command: cmd.Command, + action: Union[SucceedCommandAction, FailCommandAction], expected_location: CurrentWell, subject: PipetteStore, ) -> None: """It should save the last used pipette, labware, and well for movement commands.""" load_pipette_command = create_load_pipette_command( - pipette_id=command.params.pipetteId, # type: ignore[arg-type, union-attr] + pipette_id="pipette-id", pipette_name=PipetteNameType.P300_SINGLE, mount=MountType.LEFT, ) @@ -380,7 +441,7 @@ def test_movement_commands_update_current_well( subject.handle_action( SucceedCommandAction(private_result=None, command=load_pipette_command) ) - subject.handle_action(SucceedCommandAction(private_result=None, command=command)) + subject.handle_action(action) assert subject.state.current_location == expected_location @@ -720,67 +781,124 @@ def test_add_pipette_config( @pytest.mark.parametrize( - "command", + "action", ( - create_aspirate_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - volume=1337, - flow_rate=1.23, - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_aspirate_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + volume=1337, + flow_rate=1.23, + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_dispense_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - volume=1337, - flow_rate=1.23, - destination=DeckPoint(x=11, y=22, z=33), + FailCommandAction( + running_command=cmd.Aspirate( + params=cmd.AspirateParams( + pipetteId="pipette-id", + labwareId="labware-id", + wellName="well-name", + volume=99999, + flowRate=1.23, + ), + id="command-id", + key="command-key", + createdAt=datetime.now(), + status=cmd.CommandStatus.RUNNING, + ), + error=DefinedErrorData( + public=OverpressureError( + id="error-id", + detail="error-detail", + createdAt=datetime.now(), + ), + private=OverpressureErrorInternalData( + position=DeckPoint(x=11, y=22, z=33) + ), + ), + command_id="command-id", + error_id="error-id", + failed_at=datetime.now(), + notes=[], + type=ErrorRecoveryType.WAIT_FOR_RECOVERY, ), - create_blow_out_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - flow_rate=1.23, - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_dispense_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + volume=1337, + flow_rate=1.23, + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_pick_up_tip_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_blow_out_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + flow_rate=1.23, + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_drop_tip_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_pick_up_tip_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_touch_tip_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_drop_tip_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_move_to_well_command( - pipette_id="pipette-id", - labware_id="labware-id", - well_name="well-name", - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_touch_tip_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_move_to_coordinates_command( - pipette_id="pipette-id", - coordinates=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_move_to_well_command( + pipette_id="pipette-id", + labware_id="labware-id", + well_name="well-name", + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), - create_move_relative_command( - pipette_id="pipette-id", - destination=DeckPoint(x=11, y=22, z=33), + SucceedCommandAction( + command=create_move_to_coordinates_command( + pipette_id="pipette-id", + coordinates=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, + ), + SucceedCommandAction( + command=create_move_relative_command( + pipette_id="pipette-id", + destination=DeckPoint(x=11, y=22, z=33), + ), + private_result=None, ), ), ) def test_movement_commands_update_deck_point( - command: cmd.Command, + action: Union[SucceedCommandAction, FailCommandAction], subject: PipetteStore, ) -> None: """It should save the last used pipette, labware, and well for movement commands.""" @@ -793,7 +911,7 @@ def test_movement_commands_update_deck_point( subject.handle_action( SucceedCommandAction(private_result=None, command=load_pipette_command) ) - subject.handle_action(SucceedCommandAction(private_result=None, command=command)) + subject.handle_action(action) assert subject.state.current_deck_point == CurrentDeckPoint( mount=MountType.LEFT, deck_point=DeckPoint(x=11, y=22, z=33)