diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py index 920a7b526185..c46549f69eff 100644 --- a/api/src/opentrons/protocol_api/core/engine/instrument.py +++ b/api/src/opentrons/protocol_api/core/engine/instrument.py @@ -1010,15 +1010,15 @@ def get_hardware_state(self) -> PipetteDict: return self._sync_hardware_api.get_attached_instrument(self.get_mount()) # type: ignore[no-any-return] def get_channels(self) -> int: - return self._engine_client.state.tips.get_pipette_channels(self._pipette_id) + return self._engine_client.state.pipettes.get_channels(self._pipette_id) def get_active_channels(self) -> int: - return self._engine_client.state.tips.get_pipette_active_channels( - self._pipette_id - ) + return self._engine_client.state.pipettes.get_active_channels(self._pipette_id) def get_nozzle_map(self) -> NozzleMapInterface: - return self._engine_client.state.tips.get_pipette_nozzle_map(self._pipette_id) + return self._engine_client.state.pipettes.get_nozzle_configuration( + self._pipette_id + ) def has_tip(self) -> bool: return ( diff --git a/api/src/opentrons/protocol_engine/commands/get_next_tip.py b/api/src/opentrons/protocol_engine/commands/get_next_tip.py index 9e0cf3b50cf0..236c90001a37 100644 --- a/api/src/opentrons/protocol_engine/commands/get_next_tip.py +++ b/api/src/opentrons/protocol_engine/commands/get_next_tip.py @@ -73,8 +73,8 @@ async def execute(self, params: GetNextTipParams) -> SuccessData[GetNextTipResul pipette_id = params.pipetteId starting_tip_name = params.startingTipWell - num_tips = self._state_view.tips.get_pipette_active_channels(pipette_id) - nozzle_map = self._state_view.tips.get_pipette_nozzle_map(pipette_id) + num_tips = self._state_view.pipettes.get_active_channels(pipette_id) + nozzle_map = self._state_view.pipettes.get_nozzle_configuration(pipette_id) if ( starting_tip_name is not None diff --git a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py index 0d35312364ab..f9463573a5e5 100644 --- a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py +++ b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py @@ -121,6 +121,12 @@ async def execute( labware_id = params.labwareId well_name = params.wellName + tips_to_mark_as_used = self._state_view.tips.compute_tips_to_mark_as_used( + labware_id=labware_id, + well_name=well_name, + nozzle_map=self._state_view.pipettes.get_nozzle_configuration(pipette_id), + ) + well_location = self._state_view.geometry.convert_pick_up_tip_well_location( well_location=params.wellLocation ) @@ -152,7 +158,7 @@ async def execute( ) .set_fluid_empty(pipette_id=pipette_id, clean_tip=True) .mark_tips_as_used( - pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + labware_id=labware_id, well_names=tips_to_mark_as_used ) ) state_update = ( @@ -160,7 +166,7 @@ async def execute( update_types.StateUpdate(), move_result.state_update ) .mark_tips_as_used( - pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + labware_id=labware_id, well_names=tips_to_mark_as_used ) .set_fluid_unknown(pipette_id=pipette_id) ) @@ -186,7 +192,7 @@ async def execute( tip_geometry=tip_geometry, ) .mark_tips_as_used( - pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + labware_id=labware_id, well_names=tips_to_mark_as_used ) .set_fluid_empty(pipette_id=pipette_id, clean_tip=True) .set_pipette_ready_to_aspirate( diff --git a/api/src/opentrons/protocol_engine/commands/seal_pipette_to_tip.py b/api/src/opentrons/protocol_engine/commands/seal_pipette_to_tip.py index bd8f679d5fb6..29257739626f 100644 --- a/api/src/opentrons/protocol_engine/commands/seal_pipette_to_tip.py +++ b/api/src/opentrons/protocol_engine/commands/seal_pipette_to_tip.py @@ -249,7 +249,7 @@ async def execute( # Begin relative pickup steps for the resin tips - channels = self._state_view.tips.get_pipette_active_channels(pipette_id) + channels = self._state_view.pipettes.get_active_channels(pipette_id) mount = self._state_view.pipettes.get_mount(pipette_id) tip_pick_up_params = params.tipPickUpParams diff --git a/api/src/opentrons/protocol_engine/commands/unsafe/unsafe_drop_tip_in_place.py b/api/src/opentrons/protocol_engine/commands/unsafe/unsafe_drop_tip_in_place.py index 33542d37f60c..fcea7075473c 100644 --- a/api/src/opentrons/protocol_engine/commands/unsafe/unsafe_drop_tip_in_place.py +++ b/api/src/opentrons/protocol_engine/commands/unsafe/unsafe_drop_tip_in_place.py @@ -81,8 +81,7 @@ async def execute( pipette_id=params.pipetteId, home_after=params.homeAfter, ignore_plunger=( - self._state_view.tips.get_pipette_active_channels(params.pipetteId) - == 96 + self._state_view.pipettes.get_active_channels(params.pipetteId) == 96 ), ) diff --git a/api/src/opentrons/protocol_engine/execution/movement.py b/api/src/opentrons/protocol_engine/execution/movement.py index be8bbbb8de2f..4e0e5c430282 100644 --- a/api/src/opentrons/protocol_engine/execution/movement.py +++ b/api/src/opentrons/protocol_engine/execution/movement.py @@ -105,9 +105,7 @@ async def move_to_well( self._hs_movement_flagger.raise_if_movement_restricted( hs_movement_restrictors=hs_movement_restrictors, destination_slot=dest_slot_int, - is_multi_channel=( - self._state_store.tips.get_pipette_channels(pipette_id) > 1 - ), + is_multi_channel=(self._state_store.pipettes.get_channels(pipette_id) > 1), destination_is_tip_rack=self._state_store.labware.is_tiprack(labware_id), ) @@ -204,9 +202,7 @@ async def move_to_addressable_area( self._hs_movement_flagger.raise_if_movement_restricted( hs_movement_restrictors=hs_movement_restrictors, destination_slot=dest_slot_int, - is_multi_channel=( - self._state_store.tips.get_pipette_channels(pipette_id) > 1 - ), + is_multi_channel=(self._state_store.pipettes.get_channels(pipette_id) > 1), destination_is_tip_rack=False, ) diff --git a/api/src/opentrons/protocol_engine/execution/queue_worker.py b/api/src/opentrons/protocol_engine/execution/queue_worker.py index 015adf085c90..2f00c0ed2082 100644 --- a/api/src/opentrons/protocol_engine/execution/queue_worker.py +++ b/api/src/opentrons/protocol_engine/execution/queue_worker.py @@ -72,7 +72,14 @@ async def _run_commands(self) -> None: try: await self._command_executor.execute(command_id=command_id) except BaseException: - log.exception("Unhandled failure in command executor") + log.exception( + # The state can tear if e.g. we've finished updating PipetteStore, + # but the exception came before we could update LabwareStore. Or + # the exception could have interrupted updating a single store. + "Unhandled failure in command executor." + " This is a bug in opentrons.protocol_engine" + " and has probably left the ProtocolEngine in a torn state." + ) raise # Yield to the event loop in case we're executing a long sequence of commands # that never yields internally. For example, a long sequence of comment commands. diff --git a/api/src/opentrons/protocol_engine/state/_well_math.py b/api/src/opentrons/protocol_engine/state/_well_math.py index 2d0998580f59..5a9d19a621c7 100644 --- a/api/src/opentrons/protocol_engine/state/_well_math.py +++ b/api/src/opentrons/protocol_engine/state/_well_math.py @@ -63,7 +63,7 @@ def wells_covered_dense( # noqa: C901 row_downsample = len(target_wells_by_column[0]) // 8 if column_downsample < 1 or row_downsample < 1: raise InvalidStoredData( - "This labware cannot be used wells_covered_dense because it is less dense than an SBS 96 standard" + "This labware cannot be used with wells_covered_dense() because it is less dense than an SBS 96 standard" ) for nozzle_column in range(len(nozzle_map.columns)): @@ -126,7 +126,7 @@ def wells_covered_sparse( # noqa: C901 row_upsample = 8 // len(target_wells_by_column[0]) if column_upsample < 1 or row_upsample < 1: raise InvalidStoredData( - "This labware cannot be used with wells_covered_sparse because it is more dense than an SBS 96 standard." + "This labware cannot be used with wells_covered_sparse() because it is more dense than an SBS 96 standard." ) for nozzle_column in range(max(1, len(nozzle_map.columns) // column_upsample)): for nozzle_row in range(max(1, len(nozzle_map.rows) // row_upsample)): diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index 518753f822eb..189d312d76e8 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -650,6 +650,10 @@ def get_channels(self, pipette_id: str) -> int: """Return the max channels of the pipette.""" return self.get_config(pipette_id).channels + def get_active_channels(self, pipette_id: str) -> int: + """Get the number of channels being used in the given pipette's configuration.""" + return self.get_nozzle_configuration(pipette_id).tip_count + def get_minimum_volume(self, pipette_id: str) -> float: """Return the given pipette's minimum volume.""" return self.get_config(pipette_id).min_volume @@ -727,6 +731,10 @@ def get_primary_nozzle(self, pipette_id: str) -> str: nozzle_map = self._state.nozzle_configuration_by_id[pipette_id] return nozzle_map.starting_nozzle + def get_nozzle_configurations(self) -> Dict[str, NozzleMap]: + """Get the nozzle maps of all pipettes, keyed by pipette ID.""" + return self._state.nozzle_configuration_by_id.copy() + def get_nozzle_configuration(self, pipette_id: str) -> NozzleMap: """Get the nozzle map of the pipette.""" return self._state.nozzle_configuration_by_id[pipette_id] diff --git a/api/src/opentrons/protocol_engine/state/tips.py b/api/src/opentrons/protocol_engine/state/tips.py index 80ce7d083508..74445603ffbb 100644 --- a/api/src/opentrons/protocol_engine/state/tips.py +++ b/api/src/opentrons/protocol_engine/state/tips.py @@ -2,7 +2,7 @@ from dataclasses import dataclass from enum import Enum -from typing import Dict, Optional, List, Union +from typing import Dict, Iterable, Optional, List, Union from opentrons.types import NozzleMapInterface from opentrons.protocol_engine.state import update_types @@ -14,36 +14,22 @@ from opentrons.hardware_control.nozzle_manager import NozzleMap -class TipRackWellState(Enum): +class _TipRackWellState(Enum): """The state of a single tip in a tip rack's well.""" CLEAN = "clean" USED = "used" -TipRackStateByWellName = Dict[str, TipRackWellState] - - -# todo(mm, 2024-10-10): This info is duplicated between here and PipetteState because -# TipStore is using it to compute which tips a PickUpTip removes from the tip rack, -# given the pipette's current nozzle map. We could avoid this duplication by moving the -# computation to TipView, calling it from PickUpTipImplementation, and passing the -# precomputed list of wells to TipStore. -@dataclass -class _PipetteInfo: - channels: int - active_channels: int - nozzle_map: NozzleMap +_TipRackStateByWellName = Dict[str, _TipRackWellState] @dataclass class TipState: """State of all tips.""" - tips_by_labware_id: Dict[str, TipRackStateByWellName] - column_by_labware_id: Dict[str, List[List[str]]] - - pipette_info_by_pipette_id: Dict[str, _PipetteInfo] + tips_by_labware_id: Dict[str, _TipRackStateByWellName] + columns_by_labware_id: Dict[str, List[List[str]]] class TipStore(HasState[TipState], HandlesActions): @@ -55,8 +41,7 @@ def __init__(self) -> None: """Initialize a liquid store and its state.""" self._state = TipState( tips_by_labware_id={}, - column_by_labware_id={}, - pipette_info_by_pipette_id={}, + columns_by_labware_id={}, ) def handle_action(self, action: Action) -> None: @@ -70,44 +55,25 @@ def handle_action(self, action: Action) -> None: for well_name in self._state.tips_by_labware_id[labware_id].keys(): self._state.tips_by_labware_id[labware_id][ well_name - ] = TipRackWellState.CLEAN + ] = _TipRackWellState.CLEAN def _handle_state_update(self, state_update: update_types.StateUpdate) -> None: - if state_update.pipette_config != update_types.NO_CHANGE: - self._state.pipette_info_by_pipette_id[ - state_update.pipette_config.pipette_id - ] = _PipetteInfo( - channels=state_update.pipette_config.config.channels, - active_channels=state_update.pipette_config.config.channels, - nozzle_map=state_update.pipette_config.config.nozzle_map, - ) - if state_update.tips_used != update_types.NO_CHANGE: self._set_used_tips( - pipette_id=state_update.tips_used.pipette_id, labware_id=state_update.tips_used.labware_id, - well_name=state_update.tips_used.well_name, - ) - - if state_update.pipette_nozzle_map != update_types.NO_CHANGE: - pipette_info = self._state.pipette_info_by_pipette_id[ - state_update.pipette_nozzle_map.pipette_id - ] - pipette_info.active_channels = ( - state_update.pipette_nozzle_map.nozzle_map.tip_count + well_names=state_update.tips_used.well_names, ) - pipette_info.nozzle_map = state_update.pipette_nozzle_map.nozzle_map if state_update.loaded_labware != update_types.NO_CHANGE: labware_id = state_update.loaded_labware.labware_id definition = state_update.loaded_labware.definition if definition.parameters.isTiprack: self._state.tips_by_labware_id[labware_id] = { - well_name: TipRackWellState.CLEAN + well_name: _TipRackWellState.CLEAN for column in definition.ordering for well_name in column } - self._state.column_by_labware_id[labware_id] = [ + self._state.columns_by_labware_id[labware_id] = [ column for column in definition.ordering ] if state_update.batch_loaded_labware != update_types.NO_CHANGE: @@ -117,20 +83,18 @@ def _handle_state_update(self, state_update: update_types.StateUpdate) -> None: ] if definition.parameters.isTiprack: self._state.tips_by_labware_id[labware_id] = { - well_name: TipRackWellState.CLEAN + well_name: _TipRackWellState.CLEAN for column in definition.ordering for well_name in column } - self._state.column_by_labware_id[labware_id] = [ + self._state.columns_by_labware_id[labware_id] = [ column for column in definition.ordering ] - def _set_used_tips(self, pipette_id: str, well_name: str, labware_id: str) -> None: - columns = self._state.column_by_labware_id.get(labware_id, []) - wells = self._state.tips_by_labware_id.get(labware_id, {}) - nozzle_map = self._state.pipette_info_by_pipette_id[pipette_id].nozzle_map - for well in wells_covered_dense(nozzle_map, well_name, columns): - wells[well] = TipRackWellState.USED + def _set_used_tips(self, labware_id: str, well_names: Iterable[str]) -> None: + well_states = self._state.tips_by_labware_id.get(labware_id, {}) + for well_name in well_names: + well_states[well_name] = _TipRackWellState.USED class TipView: @@ -155,7 +119,7 @@ def get_next_tip( # noqa: C901 ) -> Optional[str]: """Get the next available clean tip. Does not support use of a starting tip if the pipette used is in a partial configuration.""" wells = self._state.tips_by_labware_id.get(labware_id, {}) - columns = self._state.column_by_labware_id.get(labware_id, []) + columns = self._state.columns_by_labware_id.get(labware_id, []) # TODO(sf): I'm pretty sure this can be replaced with wells_covered_96 but I'm not quite sure how def _identify_tip_cluster( @@ -202,9 +166,9 @@ def _identify_tip_cluster( def _validate_tip_cluster( active_columns: int, active_rows: int, tip_cluster: List[str] ) -> Union[str, int, None]: - if not any(wells[well] == TipRackWellState.USED for well in tip_cluster): + if not any(wells[well] == _TipRackWellState.USED for well in tip_cluster): return tip_cluster[0] - elif all(wells[well] == TipRackWellState.USED for well in tip_cluster): + elif all(wells[well] == _TipRackWellState.USED for well in tip_cluster): return None else: # In the case of an 8ch pipette where a column has mixed state tips we may simply progress to the next column in our search @@ -224,12 +188,12 @@ def _validate_tip_cluster( tip_cluster[(active_rows - 1) + (i * active_rows)] ) if all( - wells[well] == TipRackWellState.USED + wells[well] == _TipRackWellState.USED for well in tip_cluster_final_column ): return None elif all( - wells[well] == TipRackWellState.USED + wells[well] == _TipRackWellState.USED for well in tip_cluster_final_row ): return None @@ -386,7 +350,9 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]: starting_column_index = idx for column in columns[starting_column_index:]: - if not any(wells[well] == TipRackWellState.USED for well in column): + if not any( + wells[well] == _TipRackWellState.USED for well in column + ): return column[0] elif num_tips == len(wells.keys()): # Get next tips for 96 channel @@ -394,7 +360,7 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]: return None if not any( - tip_state == TipRackWellState.USED for tip_state in wells.values() + tip_state == _TipRackWellState.USED for tip_state in wells.values() ): return next(iter(wells)) @@ -403,29 +369,10 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]: wells = _drop_wells_before_starting_tip(wells, starting_tip_name) for well_name, tip_state in wells.items(): - if tip_state == TipRackWellState.CLEAN: + if tip_state == _TipRackWellState.CLEAN: return well_name return None - def get_pipette_channels(self, pipette_id: str) -> int: - """Return the given pipette's number of channels.""" - return self._state.pipette_info_by_pipette_id[pipette_id].channels - - def get_pipette_active_channels(self, pipette_id: str) -> int: - """Get the number of channels being used in the given pipette's configuration.""" - return self._state.pipette_info_by_pipette_id[pipette_id].active_channels - - def get_pipette_nozzle_map(self, pipette_id: str) -> NozzleMap: - """Get the current nozzle map the given pipette's configuration.""" - return self._state.pipette_info_by_pipette_id[pipette_id].nozzle_map - - def get_pipette_nozzle_maps(self) -> Dict[str, NozzleMap]: - """Get current nozzle maps keyed by pipette id.""" - return { - pipette_id: pipette_info.nozzle_map - for pipette_id, pipette_info in self._state.pipette_info_by_pipette_id.items() - } - def has_clean_tip(self, labware_id: str, well_name: str) -> bool: """Get whether a well in a labware has a clean tip. @@ -440,15 +387,31 @@ def has_clean_tip(self, labware_id: str, well_name: str) -> bool: tip_rack = self._state.tips_by_labware_id.get(labware_id) well_state = tip_rack.get(well_name) if tip_rack else None - return well_state == TipRackWellState.CLEAN + return well_state == _TipRackWellState.CLEAN + + def compute_tips_to_mark_as_used( + self, labware_id: str, well_name: str, nozzle_map: NozzleMap + ) -> list[str]: + """Compute which tips a hypothetical tip pickup should mark as "used". + + Params: + labware_id: The labware ID of the tip rack. + well_name: The single target well of the tip pickup. + nozzle_map: The nozzle configuration that the pipette will use for the pickup. + + Returns: + The well names of all the tips that the operation will use. + """ + columns = self._state.columns_by_labware_id.get(labware_id, []) + return list(wells_covered_dense(nozzle_map, well_name, columns)) def _drop_wells_before_starting_tip( - wells: TipRackStateByWellName, starting_tip_name: str -) -> TipRackStateByWellName: + wells: _TipRackStateByWellName, starting_tip_name: str +) -> _TipRackStateByWellName: """Drop any wells that come before the starting tip and return the remaining ones after.""" seen_starting_well = False - remaining_wells = {} + remaining_wells: dict[str, _TipRackWellState] = {} for well_name, tip_state in wells.items(): if well_name == starting_tip_name: seen_starting_well = True diff --git a/api/src/opentrons/protocol_engine/state/update_types.py b/api/src/opentrons/protocol_engine/state/update_types.py index 30fee20956f0..b2c145ac946a 100644 --- a/api/src/opentrons/protocol_engine/state/update_types.py +++ b/api/src/opentrons/protocol_engine/state/update_types.py @@ -127,6 +127,7 @@ class BatchLabwareLocationUpdate: """The new offsets of each id.""" +# todo(mm, 2025-04-28): Combine with BatchLoadedLabwareUpdate. @dataclasses.dataclass class LoadedLabwareUpdate: """An update that loads a new labware.""" @@ -250,16 +251,14 @@ class PipetteAspirateReadyUpdate: class TipsUsedUpdate: """Represents an update that marks tips in a tip rack as used.""" - pipette_id: str - """The pipette that did the tip pickup.""" - labware_id: str + """The labware ID of the tip rack.""" - well_name: str - """The well that the pipette's primary nozzle targeted. + well_names: list[str] + """The exact wells in the tip rack that should be marked as used. - Wells in addition to this one will also be marked as used, depending on the - pipette's nozzle layout. + This is the *full* list, which is probably more than what appeared in the pickUpTip + command's params, for multi-channel reasons. """ @@ -701,13 +700,9 @@ def update_pipette_tip_state( ) return self - def mark_tips_as_used( - self: Self, pipette_id: str, labware_id: str, well_name: str - ) -> Self: + def mark_tips_as_used(self: Self, labware_id: str, well_names: list[str]) -> Self: """Mark tips in a tip rack as used. See `TipsUsedUpdate`.""" - self.tips_used = TipsUsedUpdate( - pipette_id=pipette_id, labware_id=labware_id, well_name=well_name - ) + self.tips_used = TipsUsedUpdate(labware_id=labware_id, well_names=well_names) return self def set_liquid_loaded( diff --git a/api/src/opentrons/protocol_runner/run_orchestrator.py b/api/src/opentrons/protocol_runner/run_orchestrator.py index b45d8b9db94b..92606b1e85f4 100644 --- a/api/src/opentrons/protocol_runner/run_orchestrator.py +++ b/api/src/opentrons/protocol_runner/run_orchestrator.py @@ -429,7 +429,7 @@ def get_deck_type(self) -> DeckType: def get_nozzle_maps(self) -> Mapping[str, NozzleMapInterface]: """Get current nozzle maps keyed by pipette id.""" - return self._protocol_engine.state_view.tips.get_pipette_nozzle_maps() + return self._protocol_engine.state_view.pipettes.get_nozzle_configurations() def get_tip_attached(self) -> Dict[str, bool]: """Get current tip state keyed by pipette id.""" diff --git a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py index 97b2e49b1544..676e2dda6dd8 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py @@ -463,9 +463,7 @@ def test_drop_tip_no_location( labware_id="labware-id", engine_client=mock_engine_client, ) - decoy.when( - mock_engine_client.state.tips.get_pipette_channels("abc123") - ).then_return(8) + decoy.when(mock_engine_client.state.pipettes.get_channels("abc123")).then_return(8) subject.drop_tip(location=None, well_core=well_core, home_after=True) @@ -517,9 +515,7 @@ def test_drop_tip_with_location( ).then_return( (WellLocation(origin=WellOrigin.TOP, offset=WellOffset(x=3, y=2, z=1)), False) ) - decoy.when( - mock_engine_client.state.tips.get_pipette_channels("abc123") - ).then_return(8) + decoy.when(mock_engine_client.state.pipettes.get_channels("abc123")).then_return(8) decoy.when(mock_engine_client.state.labware.is_tiprack("labware-id")).then_return( True ) @@ -599,9 +595,7 @@ def test_drop_tip_in_waste_chute( waste_chute = decoy.mock(cls=WasteChute) decoy.when(waste_chute.offset).then_return(DisposalOffset(x=4, y=5, z=6)) - decoy.when( - mock_engine_client.state.tips.get_pipette_channels("abc123") - ).then_return(96) + decoy.when(mock_engine_client.state.pipettes.get_channels("abc123")).then_return(96) subject.drop_tip_in_disposal_location( waste_chute, home_after=True, alternate_tip_drop=True @@ -1306,9 +1300,7 @@ def test_get_channels( ) -> None: """It should get the pipette's number of channels.""" decoy.when( - mock_engine_client.state.tips.get_pipette_channels( - pipette_id=subject.pipette_id - ) + mock_engine_client.state.pipettes.get_channels(pipette_id=subject.pipette_id) ).then_return(42) assert subject.get_channels() == 42 @@ -1639,7 +1631,7 @@ def test_is_tip_tracking_available( ) -> None: """It should return whether tip tracking is available based on nozzle configuration.""" decoy.when( - mock_engine_client.state.tips.get_pipette_channels(subject.pipette_id) + mock_engine_client.state.pipettes.get_channels(subject.pipette_id) ).then_return(pipette_channels) decoy.when( mock_engine_client.state.pipettes.get_nozzle_layout_type(subject.pipette_id) diff --git a/api/tests/opentrons/protocol_engine/commands/test_get_next_tip.py b/api/tests/opentrons/protocol_engine/commands/test_get_next_tip.py index 4221cae864db..94457aeea394 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_get_next_tip.py +++ b/api/tests/opentrons/protocol_engine/commands/test_get_next_tip.py @@ -25,8 +25,8 @@ async def test_get_next_tip_implementation( ) mock_nozzle_map = decoy.mock(cls=NozzleMap) - decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42) - decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return( + decoy.when(state_view.pipettes.get_active_channels("abc")).then_return(42) + decoy.when(state_view.pipettes.get_nozzle_configuration("abc")).then_return( mock_nozzle_map ) decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL) @@ -60,8 +60,8 @@ async def test_get_next_tip_implementation_multiple_tip_racks( ) mock_nozzle_map = decoy.mock(cls=NozzleMap) - decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42) - decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return( + decoy.when(state_view.pipettes.get_active_channels("abc")).then_return(42) + decoy.when(state_view.pipettes.get_nozzle_configuration("abc")).then_return( mock_nozzle_map ) decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL) @@ -95,8 +95,8 @@ async def test_get_next_tip_implementation_no_tips( ) mock_nozzle_map = decoy.mock(cls=NozzleMap) - decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42) - decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return( + decoy.when(state_view.pipettes.get_active_channels("abc")).then_return(42) + decoy.when(state_view.pipettes.get_nozzle_configuration("abc")).then_return( mock_nozzle_map ) decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.FULL) @@ -124,8 +124,8 @@ async def test_get_next_tip_implementation_partial_with_starting_tip( ) mock_nozzle_map = decoy.mock(cls=NozzleMap) - decoy.when(state_view.tips.get_pipette_active_channels("abc")).then_return(42) - decoy.when(state_view.tips.get_pipette_nozzle_map("abc")).then_return( + decoy.when(state_view.pipettes.get_active_channels("abc")).then_return(42) + decoy.when(state_view.pipettes.get_nozzle_configuration("abc")).then_return( mock_nozzle_map ) decoy.when(mock_nozzle_map.configuration).then_return(NozzleConfigurationType.ROW) diff --git a/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py b/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py index 1c33e023069f..6b9952f83871 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py +++ b/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py @@ -78,6 +78,15 @@ async def test_success( ) ).then_return(TipGeometry(length=42, diameter=5, volume=300)) + decoy.when(state_view.pipettes.get_nozzle_configuration("pipette-id")).then_return( + sentinel.nozzle_configuration + ) + decoy.when( + state_view.tips.compute_tips_to_mark_as_used( + "labware-id", "A3", sentinel.nozzle_configuration + ) + ).then_return(sentinel.tips_to_mark_as_used) + result = await subject.execute( PickUpTipParams( pipetteId="pipette-id", @@ -105,7 +114,7 @@ async def test_success( tip_geometry=TipGeometry(length=42, diameter=5, volume=300), ), tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="labware-id", well_name="A3" + labware_id="labware-id", well_names=sentinel.tips_to_mark_as_used ), pipette_aspirated_fluid=update_types.PipetteEmptyFluidUpdate( pipette_id="pipette-id", clean_tip=True @@ -165,6 +174,15 @@ async def test_tip_physically_missing_error( decoy.when(model_utils.generate_id()).then_return(error_id) decoy.when(model_utils.get_timestamp()).then_return(error_created_at) + decoy.when(state_view.pipettes.get_nozzle_configuration(pipette_id)).then_return( + sentinel.nozzle_configuration + ) + decoy.when( + state_view.tips.compute_tips_to_mark_as_used( + labware_id, well_name, sentinel.nozzle_configuration + ) + ).then_return(sentinel.tips_to_mark_as_used) + result = await subject.execute( PickUpTipParams(pipetteId=pipette_id, labwareId=labware_id, wellName=well_name) ) @@ -182,7 +200,7 @@ async def test_tip_physically_missing_error( new_deck_point=DeckPoint(x=111, y=222, z=333), ), tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="labware-id", well_name="well-name" + labware_id="labware-id", well_names=sentinel.tips_to_mark_as_used ), pipette_aspirated_fluid=update_types.PipetteUnknownFluidUpdate( pipette_id="pipette-id" @@ -196,7 +214,7 @@ async def test_tip_physically_missing_error( pipette_id="pipette-id", clean_tip=True ), tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="labware-id", well_name="well-name" + labware_id="labware-id", well_names=sentinel.tips_to_mark_as_used ), pipette_location=update_types.PipetteLocationUpdate( pipette_id="pipette-id", diff --git a/api/tests/opentrons/protocol_engine/commands/unsafe/test_unsafe_drop_tip_in_place.py b/api/tests/opentrons/protocol_engine/commands/unsafe/test_unsafe_drop_tip_in_place.py index 27f2964ab741..6c6a875b53f8 100644 --- a/api/tests/opentrons/protocol_engine/commands/unsafe/test_unsafe_drop_tip_in_place.py +++ b/api/tests/opentrons/protocol_engine/commands/unsafe/test_unsafe_drop_tip_in_place.py @@ -49,9 +49,9 @@ async def test_drop_tip_implementation( decoy.when(state_view.motion.get_pipette_location(pipette_id="abc")).then_return( PipetteLocationData(mount=MountType.LEFT, critical_point=None) ) - decoy.when( - state_view.tips.get_pipette_active_channels(params.pipetteId) - ).then_return(channels) + decoy.when(state_view.pipettes.get_active_channels(params.pipetteId)).then_return( + channels + ) result = await subject.execute(params) diff --git a/api/tests/opentrons/protocol_engine/execution/test_movement_handler.py b/api/tests/opentrons/protocol_engine/execution/test_movement_handler.py index 73b293fdbeff..2ddc3d7263df 100644 --- a/api/tests/opentrons/protocol_engine/execution/test_movement_handler.py +++ b/api/tests/opentrons/protocol_engine/execution/test_movement_handler.py @@ -106,7 +106,7 @@ async def test_move_to_well( DeckSlotName.SLOT_1 ) - decoy.when(state_store.tips.get_pipette_channels("pipette-id")).then_return(1) + decoy.when(state_store.pipettes.get_channels("pipette-id")).then_return(1) decoy.when(state_store.labware.is_tiprack("labware-id")).then_return(False) decoy.when( @@ -221,7 +221,7 @@ async def test_move_to_well_from_starting_location( DeckSlotName.SLOT_1 ) - decoy.when(state_store.tips.get_pipette_channels("pipette-id")).then_return(1) + decoy.when(state_store.pipettes.get_channels("pipette-id")).then_return(1) decoy.when(state_store.labware.is_tiprack("labware-id")).then_return(False) decoy.when( @@ -322,7 +322,7 @@ async def test_move_to_addressable_area( state_store.addressable_areas.get_addressable_area_base_slot("area-name") ).then_return(DeckSlotName.SLOT_1) - decoy.when(state_store.tips.get_pipette_channels("pipette-id")).then_return(1) + decoy.when(state_store.pipettes.get_channels("pipette-id")).then_return(1) decoy.when( state_store.motion.get_pipette_location( diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_state.py b/api/tests/opentrons/protocol_engine/state/test_pipette_state.py new file mode 100644 index 000000000000..eb27e2a2f535 --- /dev/null +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_state.py @@ -0,0 +1,226 @@ +"""Tests for the PipetteStore+PipetteState+PipetteView trifecta. + +The trifecta is tested here as a single unit, treating PipetteState as a private +implementation detail. +""" + +from collections import OrderedDict + +import pytest + +from opentrons_shared_data.pipette import pipette_definition + +from opentrons.hardware_control.nozzle_manager import NozzleMap +from opentrons.protocol_engine import actions, commands +from opentrons.protocol_engine.resources.pipette_data_provider import ( + LoadedStaticPipetteData, +) +from opentrons.protocol_engine.state import update_types +from opentrons.protocol_engine.state.pipettes import PipetteStore, PipetteView +from opentrons.protocol_engine.types import FlowRates +from opentrons.types import Point + +from ..pipette_fixtures import ( + EIGHT_CHANNEL_COLS, + EIGHT_CHANNEL_MAP, + EIGHT_CHANNEL_ROWS, + NINETY_SIX_COLS, + NINETY_SIX_MAP, + NINETY_SIX_ROWS, +) + + +def _dummy_command() -> commands.Command: + """Return a placeholder command.""" + return commands.Comment.model_construct() # type: ignore[call-arg] + + +def test_handle_pipette_config_action( + supported_tip_fixture: pipette_definition.SupportedTipsDefinition, +) -> None: + """Should add pipette channel to state.""" + subject = PipetteStore() + + config_update = update_types.PipetteConfigUpdate( + pipette_id="pipette-id", + serial_number="pipette-serial", + config=LoadedStaticPipetteData( + channels=8, + max_volume=15, + min_volume=3, + model="gen a", + display_name="display name", + flow_rates=FlowRates( + default_aspirate={}, + default_dispense={}, + default_blow_out={}, + ), + tip_configuration_lookup_table={15: supported_tip_fixture}, + nominal_tip_overlap={}, + nozzle_offset_z=1.23, + home_position=4.56, + nozzle_map=NozzleMap.build( + physical_nozzles=EIGHT_CHANNEL_MAP, + physical_rows=EIGHT_CHANNEL_ROWS, + physical_columns=EIGHT_CHANNEL_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="H1", + valid_nozzle_maps=pipette_definition.ValidNozzleMaps( + maps={"Full": EIGHT_CHANNEL_COLS["1"]} + ), + ), + 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=pipette_definition.AvailableSensorDefinition( + sensors=["pressure", "capacitive", "environment"] + ), + ), + ) + subject.handle_action( + actions.SucceedCommandAction( + state_update=update_types.StateUpdate(pipette_config=config_update), + command=_dummy_command(), + ) + ) + + assert PipetteView(subject.state).get_channels("pipette-id") == 8 + assert PipetteView(subject.state).get_active_channels("pipette-id") == 8 + + +@pytest.mark.parametrize( + argnames=["nozzle_map", "expected_channels"], + argvalues=[ + ( + NozzleMap.build( + physical_nozzles=OrderedDict({"A1": Point(0, 0, 0)}), + physical_rows=OrderedDict({"A": ["A1"]}), + physical_columns=OrderedDict({"1": ["A1"]}), + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="A1", + valid_nozzle_maps=pipette_definition.ValidNozzleMaps( + maps={"A1": ["A1"]} + ), + ), + 1, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="H12", + valid_nozzle_maps=pipette_definition.ValidNozzleMaps( + maps={ + "Full": sum( + [ + NINETY_SIX_ROWS["A"], + NINETY_SIX_ROWS["B"], + NINETY_SIX_ROWS["C"], + NINETY_SIX_ROWS["D"], + NINETY_SIX_ROWS["E"], + NINETY_SIX_ROWS["F"], + NINETY_SIX_ROWS["G"], + NINETY_SIX_ROWS["H"], + ], + [], + ) + } + ), + ), + 96, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="E1", + valid_nozzle_maps=pipette_definition.ValidNozzleMaps( + maps={"A1_E1": ["A1", "B1", "C1", "D1", "E1"]} + ), + ), + 5, + ), + ], +) +def test_active_channels( + supported_tip_fixture: pipette_definition.SupportedTipsDefinition, + nozzle_map: NozzleMap, + expected_channels: int, +) -> None: + """Should update active channels after pipette configuration change.""" + subject = PipetteStore() + + # Load pipette to update state + config_update = update_types.PipetteConfigUpdate( + pipette_id="pipette-id", + serial_number="pipette-serial", + config=LoadedStaticPipetteData( + channels=9, + max_volume=15, + min_volume=3, + model="gen a", + display_name="display name", + flow_rates=FlowRates( + default_aspirate={}, + default_dispense={}, + default_blow_out={}, + ), + tip_configuration_lookup_table={15: supported_tip_fixture}, + nominal_tip_overlap={}, + nozzle_offset_z=1.23, + home_position=4.56, + nozzle_map=nozzle_map, + 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={}, + plunger_positions={ + "top": 0.0, + "bottom": 5.0, + "blow_out": 19.0, + "drop_tip": 20.0, + }, + shaft_ul_per_mm=5.0, + available_sensors=pipette_definition.AvailableSensorDefinition( + sensors=["pressure", "capacitive", "environment"] + ), + ), + ) + subject.handle_action( + actions.SucceedCommandAction( + state_update=update_types.StateUpdate(pipette_config=config_update), + command=_dummy_command(), + ) + ) + + # Configure nozzle for partial configuration + state_update = update_types.StateUpdate( + pipette_nozzle_map=update_types.PipetteNozzleMapUpdate( + pipette_id="pipette-id", + nozzle_map=nozzle_map, + ) + ) + subject.handle_action( + actions.SucceedCommandAction( + command=_dummy_command(), + state_update=state_update, + ) + ) + assert ( + PipetteView(subject.state).get_active_channels("pipette-id") + == expected_channels + ) diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_view_old.py b/api/tests/opentrons/protocol_engine/state/test_pipette_view_old.py index c9e180c35353..6036c6309caf 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_view_old.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_view_old.py @@ -660,6 +660,8 @@ def test_nozzle_configuration_getters() -> None: valid_nozzle_maps=ValidNozzleMaps(maps={"A1": ["A1"]}), ) subject = get_pipette_view(nozzle_layout_by_id={"pipette-id": nozzle_map}) + assert subject.get_nozzle_configuration("pipette-id") == nozzle_map + assert subject.get_nozzle_configurations() == {"pipette-id": nozzle_map} assert subject.get_nozzle_layout_type("pipette-id") == NozzleConfigurationType.FULL assert subject.get_is_partially_configured("pipette-id") is False assert subject.get_primary_nozzle("pipette-id") == "A1" 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 6fcf00db1f55..e469ac6d135e 100644 --- a/api/tests/opentrons/protocol_engine/state/test_tip_state.py +++ b/api/tests/opentrons/protocol_engine/state/test_tip_state.py @@ -1,7 +1,5 @@ """Tests for tip state store and selectors.""" -from collections import OrderedDict - import pytest from typing import Optional @@ -11,22 +9,17 @@ LabwareDefinition2, Parameters2 as LabwareDefinition2Parameters, ) -from opentrons_shared_data.pipette import pipette_definition from opentrons_shared_data.pipette.pipette_definition import ValidNozzleMaps from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_engine import actions, commands from opentrons.protocol_engine.state import update_types -from opentrons.protocol_engine.state.tips import TipStore, TipView, TipRackWellState +from opentrons.protocol_engine.state.tips import TipStore, TipView from opentrons.protocol_engine.types import ( DeckSlotLocation, - FlowRates, OFF_DECK_LOCATION, ) -from opentrons.protocol_engine.resources.pipette_data_provider import ( - LoadedStaticPipetteData, -) -from opentrons.types import DeckSlotName, Point +from opentrons.types import DeckSlotName from opentrons_shared_data.pipette.types import PipetteNameType from opentrons_shared_data.pipette.pipette_definition import ( AvailableSensorDefinition, @@ -106,49 +99,9 @@ def _dummy_command() -> commands.Command: 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) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=96, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) result = TipView(subject.state).get_next_tip( labware_id="cool-labware", @@ -165,58 +118,10 @@ def test_get_next_tip_returns_first_tip( load_labware_action: actions.SucceedCommandAction, 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) - pipette_name_type = PipetteNameType.P1000_96 - if input_tip_amount == 1: - pipette_name_type = PipetteNameType.P300_SINGLE_GEN2 - elif input_tip_amount == 8: - pipette_name_type = PipetteNameType.P300_MULTI_GEN2 - else: - pipette_name_type = PipetteNameType.P1000_96 - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=input_tip_amount, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(pipette_name_type), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=input_tip_amount, @@ -233,51 +138,10 @@ def test_get_next_tip_used_starting_tip( subject: TipStore, 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) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=input_tip_amount, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=input_tip_amount, @@ -309,13 +173,10 @@ def test_get_next_tip_skips_picked_up_tip( get_next_tip_tips: int, 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) - channels_num = input_tip_amount if input_starting_tip is not None: pipette_name_type = PipetteNameType.P1000_96 if input_tip_amount == 1: @@ -325,7 +186,6 @@ def test_get_next_tip_skips_picked_up_tip( else: pipette_name_type = PipetteNameType.P1000_96 else: - channels_num = get_next_tip_tips pipette_name_type = PipetteNameType.P1000_96 if get_next_tip_tips == 1: pipette_name_type = PipetteNameType.P300_SINGLE_GEN2 @@ -333,50 +193,15 @@ def test_get_next_tip_skips_picked_up_tip( pipette_name_type = PipetteNameType.P300_MULTI_GEN2 else: pipette_name_type = PipetteNameType.P1000_96 - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=channels_num, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(pipette_name_type), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) + + nozzle_map = get_default_nozzle_map(pipette_name_type) pick_up_tip_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", - well_name="A1", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", well_name="A1", nozzle_map=nozzle_map + ), ) ) subject.handle_action( @@ -390,7 +215,7 @@ def test_get_next_tip_skips_picked_up_tip( labware_id="cool-labware", num_tips=get_next_tip_tips, starting_tip_name=input_starting_tip, - nozzle_map=config_update.config.nozzle_map, + nozzle_map=nozzle_map, ) assert result == result_well_name @@ -399,61 +224,27 @@ def test_get_next_tip_skips_picked_up_tip( 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.""" + nozzle_map = get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2) + subject.handle_action(load_labware_action) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=1, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=1, starting_tip_name="B2", - nozzle_map=config_update.config.nozzle_map, + nozzle_map=nozzle_map, ) - assert result == "B2" pick_up_tip_state_update = update_types.StateUpdate( - tips_used=update_types.TipsUsedUpdate("pipette-id", "cool-labware", "B2") + tips_used=update_types.TipsUsedUpdate( + labware_id="cool-labware", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", well_name="B2", nozzle_map=nozzle_map + ), + ) ) subject.handle_action( actions.SucceedCommandAction( @@ -466,59 +257,19 @@ def test_get_next_tip_with_starting_tip( labware_id="cool-labware", num_tips=1, starting_tip_name="B2", - nozzle_map=config_update.config.nozzle_map, + nozzle_map=nozzle_map, ) - assert result == "C2" 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) + nozzle_map = get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=8, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) + subject.handle_action(load_labware_action) result = TipView(subject.state).get_next_tip( labware_id="cool-labware", @@ -526,12 +277,14 @@ def test_get_next_tip_with_starting_tip_8_channel( starting_tip_name="A2", nozzle_map=None, ) - assert result == "A2" pick_up_tip_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", well_name="A2" + labware_id="cool-labware", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", well_name="A2", nozzle_map=nozzle_map + ), ) ) subject.handle_action( @@ -554,107 +307,35 @@ def test_get_next_tip_with_starting_tip_8_channel( 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) + nozzle_map_1_channel = get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2) + nozzle_map_8_channel = get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=1, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - config_update_2 = update_types.PipetteConfigUpdate( - pipette_id="pipette-id2", - serial_number="pipette-serial2", - config=LoadedStaticPipetteData( - channels=8, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name2", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update_2), - command=_dummy_command(), - ) - ) + subject.handle_action(load_labware_action) result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=1, starting_tip_name=None, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), + nozzle_map=nozzle_map_1_channel, ) - assert result == "A1" - pick_up_tip_2_state_update = update_types.StateUpdate( + pick_up_tip_1_channel_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id2", labware_id="cool-labware", well_name="A1" + labware_id="cool-labware", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", + well_name="A1", + nozzle_map=nozzle_map_1_channel, + ), ) ) subject.handle_action( actions.SucceedCommandAction( command=_dummy_command(), - state_update=pick_up_tip_2_state_update, + state_update=pick_up_tip_1_channel_state_update, ) ) @@ -662,72 +343,34 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( labware_id="cool-labware", num_tips=8, starting_tip_name=None, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2), + nozzle_map=nozzle_map_8_channel, ) - assert result == "A2" 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) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=1, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=1, starting_tip_name="H12", nozzle_map=None, ) - assert result == "H12" pick_up_tip_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", well_name="H12" + labware_id="cool-labware", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", + well_name="H12", + nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), + ), ) ) subject.handle_action( @@ -750,51 +393,10 @@ def test_get_next_tip_with_starting_tip_out_of_tips( 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) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=8, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_MULTI_GEN2), - back_left_corner_offset=Point(0, 0, 0), - front_right_corner_offset=Point(0, 0, 0), - pipette_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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=8, @@ -808,60 +410,17 @@ def test_get_next_tip_with_column_and_starting_tip( 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) - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=1, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - 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={}, - 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, - ), - ) - - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - subject.handle_action( actions.SucceedCommandAction( command=_dummy_command(), state_update=update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", - well_name="A1", + well_names=["A1", "A2", "A3"], ) ), ) @@ -880,55 +439,6 @@ def get_result() -> str | None: assert get_result() == "A1" -def test_handle_pipette_config_action( - subject: TipStore, - supported_tip_fixture: pipette_definition.SupportedTipsDefinition, - available_sensors: AvailableSensorDefinition, -) -> None: - """Should add pipette channel to state.""" - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=8, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - - assert TipView(subject.state).get_pipette_channels("pipette-id") == 8 - assert TipView(subject.state).get_pipette_active_channels("pipette-id") == 8 - - @pytest.mark.parametrize( "labware_definition", [ @@ -960,303 +470,28 @@ def test_has_tip_tip_rack( assert result is True -@pytest.mark.parametrize( - argnames=["nozzle_map", "expected_channels"], - argvalues=[ - ( - NozzleMap.build( - physical_nozzles=OrderedDict({"A1": Point(0, 0, 0)}), - physical_rows=OrderedDict({"A": ["A1"]}), - physical_columns=OrderedDict({"1": ["A1"]}), - starting_nozzle="A1", - back_left_nozzle="A1", - front_right_nozzle="A1", - valid_nozzle_maps=ValidNozzleMaps(maps={"A1": ["A1"]}), - ), - 1, - ), - ( - NozzleMap.build( - physical_nozzles=NINETY_SIX_MAP, - physical_rows=NINETY_SIX_ROWS, - physical_columns=NINETY_SIX_COLS, - starting_nozzle="A1", - back_left_nozzle="A1", - front_right_nozzle="H12", - valid_nozzle_maps=ValidNozzleMaps( - maps={ - "Full": sum( - [ - NINETY_SIX_ROWS["A"], - NINETY_SIX_ROWS["B"], - NINETY_SIX_ROWS["C"], - NINETY_SIX_ROWS["D"], - NINETY_SIX_ROWS["E"], - NINETY_SIX_ROWS["F"], - NINETY_SIX_ROWS["G"], - NINETY_SIX_ROWS["H"], - ], - [], - ) - } - ), - ), - 96, - ), - ( - NozzleMap.build( - physical_nozzles=NINETY_SIX_MAP, - physical_rows=NINETY_SIX_ROWS, - physical_columns=NINETY_SIX_COLS, - starting_nozzle="A1", - back_left_nozzle="A1", - front_right_nozzle="E1", - valid_nozzle_maps=ValidNozzleMaps( - maps={"A1_E1": ["A1", "B1", "C1", "D1", "E1"]} - ), - ), - 5, - ), - ], -) -def test_active_channels( - subject: TipStore, - 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 - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=9, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=nozzle_map, - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - - # Configure nozzle for partial configuration - state_update = update_types.StateUpdate( - pipette_nozzle_map=update_types.PipetteNozzleMapUpdate( - pipette_id="pipette-id", - nozzle_map=nozzle_map, - ) - ) - subject.handle_action( - actions.SucceedCommandAction( - command=_dummy_command(), - state_update=state_update, - ) - ) - assert ( - TipView(subject.state).get_pipette_active_channels("pipette-id") - == expected_channels - ) - - -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 - subject.handle_action(load_labware_action) - - # Load pipette - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=96, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P300_SINGLE_GEN2), - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - - # Configure nozzle for partial configuration - state_update = update_types.StateUpdate( - pipette_nozzle_map=update_types.PipetteNozzleMapUpdate( - pipette_id="pipette-id", - nozzle_map=NozzleMap.build( - physical_nozzles=NINETY_SIX_MAP, - physical_rows=NINETY_SIX_ROWS, - physical_columns=NINETY_SIX_COLS, - starting_nozzle="A12", - back_left_nozzle="A12", - front_right_nozzle="H12", - valid_nozzle_maps=ValidNozzleMaps( - maps={ - "A12_H12": [ - "A12", - "B12", - "C12", - "D12", - "E12", - "F12", - "G12", - "H12", - ] - } - ), - ), - ) - ) - subject.handle_action( - actions.SucceedCommandAction( - command=_dummy_command(), - state_update=state_update, - ) - ) - # Pick up partial tips - subject.handle_action( - actions.SucceedCommandAction( - command=_dummy_command(), - state_update=update_types.StateUpdate( - tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", - labware_id="cool-labware", - well_name="A1", - ) - ), - ) - ) - - result = TipView(subject.state).get_next_tip( - labware_id="cool-labware", - num_tips=5, - starting_tip_name=None, - nozzle_map=None, - ) - assert result == "A2" - - 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 subject.handle_action(load_labware_action) - # Load pipette - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=96, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - - def _assert_and_pickup(well: str, nozzle_map: NozzleMap) -> None: + def _assert_and_pickup(expected_next_tip: str, nozzle_map: NozzleMap) -> None: result = TipView(subject.state).get_next_tip( labware_id="cool-labware", num_tips=0, starting_tip_name=None, nozzle_map=nozzle_map, ) - assert result is not None and result == well + assert result is not None and result == expected_next_tip pick_up_tip_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", - well_name=result, + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", well_name=result, nozzle_map=nozzle_map + ), ) ) @@ -1267,8 +502,8 @@ def _assert_and_pickup(well: str, nozzle_map: NozzleMap) -> None: ) ) - def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleMap: - nozzle_map = NozzleMap.build( + def _build_nozzle_map(start: str, back_l: str, front_r: str) -> NozzleMap: + return NozzleMap.build( physical_nozzles=NINETY_SIX_MAP, physical_rows=NINETY_SIX_ROWS, physical_columns=NINETY_SIX_COLS, @@ -1324,86 +559,31 @@ def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleM } ), ) - state_update = update_types.StateUpdate( - pipette_nozzle_map=update_types.PipetteNozzleMapUpdate( - pipette_id="pipette-id", - nozzle_map=nozzle_map, - ) - ) - subject.handle_action( - actions.SucceedCommandAction( - command=_dummy_command(), - state_update=state_update, - ) - ) - return nozzle_map - map = _reconfigure_nozzle_layout("A1", "A1", "H3") + map = _build_nozzle_map("A1", "A1", "H3") _assert_and_pickup("A10", map) - map = _reconfigure_nozzle_layout("A1", "A1", "F2") + map = _build_nozzle_map("A1", "A1", "F2") _assert_and_pickup("C8", map) # Configure to single tip pickups - map = _reconfigure_nozzle_layout("H12", "H12", "H12") + map = _build_nozzle_map("H12", "H12", "H12") _assert_and_pickup("A1", map) - map = _reconfigure_nozzle_layout("H1", "H1", "H1") + map = _build_nozzle_map("H1", "H1", "H1") _assert_and_pickup("A9", map) - map = _reconfigure_nozzle_layout("A12", "A12", "A12") + map = _build_nozzle_map("A12", "A12", "A12") _assert_and_pickup("H1", map) - map = _reconfigure_nozzle_layout("A1", "A1", "A1") + map = _build_nozzle_map("A1", "A1", "A1") _assert_and_pickup("B9", map) 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 subject.handle_action(load_labware_action) - # Load pipette - config_update = update_types.PipetteConfigUpdate( - pipette_id="pipette-id", - serial_number="pipette-serial", - config=LoadedStaticPipetteData( - channels=96, - max_volume=15, - min_volume=3, - model="gen a", - display_name="display name", - flow_rates=FlowRates( - default_aspirate={}, - default_dispense={}, - default_blow_out={}, - ), - tip_configuration_lookup_table={15: supported_tip_fixture}, - nominal_tip_overlap={}, - nozzle_offset_z=1.23, - home_position=4.56, - nozzle_map=get_default_nozzle_map(PipetteNameType.P1000_96), - 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={}, - 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, - ), - ) - subject.handle_action( - actions.SucceedCommandAction( - state_update=update_types.StateUpdate(pipette_config=config_update), - command=_dummy_command(), - ) - ) - def _get_next_and_pickup(nozzle_map: NozzleMap) -> str | None: result = TipView(subject.state).get_next_tip( labware_id="cool-labware", @@ -1414,7 +594,12 @@ def _get_next_and_pickup(nozzle_map: NozzleMap) -> str | None: if result is not None: pick_up_tip_state_update = update_types.StateUpdate( tips_used=update_types.TipsUsedUpdate( - pipette_id="pipette-id", labware_id="cool-labware", well_name=result + labware_id="cool-labware", + well_names=TipView(subject.state).compute_tips_to_mark_as_used( + labware_id="cool-labware", + well_name=result, + nozzle_map=nozzle_map, + ), ) ) @@ -1427,8 +612,8 @@ def _get_next_and_pickup(nozzle_map: NozzleMap) -> str | None: return result - def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleMap: - nozzle_map = NozzleMap.build( + def _build_nozzle_map(start: str, back_l: str, front_r: str) -> NozzleMap: + return NozzleMap.build( physical_nozzles=NINETY_SIX_MAP, physical_rows=NINETY_SIX_ROWS, physical_columns=NINETY_SIX_COLS, @@ -1457,39 +642,28 @@ def _reconfigure_nozzle_layout(start: str, back_l: str, front_r: str) -> NozzleM } ), ) - state_update = update_types.StateUpdate( - pipette_nozzle_map=update_types.PipetteNozzleMapUpdate( - pipette_id="pipette-id", nozzle_map=nozzle_map - ) - ) - subject.handle_action( - actions.SucceedCommandAction( - command=_dummy_command(), state_update=state_update - ) - ) - return nozzle_map - map = _reconfigure_nozzle_layout("A1", "A1", "A1") + map = _build_nozzle_map("A1", "A1", "A1") for _ in range(96): - _get_next_and_pickup(map) + assert _get_next_and_pickup(map) is not None assert _get_next_and_pickup(map) is None subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware")) - map = _reconfigure_nozzle_layout("A12", "A12", "A12") + map = _build_nozzle_map("A12", "A12", "A12") for _ in range(96): - _get_next_and_pickup(map) + assert _get_next_and_pickup(map) is not None assert _get_next_and_pickup(map) is None subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware")) - map = _reconfigure_nozzle_layout("H1", "H1", "H1") + map = _build_nozzle_map("H1", "H1", "H1") for _ in range(96): - _get_next_and_pickup(map) + assert _get_next_and_pickup(map) is not None assert _get_next_and_pickup(map) is None subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware")) - map = _reconfigure_nozzle_layout("H12", "H12", "H12") + map = _build_nozzle_map("H12", "H12", "H12") for _ in range(96): - _get_next_and_pickup(map) + assert _get_next_and_pickup(map) is not None assert _get_next_and_pickup(map) is None @@ -1525,13 +699,7 @@ def test_handle_batch_labware_loaded_update( ) ) - assert "some-labware-id" in subject._state.tips_by_labware_id - assert "some-other-labware-id" in subject._state.tips_by_labware_id - for well_state in subject._state.tips_by_labware_id["some-labware-id"].values(): - assert well_state == TipRackWellState.CLEAN - for well_state in subject._state.tips_by_labware_id[ - "some-other-labware-id" - ].values(): - assert well_state == TipRackWellState.CLEAN - assert "some-labware-id" in subject._state.column_by_labware_id - assert "some-other-labware-id" in subject._state.column_by_labware_id + # The use of has_clean_tip() is arbitrary here. We just need anything that can make + # sure each labware in the batch has actually been ingested. + assert TipView(subject.state).has_clean_tip("some-labware-id", "A1") + assert TipView(subject.state).has_clean_tip("some-other-labware-id", "A1")