diff --git a/api/src/opentrons/hardware_control/backends/flex_protocol.py b/api/src/opentrons/hardware_control/backends/flex_protocol.py index 8b81d2c66ef..e5bc7ba1905 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 87f886f1c74..a7c30677910 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, @@ -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, diff --git a/api/src/opentrons/hardware_control/dev_types.py b/api/src/opentrons/hardware_control/dev_types.py index 575a5e612d9..981e95e114e 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, @@ -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): diff --git a/api/src/opentrons/hardware_control/ot3api.py b/api/src/opentrons/hardware_control/ot3api.py index de2de9ae9ab..af170484150 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 @@ -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")) diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 2f172c8cda2..8fc707541f0 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/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() diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py index d17ab43dd4f..f110bde928d 100644 --- a/api/src/opentrons/protocol_api/core/instrument.py +++ b/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: ... 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 90a8a05c6da..76d49b40557 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 @@ -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 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 66c33aae511..f55bf05c447 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 @@ -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 diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index a8d0a4b5765..7cc2d43bac2 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/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]]: diff --git a/api/src/opentrons/protocol_engine/commands/liquid_probe.py b/api/src/opentrons/protocol_engine/commands/liquid_probe.py index b99e6ac11b1..1bf58e8be26 100644 --- a/api/src/opentrons/protocol_engine/commands/liquid_probe.py +++ b/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." 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 6387bf5dcf1..4df6b0d4d77 100644 --- a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py +++ b/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, ) diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index d20b8665318..6418f50ee90 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/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: 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 1035649b7f5..9c03bed68b2 100644 --- a/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py +++ b/api/tests/opentrons/hardware_control/backends/test_ot3_controller.py @@ -374,6 +374,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 @@ -485,6 +487,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_home(controller, axes) @@ -729,6 +733,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, @@ -1413,3 +1420,34 @@ async def test_controller_move( assert position == expected_pos assert gear_position == gear_position + + +@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 8282f660a44..1caae624377 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -85,7 +85,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 9be08a0a71b..2d8685109ed 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) @@ -70,6 +80,7 @@ async def test_configure_for_volume_implementation( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 ab46c4b03e2..34b979901aa 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py +++ b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py @@ -16,9 +16,20 @@ PipetteLiquidNotFoundError, StallOrCollisionDetectedError, ) +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 @@ -41,6 +52,8 @@ ) from opentrons.protocol_engine.resources.model_utils import ModelUtils +from ..pipette_fixtures import get_default_nozzle_map + EitherImplementationType = Union[ Type[LiquidProbeImplementation], Type[TryLiquidProbeImplementation] ] @@ -49,6 +62,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), @@ -108,6 +127,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)) @@ -157,6 +178,41 @@ async def test_liquid_probe_implementation( state_view.pipettes.get_nozzle_configuration_supports_lld("abc") ).then_return(True) + 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) + timestamp = datetime(year=2020, month=1, day=2) decoy.when(model_utils.get_timestamp()).then_return(timestamp) @@ -190,6 +246,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" @@ -212,7 +270,40 @@ 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) decoy.when( await movement.move_to_well( pipette_id=pipette_id, @@ -281,6 +372,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" @@ -302,6 +395,40 @@ 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) with pytest.raises(TipNotAttachedError): await subject.execute(data) @@ -311,6 +438,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" @@ -329,6 +458,40 @@ async def test_liquid_probe_plunger_preparedness_checking( 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + 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) @@ -339,6 +502,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" @@ -358,6 +523,40 @@ 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) decoy.when( state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) ).then_return(True) @@ -379,6 +578,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" @@ -395,6 +596,40 @@ 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) decoy.when( await movement.check_for_valid_position( mount=MountType.LEFT, @@ -415,6 +650,8 @@ async def test_liquid_probe_stall( subject: EitherImplementation, params_type: EitherParamsType, 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)) @@ -429,6 +666,40 @@ async def test_liquid_probe_stall( decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id="abc")).then_return( 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=available_sensors, + ) + ) decoy.when( state_view.pipettes.get_nozzle_configuration_supports_lld("abc") ).then_return(True) 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 570666e9c98..a251c6aef1f 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py +++ b/api/tests/opentrons/protocol_engine/commands/test_load_pipette.py @@ -10,6 +10,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 @@ -28,6 +29,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", [ @@ -49,6 +56,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) @@ -76,6 +84,7 @@ async def test_load_pipette_implementation( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) decoy.when( @@ -118,6 +127,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) @@ -151,6 +161,7 @@ async def test_load_pipette_implementation_96_channel( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 3ee027c24c1..39208184754 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( @@ -161,6 +170,7 @@ def loaded_static_pipette_data( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 cbf7fa6174e..ae3d78d2230 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( @@ -72,11 +80,13 @@ def test_get_virtual_pipette_static_config( "drop_tip": -27.0, }, shaft_ul_per_mm=0.785, + 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( @@ -108,6 +118,7 @@ def test_configure_virtual_pipette_for_volume( "drop_tip": 90.5, }, shaft_ul_per_mm=0.785, + available_sensors=available_sensors, ) subject_instance.configure_virtual_pipette_for_volume( "my-pipette", 1, result1.model @@ -141,11 +152,13 @@ def test_configure_virtual_pipette_for_volume( "drop_tip": 90.5, }, shaft_ul_per_mm=0.785, + 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( @@ -177,6 +190,7 @@ def test_load_virtual_pipette_by_model_string( "drop_tip": -33.4, }, shaft_ul_per_mm=9.621, + available_sensors=AvailableSensorDefinition(sensors=[]), ) @@ -221,6 +235,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 { @@ -276,6 +291,7 @@ def pipette_dict( }, "plunger_positions": {"top": 100, "bottom": 20, "blow_out": 10, "drop_tip": 0}, "shaft_ul_per_mm": 5.0, + "available_sensors": available_sensors, } @@ -293,6 +309,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) @@ -324,6 +341,7 @@ def test_get_pipette_static_config( }, plunger_positions={"top": 100, "bottom": 20, "blow_out": 10, "drop_tip": 0}, shaft_ul_per_mm=5.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 abfb31f5f2a..b145458649d 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) @@ -2618,6 +2627,7 @@ def test_get_next_drop_tip_location( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 60c857e4911..e88f7886b81 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_store.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_store.py @@ -53,6 +53,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.""" @@ -190,6 +198,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() @@ -227,6 +236,7 @@ def test_handles_load_pipette( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) config_update = update_types.PipetteConfigUpdate( pipette_id="pipette-id", @@ -752,6 +762,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] @@ -786,6 +797,7 @@ def test_add_pipette_config( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) subject.handle_action( @@ -831,6 +843,7 @@ def test_add_pipette_config( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 14c43bf70f6..c3addf9f1d7 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py @@ -9,7 +9,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 @@ -58,6 +61,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, current_well: Optional[CurrentPipetteLocation] = None, @@ -269,6 +278,7 @@ def test_get_aspirated_volume(decoy: Decoy) -> 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( @@ -298,6 +308,7 @@ def test_get_pipette_working_volume( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) }, ) @@ -307,6 +318,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( @@ -336,6 +348,7 @@ def test_get_pipette_working_volume_raises_if_tip_volume_is_none( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) }, ) @@ -348,7 +361,9 @@ 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 + supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + decoy: Decoy, + available_sensors: AvailableSensorDefinition, ) -> None: """It should get the available volume for a pipette.""" stack = decoy.mock(cls=fluid_stack.FluidStack) @@ -385,6 +400,7 @@ def test_get_pipette_available_volume( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), "pipette-id-none": StaticPipetteConfig( min_volume=1, @@ -408,6 +424,7 @@ def test_get_pipette_available_volume( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), }, ) @@ -503,6 +520,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( @@ -527,6 +545,7 @@ def test_get_static_config( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) subject = get_pipette_view( @@ -558,6 +577,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( @@ -585,6 +605,7 @@ def test_get_nominal_tip_overlap( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ) subject = get_pipette_view(static_config_by_id={"pipette-id": config}) @@ -986,6 +1007,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( @@ -1016,6 +1038,7 @@ def test_get_pipette_bounds_at_location( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + 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 8abcc6a24e2..7a958a37e5f 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) @@ -126,6 +136,7 @@ def test_get_next_tip_returns_none( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -151,6 +162,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) @@ -191,6 +203,7 @@ def test_get_next_tip_returns_first_tip( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -217,6 +230,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) @@ -250,6 +264,7 @@ def test_get_next_tip_used_starting_tip( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -291,6 +306,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) @@ -342,6 +358,7 @@ def test_get_next_tip_skips_picked_up_tip( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -379,6 +396,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) @@ -412,6 +430,7 @@ def test_get_next_tip_with_starting_tip( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -453,6 +472,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) @@ -486,6 +506,7 @@ def test_get_next_tip_with_starting_tip_8_channel( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -530,6 +551,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) @@ -563,6 +585,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -601,6 +624,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -645,6 +669,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) @@ -678,6 +703,7 @@ def test_get_next_tip_with_starting_tip_out_of_tips( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -722,6 +748,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) @@ -755,6 +782,7 @@ def test_get_next_tip_with_column_and_starting_tip( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -778,6 +806,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) @@ -811,6 +840,7 @@ def test_reset_tips( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) @@ -848,7 +878,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( @@ -880,6 +912,7 @@ def test_handle_pipette_config_action( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -988,6 +1021,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 @@ -1020,6 +1054,7 @@ def test_active_channels( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -1052,6 +1087,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 @@ -1087,6 +1123,7 @@ def test_next_tip_uses_active_channels( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -1157,6 +1194,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 @@ -1192,6 +1230,7 @@ def test_next_tip_automatic_tip_tracking_with_partial_configurations( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action( @@ -1316,6 +1355,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 @@ -1351,6 +1391,7 @@ def test_next_tip_automatic_tip_tracking_tiprack_limits( "drop_tip": 20.0, }, shaft_ul_per_mm=5.0, + available_sensors=available_sensors, ), ) subject.handle_action(