Skip to content

Commit

Permalink
fix(api): Fix liquidProbe error message after blowout (#16294)
Browse files Browse the repository at this point in the history 
## Overview

This is the error message fix part of RQA-3171.

## Changelog

When we're not ready to do a `liquidProbe`, be a little bit
finer-grained about why we're not ready. Give the "missing tip" case and
"plunger not ready" case their own error messages.
  • Loading branch information
@SyntaxColoring
SyntaxColoring authored Sep 19, 2024
1 parent 6293e07 commit 65b0d7f
Show file tree
Hide file tree
Showing 4 changed files with 80 additions and 95 deletions.
34 changes: 27 additions & 7 deletions api/src/opentrons/protocol_engine/commands/liquid_probe.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,11 @@

from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Type, Union
from opentrons.protocol_engine.errors.exceptions import MustHomeError, TipNotEmptyError
from opentrons.protocol_engine.errors.exceptions import (
MustHomeError,
PipetteNotReadyToAspirateError,
TipNotEmptyError,
)
from opentrons.types import MountType
from opentrons_shared_data.errors.exceptions import (
PipetteLiquidNotFoundError,
Expand Down Expand Up @@ -32,6 +36,7 @@
if TYPE_CHECKING:
from ..execution import MovementHandler, PipettingHandler
from ..resources import ModelUtils
from ..state import StateView


LiquidProbeCommandType = Literal["liquidProbe"]
Expand Down Expand Up @@ -92,11 +97,13 @@ class LiquidProbeImplementation(

def __init__(
self,
state_view: StateView,
movement: MovementHandler,
pipetting: PipettingHandler,
model_utils: ModelUtils,
**kwargs: object,
) -> None:
self._state_view = state_view
self._movement = movement
self._pipetting = pipetting
self._model_utils = model_utils
Expand All @@ -112,20 +119,30 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn:
the pipette.
TipNotEmptyError: as an undefined error, if the tip starts with liquid
in it.
PipetteNotReadyToAspirateError: as an undefined error, if the plunger is not
in a safe position to do the liquid probe.
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:
# May raise TipNotAttachedError.
aspirated_volume = self._state_view.pipettes.get_aspirated_volume(pipette_id)

if aspirated_volume is None:
# Theoretically, we could avoid raising an error by automatically preparing
# to aspirate above the well like AspirateImplementation does. However, the
# only way for this to happen is if someone tries to do a liquid probe with
# a tip that's previously held liquid, which they should avoid anyway.
raise PipetteNotReadyToAspirateError(
"The pipette cannot probe liquid because of a previous blow out."
" The plunger must be reset while the tip is somewhere away from liquid."
)
elif aspirated_volume != 0:
raise TipNotEmptyError(
message="This operation requires a tip with no liquid in it."
message="The pipette cannot probe for liquid when the tip has liquid in it."
)

if await self._movement.check_for_valid_position(mount=MountType.LEFT) is False:
Expand Down Expand Up @@ -182,11 +199,13 @@ class TryLiquidProbeImplementation(

def __init__(
self,
state_view: StateView,
movement: MovementHandler,
pipetting: PipettingHandler,
model_utils: ModelUtils,
**kwargs: object,
) -> None:
self._state_view = state_view
self._movement = movement
self._pipetting = pipetting
self._model_utils = model_utils
Expand All @@ -203,6 +222,7 @@ async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn:
# Otherwise, we return the result or propagate the exception unchanged.

original_impl = LiquidProbeImplementation(
state_view=self._state_view,
movement=self._movement,
pipetting=self._pipetting,
model_utils=self._model_utils,
Expand Down
11 changes: 0 additions & 11 deletions api/src/opentrons/protocol_engine/execution/pipetting.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,6 @@
class PipettingHandler(TypingProtocol):
"""Liquid handling commands."""

def get_is_empty(self, pipette_id: str) -> bool:
"""Get whether a pipette has an aspirated volume equal to 0."""

def get_is_ready_to_aspirate(self, pipette_id: str) -> bool:
"""Get whether a pipette is ready to aspirate."""

Expand Down Expand Up @@ -81,10 +78,6 @@ def __init__(self, state_view: StateView, hardware_api: HardwareControlAPI) -> N
self._state_view = state_view
self._hardware_api = hardware_api

def get_is_empty(self, pipette_id: str) -> bool:
"""Get whether a pipette has an aspirated volume equal to 0."""
return self._state_view.pipettes.get_aspirated_volume(pipette_id) == 0

def get_is_ready_to_aspirate(self, pipette_id: str) -> bool:
"""Get whether a pipette is ready to aspirate."""
hw_pipette = self._state_view.pipettes.get_hardware_pipette(
Expand Down Expand Up @@ -236,10 +229,6 @@ def __init__(
"""Initialize a PipettingHandler instance."""
self._state_view = state_view

def get_is_empty(self, pipette_id: str) -> bool:
"""Get whether a pipette has an aspirated volume equal to 0."""
return self._state_view.pipettes.get_aspirated_volume(pipette_id) == 0

def get_is_ready_to_aspirate(self, pipette_id: str) -> bool:
"""Get whether a pipette is ready to aspirate."""
return self._state_view.pipettes.get_aspirated_volume(pipette_id) is not None
Expand Down
2 changes: 1 addition & 1 deletion api/src/opentrons/protocol_engine/state/pipettes.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -611,7 +611,7 @@ def get_aspirated_volume(self, pipette_id: str) -> Optional[float]:
Returns:
The volume the pipette has aspirated.
None, after blow-out and the plunger is in an unsafe position or drop-tip and there is no tip attached.
None, after blow-out and the plunger is in an unsafe position.
Raises:
PipetteNotLoadedError: pipette ID does not exist.
Expand Down
128 changes: 52 additions & 76 deletions api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from opentrons.protocol_engine.errors.exceptions import (
MustHomeError,
PipetteNotReadyToAspirateError,
TipNotAttachedError,
TipNotEmptyError,
)
Expand Down Expand Up @@ -37,7 +38,6 @@
PipettingHandler,
)
from opentrons.protocol_engine.resources.model_utils import ModelUtils
from opentrons.protocol_engine.types import LoadedPipette


EitherImplementationType = Union[
Expand Down Expand Up @@ -84,12 +84,14 @@ def result_type(types: tuple[object, object, EitherResultType]) -> EitherResultT
@pytest.fixture
def subject(
implementation_type: EitherImplementationType,
state_view: StateView,
movement: MovementHandler,
pipetting: PipettingHandler,
model_utils: ModelUtils,
) -> Union[LiquidProbeImplementation, TryLiquidProbeImplementation]:
"""Get the implementation subject."""
return implementation_type(
state_view=state_view,
pipetting=pipetting,
movement=movement,
model_utils=model_utils,
Expand All @@ -99,6 +101,7 @@ def subject(
async def test_liquid_probe_implementation_no_prep(
decoy: Decoy,
movement: MovementHandler,
state_view: StateView,
pipetting: PipettingHandler,
subject: EitherImplementation,
params_type: EitherParamsType,
Expand All @@ -114,7 +117,9 @@ async def test_liquid_probe_implementation_no_prep(
wellLocation=location,
)

decoy.when(pipetting.get_is_ready_to_aspirate(pipette_id="abc")).then_return(True)
decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id="abc")).then_return(
0
)

decoy.when(
await movement.move_to_well(
Expand Down Expand Up @@ -143,69 +148,9 @@ async def test_liquid_probe_implementation_no_prep(
)


async def test_liquid_probe_implementation_with_prep(
decoy: Decoy,
state_view: StateView,
movement: MovementHandler,
pipetting: PipettingHandler,
subject: EitherImplementation,
params_type: EitherParamsType,
result_type: EitherResultType,
) -> None:
"""A Liquid Probe should have an execution implementation with preparing to aspirate."""
location = WellLocation(origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=2))

data = params_type(
pipetteId="abc",
labwareId="123",
wellName="A3",
wellLocation=location,
)

decoy.when(pipetting.get_is_ready_to_aspirate(pipette_id="abc")).then_return(False)

decoy.when(state_view.pipettes.get(pipette_id="abc")).then_return(
LoadedPipette.construct( # type:ignore[call-arg]
mount=MountType.LEFT
)
)
decoy.when(
await movement.move_to_well(
pipette_id="abc", labware_id="123", well_name="A3", well_location=location
),
).then_return(Point(x=1, y=2, z=3))

decoy.when(
await pipetting.liquid_probe_in_place(
pipette_id="abc",
labware_id="123",
well_name="A3",
well_location=location,
),
).then_return(15.0)

result = await subject.execute(data)

assert type(result.public) is result_type # Pydantic v1 only compares the fields.
assert result == SuccessData(
public=result_type(z_position=15.0, position=DeckPoint(x=1, y=2, z=3)),
private=None,
)

decoy.verify(
await movement.move_to_well(
pipette_id="abc",
labware_id="123",
well_name="A3",
well_location=WellLocation(
origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=2)
),
),
)


async def test_liquid_not_found_error(
decoy: Decoy,
state_view: StateView,
movement: MovementHandler,
pipetting: PipettingHandler,
subject: EitherImplementation,
Expand All @@ -232,9 +177,7 @@ async def test_liquid_not_found_error(
wellLocation=well_location,
)

decoy.when(pipetting.get_is_ready_to_aspirate(pipette_id=pipette_id)).then_return(
True
)
decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(0)

decoy.when(
await movement.move_to_well(
Expand Down Expand Up @@ -282,11 +225,11 @@ async def test_liquid_not_found_error(

async def test_liquid_probe_tip_checking(
decoy: Decoy,
pipetting: PipettingHandler,
state_view: StateView,
subject: EitherImplementation,
params_type: EitherParamsType,
) -> None:
"""It should return a TipNotAttached error if the hardware API indicates that."""
"""It should raise a TipNotAttached error if the state view indicates that."""
pipette_id = "pipette-id"
labware_id = "labware-id"
well_name = "well-name"
Expand All @@ -301,18 +244,42 @@ async def test_liquid_probe_tip_checking(
wellLocation=well_location,
)

decoy.when(
pipetting.get_is_ready_to_aspirate(
pipette_id=pipette_id,
),
).then_raise(TipNotAttachedError())
decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_raise(
TipNotAttachedError()
)
with pytest.raises(TipNotAttachedError):
await subject.execute(data)


async def test_liquid_probe_plunger_preparedness_checking(
decoy: Decoy,
state_view: StateView,
subject: EitherImplementation,
params_type: EitherParamsType,
) -> None:
"""It should raise a PipetteNotReadyToAspirate error if the state view indicates that."""
pipette_id = "pipette-id"
labware_id = "labware-id"
well_name = "well-name"
well_location = WellLocation(
origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1)
)

data = params_type(
pipetteId=pipette_id,
labwareId=labware_id,
wellName=well_name,
wellLocation=well_location,
)

decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(None)
with pytest.raises(PipetteNotReadyToAspirateError):
await subject.execute(data)


async def test_liquid_probe_volume_checking(
decoy: Decoy,
pipetting: PipettingHandler,
state_view: StateView,
subject: EitherImplementation,
params_type: EitherParamsType,
) -> None:
Expand All @@ -330,15 +297,23 @@ async def test_liquid_probe_volume_checking(
wellName=well_name,
wellLocation=well_location,
)

decoy.when(
pipetting.get_is_empty(pipette_id=pipette_id),
).then_return(False)
state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id),
).then_return(123)
with pytest.raises(TipNotEmptyError):
await subject.execute(data)

decoy.when(
state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id),
).then_return(None)
with pytest.raises(PipetteNotReadyToAspirateError):
await subject.execute(data)


async def test_liquid_probe_location_checking(
decoy: Decoy,
state_view: StateView,
movement: MovementHandler,
subject: EitherImplementation,
params_type: EitherParamsType,
Expand All @@ -357,6 +332,7 @@ async def test_liquid_probe_location_checking(
wellName=well_name,
wellLocation=well_location,
)
decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(0)
decoy.when(
await movement.check_for_valid_position(
mount=MountType.LEFT,
Expand Down

0 comments on commit 65b0d7f

Please sign in to comment.