diff --git a/api/src/opentrons/protocol_engine/commands/__init__.py b/api/src/opentrons/protocol_engine/commands/__init__.py index 09ad591277e..75904ab00a3 100644 --- a/api/src/opentrons/protocol_engine/commands/__init__.py +++ b/api/src/opentrons/protocol_engine/commands/__init__.py @@ -331,6 +331,11 @@ LiquidProbeCreate, LiquidProbeResult, LiquidProbeCommandType, + TryLiquidProbe, + TryLiquidProbeParams, + TryLiquidProbeCreate, + TryLiquidProbeResult, + TryLiquidProbeCommandType, ) __all__ = [ @@ -580,4 +585,9 @@ "LiquidProbeCreate", "LiquidProbeResult", "LiquidProbeCommandType", + "TryLiquidProbe", + "TryLiquidProbeParams", + "TryLiquidProbeCreate", + "TryLiquidProbeResult", + "TryLiquidProbeCommandType", ] diff --git a/api/src/opentrons/protocol_engine/commands/command_unions.py b/api/src/opentrons/protocol_engine/commands/command_unions.py index 68e59d5e3c5..d20b64f363b 100644 --- a/api/src/opentrons/protocol_engine/commands/command_unions.py +++ b/api/src/opentrons/protocol_engine/commands/command_unions.py @@ -1,6 +1,7 @@ """Union types of concrete command definitions.""" -from typing import Annotated, Iterable, Type, Union, get_type_hints +from collections.abc import Collection +from typing import Annotated, Type, Union, get_type_hints from pydantic import Field @@ -313,6 +314,11 @@ LiquidProbeCreate, LiquidProbeResult, LiquidProbeCommandType, + TryLiquidProbe, + TryLiquidProbeParams, + TryLiquidProbeCreate, + TryLiquidProbeResult, + TryLiquidProbeCommandType, ) Command = Annotated[ @@ -353,6 +359,7 @@ VerifyTipPresence, GetTipPresence, LiquidProbe, + TryLiquidProbe, heater_shaker.WaitForTemperature, heater_shaker.SetTargetTemperature, heater_shaker.DeactivateHeater, @@ -421,6 +428,7 @@ VerifyTipPresenceParams, GetTipPresenceParams, LiquidProbeParams, + TryLiquidProbeParams, heater_shaker.WaitForTemperatureParams, heater_shaker.SetTargetTemperatureParams, heater_shaker.DeactivateHeaterParams, @@ -487,6 +495,7 @@ VerifyTipPresenceCommandType, GetTipPresenceCommandType, LiquidProbeCommandType, + TryLiquidProbeCommandType, heater_shaker.WaitForTemperatureCommandType, heater_shaker.SetTargetTemperatureCommandType, heater_shaker.DeactivateHeaterCommandType, @@ -554,6 +563,7 @@ VerifyTipPresenceCreate, GetTipPresenceCreate, LiquidProbeCreate, + TryLiquidProbeCreate, heater_shaker.WaitForTemperatureCreate, heater_shaker.SetTargetTemperatureCreate, heater_shaker.DeactivateHeaterCreate, @@ -622,6 +632,7 @@ VerifyTipPresenceResult, GetTipPresenceResult, LiquidProbeResult, + TryLiquidProbeResult, heater_shaker.WaitForTemperatureResult, heater_shaker.SetTargetTemperatureResult, heater_shaker.DeactivateHeaterResult, @@ -671,12 +682,20 @@ def _map_create_types_by_params_type( - create_types: Iterable[Type[CommandCreate]], + create_types: Collection[Type[CommandCreate]], ) -> dict[Type[CommandParams], Type[CommandCreate]]: def get_params_type(create_type: Type[CommandCreate]) -> Type[CommandParams]: return get_type_hints(create_type)["params"] # type: ignore[no-any-return] - return {get_params_type(create_type): create_type for create_type in create_types} + result = {get_params_type(create_type): create_type for create_type in create_types} + + # This isn't an inherent requirement of opentrons.protocol_engine, + # but this mapping is only useful to higher-level code if this holds true. + assert len(result) == len( + create_types + ), "Param models should map to create models 1:1." + + return result CREATE_TYPES_BY_PARAMS_TYPE = _map_create_types_by_params_type( diff --git a/api/src/opentrons/protocol_engine/commands/liquid_probe.py b/api/src/opentrons/protocol_engine/commands/liquid_probe.py index abac5537e73..1c24a8b2d13 100644 --- a/api/src/opentrons/protocol_engine/commands/liquid_probe.py +++ b/api/src/opentrons/protocol_engine/commands/liquid_probe.py @@ -1,4 +1,5 @@ -"""Liquid-probe command for OT3 hardware. request, result, and implementation models.""" +"""The liquidProbe and tryLiquidProbe commands.""" + from __future__ import annotations from typing import TYPE_CHECKING, Optional, Type, Union from opentrons.protocol_engine.errors.exceptions import MustHomeError, TipNotEmptyError @@ -34,16 +35,30 @@ LiquidProbeCommandType = Literal["liquidProbe"] +TryLiquidProbeCommandType = Literal["tryLiquidProbe"] + +# Both command variants should have identical parameters. +# But we need two separate parameter model classes because +# `command_unions.CREATE_TYPES_BY_PARAMS_TYPE` needs to be a 1:1 mapping. +class _CommonParams(PipetteIdMixin, WellLocationMixin): + pass -class LiquidProbeParams(PipetteIdMixin, WellLocationMixin): - """Parameters required to liquid probe a specific well.""" + +class LiquidProbeParams(_CommonParams): + """Parameters required for a `liquidProbe` command.""" + + pass + + +class TryLiquidProbeParams(_CommonParams): + """Parameters required for a `tryLiquidProbe` command.""" pass class LiquidProbeResult(DestinationPositionResult): - """Result data from the execution of a liquid-probe command.""" + """Result data from the execution of a `liquidProbe` command.""" z_position: float = Field( ..., description="The Z coordinate, in mm, of the found liquid in deck space." @@ -51,13 +66,28 @@ class LiquidProbeResult(DestinationPositionResult): # New fields should use camelCase. z_position is snake_case for historical reasons. -_ExecuteReturn = Union[ +class TryLiquidProbeResult(DestinationPositionResult): + """Result data from the execution of a `tryLiquidProbe` command.""" + + z_position: Optional[float] = Field( + ..., + description=( + "The Z coordinate, in mm, of the found liquid in deck space." + " If no liquid was found, `null` or omitted." + ), + ) + + +_LiquidProbeExecuteReturn = Union[ SuccessData[LiquidProbeResult, None], DefinedErrorData[LiquidNotFoundError, LiquidNotFoundErrorInternalData], ] +_TryLiquidProbeExecuteReturn = SuccessData[TryLiquidProbeResult, None] -class LiquidProbeImplementation(AbstractCommandImpl[LiquidProbeParams, _ExecuteReturn]): +class LiquidProbeImplementation( + AbstractCommandImpl[LiquidProbeParams, _LiquidProbeExecuteReturn] +): """The implementation of a `liquidProbe` command.""" def __init__( @@ -71,16 +101,19 @@ def __init__( self._pipetting = pipetting self._model_utils = model_utils - async def execute(self, params: LiquidProbeParams) -> _ExecuteReturn: + async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn: """Move to and liquid probe the requested well. Return the z-position of the found liquid. + If no liquid is found, return a LiquidNotFoundError as a defined error. Raises: - TipNotAttachedError: if there is no tip attached to the pipette - MustHomeError: if the plunger is not in a valid position - TipNotEmptyError: if the tip starts with liquid in it - LiquidNotFoundError: if liquid is not found during the probe process. + TipNotAttachedError: as an undefined error, if there is not tip attached to + the pipette. + TipNotEmptyError: as an undefined error, if the tip starts with liquid + in it. + MustHomeError: as an undefined error, if the plunger is not in a valid + position. """ pipette_id = params.pipetteId labware_id = params.labwareId @@ -139,8 +172,68 @@ async def execute(self, params: LiquidProbeParams) -> _ExecuteReturn: ) -class LiquidProbe(BaseCommand[LiquidProbeParams, LiquidProbeResult, ErrorOccurrence]): - """LiquidProbe command model.""" +class TryLiquidProbeImplementation( + AbstractCommandImpl[TryLiquidProbeParams, _TryLiquidProbeExecuteReturn] +): + """The implementation of a `tryLiquidProbe` command.""" + + def __init__( + self, + movement: MovementHandler, + pipetting: PipettingHandler, + model_utils: ModelUtils, + **kwargs: object, + ) -> None: + self._movement = movement + self._pipetting = pipetting + self._model_utils = model_utils + + async def execute(self, params: _CommonParams) -> _TryLiquidProbeExecuteReturn: + """Execute a `tryLiquidProbe` command. + + `tryLiquidProbe` is identical to `liquidProbe`, except that if no liquid is + found, `tryLiquidProbe` returns a success result with `z_position=null` instead + of a defined error. + """ + # We defer to the `liquidProbe` implementation. If it returns a defined + # `liquidNotFound` error, we remap that to a success result. + # Otherwise, we return the result or propagate the exception unchanged. + + original_impl = LiquidProbeImplementation( + movement=self._movement, + pipetting=self._pipetting, + model_utils=self._model_utils, + ) + original_result = await original_impl.execute(params) + + match original_result: + case DefinedErrorData( + public=LiquidNotFoundError(), + private=LiquidNotFoundErrorInternalData() as original_private, + ): + return SuccessData( + public=TryLiquidProbeResult( + z_position=None, + position=original_private.position, + ), + private=None, + ) + case SuccessData( + public=LiquidProbeResult() as original_public, private=None + ): + return SuccessData( + public=TryLiquidProbeResult( + position=original_public.position, + z_position=original_public.z_position, + ), + private=None, + ) + + +class LiquidProbe( + BaseCommand[LiquidProbeParams, LiquidProbeResult, LiquidNotFoundError] +): + """The model for a full `liquidProbe` command.""" commandType: LiquidProbeCommandType = "liquidProbe" params: LiquidProbeParams @@ -149,10 +242,33 @@ class LiquidProbe(BaseCommand[LiquidProbeParams, LiquidProbeResult, ErrorOccurre _ImplementationCls: Type[LiquidProbeImplementation] = LiquidProbeImplementation +class TryLiquidProbe( + BaseCommand[TryLiquidProbeParams, TryLiquidProbeResult, ErrorOccurrence] +): + """The model for a full `tryLiquidProbe` command.""" + + commandType: TryLiquidProbeCommandType = "tryLiquidProbe" + params: TryLiquidProbeParams + result: Optional[TryLiquidProbeResult] + + _ImplementationCls: Type[ + TryLiquidProbeImplementation + ] = TryLiquidProbeImplementation + + class LiquidProbeCreate(BaseCommandCreate[LiquidProbeParams]): - """Create LiquidProbe command request model.""" + """The request model for a `liquidProbe` command.""" commandType: LiquidProbeCommandType = "liquidProbe" params: LiquidProbeParams _CommandCls: Type[LiquidProbe] = LiquidProbe + + +class TryLiquidProbeCreate(BaseCommandCreate[TryLiquidProbeParams]): + """The request model for a `tryLiquidProbe` command.""" + + commandType: TryLiquidProbeCommandType = "tryLiquidProbe" + params: TryLiquidProbeParams + + _CommandCls: Type[TryLiquidProbe] = TryLiquidProbe diff --git a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py index dddf1ca5e06..d5bc6571214 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py +++ b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py @@ -1,5 +1,6 @@ """Test LiquidProbe commands.""" from datetime import datetime +from typing import Type, Union from opentrons.protocol_engine.errors.exceptions import ( MustHomeError, @@ -23,6 +24,9 @@ LiquidProbeParams, LiquidProbeResult, LiquidProbeImplementation, + TryLiquidProbeParams, + TryLiquidProbeResult, + TryLiquidProbeImplementation, ) from opentrons.protocol_engine.commands.command import DefinedErrorData, SuccessData @@ -36,14 +40,56 @@ from opentrons.protocol_engine.types import LoadedPipette +EitherImplementationType = Union[ + Type[LiquidProbeImplementation], Type[TryLiquidProbeImplementation] +] +EitherImplementation = Union[LiquidProbeImplementation, TryLiquidProbeImplementation] +EitherParamsType = Union[Type[LiquidProbeParams], Type[TryLiquidProbeParams]] +EitherResultType = Union[Type[LiquidProbeResult], Type[TryLiquidProbeResult]] + + +@pytest.fixture( + params=[ + (LiquidProbeImplementation, LiquidProbeParams, LiquidProbeResult), + (TryLiquidProbeImplementation, TryLiquidProbeParams, TryLiquidProbeResult), + ] +) +def types( + request: pytest.FixtureRequest, +) -> tuple[EitherImplementationType, EitherParamsType, EitherResultType]: + """Return a tuple of types associated with a single variant of the command.""" + return request.param # type: ignore[no-any-return] + + +@pytest.fixture +def implementation_type( + types: tuple[EitherImplementationType, object, object] +) -> EitherImplementationType: + """Return an implementation type. Kept in sync with the params and result types.""" + return types[0] + + +@pytest.fixture +def params_type(types: tuple[object, EitherParamsType, object]) -> EitherParamsType: + """Return a params type. Kept in sync with the implementation and result types.""" + return types[1] + + +@pytest.fixture +def result_type(types: tuple[object, object, EitherResultType]) -> EitherResultType: + """Return a result type. Kept in sync with the implementation and params types.""" + return types[2] + + @pytest.fixture def subject( + implementation_type: EitherImplementationType, movement: MovementHandler, pipetting: PipettingHandler, model_utils: ModelUtils, -) -> LiquidProbeImplementation: +) -> Union[LiquidProbeImplementation, TryLiquidProbeImplementation]: """Get the implementation subject.""" - return LiquidProbeImplementation( + return implementation_type( pipetting=pipetting, movement=movement, model_utils=model_utils, @@ -54,12 +100,14 @@ async def test_liquid_probe_implementation_no_prep( decoy: Decoy, movement: MovementHandler, pipetting: PipettingHandler, - subject: LiquidProbeImplementation, + subject: EitherImplementation, + params_type: EitherParamsType, + result_type: EitherResultType, ) -> None: """A Liquid Probe should have an execution implementation without preparing to aspirate.""" location = WellLocation(origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1)) - data = LiquidProbeParams( + data = params_type( pipetteId="abc", labwareId="123", wellName="A3", @@ -87,8 +135,9 @@ async def test_liquid_probe_implementation_no_prep( result = await subject.execute(data) + assert type(result.public) is result_type # Pydantic v1 only compares the fields. assert result == SuccessData( - public=LiquidProbeResult(z_position=15.0, position=DeckPoint(x=1, y=2, z=3)), + public=result_type(z_position=15.0, position=DeckPoint(x=1, y=2, z=3)), private=None, ) @@ -98,12 +147,14 @@ async def test_liquid_probe_implementation_with_prep( state_view: StateView, movement: MovementHandler, pipetting: PipettingHandler, - subject: LiquidProbeImplementation, + 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=0)) - data = LiquidProbeParams( + data = params_type( pipetteId="abc", labwareId="123", wellName="A3", @@ -133,26 +184,19 @@ async def test_liquid_probe_implementation_with_prep( result = await subject.execute(data) + assert type(result.public) is result_type # Pydantic v1 only compares the fields. assert result == SuccessData( - public=LiquidProbeResult(z_position=15.0, position=DeckPoint(x=1, y=2, z=3)), + 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), - ), - ) - async def test_liquid_not_found_error( decoy: Decoy, movement: MovementHandler, pipetting: PipettingHandler, - subject: LiquidProbeImplementation, + subject: EitherImplementation, + params_type: EitherParamsType, model_utils: ModelUtils, ) -> None: """It should return a liquid not found error if the hardware API indicates that.""" @@ -168,7 +212,7 @@ async def test_liquid_not_found_error( error_id = "error-id" error_timestamp = datetime(year=2020, month=1, day=2) - data = LiquidProbeParams( + data = params_type( pipetteId=pipette_id, labwareId=labware_id, wellName=well_name, @@ -201,22 +245,32 @@ async def test_liquid_not_found_error( result = await subject.execute(data) - assert result == DefinedErrorData( - public=LiquidNotFoundError.construct( - id=error_id, createdAt=error_timestamp, wrappedErrors=[matchers.Anything()] - ), - private=LiquidNotFoundErrorInternalData( - position=DeckPoint(x=position.x, y=position.y, z=position.z) - ), - ) + if isinstance(subject, LiquidProbeImplementation): + assert result == DefinedErrorData( + public=LiquidNotFoundError.construct( + id=error_id, + createdAt=error_timestamp, + wrappedErrors=[matchers.Anything()], + ), + private=LiquidNotFoundErrorInternalData( + position=DeckPoint(x=position.x, y=position.y, z=position.z) + ), + ) + else: + assert result == SuccessData( + public=TryLiquidProbeResult( + z_position=None, + position=DeckPoint(x=position.x, y=position.y, z=position.z), + ), + private=None, + ) async def test_liquid_probe_tip_checking( decoy: Decoy, - movement: MovementHandler, pipetting: PipettingHandler, - subject: LiquidProbeImplementation, - model_utils: ModelUtils, + subject: EitherImplementation, + params_type: EitherParamsType, ) -> None: """It should return a TipNotAttached error if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -226,7 +280,7 @@ async def test_liquid_probe_tip_checking( origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1) ) - data = LiquidProbeParams( + data = params_type( pipetteId=pipette_id, labwareId=labware_id, wellName=well_name, @@ -238,21 +292,15 @@ async def test_liquid_probe_tip_checking( pipette_id=pipette_id, ), ).then_raise(TipNotAttachedError()) - try: + with pytest.raises(TipNotAttachedError): await subject.execute(data) - assert False - except TipNotAttachedError: - assert True - except Exception: - assert False async def test_liquid_probe_volume_checking( decoy: Decoy, - movement: MovementHandler, pipetting: PipettingHandler, - subject: LiquidProbeImplementation, - model_utils: ModelUtils, + subject: EitherImplementation, + params_type: EitherParamsType, ) -> None: """It should return a TipNotEmptyError if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -262,7 +310,7 @@ async def test_liquid_probe_volume_checking( origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1) ) - data = LiquidProbeParams( + data = params_type( pipetteId=pipette_id, labwareId=labware_id, wellName=well_name, @@ -271,21 +319,15 @@ async def test_liquid_probe_volume_checking( decoy.when( pipetting.get_is_empty(pipette_id=pipette_id), ).then_return(False) - try: + with pytest.raises(TipNotEmptyError): await subject.execute(data) - assert False - except TipNotEmptyError: - assert True - except Exception: - assert False async def test_liquid_probe_location_checking( decoy: Decoy, movement: MovementHandler, - pipetting: PipettingHandler, - subject: LiquidProbeImplementation, - model_utils: ModelUtils, + subject: EitherImplementation, + params_type: EitherParamsType, ) -> None: """It should return a PositionUnkownError if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -295,7 +337,7 @@ async def test_liquid_probe_location_checking( origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1) ) - data = LiquidProbeParams( + data = params_type( pipetteId=pipette_id, labwareId=labware_id, wellName=well_name, @@ -306,10 +348,5 @@ async def test_liquid_probe_location_checking( mount=MountType.LEFT, ), ).then_return(False) - try: + with pytest.raises(MustHomeError): await subject.execute(data) - assert False - except MustHomeError: - assert True - except Exception: - assert False diff --git a/shared-data/command/schemas/8.json b/shared-data/command/schemas/8.json index 4700ace9229..d985fc7b902 100644 --- a/shared-data/command/schemas/8.json +++ b/shared-data/command/schemas/8.json @@ -41,6 +41,7 @@ "verifyTipPresence": "#/definitions/VerifyTipPresenceCreate", "getTipPresence": "#/definitions/GetTipPresenceCreate", "liquidProbe": "#/definitions/LiquidProbeCreate", + "tryLiquidProbe": "#/definitions/TryLiquidProbeCreate", "heaterShaker/waitForTemperature": "#/definitions/opentrons__protocol_engine__commands__heater_shaker__wait_for_temperature__WaitForTemperatureCreate", "heaterShaker/setTargetTemperature": "#/definitions/opentrons__protocol_engine__commands__heater_shaker__set_target_temperature__SetTargetTemperatureCreate", "heaterShaker/deactivateHeater": "#/definitions/DeactivateHeaterCreate", @@ -179,6 +180,9 @@ { "$ref": "#/definitions/LiquidProbeCreate" }, + { + "$ref": "#/definitions/TryLiquidProbeCreate" + }, { "$ref": "#/definitions/opentrons__protocol_engine__commands__heater_shaker__wait_for_temperature__WaitForTemperatureCreate" }, @@ -2773,7 +2777,7 @@ }, "LiquidProbeParams": { "title": "LiquidProbeParams", - "description": "Parameters required to liquid probe a specific well.", + "description": "Parameters required for a `liquidProbe` command.", "type": "object", "properties": { "labwareId": { @@ -2805,7 +2809,7 @@ }, "LiquidProbeCreate": { "title": "LiquidProbeCreate", - "description": "Create LiquidProbe command request model.", + "description": "The request model for a `liquidProbe` command.", "type": "object", "properties": { "commandType": { @@ -2833,6 +2837,68 @@ }, "required": ["params"] }, + "TryLiquidProbeParams": { + "title": "TryLiquidProbeParams", + "description": "Parameters required for a `tryLiquidProbe` command.", + "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/WellLocation" + } + ] + }, + "pipetteId": { + "title": "Pipetteid", + "description": "Identifier of pipette to use for liquid handling.", + "type": "string" + } + }, + "required": ["labwareId", "wellName", "pipetteId"] + }, + "TryLiquidProbeCreate": { + "title": "TryLiquidProbeCreate", + "description": "The request model for a `tryLiquidProbe` command.", + "type": "object", + "properties": { + "commandType": { + "title": "Commandtype", + "default": "tryLiquidProbe", + "enum": ["tryLiquidProbe"], + "type": "string" + }, + "params": { + "$ref": "#/definitions/TryLiquidProbeParams" + }, + "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"] + }, "opentrons__protocol_engine__commands__heater_shaker__wait_for_temperature__WaitForTemperatureParams": { "title": "WaitForTemperatureParams", "description": "Input parameters to wait for a Heater-Shaker's target temperature.",