diff --git a/api/src/opentrons/hardware_control/instruments/ot2/pipette.py b/api/src/opentrons/hardware_control/instruments/ot2/pipette.py index 52522257387..990dcd8579b 100644 --- a/api/src/opentrons/hardware_control/instruments/ot2/pipette.py +++ b/api/src/opentrons/hardware_control/instruments/ot2/pipette.py @@ -120,7 +120,9 @@ def __init__( ) self._nozzle_offset = self._config.nozzle_offset self._nozzle_manager = ( - nozzle_manager.NozzleConfigurationManager.build_from_config(self._config) + nozzle_manager.NozzleConfigurationManager.build_from_config( + self._config, self._valid_nozzle_maps + ) ) self._current_volume = 0.0 self._working_volume = float(self._liquid_class.max_volume) @@ -303,7 +305,9 @@ def reset_state(self) -> None: ) self._nozzle_manager = ( - nozzle_manager.NozzleConfigurationManager.build_from_config(self._config) + nozzle_manager.NozzleConfigurationManager.build_from_config( + self._config, self._valid_nozzle_maps + ) ) def reset_pipette_offset(self, mount: Mount, to_default: bool) -> None: @@ -531,77 +535,58 @@ def remove_tip(self) -> None: def has_tip(self) -> bool: return self._has_tip - def _get_matching_approved_nozzle_map(self) -> str: - for map_key in self._valid_nozzle_maps.maps.keys(): - if self._valid_nozzle_maps.maps[map_key] == list( - self._nozzle_manager.current_configuration.map_store.keys() - ): - return map_key - raise ValueError( - "Nozzle Configuration does not match any approved map layout for the current pipette." - ) - def get_pick_up_speed_by_configuration( self, config: PressFitPickUpTipConfiguration, ) -> float: - approved_map = None - for map_key in self._valid_nozzle_maps.maps.keys(): - if self._valid_nozzle_maps.maps[map_key] == list( - self._nozzle_manager.current_configuration.map_store.keys() - ): - approved_map = map_key - if approved_map is None: - raise ValueError( - "Pick up tip speed request error. Nozzle Configuration does not match any approved map layout for the current pipette." - ) - try: - return config.configuration_by_nozzle_map[approved_map][ - pip_types.PipetteTipType(self._liquid_class.max_volume).name - ].speed + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][pip_types.PipetteTipType(self._liquid_class.max_volume).name].speed except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.speed raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_pick_up_distance_by_configuration( self, config: PressFitPickUpTipConfiguration, ) -> float: - approved_map = self._get_matching_approved_nozzle_map() - try: - return config.configuration_by_nozzle_map[approved_map][ - pip_types.PipetteTipType(self._liquid_class.max_volume).name - ].distance + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][pip_types.PipetteTipType(self._liquid_class.max_volume).name].distance except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.distance raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_pick_up_current_by_configuration( self, config: PressFitPickUpTipConfiguration, ) -> float: - approved_map = self._get_matching_approved_nozzle_map() - try: - return config.configuration_by_nozzle_map[approved_map][ - pip_types.PipetteTipType(self._liquid_class.max_volume).name - ].current + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][pip_types.PipetteTipType(self._liquid_class.max_volume).name].current except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.current raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_nominal_tip_overlap_dictionary_by_configuration( @@ -613,21 +598,22 @@ def get_nominal_tip_overlap_dictionary_by_configuration( ): if not config: continue - approved_map = self._get_matching_approved_nozzle_map() try: - return config.configuration_by_nozzle_map[approved_map][ + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][ pip_types.PipetteTipType(self._liquid_class.max_volume).name ].tip_overlap_dictionary except KeyError: try: - default = config.configuration_by_nozzle_map[approved_map].get( - "default" - ) + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.tip_overlap_dictionary raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) except KeyError: # No valid key found for the approved nozzle map under this configuration - try the next diff --git a/api/src/opentrons/hardware_control/instruments/ot3/pipette.py b/api/src/opentrons/hardware_control/instruments/ot3/pipette.py index 070e856636c..bffbb29c864 100644 --- a/api/src/opentrons/hardware_control/instruments/ot3/pipette.py +++ b/api/src/opentrons/hardware_control/instruments/ot3/pipette.py @@ -103,7 +103,9 @@ def __init__( ) self._nozzle_offset = self._config.nozzle_offset self._nozzle_manager = ( - nozzle_manager.NozzleConfigurationManager.build_from_config(self._config) + nozzle_manager.NozzleConfigurationManager.build_from_config( + self._config, self._valid_nozzle_maps + ) ) self._current_volume = 0.0 self._working_volume = float(self._liquid_class.max_volume) @@ -265,7 +267,9 @@ def reset_state(self) -> None: self.get_nominal_tip_overlap_dictionary_by_configuration() ) self._nozzle_manager = ( - nozzle_manager.NozzleConfigurationManager.build_from_config(self._config) + nozzle_manager.NozzleConfigurationManager.build_from_config( + self._config, self._valid_nozzle_maps + ) ) def reset_pipette_offset(self, mount: OT3Mount, to_default: bool) -> None: @@ -669,18 +673,8 @@ def set_tip_type(self, tip_type: pip_types.PipetteTipType) -> None: ) self._working_volume = min(tip_type.value, self.liquid_class.max_volume) - def _get_matching_approved_nozzle_map(self) -> str: - for map_key in self._valid_nozzle_maps.maps.keys(): - if self._valid_nozzle_maps.maps[map_key] == list( - self._nozzle_manager.current_configuration.map_store.keys() - ): - return map_key - raise ValueError( - "Nozzle Configuration does not match any approved map layout for the current pipette." - ) - - def get_pick_up_configuration_for_tip_count( - self, count: int + def get_pick_up_configuration( # noqa: C901 + self, ) -> Union[CamActionPickUpTipConfiguration, PressFitPickUpTipConfiguration]: for config in ( self._config.pick_up_tip_configurations.press_fit, @@ -688,112 +682,91 @@ def get_pick_up_configuration_for_tip_count( ): if not config: continue - approved_map = self._get_matching_approved_nozzle_map() + config_values = None try: + config_values = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][self._active_tip_setting_name.name] + except KeyError: + try: + config_values = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") + if config_values is None: + raise KeyError( + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." + ) + except KeyError: + # No valid key found for the approved nozzle map under this configuration - try the next + continue + if config_values is not None: if isinstance(config, PressFitPickUpTipConfiguration) and all( [ - config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].speed, - config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].distance, - config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].current, + config_values.speed, + config_values.distance, + config_values.current, ] ): return config - elif ( - config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].current - is not None - ): + elif config_values.current is not None: return config - except KeyError: - try: - if isinstance(config, PressFitPickUpTipConfiguration) and all( - [ - config.configuration_by_nozzle_map[approved_map] - .get("default") - .speed, - config.configuration_by_nozzle_map[approved_map] - .get("default") - .distance, - config.configuration_by_nozzle_map[approved_map] - .get("default") - .current, - ] - ): - return config - elif ( - config.configuration_by_nozzle_map[approved_map] - .get("default") - .current - is not None - ): - return config - except KeyError: - # No valid key found for the approved nozzle map under this configuration - try the next - continue raise CommandPreconditionViolated( - message=f"No pick up tip configuration for {count} tips", + message="No valid pick up tip configuration values found in instrument definition.", ) def get_pick_up_speed_by_configuration( self, config: Union[CamActionPickUpTipConfiguration, PressFitPickUpTipConfiguration], ) -> float: - approved_map = self._get_matching_approved_nozzle_map() - try: - return config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].speed + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][self._active_tip_setting_name.name].speed except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.speed raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_pick_up_distance_by_configuration( self, config: Union[CamActionPickUpTipConfiguration, PressFitPickUpTipConfiguration], ) -> float: - approved_map = self._get_matching_approved_nozzle_map() - try: - return config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].distance + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][self._active_tip_setting_name.name].distance except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.distance raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_pick_up_current_by_configuration( self, config: Union[CamActionPickUpTipConfiguration, PressFitPickUpTipConfiguration], ) -> float: - approved_map = self._get_matching_approved_nozzle_map() - try: - return config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].current + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][self._active_tip_setting_name.name].current except KeyError: - default = config.configuration_by_nozzle_map[approved_map].get("default") + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.current raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) def get_nominal_tip_overlap_dictionary_by_configuration(self) -> Dict[str, float]: @@ -803,21 +776,20 @@ def get_nominal_tip_overlap_dictionary_by_configuration(self) -> Dict[str, float ): if not config: continue - approved_map = self._get_matching_approved_nozzle_map() try: - return config.configuration_by_nozzle_map[approved_map][ - self._active_tip_setting_name.name - ].tip_overlap_dictionary + return config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ][self._active_tip_setting_name.name].tip_overlap_dictionary except KeyError: try: - default = config.configuration_by_nozzle_map[approved_map].get( - "default" - ) + default = config.configuration_by_nozzle_map[ + self._nozzle_manager.current_configuration.valid_map_key + ].get("default") if default is not None: return default.tip_overlap_dictionary raise KeyError( - f"Default tip type configuration values do not exist for Nozzle Map {approved_map}." + f"Default tip type configuration values do not exist for Nozzle Map {self._nozzle_manager.current_configuration.valid_map_key}." ) except KeyError: # No valid key found for the approved nozzle map under this configuration - try the next diff --git a/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py b/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py index 601a8cf7ea0..8133eb081c9 100644 --- a/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py +++ b/api/src/opentrons/hardware_control/instruments/ot3/pipette_handler.py @@ -747,7 +747,7 @@ def plan_ht_pick_up_tip(self, tip_count: int) -> TipActionSpec: raise UnexpectedTipAttachError("pick_up_tip", instrument.name, mount.name) self._ihp_log.debug(f"Picking up tip on {mount.name}") - pick_up_config = instrument.get_pick_up_configuration_for_tip_count(tip_count) + pick_up_config = instrument.get_pick_up_configuration() if not isinstance(pick_up_config, CamActionPickUpTipConfiguration): raise CommandPreconditionViolated( f"Low-throughput pick up tip got wrong config for {instrument.name} on {mount.name}" @@ -788,7 +788,7 @@ def plan_lt_pick_up_tip( raise UnexpectedTipAttachError("pick_up_tip", instrument.name, mount.name) self._ihp_log.debug(f"Picking up tip on {mount.name}") - pick_up_config = instrument.get_pick_up_configuration_for_tip_count(tip_count) + pick_up_config = instrument.get_pick_up_configuration() if not isinstance(pick_up_config, PressFitPickUpTipConfiguration): raise CommandPreconditionViolated( f"Low-throughput pick up tip got wrong config for {instrument.name} on {mount.name}" diff --git a/api/src/opentrons/hardware_control/nozzle_manager.py b/api/src/opentrons/hardware_control/nozzle_manager.py index a25e5e57319..edea61165ec 100644 --- a/api/src/opentrons/hardware_control/nozzle_manager.py +++ b/api/src/opentrons/hardware_control/nozzle_manager.py @@ -9,6 +9,7 @@ from opentrons_shared_data.pipette.pipette_definition import ( PipetteGeometryDefinition, PipetteRowDefinition, + ValidNozzleMaps, ) from opentrons_shared_data.errors import ErrorCodes, GeneralError, PythonException @@ -98,6 +99,8 @@ class NozzleMap: # evaluate them to generate serdes code so please only use ordered dicts here map_store: Dict[str, Point] #: A map of all of the nozzles active in this configuration + valid_map_key: str + #: A key indicating which valid nozzle map from the pipette definition represents this configuration rows: Dict[str, List[str]] #: A map of all the rows active in this configuration columns: Dict[str, List[str]] @@ -222,6 +225,7 @@ def build( starting_nozzle: str, back_left_nozzle: str, front_right_nozzle: str, + valid_nozzle_maps: ValidNozzleMaps, ) -> "NozzleMap": try: back_left_row_index, back_left_column_index = _row_col_indices_for_nozzle( @@ -280,9 +284,20 @@ def build( f"Partial Nozzle Layouts may not be configured to contain more than {MAXIMUM_NOZZLE_COUNT} channels." ) + validated_map_key = None + for map_key in valid_nozzle_maps.maps.keys(): + if valid_nozzle_maps.maps[map_key] == list(map_store.keys()): + validated_map_key = map_key + break + if validated_map_key is None: + raise IncompatibleNozzleConfiguration( + "Attempted Nozzle Configuration does not match any approved map layout for the current pipette." + ) + return cls( starting_nozzle=starting_nozzle, map_store=map_store, + valid_map_key=validated_map_key, rows=rows, full_instrument_map_store=physical_nozzles, full_instrument_rows=physical_rows, @@ -313,15 +328,17 @@ def __init__( class NozzleConfigurationManager: def __init__( - self, - nozzle_map: NozzleMap, + self, nozzle_map: NozzleMap, valid_nozzle_maps: ValidNozzleMaps ) -> None: self._physical_nozzle_map = nozzle_map self._current_nozzle_configuration = nozzle_map + self._valid_nozzle_maps = valid_nozzle_maps @classmethod def build_from_config( - cls, pipette_geometry: PipetteGeometryDefinition + cls, + pipette_geometry: PipetteGeometryDefinition, + valid_nozzle_maps: ValidNozzleMaps, ) -> "NozzleConfigurationManager": sorted_nozzle_map = OrderedDict( ( @@ -346,8 +363,9 @@ def build_from_config( starting_nozzle=back_left, back_left_nozzle=back_left, front_right_nozzle=front_right, + valid_nozzle_maps=valid_nozzle_maps, ) - return cls(starting_nozzle_config) + return cls(starting_nozzle_config, valid_nozzle_maps) @property def starting_nozzle_offset(self) -> Point: @@ -380,6 +398,7 @@ def update_nozzle_configuration( starting_nozzle=starting_nozzle or back_left_nozzle, back_left_nozzle=back_left_nozzle, front_right_nozzle=front_right_nozzle, + valid_nozzle_maps=self._valid_nozzle_maps, ) def get_tip_count(self) -> int: 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 6b322e2144e..b237cdc412e 100644 --- a/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py +++ b/api/src/opentrons/protocol_engine/resources/pipette_data_provider.py @@ -10,7 +10,6 @@ pipette_definition, ) - from opentrons.hardware_control.dev_types import PipetteDict from opentrons.hardware_control.nozzle_manager import ( NozzleConfigurationManager, @@ -63,7 +62,15 @@ def configure_virtual_pipette_nozzle_layout( config = self._get_virtual_pipette_full_config_by_model_string( pipette_model_string ) - new_nozzle_manager = NozzleConfigurationManager.build_from_config(config) + + valid_nozzle_maps = load_pipette_data.load_valid_nozzle_maps( + config.pipette_type, + config.channels, + config.version, + ) + new_nozzle_manager = NozzleConfigurationManager.build_from_config( + config, valid_nozzle_maps + ) if back_left_nozzle and front_right_nozzle: new_nozzle_manager.update_nozzle_configuration( back_left_nozzle, front_right_nozzle, starting_nozzle @@ -155,7 +162,9 @@ def _get_virtual_pipette_static_config_by_model( # noqa: C901 pipette_model.pipette_channels, pipette_model.pipette_version, ) - nozzle_manager = NozzleConfigurationManager.build_from_config(config) + nozzle_manager = NozzleConfigurationManager.build_from_config( + config, valid_nozzle_maps + ) tip_overlap_dict_for_tip_type = None for configuration in ( @@ -179,15 +188,15 @@ def _get_virtual_pipette_static_config_by_model( # noqa: C901 if configuration is not None: try: tip_overlap_dict_for_tip_type = ( - configuration.configuration_by_nozzle_map[approved_map][ - tip_type.name - ].tip_overlap_dictionary + configuration.configuration_by_nozzle_map[ + nozzle_manager.current_configuration.valid_map_key + ][tip_type.name].tip_overlap_dictionary ) break except KeyError: try: default = configuration.configuration_by_nozzle_map[ - approved_map + nozzle_manager.current_configuration.valid_map_key ].get("default") if default is not None: tip_overlap_dict_for_tip_type = ( diff --git a/api/tests/opentrons/hardware_control/test_pipette_handler.py b/api/tests/opentrons/hardware_control/test_pipette_handler.py index 1134a09b807..7080b14ef06 100644 --- a/api/tests/opentrons/hardware_control/test_pipette_handler.py +++ b/api/tests/opentrons/hardware_control/test_pipette_handler.py @@ -160,9 +160,7 @@ def test_plan_check_pick_up_tip_with_presses_argument_ot3( increment = 1 decoy.when(mock_pipette_ot3.has_tip).then_return(False) - decoy.when( - mock_pipette_ot3.get_pick_up_configuration_for_tip_count(channels) - ).then_return( + decoy.when(mock_pipette_ot3.get_pick_up_configuration()).then_return( CamActionPickUpTipConfiguration( distance=10, speed=5.5,