Merge branch 'edge' into opentrons-ai-client-modules

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -104,6 +104,19 @@ def set_default_speed(self, speed: float) -> None:
             pipette_id=self._pipette_id, speed=speed
         )
 
+    def air_gap_in_place(self, volume: float, flow_rate: float) -> None:
+        """Aspirate a given volume of air from the current location of the pipette.
+
+        Args:
+            volume: The volume of air to aspirate, in microliters.
+            folw_rate: The flow rate of air into the pipette, in microliters/s
+        """
+        self._engine_client.execute_command(
+            cmd.AirGapInPlaceParams(
+                pipetteId=self._pipette_id, volume=volume, flowRate=flow_rate
+            )
+        )
+
     def aspirate(
         self,
         location: Location,

api/src/opentrons/protocol_api/core/instrument.py

@@ -24,6 +24,14 @@ def get_default_speed(self) -> float:
     def set_default_speed(self, speed: float) -> None:
         ...
 
+    @abstractmethod
+    def air_gap_in_place(self, volume: float, flow_rate: float) -> None:
+        """Aspirate a given volume of air from the current location of the pipette.
+        Args:
+            volume: The volume of air to aspirate, in microliters.
+            flow_rate: The flow rate of air into the pipette, in microliters.
+        """
+
     @abstractmethod
     def aspirate(
         self,

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -72,6 +72,9 @@ def set_default_speed(self, speed: float) -> None:
         """Sets the speed at which the robot's gantry moves."""
         self._default_speed = speed
 
+    def air_gap_in_place(self, volume: float, flow_rate: float) -> None:
+        assert False, "Air gap tracking only available in API version 2.22 and later"
+
     def aspirate(
         self,
         location: types.Location,

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -83,6 +83,9 @@ def get_default_speed(self) -> float:
     def set_default_speed(self, speed: float) -> None:
         self._default_speed = speed
 
+    def air_gap_in_place(self, volume: float, flow_rate: float) -> None:
+        assert False, "Air gap tracking only available in API version 2.22 and later"
+
     def aspirate(
         self,
         location: types.Location,

api/src/opentrons/protocol_api/instrument_context.py

@@ -60,6 +60,8 @@
 """The version after which offsets for deck configured trash containers and changes to alternating tip drop behavior were introduced."""
 _PARTIAL_NOZZLE_CONFIGURATION_SINGLE_ROW_PARTIAL_COLUMN_ADDED_IN = APIVersion(2, 20)
 """The version after which partial nozzle configurations of single, row, and partial column layouts became available."""
+_AIR_GAP_TRACKING_ADDED_IN = APIVersion(2, 22)
+"""The version after which air gaps should be implemented with a separate call instead of an aspirate for better liquid volume tracking."""
 
 
 class InstrumentContext(publisher.CommandPublisher):
@@ -753,7 +755,12 @@ def air_gap(
             ``pipette.air_gap(height=2)``. If you call ``air_gap`` with a single,
             unnamed argument, it will always be interpreted as a volume.
 
+        .. note::
 
+           Before API version 2.22, this function was implemented as an aspirate, and
+           dispensing into a well would add the air gap volume to the liquid tracked in
+           the well. At or above API version 2.22, air gap volume is not counted as liquid
+           when dispensing into a well.
         """
         if not self._core.has_tip():
             raise UnexpectedTipRemovalError("air_gap", self.name, self.mount)
@@ -765,7 +772,12 @@ def air_gap(
             raise RuntimeError("No previous Well cached to perform air gap")
         target = loc.labware.as_well().top(height)
         self.move_to(target, publish=False)
-        self.aspirate(volume)
+        if self.api_version >= _AIR_GAP_TRACKING_ADDED_IN:
+            c_vol = self._core.get_available_volume() if volume is None else volume
+            flow_rate = self._core.get_aspirate_flow_rate()
+            self._core.air_gap_in_place(c_vol, flow_rate)
+        else:
+            self.aspirate(volume)
         return self
 
     @publisher.publish(command=cmds.return_tip)

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

@@ -41,6 +41,14 @@
     CommandDefinedErrorData,
 )
 
+from .air_gap_in_place import (
+    AirGapInPlace,
+    AirGapInPlaceParams,
+    AirGapInPlaceCreate,
+    AirGapInPlaceResult,
+    AirGapInPlaceCommandType,
+)
+
 from .aspirate import (
     Aspirate,
     AspirateParams,
@@ -355,6 +363,12 @@
     "hash_protocol_command_params",
     # command schema generation
     "generate_command_schema",
+    # air gap command models
+    "AirGapInPlace",
+    "AirGapInPlaceCreate",
+    "AirGapInPlaceParams",
+    "AirGapInPlaceResult",
+    "AirGapInPlaceCommandType",
     # aspirate command models
     "Aspirate",
     "AspirateCreate",

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

@@ -0,0 +1,160 @@
+"""AirGap in place command request, result, and implementation models."""
+
+from __future__ import annotations
+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,
+)
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    SuccessData,
+    DefinedErrorData,
+)
+from ..errors.error_occurrence import ErrorOccurrence
+from ..errors.exceptions import PipetteNotReadyToAspirateError
+from ..state.update_types import StateUpdate
+from ..types import AspiratedFluid, FluidKind
+
+if TYPE_CHECKING:
+    from ..execution import PipettingHandler, GantryMover
+    from ..resources import ModelUtils
+    from ..state.state import StateView
+    from ..notes import CommandNoteAdder
+
+AirGapInPlaceCommandType = Literal["airGapInPlace"]
+
+
+class AirGapInPlaceParams(PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin):
+    """Payload required to air gap in place."""
+
+    pass
+
+
+class AirGapInPlaceResult(BaseLiquidHandlingResult):
+    """Result data from the execution of a AirGapInPlace command."""
+
+    pass
+
+
+_ExecuteReturn = Union[
+    SuccessData[AirGapInPlaceResult],
+    DefinedErrorData[OverpressureError],
+]
+
+
+class AirGapInPlaceImplementation(
+    AbstractCommandImpl[AirGapInPlaceParams, _ExecuteReturn]
+):
+    """AirGapInPlace command implementation."""
+
+    def __init__(
+        self,
+        pipetting: PipettingHandler,
+        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: AirGapInPlaceParams) -> _ExecuteReturn:
+        """Air gap without moving the pipette.
+
+        Raises:
+            TipNotAttachedError: if no tip is attached to the pipette.
+            PipetteNotReadyToAirGapError: pipette plunger is not ready.
+        """
+        ready_to_aspirate = self._pipetting.get_is_ready_to_aspirate(
+            pipette_id=params.pipetteId,
+        )
+
+        if not ready_to_aspirate:
+            raise PipetteNotReadyToAspirateError(
+                "Pipette cannot air gap in place because of a previous blow out."
+                " The first aspirate following a blow-out must be from a specific well"
+                " so the plunger can be reset in a known safe position."
+            )
+
+        state_update = StateUpdate()
+
+        try:
+            current_position = await self._gantry_mover.get_position(params.pipetteId)
+            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:
+            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,
+                            )
+                        }
+                    ),
+                ),
+                state_update=state_update,
+            )
+        else:
+            state_update.set_fluid_aspirated(
+                pipette_id=params.pipetteId,
+                fluid=AspiratedFluid(kind=FluidKind.AIR, volume=volume),
+            )
+            return SuccessData(
+                public=AirGapInPlaceResult(volume=volume),
+                state_update=state_update,
+            )
+
+
+class AirGapInPlace(
+    BaseCommand[AirGapInPlaceParams, AirGapInPlaceResult, OverpressureError]
+):
+    """AirGapInPlace command model."""
+
+    commandType: AirGapInPlaceCommandType = "airGapInPlace"
+    params: AirGapInPlaceParams
+    result: Optional[AirGapInPlaceResult]
+
+    _ImplementationCls: Type[AirGapInPlaceImplementation] = AirGapInPlaceImplementation
+
+
+class AirGapInPlaceCreate(BaseCommandCreate[AirGapInPlaceParams]):
+    """AirGapInPlace command request model."""
+
+    commandType: AirGapInPlaceCommandType = "airGapInPlace"
+    params: AirGapInPlaceParams
+
+    _CommandCls: Type[AirGapInPlace] = AirGapInPlace

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

@@ -25,7 +25,14 @@
 from opentrons.hardware_control import HardwareControlAPI
 
 from ..state.update_types import StateUpdate, CLEAR
-from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint
+from ..types import (
+    WellLocation,
+    WellOrigin,
+    CurrentWell,
+    DeckPoint,
+    AspiratedFluid,
+    FluidKind,
+)
 
 if TYPE_CHECKING:
     from ..execution import MovementHandler, PipettingHandler
@@ -141,6 +148,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
                 volume_added=CLEAR,
             )
+            state_update.set_fluid_unknown(pipette_id=params.pipetteId)
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -162,6 +170,10 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 well_name=well_name,
                 volume_added=-volume_aspirated,
             )
+            state_update.set_fluid_aspirated(
+                pipette_id=params.pipetteId,
+                fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume_aspirated),
+            )
             return SuccessData(
                 public=AspirateResult(
                     volume=volume_aspirated,

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

@@ -25,7 +25,7 @@
 from ..errors.error_occurrence import ErrorOccurrence
 from ..errors.exceptions import PipetteNotReadyToAspirateError
 from ..state.update_types import StateUpdate, CLEAR
-from ..types import CurrentWell
+from ..types import CurrentWell, AspiratedFluid, FluidKind
 
 if TYPE_CHECKING:
     from ..execution import PipettingHandler, GantryMover
@@ -115,6 +115,7 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                     well_name=current_location.well_name,
                     volume_added=CLEAR,
                 )
+            state_update.set_fluid_unknown(pipette_id=params.pipetteId)
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -139,6 +140,10 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                 state_update=state_update,
             )
         else:
+            state_update.set_fluid_aspirated(
+                pipette_id=params.pipetteId,
+                fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume),
+            )
             if (
                 isinstance(current_location, CurrentWell)
                 and current_location.pipette_id == params.pipetteId
@@ -148,6 +153,7 @@ async def execute(self, params: AspirateInPlaceParams) -> _ExecuteReturn:
                     well_name=current_location.well_name,
                     volume_added=-volume,
                 )
+
             return SuccessData(
                 public=AspirateInPlaceResult(volume=volume),
                 state_update=state_update,

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

@@ -93,6 +93,7 @@ async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
                 pipette_id=params.pipetteId, flow_rate=params.flowRate
             )
         except PipetteOverpressureError as e:
+            state_update.set_fluid_unknown(pipette_id=params.pipetteId)
             return DefinedErrorData(
                 public=OverpressureError(
                     id=self._model_utils.generate_id(),
@@ -112,8 +113,10 @@ async def execute(self, params: BlowOutParams) -> _ExecuteReturn:
                         )
                     },
                 ),
+                state_update=state_update,
             )
         else:
+            state_update.set_fluid_empty(pipette_id=params.pipetteId)
             return SuccessData(
                 public=BlowOutResult(position=deck_point),
                 state_update=state_update,