Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(api): Handle overpressures in AspirateInPlace, DispenseInPlace #15791

Merged
merged 2 commits into from
Jul 26, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 69 additions & 15 deletions api/src/opentrons/protocol_engine/commands/aspirate_in_place.py
Original file line number Diff line number Diff line change
@@ -1,23 +1,35 @@
"""Aspirate in place 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 typing_extensions import Literal

from opentrons_shared_data.errors.exceptions import PipetteOverpressureError

from opentrons.hardware_control import HardwareControlAPI

from .pipetting_common import (
PipetteIdMixin,
AspirateVolumeMixin,
FlowRateMixin,
BaseLiquidHandlingResult,
OverpressureError,
OverpressureErrorInternalData,
)
from .command import (
AbstractCommandImpl,
BaseCommand,
BaseCommandCreate,
SuccessData,
DefinedErrorData,
)
from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
from ..errors.error_occurrence import ErrorOccurrence
from ..errors.exceptions import PipetteNotReadyToAspirateError
from ..types import DeckPoint

if TYPE_CHECKING:
from ..execution import PipettingHandler
from ..execution import PipettingHandler, GantryMover
from ..resources import ModelUtils
from ..state import StateView
from ..notes import CommandNoteAdder

Expand All @@ -36,8 +48,14 @@ class AspirateInPlaceResult(BaseLiquidHandlingResult):
pass


_ExecuteReturn = Union[
SuccessData[AspirateInPlaceResult, None],
DefinedErrorData[OverpressureError, OverpressureErrorInternalData],
]


class AspirateInPlaceImplementation(
AbstractCommandImpl[AspirateInPlaceParams, SuccessData[AspirateInPlaceResult, None]]
AbstractCommandImpl[AspirateInPlaceParams, _ExecuteReturn]
):
"""AspirateInPlace command implementation."""

Expand All @@ -47,16 +65,18 @@ def __init__(
hardware_api: HardwareControlAPI,
state_view: StateView,
command_note_adder: CommandNoteAdder,
model_utils: ModelUtils,
gantry_mover: GantryMover,
**kwargs: object,
) -> None:
self._pipetting = pipetting
self._state_view = state_view
self._hardware_api = hardware_api
self._command_note_adder = command_note_adder
self._model_utils = model_utils
self._gantry_mover = gantry_mover

async def execute(
self, params: AspirateInPlaceParams
) -> SuccessData[AspirateInPlaceResult, None]:
async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
"""Aspirate without moving the pipette.

Raises:
Expand All @@ -73,14 +93,48 @@ async def execute(
" The first aspirate following a blow-out must be from a specific well"
" so the plunger can be reset in a known safe position."
)
volume = await self._pipetting.aspirate_in_place(
pipette_id=params.pipetteId,
volume=params.volume,
flow_rate=params.flowRate,
command_note_adder=self._command_note_adder,
)

return SuccessData(public=AspirateInPlaceResult(volume=volume), private=None)
try:
volume = await self._pipetting.aspirate_in_place(
pipette_id=params.pipetteId,
volume=params.volume,
flow_rate=params.flowRate,
command_note_adder=self._command_note_adder,
)
except PipetteOverpressureError as e:
current_position = await self._gantry_mover.get_position(params.pipetteId)
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": (
current_position.x,
current_position.y,
current_position.z,
)
}
),
),
private=OverpressureErrorInternalData(
position=DeckPoint(
x=current_position.x,
y=current_position.y,
z=current_position.z,
),
),
)
else:
return SuccessData(
public=AspirateInPlaceResult(volume=volume), private=None
)


class AspirateInPlace(
Expand Down
90 changes: 74 additions & 16 deletions api/src/opentrons/protocol_engine/commands/dispense_in_place.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,32 @@
"""Dispense-in-place 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 typing_extensions import Literal

from pydantic import Field

from opentrons_shared_data.errors.exceptions import PipetteOverpressureError

from .pipetting_common import (
PipetteIdMixin,
DispenseVolumeMixin,
FlowRateMixin,
BaseLiquidHandlingResult,
OverpressureError,
OverpressureErrorInternalData,
)
from .command import (
AbstractCommandImpl,
BaseCommand,
BaseCommandCreate,
SuccessData,
DefinedErrorData,
)
from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
from ..errors.error_occurrence import ErrorOccurrence
from ..types import DeckPoint

if TYPE_CHECKING:
from ..execution import PipettingHandler
from ..execution import PipettingHandler, GantryMover
from ..resources import ModelUtils


DispenseInPlaceCommandType = Literal["dispenseInPlace"]
Expand All @@ -36,25 +47,72 @@ class DispenseInPlaceResult(BaseLiquidHandlingResult):
pass


_ExecuteReturn = Union[
SuccessData[DispenseInPlaceResult, None],
DefinedErrorData[OverpressureError, OverpressureErrorInternalData],
]


class DispenseInPlaceImplementation(
AbstractCommandImpl[DispenseInPlaceParams, SuccessData[DispenseInPlaceResult, None]]
AbstractCommandImpl[DispenseInPlaceParams, _ExecuteReturn]
):
"""DispenseInPlace command implementation."""

def __init__(self, pipetting: PipettingHandler, **kwargs: object) -> None:
def __init__(
self,
pipetting: PipettingHandler,
gantry_mover: GantryMover,
model_utils: ModelUtils,
**kwargs: object,
) -> None:
self._pipetting = pipetting
self._gantry_mover = gantry_mover
self._model_utils = model_utils

async def execute(
self, params: DispenseInPlaceParams
) -> SuccessData[DispenseInPlaceResult, None]:
async def execute(self, params: DispenseInPlaceParams) -> _ExecuteReturn:
"""Dispense without moving the pipette."""
volume = await self._pipetting.dispense_in_place(
pipette_id=params.pipetteId,
volume=params.volume,
flow_rate=params.flowRate,
push_out=params.pushOut,
)
return SuccessData(public=DispenseInPlaceResult(volume=volume), private=None)
try:
volume = await self._pipetting.dispense_in_place(
pipette_id=params.pipetteId,
volume=params.volume,
flow_rate=params.flowRate,
push_out=params.pushOut,
)
except PipetteOverpressureError as e:
current_position = await self._gantry_mover.get_position(params.pipetteId)
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": (
current_position.x,
current_position.y,
current_position.z,
)
}
),
),
private=OverpressureErrorInternalData(
position=DeckPoint(
x=current_position.x,
y=current_position.y,
z=current_position.z,
),
),
)
else:
return SuccessData(
public=DispenseInPlaceResult(volume=volume), private=None
)


class DispenseInPlace(
Expand Down
7 changes: 6 additions & 1 deletion api/src/opentrons/protocol_engine/state/pipettes.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
from opentrons.protocol_engine.actions.actions import FailCommandAction
from opentrons.protocol_engine.commands.aspirate import Aspirate
from opentrons.protocol_engine.commands.dispense import Dispense
from opentrons.protocol_engine.commands.aspirate_in_place import AspirateInPlace
from opentrons.protocol_engine.commands.dispense_in_place import DispenseInPlace
from opentrons.protocol_engine.commands.command import DefinedErrorData
from opentrons.protocol_engine.commands.pipetting_common import (
OverpressureError,
Expand Down Expand Up @@ -413,7 +415,10 @@ def _update_deck_point(
)
elif (
isinstance(action, FailCommandAction)
and isinstance(action.running_command, (Aspirate, Dispense))
and isinstance(
action.running_command,
(Aspirate, Dispense, AspirateInPlace, DispenseInPlace),
)
and isinstance(action.error, DefinedErrorData)
and isinstance(action.error.public, OverpressureError)
):
Expand Down
7 changes: 7 additions & 0 deletions api/tests/opentrons/protocol_engine/commands/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
LabwareMovementHandler,
StatusBarHandler,
TipHandler,
GantryMover,
)
from opentrons.protocol_engine.resources.model_utils import ModelUtils
from opentrons.protocol_engine.state import StateView
Expand Down Expand Up @@ -76,3 +77,9 @@ def model_utils(decoy: Decoy) -> ModelUtils:
def status_bar(decoy: Decoy) -> StatusBarHandler:
"""Get a mocked out StatusBarHandler."""
return decoy.mock(cls=StatusBarHandler)


@pytest.fixture
def gantry_mover(decoy: Decoy) -> GantryMover:
"""Get a mocked out GantryMover."""
return decoy.mock(cls=GantryMover)
Loading
Loading