feat(api): Add prepare to aspirate to PAPI and engine (#13827)

Add the ability to explicitly call prepare_to_aspirate() to the python protocol API and the json protocol command schema. We make sure the pipette is prepared to aspirate before any aspiration, which is good, but the way we make this absolute is that if the pipette isn't ready for aspirate when you call aspirate(), we move it to the top of the current well (if necessary) to move the plunger up. This is a problem for users who are, for instance, building explicit prewetting behavior out of protocol API calls. It's common to move the pipette into the liquid and then delay to let the liquid settle before pipetting; if the pipette then moves up to prepare for aspirate before coming back down, that settling is undone. By adding the ability to explicitly prepare_for_aspirate(), the user can call it while they know the pipette is above the well. --------- Co-authored-by: Edward Cormany <[email protected]> Co-authored-by: Max Marrone <[email protected]>
Opentrons · Nov 1, 2023 · f8c53e8 · f8c53e8
1 parent 5dc8819
commit f8c53e8
Show file tree

Hide file tree

Showing 17 changed files with 340 additions and 10 deletions.
diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py
@@ -584,3 +584,6 @@ def configure_for_volume(self, volume: float) -> None:
         self._engine_client.configure_for_volume(
             pipette_id=self._pipette_id, volume=volume
         )
+
+    def prepare_to_aspirate(self) -> None:
+        self._engine_client.prepare_to_aspirate(pipette_id=self._pipette_id)
diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py
@@ -243,5 +243,9 @@ def configure_for_volume(self, volume: float) -> None:
         """
         ...
 
+    def prepare_to_aspirate(self) -> None:
+        """Prepare the pipette to aspirate."""
+        ...
+
 
 InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any])
diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
@@ -102,7 +102,7 @@ def aspirate(
                         "cause over aspiration if the previous command is a "
                         "blow_out."
                     )
-                self.prepare_for_aspirate()
+                self.prepare_to_aspirate()
             self.move_to(location=location)
         elif not in_place:
             self.move_to(location=location)
@@ -443,7 +443,7 @@ def has_tip(self) -> bool:
     def is_ready_to_aspirate(self) -> bool:
         return self.get_hardware_state()["ready_to_aspirate"]
 
-    def prepare_for_aspirate(self) -> None:
+    def prepare_to_aspirate(self) -> None:
         self._protocol_interface.get_hardware().prepare_for_aspirate(self._mount)
 
     def get_return_height(self) -> float:

diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
@@ -106,7 +106,7 @@ def aspirate(
                         "cause over aspiration if the previous command is a "
                         "blow_out."
                     )
-                self.prepare_for_aspirate()
+                self.prepare_to_aspirate()
             self.move_to(location=location, well_core=well_core)
         elif location != self._protocol_interface.get_last_location():
             self.move_to(location=location, well_core=well_core)
@@ -334,7 +334,7 @@ def has_tip(self) -> bool:
     def is_ready_to_aspirate(self) -> bool:
         return self._pipette_dict["ready_to_aspirate"]
 
-    def prepare_for_aspirate(self) -> None:
+    def prepare_to_aspirate(self) -> None:
         self._raise_if_no_tip(HardwareAction.PREPARE_ASPIRATE.name)
 
     def get_return_height(self) -> float:

diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py
@@ -1614,3 +1614,47 @@ def configure_for_volume(self, volume: float) -> None:
         if last_location and isinstance(last_location.labware, labware.Well):
             self.move_to(last_location.labware.top())
         self._core.configure_for_volume(volume)
+
+    @requires_version(2, 16)
+    def prepare_to_aspirate(self) -> None:
+        """Prepare a pipette for aspiration.
+
+        Before a pipette can aspirate into an empty tip, the plunger must be in its
+        bottom position. After dropping a tip or blowing out, the plunger will be in a
+        different position. This function moves the plunger to the bottom position,
+        regardless of its current position, to make sure that the pipette is ready to
+        aspirate.
+
+        You rarely need to call this function. The API automatically prepares the
+        pipette for aspiration as part of other commands:
+
+            - After picking up a tip with :py:meth:`.pick_up_tip`.
+            - When calling :py:meth:`.aspirate`, if the pipette isn't already prepared.
+              If the pipette is in a well, it will move out of the well, move the plunger,
+              and then move back.
+
+        Use ``prepare_to_aspirate`` when you need to control exactly when the plunger
+        motion will happen. A common use case is a pre-wetting routine, which requires
+        preparing for aspiration, moving into a well, and then aspirating *without
+        leaving the well*::
+
+             pipette.move_to(well.bottom(z=2))
+             pipette.delay(5)
+             pipette.mix(10, 10)
+             pipette.move_to(well.top(z=5))
+             pipette.blow_out()
+             pipette.prepare_to_aspirate()
+             pipette.move_to(well.bottom(z=2))
+             pipette.delay(5)
+             pipette.aspirate(10, well.bottom(z=2))
+
+        The call to ``prepare_to_aspirate()`` means that the plunger will be in the
+        bottom position before the call to ``aspirate()``. Since it doesn't need to
+        prepare again, it will not move up out of the well to move the plunger. It will
+        aspirate in place.
+        """
+        if self._core.get_current_volume():
+            raise CommandPreconditionViolated(
+                message=f"Cannot prepare {str(self)} for aspirate while it contains liquid."
+            )
+        self._core.prepare_to_aspirate()
diff --git a/api/src/opentrons/protocol_engine/clients/sync_client.py b/api/src/opentrons/protocol_engine/clients/sync_client.py
@@ -284,6 +284,14 @@ def configure_for_volume(
         result = self._transport.execute_command(request=request)
         return cast(commands.ConfigureForVolumeResult, result)
 
+    def prepare_to_aspirate(self, pipette_id: str) -> commands.PrepareToAspirateResult:
+        """Execute a PrepareToAspirate command."""
+        request = commands.PrepareToAspirateCreate(
+            params=commands.PrepareToAspirateParams(pipetteId=pipette_id)
+        )
+        result = self._transport.execute_command(request=request)
+        return cast(commands.PrepareToAspirateResult, result)
+
     def aspirate(
         self,
         pipette_id: str,

diff --git a/api/src/opentrons/protocol_engine/commands/__init__.py b/api/src/opentrons/protocol_engine/commands/__init__.py
@@ -266,6 +266,14 @@
     ConfigureForVolumeCommandType,
 )
 
+from .prepare_to_aspirate import (
+    PrepareToAspirate,
+    PrepareToAspirateCreate,
+    PrepareToAspirateParams,
+    PrepareToAspirateResult,
+    PrepareToAspirateCommandType,
+)
+
 __all__ = [
     # command type unions
     "Command",
@@ -463,4 +471,10 @@
     "ConfigureForVolumeParams",
     "ConfigureForVolumeResult",
     "ConfigureForVolumeCommandType",
+    # prepare pipette for aspirate command bundle
+    "PrepareToAspirate",
+    "PrepareToAspirateCreate",
+    "PrepareToAspirateParams",
+    "PrepareToAspirateResult",
+    "PrepareToAspirateCommandType",
 ]
diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py
@@ -235,6 +235,14 @@
     ConfigureForVolumePrivateResult,
 )
 
+from .prepare_to_aspirate import (
+    PrepareToAspirate,
+    PrepareToAspirateParams,
+    PrepareToAspirateCreate,
+    PrepareToAspirateResult,
+    PrepareToAspirateCommandType,
+)
+
 Command = Union[
     Aspirate,
     AspirateInPlace,
@@ -257,6 +265,7 @@
     MoveRelative,
     MoveToCoordinates,
     MoveToWell,
+    PrepareToAspirate,
     WaitForResume,
     WaitForDuration,
     PickUpTip,
@@ -313,6 +322,7 @@
     MoveRelativeParams,
     MoveToCoordinatesParams,
     MoveToWellParams,
+    PrepareToAspirateParams,
     WaitForResumeParams,
     WaitForDurationParams,
     PickUpTipParams,
@@ -370,6 +380,7 @@
     MoveRelativeCommandType,
     MoveToCoordinatesCommandType,
     MoveToWellCommandType,
+    PrepareToAspirateCommandType,
     WaitForResumeCommandType,
     WaitForDurationCommandType,
     PickUpTipCommandType,
@@ -426,6 +437,7 @@
     MoveRelativeCreate,
     MoveToCoordinatesCreate,
     MoveToWellCreate,
+    PrepareToAspirateCreate,
     WaitForResumeCreate,
     WaitForDurationCreate,
     PickUpTipCreate,
@@ -482,6 +494,7 @@
     MoveRelativeResult,
     MoveToCoordinatesResult,
     MoveToWellResult,
+    PrepareToAspirateResult,
     WaitForResumeResult,
     WaitForDurationResult,
     PickUpTipResult,

diff --git a/api/src/opentrons/protocol_engine/commands/prepare_to_aspirate.py b/api/src/opentrons/protocol_engine/commands/prepare_to_aspirate.py
@@ -0,0 +1,73 @@
+"""Prepare to aspirate command request, result, and implementation models."""
+
+from __future__ import annotations
+from pydantic import BaseModel
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+
+from .pipetting_common import (
+    PipetteIdMixin,
+)
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+)
+
+if TYPE_CHECKING:
+    from ..execution.pipetting import PipettingHandler
+
+PrepareToAspirateCommandType = Literal["prepareToAspirate"]
+
+
+class PrepareToAspirateParams(PipetteIdMixin):
+    """Parameters required to prepare a specific pipette for aspiration."""
+
+    pass
+
+
+class PrepareToAspirateResult(BaseModel):
+    """Result data from execution of an PrepareToAspirate command."""
+
+    pass
+
+
+class PrepareToAspirateImplementation(
+    AbstractCommandImpl[
+        PrepareToAspirateParams,
+        PrepareToAspirateResult,
+    ]
+):
+    """Prepare for aspirate command implementation."""
+
+    def __init__(self, pipetting: PipettingHandler, **kwargs: object) -> None:
+        self._pipetting_handler = pipetting
+
+    async def execute(self, params: PrepareToAspirateParams) -> PrepareToAspirateResult:
+        """Prepare the pipette to aspirate."""
+        await self._pipetting_handler.prepare_for_aspirate(
+            pipette_id=params.pipetteId,
+        )
+
+        return PrepareToAspirateResult()
+
+
+class PrepareToAspirate(BaseCommand[PrepareToAspirateParams, PrepareToAspirateResult]):
+    """Prepare for aspirate command model."""
+
+    commandType: PrepareToAspirateCommandType = "prepareToAspirate"
+    params: PrepareToAspirateParams
+    result: Optional[PrepareToAspirateResult]
+
+    _ImplementationCls: Type[
+        PrepareToAspirateImplementation
+    ] = PrepareToAspirateImplementation
+
+
+class PrepareToAspirateCreate(BaseCommandCreate[PrepareToAspirateParams]):
+    """Prepare for aspirate command creation request model."""
+
+    commandType: PrepareToAspirateCommandType = "prepareToAspirate"
+    params: PrepareToAspirateParams
+
+    _CommandCls: Type[PrepareToAspirate] = PrepareToAspirate
diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py
@@ -37,6 +37,7 @@
     thermocycler,
     heater_shaker,
     CommandPrivateResult,
+    PrepareToAspirateResult,
 )
 from ..commands.configuring_common import PipetteConfigUpdateResultMixin
 from ..actions import (
@@ -221,6 +222,10 @@ def _handle_command(  # noqa: C901
             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_well(self, command: Command) -> None:
         # These commands leave the pipette in a new well.
         # Update current_well to reflect that.

diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py
@@ -28,6 +28,10 @@
 )
 from opentrons.types import Location, Mount, Point
 
+from opentrons_shared_data.errors.exceptions import (
+    CommandPreconditionViolated,
+)
+
 
 @pytest.fixture(autouse=True)
 def _mock_validation_module(decoy: Decoy, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -912,3 +916,21 @@ def test_plunger_speed_removed(subject: InstrumentContext) -> None:
     """It should raise an error on PAPI >= v2.14."""
     with pytest.raises(APIVersionError):
         subject.speed
+
+
+def test_prepare_to_aspirate(
+    subject: InstrumentContext, decoy: Decoy, mock_instrument_core: InstrumentCore
+) -> None:
+    """It should call the core function."""
+    decoy.when(mock_instrument_core.get_current_volume()).then_return(0)
+    subject.prepare_to_aspirate()
+    decoy.verify(mock_instrument_core.prepare_to_aspirate(), times=1)
+
+
+def test_prepare_to_aspirate_checks_volume(
+    subject: InstrumentContext, decoy: Decoy, mock_instrument_core: InstrumentCore
+) -> None:
+    """It should raise an error if you prepare for aspirate with liquid in the pipette."""
+    decoy.when(mock_instrument_core.get_current_volume()).then_return(10)
+    with pytest.raises(CommandPreconditionViolated):
+        subject.prepare_to_aspirate()
diff --git a/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py b/api/tests/opentrons/protocol_api_old/core/simulator/test_instrument_context.py
@@ -43,7 +43,7 @@ def test_prepare_to_aspirate_no_tip(subject: InstrumentCore) -> None:
     with pytest.raises(
         UnexpectedTipRemovalError, match="Cannot perform PREPARE_ASPIRATE"
     ):
-        subject.prepare_for_aspirate()  # type: ignore[attr-defined]
+        subject.prepare_to_aspirate()
 
 
 def test_dispense_no_tip(subject: InstrumentCore) -> None:
@@ -161,7 +161,7 @@ def test_aspirate_too_much(
         increment=None,
         prep_after=False,
     )
-    subject.prepare_for_aspirate()  # type: ignore[attr-defined]
+    subject.prepare_to_aspirate()
     with pytest.raises(
         AssertionError, match="Cannot aspirate more than pipette max volume"
     ):
@@ -215,7 +215,7 @@ def test_pipette_dict(
 
 def _aspirate(i: InstrumentCore, labware: LabwareCore) -> None:
     """pipette dict with tip fixture."""
-    i.prepare_for_aspirate()  # type: ignore[attr-defined]
+    i.prepare_to_aspirate()
     i.aspirate(
         location=Location(point=Point(1, 2, 3), labware=None),
         well_core=labware.get_well_core("A1"),
@@ -228,7 +228,7 @@ def _aspirate(i: InstrumentCore, labware: LabwareCore) -> None:
 
 def _aspirate_dispense(i: InstrumentCore, labware: LabwareCore) -> None:
     """pipette dict with tip fixture."""
-    i.prepare_for_aspirate()  # type: ignore[attr-defined]
+    i.prepare_to_aspirate()
     i.aspirate(
         location=Location(point=Point(1, 2, 3), labware=None),
         well_core=labware.get_well_core("A1"),
@@ -250,7 +250,7 @@ def _aspirate_dispense(i: InstrumentCore, labware: LabwareCore) -> None:
 
 def _aspirate_blowout(i: InstrumentCore, labware: LabwareCore) -> None:
     """pipette dict with tip fixture."""
-    i.prepare_for_aspirate()  # type: ignore[attr-defined]
+    i.prepare_to_aspirate()
     i.aspirate(
         location=Location(point=Point(1, 2, 3), labware=None),
         well_core=labware.get_well_core("A1"),

diff --git a/api/tests/opentrons/protocol_engine/commands/test_prepare_to_aspirate.py b/api/tests/opentrons/protocol_engine/commands/test_prepare_to_aspirate.py
@@ -0,0 +1,29 @@
+"""Test prepare to aspirate commands."""
+
+from decoy import Decoy
+
+from opentrons.protocol_engine.execution import (
+    PipettingHandler,
+)
+
+from opentrons.protocol_engine.commands.prepare_to_aspirate import (
+    PrepareToAspirateParams,
+    PrepareToAspirateImplementation,
+    PrepareToAspirateResult,
+)
+
+
+async def test_prepare_to_aspirate_implmenetation(
+    decoy: Decoy, pipetting: PipettingHandler
+) -> None:
+    """A PrepareToAspirate command should have an executing implementation."""
+    subject = PrepareToAspirateImplementation(pipetting=pipetting)
+
+    data = PrepareToAspirateParams(pipetteId="some id")
+
+    decoy.when(await pipetting.prepare_for_aspirate(pipette_id="some id")).then_return(
+        None
+    )
+
+    result = await subject.execute(data)
+    assert isinstance(result, PrepareToAspirateResult)