From 1e5856f4de4ef2d27b9aba2b02fa4c21e8a41d99 Mon Sep 17 00:00:00 2001 From: Seth Foster Date: Mon, 28 Oct 2024 10:18:11 -0400 Subject: [PATCH] feat(api): add air gap PE command Up until now, we've implemented air gap with an aspirate command. We can't do that anymore because we need to be able to know when to remove liquid from the target well and when not to - and when you're air gapping, you're not removing anything from the well. Closes EXEC-792 --- .../protocol_engine/commands/air_gap.py | 180 ++++++++++++++++++ .../commands/air_gap_in_place.py | 158 +++++++++++++++ .../commands/command_unions.py | 26 +++ shared-data/command/schemas/10.json | 141 +++++++++++++- 4 files changed, 503 insertions(+), 2 deletions(-) create mode 100644 api/src/opentrons/protocol_engine/commands/air_gap.py create mode 100644 api/src/opentrons/protocol_engine/commands/air_gap_in_place.py diff --git a/api/src/opentrons/protocol_engine/commands/air_gap.py b/api/src/opentrons/protocol_engine/commands/air_gap.py new file mode 100644 index 000000000000..2aa27022b226 --- /dev/null +++ b/api/src/opentrons/protocol_engine/commands/air_gap.py @@ -0,0 +1,180 @@ +"""AirGap command request, result, and implementation models.""" +from __future__ import annotations +from typing import TYPE_CHECKING, Optional, Type, Union +from opentrons_shared_data.errors.exceptions import PipetteOverpressureError +from typing_extensions import Literal + +from .pipetting_common import ( + OverpressureError, + PipetteIdMixin, + AspirateVolumeMixin, + FlowRateMixin, + LiquidHandlingWellLocationMixin, + BaseLiquidHandlingResult, + DestinationPositionResult, +) +from .command import ( + AbstractCommandImpl, + BaseCommand, + BaseCommandCreate, + DefinedErrorData, + SuccessData, +) +from ..errors.error_occurrence import ErrorOccurrence + +from opentrons.hardware_control import HardwareControlAPI + +from ..state.update_types import StateUpdate +from ..types import WellLocation, WellOrigin, CurrentWell, DeckPoint + +if TYPE_CHECKING: + from ..execution import MovementHandler, PipettingHandler + from ..resources import ModelUtils + from ..state.state import StateView + from ..notes import CommandNoteAdder + + +AirGapCommandType = Literal["airGap"] + + +class AirGapParams( + PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin +): + """Parameters required to aspirate from a specific well.""" + + pass + + +class AirGapResult(BaseLiquidHandlingResult, DestinationPositionResult): + """Result data from execution of an AirGap command.""" + + pass + + +_ExecuteReturn = Union[ + SuccessData[AirGapResult, None], + DefinedErrorData[OverpressureError], +] + + +class AirGapImplementation(AbstractCommandImpl[AirGapParams, _ExecuteReturn]): + """AirGap command implementation.""" + + def __init__( + self, + pipetting: PipettingHandler, + state_view: StateView, + hardware_api: HardwareControlAPI, + movement: MovementHandler, + command_note_adder: CommandNoteAdder, + model_utils: ModelUtils, + **kwargs: object, + ) -> None: + self._pipetting = pipetting + self._state_view = state_view + self._hardware_api = hardware_api + self._movement = movement + self._command_note_adder = command_note_adder + self._model_utils = model_utils + + async def execute(self, params: AirGapParams) -> _ExecuteReturn: + """Move to and aspirate from the requested well. + + Raises: + TipNotAttachedError: if no tip is attached to the pipette. + """ + pipette_id = params.pipetteId + labware_id = params.labwareId + well_name = params.wellName + + ready_to_aspirate = self._pipetting.get_is_ready_to_aspirate( + pipette_id=pipette_id + ) + + current_well = None + state_update = StateUpdate() + + if not ready_to_aspirate: + await self._movement.move_to_well( + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + well_location=WellLocation(origin=WellOrigin.TOP), + ) + + await self._pipetting.prepare_for_aspirate(pipette_id=pipette_id) + + # set our current deck location to the well now that we've made + # an intermediate move for the "prepare for aspirate" step + current_well = CurrentWell( + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + ) + + position = await self._movement.move_to_well( + pipette_id=pipette_id, + labware_id=labware_id, + well_name=well_name, + well_location=well_location, + current_well=current_well, + ) + deck_point = DeckPoint.construct(x=position.x, y=position.y, z=position.z) + state_update.set_pipette_location( + pipette_id=pipette_id, + new_labware_id=labware_id, + new_well_name=well_name, + new_deck_point=deck_point, + ) + + try: + volume_aspirated = await self._pipetting.aspirate_in_place( + pipette_id=pipette_id, + 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": (position.x, position.y, position.z)}, + ), + state_update=state_update, + ) + else: + return SuccessData( + public=AirGapResult( + volume=volume_aspirated, + position=deck_point, + ), + private=None, + state_update=state_update, + ) + + +class AirGap(BaseCommand[AirGapParams, AirGapResult, OverpressureError]): + """AirGap command model.""" + + commandType: AirGapCommandType = "airGap" + params: AirGapParams + result: Optional[AirGapResult] + + _ImplementationCls: Type[AirGapImplementation] = AirGapImplementation + + +class AirGapCreate(BaseCommandCreate[AirGapParams]): + """Create aspirate command request model.""" + + commandType: AirGapCommandType = "airGap" + params: AirGapParams + + _CommandCls: Type[AirGap] = AirGap diff --git a/api/src/opentrons/protocol_engine/commands/air_gap_in_place.py b/api/src/opentrons/protocol_engine/commands/air_gap_in_place.py new file mode 100644 index 000000000000..509587acc120 --- /dev/null +++ b/api/src/opentrons/protocol_engine/commands/air_gap_in_place.py @@ -0,0 +1,158 @@ +"""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 CurrentWell + +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 aspirate in place.""" + + pass + + +class AirGapInPlaceResult(BaseLiquidHandlingResult): + """Result data from the execution of a AirGapInPlace command.""" + + pass + + +_ExecuteReturn = Union[ + SuccessData[AirGapInPlaceResult, None], + 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: + """AirGap 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() + current_location = self._state_view.pipettes.get_current_location() + + 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: + return SuccessData( + public=AirGapInPlaceResult(volume=volume), + private=None, + 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 diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py index 7623cc09f686..c5501e8764d5 100644 --- a/api/src/opentrons/protocol_engine/commands/command_unions.py +++ b/api/src/opentrons/protocol_engine/commands/command_unions.py @@ -31,6 +31,22 @@ SetRailLightsResult, ) +from .air_gap import ( + AirGap, + AirGapParams, + AirGapCreate, + AirGapResult, + AirGapCommandType, +) + +from .air_gap_in_place import ( + AirGapInPlace, + AirGapInPlaceParams, + AirGapInPlaceCreate, + AirGapInPlaceResult, + AirGapInPlaceCommandType, +) + from .aspirate import ( Aspirate, AspirateParams, @@ -322,6 +338,8 @@ Command = Annotated[ Union[ + AirGap, + AirGapInPlace, Aspirate, AspirateInPlace, Comment, @@ -399,6 +417,8 @@ ] CommandParams = Union[ + AirGapParams, + AirGapInPlaceParams, AspirateParams, AspirateInPlaceParams, CommentParams, @@ -474,6 +494,8 @@ ] CommandType = Union[ + AirGapCommandType, + AirGapInPlaceCommandType, AspirateCommandType, AspirateInPlaceCommandType, CommentCommandType, @@ -550,6 +572,8 @@ CommandCreate = Annotated[ Union[ + AirGapCreate, + AirGapInPlaceCreate, AspirateCreate, AspirateInPlaceCreate, CommentCreate, @@ -627,6 +651,8 @@ ] CommandResult = Union[ + AirGapResult, + AirGapInPlaceResult, AspirateResult, AspirateInPlaceResult, CommentResult, diff --git a/shared-data/command/schemas/10.json b/shared-data/command/schemas/10.json index be8e870c5bb1..a0aa4c2d5c1e 100644 --- a/shared-data/command/schemas/10.json +++ b/shared-data/command/schemas/10.json @@ -4,6 +4,8 @@ "discriminator": { "propertyName": "commandType", "mapping": { + "airGap": "#/definitions/AirGapCreate", + "airGapInPlace": "#/definitions/AirGapInPlaceCreate", "aspirate": "#/definitions/AspirateCreate", "aspirateInPlace": "#/definitions/AspirateInPlaceCreate", "comment": "#/definitions/CommentCreate", @@ -80,6 +82,12 @@ } }, "oneOf": [ + { + "$ref": "#/definitions/AirGapCreate" + }, + { + "$ref": "#/definitions/AirGapInPlaceCreate" + }, { "$ref": "#/definitions/AspirateCreate" }, @@ -358,8 +366,8 @@ } } }, - "AspirateParams": { - "title": "AspirateParams", + "AirGapParams": { + "title": "AirGapParams", "description": "Parameters required to aspirate from a specific well.", "type": "object", "properties": { @@ -408,6 +416,135 @@ "enum": ["protocol", "setup", "fixit"], "type": "string" }, + "AirGapCreate": { + "title": "AirGapCreate", + "description": "Create aspirate command request model.", + "type": "object", + "properties": { + "commandType": { + "title": "Commandtype", + "default": "airGap", + "enum": ["airGap"], + "type": "string" + }, + "params": { + "$ref": "#/definitions/AirGapParams" + }, + "intent": { + "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.", + "allOf": [ + { + "$ref": "#/definitions/CommandIntent" + } + ] + }, + "key": { + "title": "Key", + "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.", + "type": "string" + } + }, + "required": ["params"] + }, + "AirGapInPlaceParams": { + "title": "AirGapInPlaceParams", + "description": "Payload required to aspirate in place.", + "type": "object", + "properties": { + "flowRate": { + "title": "Flowrate", + "description": "Speed in \u00b5L/s configured for the pipette", + "exclusiveMinimum": 0, + "type": "number" + }, + "volume": { + "title": "Volume", + "description": "The amount of liquid to aspirate, in \u00b5L. Must not be greater than the remaining available amount, which depends on the pipette (see `loadPipette`), its configuration (see `configureForVolume`), the tip (see `pickUpTip`), and the amount you've aspirated so far. There is some tolerance for floating point rounding errors.", + "minimum": 0, + "type": "number" + }, + "pipetteId": { + "title": "Pipetteid", + "description": "Identifier of pipette to use for liquid handling.", + "type": "string" + } + }, + "required": ["flowRate", "volume", "pipetteId"] + }, + "AirGapInPlaceCreate": { + "title": "AirGapInPlaceCreate", + "description": "AirGapInPlace command request model.", + "type": "object", + "properties": { + "commandType": { + "title": "Commandtype", + "default": "airGapInPlace", + "enum": ["airGapInPlace"], + "type": "string" + }, + "params": { + "$ref": "#/definitions/AirGapInPlaceParams" + }, + "intent": { + "description": "The reason the command was added. If not specified or `protocol`, the command will be treated as part of the protocol run itself, and added to the end of the existing command queue.\n\nIf `setup`, the command will be treated as part of run setup. A setup command may only be enqueued if the run has not started.\n\nUse setup commands for activities like pre-run calibration checks and module setup, like pre-heating.", + "allOf": [ + { + "$ref": "#/definitions/CommandIntent" + } + ] + }, + "key": { + "title": "Key", + "description": "A key value, unique in this run, that can be used to track the same logical command across multiple runs of the same protocol. If a value is not provided, one will be generated.", + "type": "string" + } + }, + "required": ["params"] + }, + "AspirateParams": { + "title": "AspirateParams", + "description": "Parameters required to aspirate from a specific well.", + "type": "object", + "properties": { + "labwareId": { + "title": "Labwareid", + "description": "Identifier of labware to use.", + "type": "string" + }, + "wellName": { + "title": "Wellname", + "description": "Name of well to use in labware.", + "type": "string" + }, + "wellLocation": { + "title": "Welllocation", + "description": "Relative well location at which to perform the operation", + "allOf": [ + { + "$ref": "#/definitions/LiquidHandlingWellLocation" + } + ] + }, + "flowRate": { + "title": "Flowrate", + "description": "Speed in \u00b5L/s configured for the pipette", + "exclusiveMinimum": 0, + "type": "number" + }, + "volume": { + "title": "Volume", + "description": "The amount of liquid to aspirate, in \u00b5L. Must not be greater than the remaining available amount, which depends on the pipette (see `loadPipette`), its configuration (see `configureForVolume`), the tip (see `pickUpTip`), and the amount you've aspirated so far. There is some tolerance for floating point rounding errors.", + "minimum": 0, + "type": "number" + }, + "pipetteId": { + "title": "Pipetteid", + "description": "Identifier of pipette to use for liquid handling.", + "type": "string" + } + }, + "required": ["labwareId", "wellName", "flowRate", "volume", "pipetteId"] + }, "AspirateCreate": { "title": "AspirateCreate", "description": "Create aspirate command request model.",