feat(api): create instr ctx methods for checking liquid presence

analyses-snapshot-testing/tests/__snapshots__/analyses_snapshot_test/test_analysis_snapshot[e42e36e3ca][OT2_X_v2_13_None_None_PythonSyntaxError].json

@@ -28,7 +28,7 @@
             "msg": "No module named 'superspecialmagic'",
             "name": "superspecialmagic",
             "path": "None",
-            "traceback": "  File \"/usr/local/lib/python3.10/site-packages/opentrons/cli/analyze.py\", line N, in _do_analyze\n    await runner.load(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/protocol_runner.py\", line N, in load\n    self._protocol_executor.extract_run_parameters(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/python_protocol_wrappers.py\", line N, in extract_run_parameters\n    return exec_add_parameters(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_add_parameters\n    exec(protocol.contents, new_globs)\n\n  File \"OT2_X_v2_13_None_None_PythonSyntaxError.py\", line N, in <module>\n"
+            "traceback": "  File \"/usr/local/lib/python3.10/site-packages/opentrons/cli/analyze.py\", line N, in _do_analyze\n    await orchestrator.load(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/run_orchestrator.py\", line N, in load\n    await self._protocol_runner.load(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/protocol_runner.py\", line N, in load\n    self._protocol_executor.extract_run_parameters(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/python_protocol_wrappers.py\", line N, in extract_run_parameters\n    return exec_add_parameters(\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_add_parameters\n    exec(protocol.contents, new_globs)\n\n  File \"OT2_X_v2_13_None_None_PythonSyntaxError.py\", line N, in <module>\n"
           },
           "errorType": "PythonException",
           "id": "UUID",

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

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from typing import Optional, TYPE_CHECKING, cast, Union
+from opentrons.protocol_engine.commands.liquid_probe import LiquidProbeResult
 from opentrons.protocols.api_support.types import APIVersion
 
 from opentrons.types import Location, Mount
@@ -31,6 +32,7 @@
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
 from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
 
+from opentrons_shared_data.errors.exceptions import PipetteLiquidNotFoundError
 from opentrons_shared_data.pipette.dev_types import PipetteNameType
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
@@ -843,3 +845,24 @@ def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         z_axis = self._engine_client.state.pipettes.get_z_axis(self._pipette_id)
         self._engine_client.execute_command(cmd.HomeParams(axes=[z_axis]))
+
+    def find_liquid_level(self, well_core: WellCore) -> float:
+        labware_id = well_core.labware_id
+        well_name = well_core.get_name()
+        well_location = WellLocation(
+            origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=0)
+        )
+        result = self._engine_client.execute_command_with_result(
+            cmd.LiquidProbeParams(
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+                pipetteId=self.pipette_id,
+            )
+        )
+        if result is not None and isinstance(result, LiquidProbeResult):
+            return result.z_position
+        # should never get here
+        raise PipetteLiquidNotFoundError(
+            "Error while trying to find liquid level.",
+        )

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

@@ -298,9 +298,15 @@ def configure_nozzle_layout(
     def is_tip_tracking_available(self) -> bool:
         """Return whether auto tip tracking is available for the pipette's current nozzle configuration."""
 
+    @abstractmethod
     def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         ...
 
+    @abstractmethod
+    def find_liquid_level(self, well_core: WellCoreType) -> float:
+        """Do a liquid probe to find the level of the liquid in the well."""
+        ...
+
 
 InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any])

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

@@ -6,6 +6,7 @@
 from opentrons import types
 from opentrons.hardware_control import CriticalPoint
 from opentrons.hardware_control.dev_types import PipetteDict
+from opentrons.protocol_api.core.common import WellCore
 from opentrons.protocols.api_support import instrument as instrument_support
 from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
 from opentrons.protocols.api_support.labware_like import LabwareLike
@@ -569,3 +570,7 @@ def is_tip_tracking_available(self) -> bool:
     def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]
+
+    def find_liquid_level(self, well_core: WellCore) -> float:
+        """This will never be called because it was added in API 2.20."""
+        assert False, "find_liquid_level only supported in API 2.20 & later"

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

@@ -6,6 +6,7 @@
 from opentrons import types
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.hardware_control.types import HardwareAction
+from opentrons.protocol_api.core.common import WellCore
 from opentrons.protocols.api_support import instrument as instrument_support
 from opentrons.protocols.api_support.labware_like import LabwareLike
 from opentrons.protocols.api_support.types import APIVersion
@@ -487,3 +488,7 @@ def is_tip_tracking_available(self) -> bool:
     def retract(self) -> None:
         """Retract this instrument to the top of the gantry."""
         self._protocol_interface.get_hardware.retract(self._mount)  # type: ignore [attr-defined]
+
+    def find_liquid_level(self, well_core: WellCore) -> float:
+        """This will never be called because it was added in API 2.20."""
+        assert False, "find_liquid_level only supported in API 2.20 & later"

api/src/opentrons/protocol_api/instrument_context.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
-
 import logging
 from contextlib import ExitStack
 from typing import Any, List, Optional, Sequence, Union, cast, Dict
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
+    PipetteLiquidNotFoundError,
     UnexpectedTipRemovalError,
 )
+from opentrons.protocol_engine.errors.exceptions import WellDoesNotExistError
 from opentrons.legacy_broker import LegacyBroker
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons import types
@@ -2046,3 +2047,44 @@ def configure_nozzle_layout(
         )
         # TODO (spp, 2023-12-05): verify that tipracks are on adapters for only full 96 channel config
         self._tip_racks = tip_racks or []
+
+    @requires_version(2, 20)
+    def detect_liquid_presence(self, well: labware.Well) -> bool:
+        """Check if there is liquid in a well.
+
+        :returns: A boolean.
+        """
+        if not isinstance(well, labware.Well):
+            raise WellDoesNotExistError("You must provide a valid well to check.")
+        try:
+            height = self._core.find_liquid_level(well._core)
+            if height > 0:
+                return True
+            return False  # it should never get here
+        except PipetteLiquidNotFoundError:
+            return False
+        except Exception as e:
+            raise e
+
+    @requires_version(2, 20)
+    def require_liquid_presence(self, well: labware.Well) -> None:
+        """If there is no liquid in a well, raise an error.
+
+        :returns: None.
+        """
+        if not isinstance(well, labware.Well):
+            raise WellDoesNotExistError("You must provide a valid well to check.")
+
+        self._core.find_liquid_level(well._core)
+
+    @requires_version(2, 20)
+    def measure_liquid_height(self, well: labware.Well) -> float:
+        """Check the height of the liquid within a well.
+
+        :returns: The height, in mm, of the liquid from the deck.
+        """
+        if not isinstance(well, labware.Well):
+            raise WellDoesNotExistError("You must provide a valid well to check.")
+
+        height = self._core.find_liquid_level(well._core)
+        return float(height)

api/src/opentrons/protocol_engine/clients/sync_client.py

@@ -2,6 +2,7 @@
 
 from typing import cast, Any, Optional, overload
 
+from opentrons.protocol_engine.errors.error_occurrence import ProtocolCommandFailedError
 from opentrons_shared_data.labware.dev_types import LabwareUri
 from opentrons_shared_data.labware.labware_definition import LabwareDefinition
 
@@ -89,6 +90,26 @@ def execute_command_without_recovery(
         create_request = CreateType(params=cast(Any, params))
         return self._transport.execute_command(create_request)
 
+    def execute_command_with_result(
+        self, params: commands.CommandParams
+    ) -> Optional[commands.CommandResult]:
+        """Execute a ProtocolEngine command, including error recovery, and return a result.
+
+        See `ChildThreadTransport.execute_command_wait_for_recovery()` for exact
+        behavior.
+        """
+        CreateType = CREATE_TYPES_BY_PARAMS_TYPE[type(params)]
+        create_request = CreateType(params=cast(Any, params))
+        result = self._transport.execute_command_wait_for_recovery(create_request)
+        if result.error is None:
+            return result.result
+        if isinstance(result.error, BaseException):  # necessary to pass lint
+            raise result.error
+        raise ProtocolCommandFailedError(
+            original_error=result.error,
+            message=f"{result.error.errorType}: {result.error.detail}",
+        )
+
     @property
     def state(self) -> StateView:
         """Get a view of the engine's state."""

api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py

@@ -1,6 +1,7 @@
 """Test for the ProtocolEngine-based instrument API core."""
 from typing import cast, Optional, Union
 
+from opentrons_shared_data.errors.exceptions import PipetteLiquidNotFoundError
 import pytest
 from decoy import Decoy
 
@@ -1288,3 +1289,33 @@ def test_configure_for_volume_post_219(
             )
         )
     )
+
+
+@pytest.mark.parametrize("version", versions_at_or_above(APIVersion(2, 20)))
+def test_find_liquid_level(
+    decoy: Decoy,
+    mock_engine_client: EngineClient,
+    mock_protocol_core: ProtocolCore,
+    subject: InstrumentCore,
+    version: APIVersion,
+) -> None:
+    """It should raise an exception on an empty well."""
+    well_core = WellCore(
+        name="my cool well", labware_id="123abc", engine_client=mock_engine_client
+    )
+    try:
+        subject.find_liquid_level(well_core)
+    except PipetteLiquidNotFoundError:
+        assert True
+    decoy.verify(
+        mock_engine_client.execute_command_with_result(
+            cmd.LiquidProbeParams(
+                pipetteId=subject.pipette_id,
+                wellLocation=WellLocation(
+                    origin=WellOrigin.TOP, offset=WellOffset(x=0, y=0, z=0)
+                ),
+                wellName=well_core.get_name(),
+                labwareId=well_core.labware_id,
+            )
+        )
+    )

api/tests/opentrons/protocol_api/test_instrument_context.py

@@ -1,7 +1,6 @@
 """Tests for the InstrumentContext public interface."""
 from collections import OrderedDict
 import inspect
-
 import pytest
 from pytest_lazyfixture import lazy_fixture  # type: ignore[import-untyped]
 from decoy import Decoy
@@ -39,6 +38,7 @@
 
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
+    PipetteLiquidNotFoundError,
 )
 
 
@@ -1268,3 +1268,53 @@ def test_aspirate_0_volume_means_aspirate_nothing(
         ),
         times=1,
     )
+
+
+@pytest.mark.parametrize("api_version", [APIVersion(2, 20)])
+def test_detect_liquid_presence(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """It should only return booleans. Not raise an exception."""
+    mock_well = decoy.mock(cls=Well)
+    decoy.when(mock_instrument_core.find_liquid_level(mock_well._core)).then_raise(
+        PipetteLiquidNotFoundError()
+    )
+    result = subject.detect_liquid_presence(mock_well)
+    assert isinstance(result, bool)
+
+
+@pytest.mark.parametrize("api_version", [APIVersion(2, 20)])
+def test_require_liquid_presence(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """It should raise an exception when called."""
+    mock_well = decoy.mock(cls=Well)
+    decoy.when(mock_instrument_core.find_liquid_level(mock_well._core))
+    subject.require_liquid_presence(mock_well)
+    decoy.when(mock_instrument_core.find_liquid_level(mock_well._core)).then_raise(
+        PipetteLiquidNotFoundError()
+    )
+    with pytest.raises(PipetteLiquidNotFoundError):
+        subject.require_liquid_presence(mock_well)
+
+
+@pytest.mark.parametrize("api_version", [APIVersion(2, 20)])
+def test_measure_liquid_height(
+    decoy: Decoy,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    mock_protocol_core: ProtocolCore,
+) -> None:
+    """It should raise an exception when called."""
+    mock_well = decoy.mock(cls=Well)
+    decoy.when(mock_instrument_core.find_liquid_level(mock_well._core)).then_raise(
+        PipetteLiquidNotFoundError()
+    )
+    with pytest.raises(PipetteLiquidNotFoundError):
+        subject.measure_liquid_height(mock_well)