feat(api): disable pressure sensor capabilities for PEEK pipettes

api/src/opentrons/hardware_control/backends/flex_protocol.py

@@ -60,6 +60,14 @@ def restore_system_constraints(self) -> AsyncIterator[None]:
     def grab_pressure(self, channels: int, mount: OT3Mount) -> AsyncIterator[None]:
         ...
 
+    def set_pressure_sensor_available(
+        self, pipette_axis: Axis, available: bool
+    ) -> None:
+        ...
+
+    def get_pressure_sensor_available(self, pipette_axis: Axis) -> bool:
+        ...
+
     def update_constraints_for_gantry_load(self, gantry_load: GantryLoad) -> None:
         ...
 

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -198,6 +198,7 @@
     PipetteLiquidNotFoundError,
     CommunicationError,
     PythonException,
+    UnsupportedHardwareCommand,
 )
 
 from .subsystem_manager import SubsystemManager
@@ -363,6 +364,7 @@ def __init__(
                 self._configuration.motion_settings, GantryLoad.LOW_THROUGHPUT
             )
         )
+        self._pressure_sensor_available: Dict[NodeId, bool] = {}
 
     @asynccontextmanager
     async def restore_system_constraints(self) -> AsyncIterator[None]:
@@ -381,6 +383,16 @@ async def grab_pressure(
         async with grab_pressure(channels, tool, self._messenger):
             yield
 
+    def set_pressure_sensor_available(
+        self, pipette_axis: Axis, available: bool
+    ) -> None:
+        pip_node = axis_to_node(pipette_axis)
+        self._pressure_sensor_available[pip_node] = available
+
+    def get_pressure_sensor_available(self, pipette_axis: Axis) -> bool:
+        pip_node = axis_to_node(pipette_axis)
+        return self._pressure_sensor_available[pip_node]
+
     def update_constraints_for_calibration_with_gantry_load(
         self,
         gantry_load: GantryLoad,
@@ -775,7 +787,8 @@ async def _runner_coroutine(
             for runner, is_gear_move in maybe_runners
             if runner
         ]
-        async with self._monitor_overpressure(pipettes_moving):
+        checked_moving_pipettes = self._pipettes_to_monitor_pressure(pipettes_moving)
+        async with self._monitor_overpressure(checked_moving_pipettes):
             all_positions = await asyncio.gather(*coros)
 
         for positions, handle_gear_move in all_positions:
@@ -884,7 +897,8 @@ async def home(
         moving_pipettes = [
             axis_to_node(ax) for ax in checked_axes if ax in Axis.pipette_axes()
         ]
-        async with self._monitor_overpressure(moving_pipettes):
+        checked_moving_pipettes = self._pipettes_to_monitor_pressure(moving_pipettes)
+        async with self._monitor_overpressure(checked_moving_pipettes):
             positions = await asyncio.gather(*coros)
         # TODO(CM): default gear motor homing routine to have some acceleration
         if Axis.Q in checked_axes:
@@ -899,6 +913,9 @@ async def home(
             self._handle_motor_status_response(position)
         return axis_convert(self._position, 0.0)
 
+    def _pipettes_to_monitor_pressure(self, pipettes: List[NodeId]) -> List[NodeId]:
+        return [pip for pip in pipettes if self._pressure_sensor_available[pip]]
+
     def _filter_move_group(self, move_group: MoveGroup) -> MoveGroup:
         new_group: MoveGroup = []
         for step in move_group:
@@ -1486,6 +1503,11 @@ async def liquid_probe(
     ) -> float:
         head_node = axis_to_node(Axis.by_mount(mount))
         tool = sensor_node_for_pipette(OT3Mount(mount.value))
+        if tool not in self._pipettes_to_monitor_pressure([tool]):
+            raise UnsupportedHardwareCommand(
+                "Liquid Presence Detection not available on this pipette."
+            )
+
         positions = await liquid_probe(
             messenger=self._messenger,
             tool=tool,

api/src/opentrons/hardware_control/dev_types.py

@@ -20,6 +20,7 @@
     PipetteConfigurations,
     SupportedTipsDefinition,
     PipetteBoundingBoxOffsetDefinition,
+    AvailableSensorDefinition,
 )
 from opentrons_shared_data.gripper import (
     GripperModel,
@@ -102,6 +103,7 @@ class PipetteDict(InstrumentDict):
     lld_settings: Optional[Dict[str, Dict[str, float]]]
     plunger_positions: Dict[str, float]
     shaft_ul_per_mm: float
+    available_sensors: AvailableSensorDefinition
 
 
 class PipetteStateDict(TypedDict):

api/src/opentrons/hardware_control/ot3api.py

@@ -32,6 +32,7 @@
 )
 from opentrons_shared_data.pipette import (
     pipette_load_name_conversions as pipette_load_name,
+    pipette_definition,
 )
 from opentrons_shared_data.robot.types import RobotType
 
@@ -634,8 +635,13 @@ async def cache_pipette(
             self._feature_flags.use_old_aspiration_functions,
         )
         self._pipette_handler.hardware_instruments[mount] = p
+
         if self._pipette_handler.has_pipette(mount):
             self._confirm_pipette_motion_constraints(mount)
+
+        if config is not None:
+            self._set_pressure_sensor_available(mount, instrument_config=config)
+
         # TODO (lc 12-5-2022) Properly support backwards compatibility
         # when applicable
         return skipped
@@ -649,6 +655,23 @@ def _confirm_pipette_motion_constraints(
                 mount, self.gantry_load
             )
 
+    def get_pressure_sensor_available(self, mount: OT3Mount) -> bool:
+        pip_axis = Axis.of_main_tool_actuator(mount)
+        return self._backend.get_pressure_sensor_available(pip_axis)
+
+    def _set_pressure_sensor_available(
+        self,
+        mount: OT3Mount,
+        instrument_config: pipette_definition.PipetteConfigurations,
+    ) -> None:
+        pressure_sensor_available = (
+            "pressure" in instrument_config.available_sensors.sensors
+        )
+        pip_axis = Axis.of_main_tool_actuator(mount)
+        self._backend.set_pressure_sensor_available(
+            pipette_axis=pip_axis, available=pressure_sensor_available
+        )
+
     async def cache_gripper(self, instrument_data: AttachedGripper) -> bool:
         """Set up gripper based on scanned information."""
         grip_cal = load_gripper_calibration_offset(instrument_data.get("id"))

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

@@ -32,6 +32,9 @@
 from opentrons.protocol_engine.clients import SyncClient as EngineClient
 from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
 from opentrons_shared_data.pipette.types import PipetteNameType
+from opentrons_shared_data.errors.exceptions import (
+    UnsupportedHardwareCommand,
+)
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from . import overlap_versions, pipette_movement_conflict
 
@@ -85,6 +88,13 @@ def __init__(
         self._liquid_presence_detection = bool(
             self._engine_client.state.pipettes.get_liquid_presence_detection(pipette_id)
         )
+        if (
+            self._liquid_presence_detection
+            and not self._pressure_supported_by_pipette()
+        ):
+            raise UnsupportedHardwareCommand(
+                "Pressure sensor not available for this pipette"
+            )
 
     @property
     def pipette_id(self) -> str:
@@ -859,6 +869,11 @@ def retract(self) -> None:
         z_axis = self._engine_client.state.pipettes.get_z_axis(self._pipette_id)
         self._engine_client.execute_command(cmd.HomeParams(axes=[z_axis]))
 
+    def _pressure_supported_by_pipette(self) -> bool:
+        return self._engine_client.state.pipettes.get_pipette_supports_pressure(
+            self.pipette_id
+        )
+
     def detect_liquid_presence(self, well_core: WellCore, loc: Location) -> bool:
         labware_id = well_core.labware_id
         well_name = well_core.get_name()

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

@@ -260,6 +260,10 @@ def get_blow_out_flow_rate(self, rate: float = 1.0) -> float:
     def get_liquid_presence_detection(self) -> bool:
         ...
 
+    @abstractmethod
+    def _pressure_supported_by_pipette(self) -> bool:
+        ...
+
     @abstractmethod
     def set_liquid_presence_detection(self, enable: bool) -> None:
         ...

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

@@ -586,6 +586,9 @@ def liquid_probe_without_recovery(
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
 
+    def _pressure_supported_by_pipette(self) -> bool:
+        return False
+
     def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""
         return False

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

@@ -504,6 +504,9 @@ def liquid_probe_without_recovery(
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
 
+    def _pressure_supported_by_pipette(self) -> bool:
+        return False
+
     def nozzle_configuration_valid_for_lld(self) -> bool:
         """Check if the nozzle configuration currently supports LLD."""
         return False

api/src/opentrons/protocol_api/instrument_context.py

@@ -6,6 +6,7 @@
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
     UnexpectedTipRemovalError,
+    UnsupportedHardwareCommand,
 )
 from opentrons.legacy_broker import LegacyBroker
 from opentrons.hardware_control.dev_types import PipetteDict
@@ -259,6 +260,7 @@ def aspirate(
             and self._core.nozzle_configuration_valid_for_lld()
             and self._core.get_current_volume() == 0
         ):
+            self._raise_if_pressure_not_supported_by_pipette()
             self.require_liquid_presence(well=well)
 
         with publisher.publish_context(
@@ -1705,6 +1707,8 @@ def liquid_presence_detection(self) -> bool:
     @liquid_presence_detection.setter
     @requires_version(2, 20)
     def liquid_presence_detection(self, enable: bool) -> None:
+        if enable:
+            self._raise_if_pressure_not_supported_by_pipette()
         self._core.set_liquid_presence_detection(enable)
 
     @property
@@ -2141,6 +2145,7 @@ def detect_liquid_presence(self, well: labware.Well) -> bool:
         .. note::
             The pressure sensors for the Flex 8-channel pipette are on channels 1 and 8 (positions A1 and H1). For the Flex 96-channel pipette, the pressure sensors are on channels 1 and 96 (positions A1 and H12). Other channels on multi-channel pipettes do not have sensors and cannot detect liquid.
         """
+        self._raise_if_pressure_not_supported_by_pipette()
         loc = well.top()
         return self._core.detect_liquid_presence(well._core, loc)
 
@@ -2153,6 +2158,7 @@ def require_liquid_presence(self, well: labware.Well) -> None:
         .. note::
             The pressure sensors for the Flex 8-channel pipette are on channels 1 and 8 (positions A1 and H1). For the Flex 96-channel pipette, the pressure sensors are on channels 1 and 96 (positions A1 and H12). Other channels on multi-channel pipettes do not have sensors and cannot detect liquid.
         """
+        self._raise_if_pressure_not_supported_by_pipette()
         loc = well.top()
         self._core.liquid_probe_with_recovery(well._core, loc)
 
@@ -2166,7 +2172,7 @@ def measure_liquid_height(self, well: labware.Well) -> float:
 
         This is intended for Opentrons internal use only and is not a guaranteed API.
         """
-
+        self._raise_if_pressure_not_supported_by_pipette()
         loc = well.top()
         height = self._core.liquid_probe_without_recovery(well._core, loc)
         return height
@@ -2187,6 +2193,12 @@ def _raise_if_configuration_not_supported_by_pipette(
                     )
             # SINGLE, QUADRANT and ALL are supported by all pipettes
 
+    def _raise_if_pressure_not_supported_by_pipette(self) -> None:
+        if not self._core._pressure_supported_by_pipette():
+            raise UnsupportedHardwareCommand(
+                "Pressure sensor not available for this pipette"
+            )
+
     def _handle_aspirate_target(
         self, target: validation.ValidTarget
     ) -> tuple[types.Location, Optional[labware.Well], Optional[bool]]:

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

@@ -17,6 +17,7 @@
 from opentrons.types import MountType
 from opentrons_shared_data.errors.exceptions import (
     PipetteLiquidNotFoundError,
+    UnsupportedHardwareCommand,
 )
 
 from ..types import DeckPoint
@@ -119,6 +120,14 @@ async def _execute_common(
     pipette_id = params.pipetteId
     labware_id = params.labwareId
     well_name = params.wellName
+    if (
+        "pressure"
+        not in state_view.pipettes.get_config(pipette_id).available_sensors.sensors
+    ):
+        raise UnsupportedHardwareCommand(
+            "Pressure sensor not available for this pipette"
+        )
+
     if not state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id):
         raise TipNotAttachedError(
             "Either the front right or back left nozzle must have a tip attached to probe liquid height."

api/src/opentrons/protocol_engine/resources/pipette_data_provider.py

@@ -69,6 +69,7 @@ class LoadedStaticPipetteData:
     pipette_lld_settings: Optional[Dict[str, Dict[str, float]]]
     plunger_positions: Dict[str, float]
     shaft_ul_per_mm: float
+    available_sensors: pipette_definition.AvailableSensorDefinition
 
 
 class VirtualPipetteDataProvider:
@@ -290,6 +291,8 @@ def _get_virtual_pipette_static_config_by_model(  # noqa: C901
                 "drop_tip": plunger_positions.drop_tip,
             },
             shaft_ul_per_mm=config.shaft_ul_per_mm,
+            available_sensors=config.available_sensors
+            or pipette_definition.AvailableSensorDefinition(sensors=[]),
         )
 
     def get_virtual_pipette_static_config(
@@ -308,6 +311,11 @@ def get_pipette_static_config(
     """Get the config for a pipette, given the state/config object from the HW API."""
     back_left_offset = pipette_dict["pipette_bounding_box_offsets"].back_left_corner
     front_right_offset = pipette_dict["pipette_bounding_box_offsets"].front_right_corner
+    available_sensors = (
+        pipette_dict["available_sensors"]
+        if "available_sensors" in pipette_dict.keys()
+        else pipette_definition.AvailableSensorDefinition(sensors=[])
+    )
     return LoadedStaticPipetteData(
         model=pipette_dict["model"],
         display_name=pipette_dict["display_name"],
@@ -339,6 +347,7 @@ def get_pipette_static_config(
         pipette_lld_settings=pipette_dict["lld_settings"],
         plunger_positions=pipette_dict["plunger_positions"],
         shaft_ul_per_mm=pipette_dict["shaft_ul_per_mm"],
+        available_sensors=available_sensors,
     )
 
 

api/src/opentrons/protocol_engine/state/pipettes.py

@@ -105,6 +105,7 @@ class StaticPipetteConfig:
     lld_settings: Optional[Dict[str, Dict[str, float]]]
     plunger_positions: Dict[str, float]
     shaft_ul_per_mm: float
+    available_sensors: pipette_definition.AvailableSensorDefinition
 
 
 @dataclasses.dataclass
@@ -296,6 +297,7 @@ def _update_pipette_config(self, state_update: update_types.StateUpdate) -> None
                 lld_settings=config.pipette_lld_settings,
                 plunger_positions=config.plunger_positions,
                 shaft_ul_per_mm=config.shaft_ul_per_mm,
+                available_sensors=config.available_sensors,
             )
             self._state.flow_rates_by_id[
                 state_update.pipette_config.pipette_id
@@ -761,6 +763,13 @@ def get_pipette_bounds_at_specified_move_to_position(
             pip_front_left_bound,
         )
 
+    def get_pipette_supports_pressure(self, pipette_id: str) -> bool:
+        """Return if this pipette supports a pressure sensor."""
+        return (
+            "pressure"
+            in self._state.static_config_by_id[pipette_id].available_sensors.sensors
+        )
+
     def get_liquid_presence_detection(self, pipette_id: str) -> bool:
         """Determine if liquid presence detection is enabled for this pipette."""
         try: