diff --git a/api/src/opentrons/hardware_control/backends/flex_protocol.py b/api/src/opentrons/hardware_control/backends/flex_protocol.py index 7b7269624f2..7665a674500 100644 --- a/api/src/opentrons/hardware_control/backends/flex_protocol.py +++ b/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: ... diff --git a/api/src/opentrons/hardware_control/backends/ot3controller.py b/api/src/opentrons/hardware_control/backends/ot3controller.py index 505bce73365..3198f8dd6fc 100644 --- a/api/src/opentrons/hardware_control/backends/ot3controller.py +++ b/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, @@ -692,7 +704,8 @@ async def move( pipettes_moving = moving_pipettes_in_move_group(move_group) - 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): positions = await runner.run(can_messenger=self._messenger) self._handle_motor_status_response(positions) @@ -799,7 +812,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: @@ -813,6 +827,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: @@ -1392,6 +1409,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, diff --git a/api/src/opentrons/hardware_control/dev_types.py b/api/src/opentrons/hardware_control/dev_types.py index a6773cb9184..90bc7dca40d 100644 --- a/api/src/opentrons/hardware_control/dev_types.py +++ b/api/src/opentrons/hardware_control/dev_types.py @@ -20,6 +20,7 @@ PipetteConfigurations, SupportedTipsDefinition, PipetteBoundingBoxOffsetDefinition, + AvailableSensorDefinition, ) from opentrons_shared_data.gripper import ( GripperModel, @@ -100,6 +101,7 @@ class PipetteDict(InstrumentDict): pipette_bounding_box_offsets: PipetteBoundingBoxOffsetDefinition current_nozzle_map: NozzleMap lld_settings: Optional[Dict[str, Dict[str, float]]] + available_sensors: AvailableSensorDefinition class PipetteStateDict(TypedDict): diff --git a/api/src/opentrons/hardware_control/ot3api.py b/api/src/opentrons/hardware_control/ot3api.py index adcf5b9d2b7..6bbb7c83c6a 100644 --- a/api/src/opentrons/hardware_control/ot3api.py +++ b/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 @@ -633,8 +634,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 @@ -648,6 +654,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")) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 825d45bfded..11a13d105b8 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -31,6 +31,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 opentrons.hardware_control.nozzle_manager import NozzleConfigurationType from opentrons.hardware_control.nozzle_manager import NozzleMap @@ -86,6 +89,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: @@ -847,6 +857,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() diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index 7d1816e1044..0fcedfa332a 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/api/src/opentrons/protocol_api/core/instrument.py @@ -253,6 +253,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: ... diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py index ed1e0d607c9..c6991e628a7 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py @@ -583,3 +583,7 @@ def liquid_probe_without_recovery( ) -> float: """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 + diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py index 55bde6c0a75..c933dd2dd43 100644 --- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py +++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py @@ -501,3 +501,6 @@ def liquid_probe_without_recovery( ) -> float: """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 diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 880626b53c9..adb48df08e6 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -7,6 +7,7 @@ CommandPreconditionViolated, CommandParameterLimitViolated, UnexpectedTipRemovalError, + UnsupportedHardwareCommand, ) from opentrons.legacy_broker import LegacyBroker from opentrons.hardware_control.dev_types import PipetteDict @@ -260,6 +261,7 @@ def aspirate( and self._96_tip_config_valid() 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( @@ -1694,6 +1696,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 @@ -2143,6 +2147,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() self._96_tip_config_valid() return self._core.detect_liquid_presence(well._core, loc) @@ -2156,6 +2161,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._96_tip_config_valid() self._core.liquid_probe_with_recovery(well._core, loc) @@ -2170,7 +2176,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() self._96_tip_config_valid() height = self._core.liquid_probe_without_recovery(well._core, loc) @@ -2192,6 +2198,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]]: diff --git a/api/src/opentrons/protocol_engine/commands/liquid_probe.py b/api/src/opentrons/protocol_engine/commands/liquid_probe.py index f78cd5bb55c..3236ca5d931 100644 --- a/api/src/opentrons/protocol_engine/commands/liquid_probe.py +++ b/api/src/opentrons/protocol_engine/commands/liquid_probe.py @@ -16,6 +16,7 @@ from opentrons.types import MountType from opentrons_shared_data.errors.exceptions import ( PipetteLiquidNotFoundError, + UnsupportedHardwareCommand, ) from ..types import DeckPoint @@ -113,6 +114,13 @@ async def _execute_common( well_name = params.wellName state_update = update_types.StateUpdate() + if ( + "pressure" + not in state_view.pipettes.get_config(pipette_id).available_sensors.sensors + ): + raise UnsupportedHardwareCommand( + "Pressure sensor not available for this pipette" + ) # May raise TipNotAttachedError. aspirated_volume = state_view.pipettes.get_aspirated_volume(pipette_id) diff --git a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py b/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py index d3998c69bd1..0d6e979fc9d 100644 --- a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py +++ b/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py @@ -67,6 +67,7 @@ class LoadedStaticPipetteData: back_left_corner_offset: Point front_right_corner_offset: Point pipette_lld_settings: Optional[Dict[str, Dict[str, float]]] + available_sensors: pipette_definition.AvailableSensorDefinition class VirtualPipetteDataProvider: @@ -280,6 +281,8 @@ def _get_virtual_pipette_static_config_by_model( # noqa: C901 pip_front_right[0], pip_front_right[1], pip_front_right[2] ), pipette_lld_settings=config.lld_settings, + available_sensors=config.available_sensors + or pipette_definition.AvailableSensorDefinition(sensors=[]), ) def get_virtual_pipette_static_config( @@ -298,6 +301,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"], @@ -327,6 +335,7 @@ def get_pipette_static_config( front_right_offset[0], front_right_offset[1], front_right_offset[2] ), pipette_lld_settings=pipette_dict["lld_settings"], + available_sensors=available_sensors, ) diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index bb90e067ec6..a27e70ebe44 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -98,6 +98,7 @@ class StaticPipetteConfig: bounding_nozzle_offsets: BoundingNozzlesOffsets default_nozzle_map: NozzleMap # todo(mm, 2024-10-14): unused, remove? lld_settings: Optional[Dict[str, Dict[str, float]]] + available_sensors: pipette_definition.AvailableSensorDefinition @dataclasses.dataclass @@ -292,6 +293,7 @@ def _update_pipette_config(self, state_update: update_types.StateUpdate) -> None ), default_nozzle_map=config.nozzle_map, lld_settings=config.pipette_lld_settings, + available_sensors=config.available_sensors, ) self._state.flow_rates_by_id[ state_update.pipette_config.pipette_id @@ -723,6 +725,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: diff --git a/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py b/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py index 5ffee581de4..a180455325c 100644 --- a/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py +++ b/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py @@ -373,6 +373,8 @@ async def test_home_execute( **config ) as mock_runner: present_axes = set(ax for ax in axes if controller.axis_is_present(ax)) + controller.set_pressure_sensor_available(Axis.P_L, True) + controller.set_pressure_sensor_available(Axis.P_R, True) # nothing has been homed assert not controller._motor_status @@ -484,6 +486,8 @@ async def test_home_only_present_devices( homed_position = {} controller._position = starting_position + controller.set_pressure_sensor_available(Axis.P_L, True) + controller.set_pressure_sensor_available(Axis.P_R, True) mock_move_group_run.side_effect = move_group_run_side_effect(controller, axes) @@ -728,6 +732,9 @@ async def test_liquid_probe( mock_move_group_run.side_effect = probe_move_group_run_side_effect( head_node, tool_node ) + controller._pipettes_to_monitor_pressure = mock.MagicMock( # type: ignore[method-assign] + return_value=[sensor_node_for_mount(mount)] + ) try: await controller.liquid_probe( mount=mount, @@ -1292,3 +1299,33 @@ def test_grip_error_detection( hard_max, hard_min, ) + +@pytest.mark.parametrize( + argnames=["axes", "pipette_has_sensor"], + argvalues=[[[Axis.P_L, Axis.P_R], True], [[Axis.P_L, Axis.P_R], False]], +) +async def test_pressure_disable( + controller: OT3Controller, + axes: List[Axis], + mock_present_devices: None, + mock_check_overpressure: None, + pipette_has_sensor: bool, +) -> None: + config = {"run.side_effect": move_group_run_side_effect_home(controller, axes)} + with mock.patch( # type: ignore [call-overload] + "opentrons.hardware_control.backends.ot3controller.MoveGroupRunner", + spec=MoveGroupRunner, + **config + ): + with mock.patch.object(controller, "_monitor_overpressure") as monitor: + controller.set_pressure_sensor_available(Axis.P_L, pipette_has_sensor) + controller.set_pressure_sensor_available(Axis.P_R, True) + + await controller.home(axes, GantryLoad.LOW_THROUGHPUT) + + if pipette_has_sensor: + monitor.assert_called_once_with( + [NodeId.pipette_left, NodeId.pipette_right] + ) + else: + monitor.assert_called_once_with([NodeId.pipette_right]) diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 069330036ec..3384c05203c 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -89,7 +89,7 @@ def mock_instrument_core(decoy: Decoy) -> InstrumentCore: """Get a mock instrument implementation core.""" instrument_core = decoy.mock(cls=InstrumentCore) decoy.when(instrument_core.get_mount()).then_return(Mount.LEFT) - + decoy.when(instrument_core._pressure_supported_by_pipette()).then_return(True) # we need to add this for the mock of liquid_presence detection to actually work # this replaces the mock with a a property again instrument_core._liquid_presence_detection = False # type: ignore[attr-defined] diff --git a/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py b/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py index d237c9e6090..6207c368da1 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py +++ b/api/tests/opentrons/protocol_engine/commands/test_configure_for_volume.py @@ -22,10 +22,17 @@ ConfigureForVolumeImplementation, ) from opentrons_shared_data.pipette.types import PipetteNameType +from opentrons_shared_data.pipette.pipette_definition import AvailableSensorDefinition from ..pipette_fixtures import get_default_nozzle_map from opentrons.types import Point +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + @pytest.mark.parametrize( "data", [ @@ -41,7 +48,10 @@ ], ) async def test_configure_for_volume_implementation( - decoy: Decoy, equipment: EquipmentHandler, data: ConfigureForVolumeParams + decoy: Decoy, + equipment: EquipmentHandler, + data: ConfigureForVolumeParams, + available_sensors: AvailableSensorDefinition, ) -> None: """A ConfigureForVolume command should have an execution implementation.""" subject = ConfigureForVolumeImplementation(equipment=equipment) @@ -63,6 +73,7 @@ async def test_configure_for_volume_implementation( back_left_corner_offset=Point(10, 20, 30), front_right_corner_offset=Point(40, 50, 60), pipette_lld_settings={}, + available_sensors=available_sensors, ) decoy.when( 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 2cada4f3e24..f9c72136822 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py +++ b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py @@ -13,9 +13,20 @@ ) from decoy import matchers, Decoy import pytest +from opentrons_shared_data.pipette.pipette_definition import ( + AvailableSensorDefinition, + SupportedTipsDefinition, +) + +from opentrons_shared_data.pipette.types import PipetteNameType from opentrons.protocol_engine.commands.pipetting_common import LiquidNotFoundError from opentrons.protocol_engine.state.state import StateView +from opentrons.protocol_engine.state.pipettes import ( + StaticPipetteConfig, + BoundingNozzlesOffsets, + PipetteBoundingBoxOffsets, +) from opentrons.protocol_engine.state import update_types from opentrons.types import MountType, Point from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset, DeckPoint @@ -37,6 +48,7 @@ ) from opentrons.protocol_engine.resources.model_utils import ModelUtils +from ..pipette_fixtures import get_default_nozzle_map EitherImplementationType = Union[ Type[LiquidProbeImplementation], Type[TryLiquidProbeImplementation] @@ -46,6 +58,12 @@ EitherResultType = Union[Type[LiquidProbeResult], Type[TryLiquidProbeResult]] +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + @pytest.fixture( params=[ (LiquidProbeImplementation, LiquidProbeParams, LiquidProbeResult), @@ -105,6 +123,8 @@ async def test_liquid_probe_implementation( params_type: EitherParamsType, result_type: EitherResultType, model_utils: ModelUtils, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should move to the destination and do a liquid probe there.""" location = WellLocation(origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1)) @@ -146,6 +166,34 @@ async def test_liquid_probe_implementation( ), ).then_return(30.0) + decoy.when(state_view.pipettes.get_config("abc")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) + timestamp = datetime(year=2020, month=1, day=2) decoy.when(model_utils.get_timestamp()).then_return(timestamp) @@ -179,6 +227,8 @@ async def test_liquid_not_found_error( subject: EitherImplementation, params_type: EitherParamsType, model_utils: ModelUtils, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should return a liquid not found error if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -201,7 +251,33 @@ async def test_liquid_not_found_error( ) decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(0) - + decoy.when(state_view.pipettes.get_config("pipette-id")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) decoy.when( await movement.move_to_well( pipette_id=pipette_id, @@ -263,6 +339,8 @@ async def test_liquid_probe_tip_checking( state_view: StateView, subject: EitherImplementation, params_type: EitherParamsType, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should raise a TipNotAttached error if the state view indicates that.""" pipette_id = "pipette-id" @@ -282,6 +360,33 @@ async def test_liquid_probe_tip_checking( decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_raise( TipNotAttachedError() ) + decoy.when(state_view.pipettes.get_config("pipette-id")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) with pytest.raises(TipNotAttachedError): await subject.execute(data) @@ -291,6 +396,8 @@ async def test_liquid_probe_plunger_preparedness_checking( state_view: StateView, subject: EitherImplementation, params_type: EitherParamsType, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should raise a PipetteNotReadyToAspirate error if the state view indicates that.""" pipette_id = "pipette-id" @@ -307,6 +414,36 @@ async def test_liquid_probe_plunger_preparedness_checking( wellLocation=well_location, ) + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) + decoy.when(state_view.pipettes.get_config("pipette-id")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(None) with pytest.raises(PipetteNotReadyToAspirateError): await subject.execute(data) @@ -317,6 +454,8 @@ async def test_liquid_probe_volume_checking( state_view: StateView, subject: EitherImplementation, params_type: EitherParamsType, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should return a TipNotEmptyError if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -336,6 +475,37 @@ async def test_liquid_probe_volume_checking( decoy.when( state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id), ).then_return(123) + decoy.when(state_view.pipettes.get_config("pipette-id")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) + with pytest.raises(TipNotEmptyError): await subject.execute(data) @@ -352,6 +522,8 @@ async def test_liquid_probe_location_checking( movement: MovementHandler, subject: EitherImplementation, params_type: EitherParamsType, + available_sensors: AvailableSensorDefinition, + supported_tip_fixture: SupportedTipsDefinition, ) -> None: """It should return a PositionUnkownError if the hardware API indicates that.""" pipette_id = "pipette-id" @@ -368,6 +540,33 @@ async def test_liquid_probe_location_checking( wellLocation=well_location, ) decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(0) + decoy.when(state_view.pipettes.get_config("pipette-id")).then_return( + StaticPipetteConfig( + min_volume=1, + max_volume=9001, + channels=1, + model="blah", + display_name="bleh", + serial_number="", + tip_configuration_lookup_table={9001: supported_tip_fixture}, + nominal_tip_overlap={}, + home_position=0, + nozzle_offset_z=0, + bounding_nozzle_offsets=BoundingNozzlesOffsets( + back_left_offset=Point(x=10, y=20, z=30), + front_right_offset=Point(x=40, y=50, z=60), + ), + default_nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), + pipette_bounding_box_offsets=PipetteBoundingBoxOffsets( + back_left_corner=Point(x=10, y=20, z=30), + front_right_corner=Point(x=40, y=50, z=60), + front_left_corner=Point(x=10, y=50, z=60), + back_right_corner=Point(x=40, y=20, z=60), + ), + lld_settings={}, + available_sensors=available_sensors, + ) + ) decoy.when( await movement.check_for_valid_position( mount=MountType.LEFT, @@ -375,3 +574,4 @@ async def test_liquid_probe_location_checking( ).then_return(False) with pytest.raises(MustHomeError): await subject.execute(data) + diff --git a/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py b/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py index 44a9db61863..0e29bf2c663 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py +++ b/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py @@ -9,6 +9,7 @@ from opentrons_shared_data.pipette.types import PipetteNameType from opentrons_shared_data.robot.types import RobotType +from opentrons_shared_data.pipette.pipette_definition import AvailableSensorDefinition from opentrons.types import MountType, Point from opentrons.protocol_engine.errors import InvalidSpecificationForRobotTypeError @@ -27,6 +28,12 @@ from ..pipette_fixtures import get_default_nozzle_map +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + @pytest.mark.parametrize( "data", [ @@ -48,6 +55,7 @@ async def test_load_pipette_implementation( equipment: EquipmentHandler, state_view: StateView, data: LoadPipetteParams, + available_sensors: AvailableSensorDefinition, ) -> None: """A LoadPipette command should have an execution implementation.""" subject = LoadPipetteImplementation(equipment=equipment, state_view=state_view) @@ -68,6 +76,7 @@ async def test_load_pipette_implementation( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ) decoy.when( @@ -109,6 +118,7 @@ async def test_load_pipette_implementation_96_channel( decoy: Decoy, equipment: EquipmentHandler, state_view: StateView, + available_sensors: AvailableSensorDefinition, ) -> None: """A LoadPipette command should have an execution implementation.""" subject = LoadPipetteImplementation(equipment=equipment, state_view=state_view) @@ -135,6 +145,7 @@ async def test_load_pipette_implementation_96_channel( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ) decoy.when( diff --git a/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py b/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py index b7a020c2d35..4b22b0d32bd 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py +++ b/api/tests/opentrons/protocol_engine/execution/test_equipment_handler.py @@ -69,6 +69,14 @@ def _make_config(use_virtual_modules: bool) -> Config: ) +@pytest.fixture +def available_sensors() -> pipette_definition.AvailableSensorDefinition: + """Provide a list of sensors.""" + return pipette_definition.AvailableSensorDefinition( + sensors=["pressure", "capacitive", "environment"] + ) + + @pytest.fixture(autouse=True) def patch_mock_pipette_data_provider( decoy: Decoy, @@ -133,6 +141,7 @@ def tip_overlap_versions(request: SubRequest) -> str: def loaded_static_pipette_data( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, target_tip_overlap_data: Dict[str, float], + available_sensors: pipette_definition.AvailableSensorDefinition, ) -> LoadedStaticPipetteData: """Get a pipette config data value object.""" return LoadedStaticPipetteData( @@ -154,6 +163,7 @@ def loaded_static_pipette_data( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ) diff --git a/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py b/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py index 086b3ec297b..1d58057c090 100644 --- a/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py +++ b/api/tests/opentrons/protocol_engine/resources/test_pipette_data_provider.py @@ -7,6 +7,7 @@ from opentrons_shared_data.pipette.pipette_definition import ( PipetteBoundingBoxOffsetDefinition, TIP_OVERLAP_VERSION_MAXIMUM, + AvailableSensorDefinition, ) from opentrons.hardware_control.dev_types import PipetteDict @@ -24,6 +25,12 @@ from opentrons.types import Point +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + @pytest.fixture def subject_instance() -> VirtualPipetteDataProvider: """Instance of a VirtualPipetteDataProvider for test.""" @@ -32,6 +39,7 @@ def subject_instance() -> VirtualPipetteDataProvider: def test_get_virtual_pipette_static_config( subject_instance: VirtualPipetteDataProvider, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return config data given a pipette name.""" result = subject_instance.get_virtual_pipette_static_config( @@ -65,11 +73,13 @@ def test_get_virtual_pipette_static_config( back_left_corner_offset=Point(0, 0, 10.45), front_right_corner_offset=Point(0, 0, 10.45), pipette_lld_settings={}, + available_sensors=AvailableSensorDefinition(sensors=[]), ) def test_configure_virtual_pipette_for_volume( subject_instance: VirtualPipetteDataProvider, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return an updated config if the liquid class changes.""" result1 = subject_instance.get_virtual_pipette_static_config( @@ -94,6 +104,7 @@ def test_configure_virtual_pipette_for_volume( back_left_corner_offset=Point(-8.0, -22.0, -259.15), front_right_corner_offset=Point(-8.0, -22.0, -259.15), pipette_lld_settings={"t50": {"minHeight": 1.0, "minVolume": 0.0}}, + available_sensors=available_sensors, ) subject_instance.configure_virtual_pipette_for_volume( "my-pipette", 1, result1.model @@ -120,11 +131,13 @@ def test_configure_virtual_pipette_for_volume( back_left_corner_offset=Point(-8.0, -22.0, -259.15), front_right_corner_offset=Point(-8.0, -22.0, -259.15), pipette_lld_settings={"t50": {"minHeight": 1.0, "minVolume": 0.0}}, + available_sensors=available_sensors, ) def test_load_virtual_pipette_by_model_string( subject_instance: VirtualPipetteDataProvider, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return config data given a pipette model.""" result = subject_instance.get_virtual_pipette_static_config_by_model_string( @@ -149,6 +162,7 @@ def test_load_virtual_pipette_by_model_string( back_left_corner_offset=Point(-16.0, 43.15, 35.52), front_right_corner_offset=Point(16.0, -43.15, 35.52), pipette_lld_settings={}, + available_sensors=AvailableSensorDefinition(sensors=[]), ) @@ -193,6 +207,7 @@ def test_load_virtual_pipette_nozzle_layout( @pytest.fixture def pipette_dict( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> PipetteDict: """Get a pipette dict.""" return { @@ -246,6 +261,7 @@ def pipette_dict( "t200": {"minHeight": 0.5, "minVolume": 0}, "t1000": {"minHeight": 0.5, "minVolume": 0}, }, + "available_sensors": available_sensors, } @@ -263,6 +279,7 @@ def test_get_pipette_static_config( pipette_dict: PipetteDict, tip_overlap_version: str, overlap_data: Dict[str, float], + available_sensors: AvailableSensorDefinition, ) -> None: """It should return config data given a PipetteDict.""" result = subject.get_pipette_static_config(pipette_dict, tip_overlap_version) @@ -292,6 +309,7 @@ def test_get_pipette_static_config( "t200": {"minHeight": 0.5, "minVolume": 0}, "t1000": {"minHeight": 0.5, "minVolume": 0}, }, + available_sensors=available_sensors, ) diff --git a/api/tests/opentrons/protocol_engine/state/test_geometry_view.py b/api/tests/opentrons/protocol_engine/state/test_geometry_view.py index 42ee037c1ce..ac125eebd41 100644 --- a/api/tests/opentrons/protocol_engine/state/test_geometry_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_geometry_view.py @@ -101,6 +101,14 @@ from ...protocol_runner.test_json_translator import _load_labware_definition_data +@pytest.fixture +def available_sensors() -> pipette_definition.AvailableSensorDefinition: + """Provide a list of sensors.""" + return pipette_definition.AvailableSensorDefinition( + sensors=["pressure", "capacitive", "environment"] + ) + + @pytest.fixture def mock_labware_view(decoy: Decoy) -> LabwareView: """Get a mock in the shape of a LabwareView.""" @@ -2575,6 +2583,7 @@ def test_get_next_drop_tip_location( pipette_mount: MountType, expected_locations: List[DropTipWellLocation], supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: pipette_definition.AvailableSensorDefinition, ) -> None: """It should provide the next location to drop tips into within a labware.""" decoy.when(mock_labware_view.is_fixed_trash(labware_id="abc")).then_return(True) @@ -2611,6 +2620,7 @@ def test_get_next_drop_tip_location( back_right_corner=Point(x=40, y=20, z=60), ), lld_settings={}, + available_sensors=available_sensors, ) ) decoy.when(mock_pipette_view.get_mount("pip-123")).then_return(pipette_mount) diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py index c8eab566abe..64d3febed45 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py @@ -50,6 +50,14 @@ from ..pipette_fixtures import get_default_nozzle_map +@pytest.fixture +def available_sensors() -> pipette_definition.AvailableSensorDefinition: + """Provide a list of sensors.""" + return pipette_definition.AvailableSensorDefinition( + sensors=["pressure", "capacitive", "environment"] + ) + + @pytest.fixture def subject() -> PipetteStore: """Get a PipetteStore test subject for all subsequent tests.""" @@ -187,6 +195,7 @@ def test_location_state_update(subject: PipetteStore) -> None: def test_handles_load_pipette( subject: PipetteStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: pipette_definition.AvailableSensorDefinition, ) -> None: """It should add the pipette data to the state.""" dummy_command = create_succeeded_command() @@ -217,6 +226,7 @@ def test_handles_load_pipette( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ) config_update = update_types.PipetteConfigUpdate( pipette_id="pipette-id", @@ -573,6 +583,7 @@ def test_set_movement_speed(subject: PipetteStore) -> None: def test_add_pipette_config( subject: PipetteStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: pipette_definition.AvailableSensorDefinition, ) -> None: """It should update state from any pipette config private result.""" command = cmd.LoadPipette.construct( # type: ignore[call-arg] @@ -600,6 +611,7 @@ def test_add_pipette_config( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ) subject.handle_action( @@ -638,6 +650,7 @@ def test_add_pipette_config( back_right_corner=Point(x=4, y=2, z=3), ), lld_settings={}, + available_sensors=available_sensors, ) assert subject.state.flow_rates_by_id["pipette-id"].default_aspirate == {"a": 1.0} assert subject.state.flow_rates_by_id["pipette-id"].default_dispense == {"b": 2.0} diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py index 3b4d04bd967..260bef6a054 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py @@ -6,7 +6,10 @@ from opentrons_shared_data.pipette.types import PipetteNameType from opentrons_shared_data.pipette import pipette_definition -from opentrons_shared_data.pipette.pipette_definition import ValidNozzleMaps +from opentrons_shared_data.pipette.pipette_definition import ( + ValidNozzleMaps, + AvailableSensorDefinition, +) from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE from opentrons.hardware_control import CriticalPoint @@ -54,6 +57,12 @@ ) +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + def get_pipette_view( pipettes_by_id: Optional[Dict[str, LoadedPipette]] = None, aspirated_volume_by_id: Optional[Dict[str, Optional[float]]] = None, @@ -261,6 +270,7 @@ def test_get_aspirated_volume() -> None: def test_get_pipette_working_volume( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should get the minimum value of tip volume and max volume.""" subject = get_pipette_view( @@ -283,6 +293,7 @@ def test_get_pipette_working_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ) }, ) @@ -292,6 +303,7 @@ def test_get_pipette_working_volume( def test_get_pipette_working_volume_raises_if_tip_volume_is_none( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """Should raise an exception that no tip is attached.""" subject = get_pipette_view( @@ -314,6 +326,7 @@ def test_get_pipette_working_volume_raises_if_tip_volume_is_none( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ) }, ) @@ -327,6 +340,8 @@ def test_get_pipette_working_volume_raises_if_tip_volume_is_none( def test_get_pipette_available_volume( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + decoy: Decoy, + available_sensors: AvailableSensorDefinition, ) -> None: """It should get the available volume for a pipette.""" subject = get_pipette_view( @@ -354,6 +369,7 @@ def test_get_pipette_available_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ), "pipette-id-none": StaticPipetteConfig( min_volume=1, @@ -370,6 +386,7 @@ def test_get_pipette_available_volume( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ), }, ) @@ -465,6 +482,7 @@ def test_get_deck_point( def test_get_static_config( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the static pipette configuration that was set for the given pipette.""" config = StaticPipetteConfig( @@ -482,6 +500,7 @@ def test_get_static_config( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ) subject = get_pipette_view( @@ -513,6 +532,7 @@ def test_get_static_config( def test_get_nominal_tip_overlap( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the static pipette configuration that was set for the given pipette.""" config = StaticPipetteConfig( @@ -533,6 +553,7 @@ def test_get_nominal_tip_overlap( default_nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE), pipette_bounding_box_offsets=_SAMPLE_PIPETTE_BOUNDING_BOX_OFFSETS, lld_settings={}, + available_sensors=available_sensors, ) subject = get_pipette_view(static_config_by_id={"pipette-id": config}) @@ -934,6 +955,7 @@ def test_get_pipette_bounds_at_location( destination_position: Point, critical_point: Optional[CriticalPoint], pipette_bounds_result: Tuple[Point, Point, Point, Point], + available_sensors: AvailableSensorDefinition, ) -> None: """It should get the pipette's nozzle's bounds at the given location.""" subject = get_pipette_view( @@ -957,6 +979,7 @@ def test_get_pipette_bounds_at_location( bounding_nozzle_offsets=_SAMPLE_NOZZLE_BOUNDS_OFFSETS, pipette_bounding_box_offsets=bounding_box_offsets, lld_settings={}, + available_sensors=available_sensors, ) }, ) diff --git a/api/tests/opentrons/protocol_engine/state/test_tip_state.py b/api/tests/opentrons/protocol_engine/state/test_tip_state.py index abb408d7418..f10456549c8 100644 --- a/api/tests/opentrons/protocol_engine/state/test_tip_state.py +++ b/api/tests/opentrons/protocol_engine/state/test_tip_state.py @@ -22,6 +22,9 @@ ) from opentrons.types import DeckSlotName, Point from opentrons_shared_data.pipette.types import PipetteNameType +from opentrons_shared_data.pipette.pipette_definition import ( + AvailableSensorDefinition, +) from ..pipette_fixtures import ( NINETY_SIX_MAP, NINETY_SIX_COLS, @@ -32,6 +35,12 @@ _tip_rack_parameters = LabwareParameters.construct(isTiprack=True) # type: ignore[call-arg] +@pytest.fixture +def available_sensors() -> AvailableSensorDefinition: + """Provide a list of sensors.""" + return AvailableSensorDefinition(sensors=["pressure", "capacitive", "environment"]) + + @pytest.fixture def subject() -> TipStore: """Get a TipStore test subject.""" @@ -94,6 +103,7 @@ def test_get_next_tip_returns_none( load_labware_action: actions.SucceedCommandAction, subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should start at the first tip in the labware.""" subject.handle_action(load_labware_action) @@ -119,6 +129,7 @@ def test_get_next_tip_returns_none( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -144,6 +155,7 @@ def test_get_next_tip_returns_first_tip( subject: TipStore, input_tip_amount: int, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should start at the first tip in the labware.""" subject.handle_action(load_labware_action) @@ -177,6 +189,7 @@ def test_get_next_tip_returns_first_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -203,6 +216,7 @@ def test_get_next_tip_used_starting_tip( input_tip_amount: int, result_well_name: str, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should start searching at the given starting tip.""" subject.handle_action(load_labware_action) @@ -229,6 +243,7 @@ def test_get_next_tip_used_starting_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -270,6 +285,7 @@ def test_get_next_tip_skips_picked_up_tip( input_starting_tip: Optional[str], result_well_name: Optional[str], supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should get the next tip in the column if one has been picked up.""" subject.handle_action(load_labware_action) @@ -314,6 +330,7 @@ def test_get_next_tip_skips_picked_up_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -351,6 +368,7 @@ def test_get_next_tip_with_starting_tip( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the starting tip, and then the following tip after that.""" subject.handle_action(load_labware_action) @@ -377,6 +395,7 @@ def test_get_next_tip_with_starting_tip( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -418,6 +437,7 @@ def test_get_next_tip_with_starting_tip_8_channel( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the starting tip, and then the following tip after that.""" subject.handle_action(load_labware_action) @@ -444,6 +464,7 @@ def test_get_next_tip_with_starting_tip_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -488,6 +509,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the first tip of column 2 for the 8 channel after performing a single tip pickup on column 1.""" subject.handle_action(load_labware_action) @@ -514,6 +536,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -545,6 +568,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -589,6 +613,7 @@ def test_get_next_tip_with_starting_tip_out_of_tips( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the starting tip of H12 and then None after that.""" subject.handle_action(load_labware_action) @@ -615,6 +640,7 @@ def test_get_next_tip_with_starting_tip_out_of_tips( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -659,6 +685,7 @@ def test_get_next_tip_with_column_and_starting_tip( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should return the first tip in a column, taking starting tip into account.""" subject.handle_action(load_labware_action) @@ -685,6 +712,7 @@ def test_get_next_tip_with_column_and_starting_tip( back_left_corner_offset=Point(0, 0, 0), front_right_corner_offset=Point(0, 0, 0), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -708,6 +736,7 @@ def test_reset_tips( subject: TipStore, load_labware_action: actions.SucceedCommandAction, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """It should be able to reset tip tracking state.""" subject.handle_action(load_labware_action) @@ -734,6 +763,7 @@ def test_reset_tips( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) @@ -771,7 +801,9 @@ def get_result() -> str | None: def test_handle_pipette_config_action( - subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition + subject: TipStore, + supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + available_sensors: AvailableSensorDefinition, ) -> None: """Should add pipette channel to state.""" config_update = update_types.PipetteConfigUpdate( @@ -796,6 +828,7 @@ def test_handle_pipette_config_action( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -904,6 +937,7 @@ def test_active_channels( supported_tip_fixture: pipette_definition.SupportedTipsDefinition, nozzle_map: NozzleMap, expected_channels: int, + available_sensors: AvailableSensorDefinition, ) -> None: """Should update active channels after pipette configuration change.""" # Load pipette to update state @@ -929,6 +963,7 @@ def test_active_channels( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -961,6 +996,7 @@ def test_next_tip_uses_active_channels( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_action: actions.SucceedCommandAction, + available_sensors: AvailableSensorDefinition, ) -> None: """Test that tip tracking logic uses pipette's active channels.""" # Load labware @@ -989,6 +1025,7 @@ def test_next_tip_uses_active_channels( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -1059,6 +1096,7 @@ def test_next_tip_automatic_tip_tracking_with_partial_configurations( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_action: actions.SucceedCommandAction, + available_sensors: AvailableSensorDefinition, ) -> None: """Test tip tracking logic using multiple pipette configurations.""" # Load labware @@ -1087,6 +1125,7 @@ def test_next_tip_automatic_tip_tracking_with_partial_configurations( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -1211,6 +1250,7 @@ def test_next_tip_automatic_tip_tracking_tiprack_limits( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_action: actions.SucceedCommandAction, + available_sensors: AvailableSensorDefinition, ) -> None: """Test tip tracking logic to ensure once a tiprack is consumed it returns None when consuming tips using multiple pipette configurations.""" # Load labware @@ -1239,6 +1279,7 @@ def test_next_tip_automatic_tip_tracking_tiprack_limits( back_left_corner_offset=Point(x=1, y=2, z=3), front_right_corner_offset=Point(x=4, y=5, z=6), pipette_lld_settings={}, + available_sensors=available_sensors, ), ) subject.handle_action(