diff --git a/api/src/opentrons/protocol_engine/commands/drop_tip.py b/api/src/opentrons/protocol_engine/commands/drop_tip.py index 021c7a47bc5b..a006bd73dd3e 100644 --- a/api/src/opentrons/protocol_engine/commands/drop_tip.py +++ b/api/src/opentrons/protocol_engine/commands/drop_tip.py @@ -114,7 +114,9 @@ async def execute(self, params: DropTipParams) -> SuccessData[DropTipResult, Non await self._tip_handler.drop_tip(pipette_id=pipette_id, home_after=home_after) - state_update.update_tip_state(pipette_id=params.pipetteId, tip_geometry=None) + state_update.update_pipette_tip_state( + pipette_id=params.pipetteId, tip_geometry=None + ) return SuccessData( public=DropTipResult(position=deck_point), diff --git a/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py b/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py index a8e62354f404..c414df864283 100644 --- a/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py +++ b/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py @@ -57,7 +57,9 @@ async def execute( state_update = update_types.StateUpdate() - state_update.update_tip_state(pipette_id=params.pipetteId, tip_geometry=None) + state_update.update_pipette_tip_state( + pipette_id=params.pipetteId, tip_geometry=None + ) return SuccessData( public=DropTipInPlaceResult(), private=None, state_update=state_update 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 465ede2f86f1..c5019b3c590d 100644 --- a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py +++ b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py @@ -71,6 +71,10 @@ class TipPhysicallyMissingError(ErrorOccurrence): of the pipette. """ + # The thing above about marking the tips as used makes it so that + # when the protocol is resumed and the Python Protocol API calls + # `get_next_tip()`, we'll move on to other tips as expected. + isDefined: bool = True errorType: Literal["tipPhysicallyMissing"] = "tipPhysicallyMissing" errorCode: str = ErrorCodes.TIP_PICKUP_FAILED.value.code @@ -130,11 +134,10 @@ async def execute( labware_id=labware_id, well_name=well_name, ) - state_update.update_tip_state( - pipette_id=pipette_id, - tip_geometry=tip_geometry, - ) except TipNotAttachedError as e: + state_update.mark_tips_as_used( + pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + ) return DefinedErrorData( public=TipPhysicallyMissingError( id=self._model_utils.generate_id(), @@ -150,6 +153,13 @@ async def execute( state_update=state_update, ) else: + state_update.update_pipette_tip_state( + pipette_id=pipette_id, + tip_geometry=tip_geometry, + ) + state_update.mark_tips_as_used( + pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + ) return SuccessData( public=PickUpTipResult( tipVolume=tip_geometry.volume, 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 e27a118ea60a..33d4baebeea8 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 @@ -74,7 +74,9 @@ async def execute( ) state_update = StateUpdate() - state_update.update_tip_state(pipette_id=params.pipetteId, tip_geometry=None) + state_update.update_pipette_tip_state( + pipette_id=params.pipetteId, tip_geometry=None + ) return SuccessData( public=UnsafeDropTipInPlaceResult(), private=None, state_update=state_update diff --git a/api/src/opentrons/protocol_engine/state/tips.py b/api/src/opentrons/protocol_engine/state/tips.py index 1e14843114ca..4ed78c1df963 100644 --- a/api/src/opentrons/protocol_engine/state/tips.py +++ b/api/src/opentrons/protocol_engine/state/tips.py @@ -3,27 +3,18 @@ from enum import Enum from typing import Dict, Optional, List, Union +from opentrons.protocol_engine.state import update_types + from ._abstract_store import HasState, HandlesActions -from ..actions import ( - Action, - SucceedCommandAction, - FailCommandAction, - ResetTipsAction, -) +from ..actions import Action, SucceedCommandAction, ResetTipsAction, get_state_update from ..commands import ( Command, LoadLabwareResult, - PickUpTip, - PickUpTipResult, - DropTipResult, - DropTipInPlaceResult, - unsafe, ) from ..commands.configuring_common import ( PipetteConfigUpdateResultMixin, PipetteNozzleLayoutResultMixin, ) -from ..error_recovery_policy import ErrorRecoveryType from opentrons.hardware_control.nozzle_manager import NozzleMap @@ -38,6 +29,18 @@ class TipRackWellState(Enum): 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 + + @dataclass class TipState: """State of all tips.""" @@ -45,9 +48,7 @@ class TipState: tips_by_labware_id: Dict[str, TipRackStateByWellName] column_by_labware_id: Dict[str, List[List[str]]] - channels_by_pipette_id: Dict[str, int] - active_channels_by_pipette_id: Dict[str, int] - nozzle_map_by_pipette_id: Dict[str, NozzleMap] + pipette_info_by_pipette_id: Dict[str, _PipetteInfo] class TipStore(HasState[TipState], HandlesActions): @@ -60,37 +61,33 @@ def __init__(self) -> None: self._state = TipState( tips_by_labware_id={}, column_by_labware_id={}, - channels_by_pipette_id={}, - active_channels_by_pipette_id={}, - nozzle_map_by_pipette_id={}, + pipette_info_by_pipette_id={}, ) def handle_action(self, action: Action) -> None: """Modify state in reaction to an action.""" + state_update = get_state_update(action) + if state_update is not None: + self._handle_state_update(state_update) + if isinstance(action, SucceedCommandAction): if isinstance(action.private_result, PipetteConfigUpdateResultMixin): pipette_id = action.private_result.pipette_id config = action.private_result.config - self._state.channels_by_pipette_id[pipette_id] = config.channels - self._state.active_channels_by_pipette_id[pipette_id] = config.channels - self._state.nozzle_map_by_pipette_id[pipette_id] = config.nozzle_map + self._state.pipette_info_by_pipette_id[pipette_id] = _PipetteInfo( + channels=config.channels, + active_channels=config.channels, + nozzle_map=config.nozzle_map, + ) + self._handle_succeeded_command(action.command) if isinstance(action.private_result, PipetteNozzleLayoutResultMixin): pipette_id = action.private_result.pipette_id nozzle_map = action.private_result.nozzle_map - if nozzle_map: - self._state.active_channels_by_pipette_id[ - pipette_id - ] = nozzle_map.tip_count - self._state.nozzle_map_by_pipette_id[pipette_id] = nozzle_map - else: - self._state.active_channels_by_pipette_id[ - pipette_id - ] = self._state.channels_by_pipette_id[pipette_id] - - elif isinstance(action, FailCommandAction): - self._handle_failed_command(action) + pipette_info = self._state.pipette_info_by_pipette_id[pipette_id] + pipette_info.active_channels = nozzle_map.tip_count + pipette_info.nozzle_map = nozzle_map elif isinstance(action, ResetTipsAction): labware_id = action.labware_id @@ -116,48 +113,20 @@ def _handle_succeeded_command(self, command: Command) -> None: column for column in definition.ordering ] - elif isinstance(command.result, PickUpTipResult): - labware_id = command.params.labwareId - well_name = command.params.wellName - pipette_id = command.params.pipetteId - self._set_used_tips( - pipette_id=pipette_id, well_name=well_name, labware_id=labware_id - ) - - elif isinstance( - command.result, - (DropTipResult, DropTipInPlaceResult, unsafe.UnsafeDropTipInPlaceResult), - ): - pipette_id = command.params.pipetteId - - def _handle_failed_command( - self, - action: FailCommandAction, - ) -> None: - # If a pickUpTip command fails recoverably, mark the tips as used. This way, - # when the protocol is resumed and the Python Protocol API calls - # `get_next_tip()`, we'll move on to other tips as expected. - # - # We don't attempt this for nonrecoverable errors because maybe the failure - # was due to a bad labware ID or well name. - if ( - isinstance(action.running_command, PickUpTip) - and action.type != ErrorRecoveryType.FAIL_RUN - ): + def _handle_state_update(self, state_update: update_types.StateUpdate) -> None: + if state_update.tips_used != update_types.NO_CHANGE: self._set_used_tips( - pipette_id=action.running_command.params.pipetteId, - labware_id=action.running_command.params.labwareId, - well_name=action.running_command.params.wellName, + pipette_id=state_update.tips_used.pipette_id, + labware_id=state_update.tips_used.labware_id, + well_name=state_update.tips_used.well_name, ) - # Note: We're logically removing the tip from the tip rack, - # but we're not logically updating the pipette to have that tip on it. def _set_used_tips( # noqa: C901 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.nozzle_map_by_pipette_id[pipette_id] + nozzle_map = self._state.pipette_info_by_pipette_id[pipette_id].nozzle_map # TODO (cb, 02-28-2024): Transition from using partial nozzle map to full instrument map for the set used logic num_nozzle_cols = len(nozzle_map.columns) @@ -225,7 +194,7 @@ def _identify_tip_cluster( critical_row: int, entry_well: str, ) -> Optional[List[str]]: - tip_cluster = [] + tip_cluster: list[str] = [] for i in range(active_columns): if entry_well == "A1" or entry_well == "H1": if critical_column - i >= 0: @@ -276,12 +245,12 @@ def _validate_tip_cluster( # In the case of a 96ch we can attempt to index in by singular rows and columns assuming that indexed direction is safe # The tip cluster list is ordered: Each row from a column in order by columns - tip_cluster_final_column = [] + tip_cluster_final_column: list[str] = [] for i in range(active_rows): tip_cluster_final_column.append( tip_cluster[((active_columns * active_rows) - 1) - i] ) - tip_cluster_final_row = [] + tip_cluster_final_row: list[str] = [] for i in range(active_columns): tip_cluster_final_row.append( tip_cluster[(active_rows - 1) + (i * active_rows)] @@ -472,19 +441,22 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]: def get_pipette_channels(self, pipette_id: str) -> int: """Return the given pipette's number of channels.""" - return self._state.channels_by_pipette_id[pipette_id] + 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.active_channels_by_pipette_id[pipette_id] + 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.nozzle_map_by_pipette_id[pipette_id] + 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 self._state.nozzle_map_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. diff --git a/api/src/opentrons/protocol_engine/state/update_types.py b/api/src/opentrons/protocol_engine/state/update_types.py index 91cdf0194a3d..0bf00cfdd86a 100644 --- a/api/src/opentrons/protocol_engine/state/update_types.py +++ b/api/src/opentrons/protocol_engine/state/update_types.py @@ -136,6 +136,23 @@ class PipetteTipStateUpdate: tip_geometry: typing.Optional[TipGeometry] +@dataclasses.dataclass +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 + + well_name: str + """The well that the pipette's primary nozzle targeted. + + Wells in addition to this one will also be marked as used, depending on the + pipette's nozzle layout. + """ + + @dataclasses.dataclass class StateUpdate: """Represents an update to perform on engine state.""" @@ -154,8 +171,10 @@ class StateUpdate: loaded_labware: LoadedLabwareUpdate | NoChangeType = NO_CHANGE + tips_used: TipsUsedUpdate | NoChangeType = NO_CHANGE + # These convenience functions let the caller avoid the boilerplate of constructing a - # complicated dataclass tree, and they give us a + # complicated dataclass tree. @typing.overload def set_pipette_location( @@ -207,6 +226,10 @@ def set_pipette_location( # noqa: D102 new_deck_point=new_deck_point, ) + def clear_all_pipette_locations(self) -> None: + """Mark all pipettes as having an unknown location.""" + self.pipette_location = CLEAR + def set_labware_location( self, *, @@ -238,10 +261,6 @@ def set_loaded_labware( display_name=display_name, ) - def clear_all_pipette_locations(self) -> None: - """Mark all pipettes as having an unknown location.""" - self.pipette_location = CLEAR - def set_load_pipette( self, pipette_id: str, @@ -274,10 +293,18 @@ def update_pipette_nozzle(self, pipette_id: str, nozzle_map: NozzleMap) -> None: pipette_id=pipette_id, nozzle_map=nozzle_map ) - def update_tip_state( + def update_pipette_tip_state( self, pipette_id: str, tip_geometry: typing.Optional[TipGeometry] ) -> None: """Update tip state.""" self.pipette_tip_state = PipetteTipStateUpdate( pipette_id=pipette_id, tip_geometry=tip_geometry ) + + def mark_tips_as_used( + self, pipette_id: str, labware_id: str, well_name: str + ) -> None: + """Mark tips in a tip rack as used. See `MarkTipsUsedState`.""" + self.tips_used = TipsUsedUpdate( + pipette_id=pipette_id, labware_id=labware_id, well_name=well_name + ) 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 c4ff37c501d6..55a4504d5a3d 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 @@ -83,6 +83,9 @@ async def test_success( pipette_id="pipette-id", 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" + ), ), ) @@ -139,6 +142,9 @@ async def test_tip_physically_missing_error( labware_id="labware-id", well_name="well-name" ), 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" + ), ), ) 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 bdc5cc639f4b..dc603ac4ca8a 100644 --- a/api/tests/opentrons/protocol_engine/state/test_tip_state.py +++ b/api/tests/opentrons/protocol_engine/state/test_tip_state.py @@ -14,8 +14,9 @@ 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 -from opentrons.protocol_engine.types import FlowRates, DeckPoint +from opentrons.protocol_engine.types import FlowRates from opentrons.protocol_engine.resources.pipette_data_provider import ( LoadedStaticPipetteData, ) @@ -70,54 +71,9 @@ def load_labware_command(labware_definition: LabwareDefinition) -> commands.Load ) -@pytest.fixture -def pick_up_tip_command() -> commands.PickUpTip: - """Get a pick-up tip command value object.""" - return commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName="A1", - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), - ) - - -@pytest.fixture -def drop_tip_command() -> commands.DropTip: - """Get a drop tip command value object.""" - return commands.DropTip.construct( # type: ignore[call-arg] - params=commands.DropTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName="A1", - ), - result=commands.DropTipResult.construct(position=DeckPoint(x=0, y=0, z=0)), - ) - - -@pytest.fixture -def drop_tip_in_place_command() -> commands.DropTipInPlace: - """Get a drop tip in place command object.""" - return commands.DropTipInPlace.construct( # type: ignore[call-arg] - params=commands.DropTipInPlaceParams.construct( - pipetteId="pipette-id", - ), - result=commands.DropTipInPlaceResult.construct(), - ) - - -@pytest.fixture -def unsafe_drop_tip_in_place_command() -> commands.unsafe.UnsafeDropTipInPlace: - """Get an unsafe drop-tip-in-place command.""" - return commands.unsafe.UnsafeDropTipInPlace.construct( # type: ignore[call-arg] - params=commands.unsafe.UnsafeDropTipInPlaceParams.construct( - pipetteId="pipette-id" - ), - result=commands.unsafe.UnsafeDropTipInPlaceResult.construct(), - ) +def _dummy_command() -> commands.Command: + """Return a placeholder command.""" + return commands.Comment.construct() # type: ignore[call-arg] @pytest.mark.parametrize( @@ -310,7 +266,6 @@ def test_get_next_tip_used_starting_tip( ) def test_get_next_tip_skips_picked_up_tip( load_labware_command: commands.LoadLabware, - pick_up_tip_command: commands.PickUpTip, subject: TipStore, input_tip_amount: int, get_next_tip_tips: int, @@ -322,6 +277,7 @@ def test_get_next_tip_skips_picked_up_tip( subject.handle_action( actions.SucceedCommandAction(private_result=None, command=load_labware_command) ) + load_pipette_command = commands.LoadPipette.construct( # type: ignore[call-arg] result=commands.LoadPipetteResult(pipetteId="pipette-id") ) @@ -372,8 +328,20 @@ def test_get_next_tip_skips_picked_up_tip( private_result=load_pipette_private_result, command=load_pipette_command ) ) + + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", + labware_id="cool-labware", + well_name="A1", + ) + ) subject.handle_action( - actions.SucceedCommandAction(command=pick_up_tip_command, private_result=None) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) result = TipView(subject.state).get_next_tip( @@ -436,19 +404,15 @@ def test_get_next_tip_with_starting_tip( assert result == "B2" - pick_up_tip = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName="B2", - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate("pipette-id", "cool-labware", "B2") ) - subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) result = TipView(subject.state).get_next_tip( @@ -512,19 +476,17 @@ def test_get_next_tip_with_starting_tip_8_channel( assert result == "A2" - pick_up_tip = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName="A2", - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", labware_id="cool-labware", well_name="A2" + ) ) - subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) result = TipView(subject.state).get_next_tip( @@ -578,10 +540,10 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( private_result=load_pipette_private_result, command=load_pipette_command ) ) - load_pipette_command2 = commands.LoadPipette.construct( # type: ignore[call-arg] + load_pipette_command_2 = commands.LoadPipette.construct( # type: ignore[call-arg] result=commands.LoadPipetteResult(pipetteId="pipette-id2") ) - load_pipette_private_result2 = commands.LoadPipettePrivateResult( + load_pipette_private_result_2 = commands.LoadPipettePrivateResult( pipette_id="pipette-id2", serial_number="pipette-serial2", config=LoadedStaticPipetteData( @@ -607,7 +569,7 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( ) subject.handle_action( actions.SucceedCommandAction( - private_result=load_pipette_private_result2, command=load_pipette_command2 + private_result=load_pipette_private_result_2, command=load_pipette_command_2 ) ) @@ -620,19 +582,17 @@ def test_get_next_tip_with_1_channel_followed_by_8_channel( assert result == "A1" - pick_up_tip2 = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id2", - labwareId="cool-labware", - wellName="A1", - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_2_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id2", labware_id="cool-labware", well_name="A1" + ) ) - subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip2) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_2_state_update, + ) ) result = TipView(subject.state).get_next_tip( @@ -696,19 +656,17 @@ def test_get_next_tip_with_starting_tip_out_of_tips( assert result == "H12" - pick_up_tip = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName="H12", - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", labware_id="cool-labware", well_name="H12" + ) ) - subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) result = TipView(subject.state).get_next_tip( @@ -776,7 +734,6 @@ def test_get_next_tip_with_column_and_starting_tip( def test_reset_tips( subject: TipStore, load_labware_command: commands.LoadLabware, - pick_up_tip_command: commands.PickUpTip, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, ) -> None: """It should be able to reset tip tracking state.""" @@ -818,18 +775,30 @@ def test_reset_tips( ) subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip_command) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", + labware_id="cool-labware", + well_name="A1", + ) + ), + ) ) - subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware")) - result = TipView(subject.state).get_next_tip( - labware_id="cool-labware", - num_tips=1, - starting_tip_name=None, - nozzle_map=None, - ) + def get_result() -> str | None: + return TipView(subject.state).get_next_tip( + labware_id="cool-labware", + num_tips=1, + starting_tip_name=None, + nozzle_map=None, + ) - assert result == "A1" + assert get_result() != "A1" + subject.handle_action(actions.ResetTipsAction(labware_id="cool-labware")) + assert get_result() == "A1" def test_handle_pipette_config_action( @@ -1032,7 +1001,6 @@ def test_next_tip_uses_active_channels( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_command: commands.LoadLabware, - pick_up_tip_command: commands.PickUpTip, ) -> None: """Test that tip tracking logic uses pipette's active channels.""" # Load labware @@ -1102,7 +1070,17 @@ def test_next_tip_uses_active_channels( ) # Pick up partial tips subject.handle_action( - actions.SucceedCommandAction(command=pick_up_tip_command, private_result=None) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + 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( @@ -1118,7 +1096,6 @@ def test_next_tip_automatic_tip_tracking_with_partial_configurations( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_command: commands.LoadLabware, - pick_up_tip_command: commands.PickUpTip, ) -> None: """Test tip tracking logic using multiple pipette configurations.""" # Load labware @@ -1167,21 +1144,22 @@ def _assert_and_pickup(well: str, nozzle_map: NozzleMap) -> None: starting_tip_name=None, nozzle_map=nozzle_map, ) - assert result == well + assert result is not None and result == well - pick_up_tip = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName=result, - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", + labware_id="cool-labware", + well_name=result, + ) ) subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) # Configure nozzle for partial configurations @@ -1277,7 +1255,6 @@ def test_next_tip_automatic_tip_tracking_tiprack_limits( subject: TipStore, supported_tip_fixture: pipette_definition.SupportedTipsDefinition, load_labware_command: commands.LoadLabware, - pick_up_tip_command: commands.PickUpTip, ) -> None: """Test tip tracking logic to ensure once a tiprack is consumed it returns None when consuming tips using multiple pipette configurations.""" # Load labware @@ -1327,19 +1304,18 @@ def _get_next_and_pickup(nozzle_map: NozzleMap) -> str | None: nozzle_map=nozzle_map, ) if result is not None: - pick_up_tip = commands.PickUpTip.construct( # type: ignore[call-arg] - params=commands.PickUpTipParams.construct( - pipetteId="pipette-id", - labwareId="cool-labware", - wellName=result, - ), - result=commands.PickUpTipResult.construct( - position=DeckPoint(x=0, y=0, z=0), tipLength=1.23 - ), + pick_up_tip_state_update = update_types.StateUpdate( + tips_used=update_types.TipsUsedUpdate( + pipette_id="pipette-id", labware_id="cool-labware", well_name=result + ) ) subject.handle_action( - actions.SucceedCommandAction(private_result=None, command=pick_up_tip) + actions.SucceedCommandAction( + command=_dummy_command(), + private_result=None, + state_update=pick_up_tip_state_update, + ) ) return result