From 94b7106feb65ba68b516f8d7bacf49eaf1f1cd44 Mon Sep 17 00:00:00 2001 From: CaseyBatten Date: Thu, 22 Feb 2024 17:14:26 -0500 Subject: [PATCH] utilize loaded static pipette --- .../protocol_api/core/engine/labware.py | 28 ++++++++---- .../opentrons/protocol_api/core/labware.py | 3 +- .../core/legacy/legacy_labware_core.py | 3 +- .../protocol_api/instrument_context.py | 4 ++ api/src/opentrons/protocol_api/labware.py | 13 ++++-- .../opentrons/protocol_engine/state/tips.py | 44 +++++++++++-------- 6 files changed, 62 insertions(+), 33 deletions(-) diff --git a/api/src/opentrons/protocol_api/core/engine/labware.py b/api/src/opentrons/protocol_api/core/engine/labware.py index 68ba604dd91..1936a2c6591 100644 --- a/api/src/opentrons/protocol_api/core/engine/labware.py +++ b/api/src/opentrons/protocol_api/core/engine/labware.py @@ -8,9 +8,10 @@ from opentrons_shared_data.labware.labware_definition import LabwareRole +from opentrons.protocol_engine import LoadedPipette from opentrons.protocol_engine.errors import LabwareNotOnDeckError, ModuleNotOnDeckError from opentrons.protocol_engine.clients import SyncClient as ProtocolEngineClient -from opentrons.types import DeckSlotName, Point +from opentrons.types import DeckSlotName, Point, Mount, MountType from ..labware import AbstractLabware, LabwareLoadParams from .well import WellCore @@ -123,18 +124,27 @@ def reset_tips(self) -> None: def get_next_tip( self, + mount: Mount, num_tips: int, starting_tip: Optional[WellCore], ) -> Optional[str]: - return self._engine_client.state.tips.get_next_tip( - labware_id=self._labware_id, - num_tips=num_tips, - starting_tip_name=( - starting_tip.get_name() - if starting_tip and starting_tip.labware_id == self._labware_id - else None - ), + pipette = self._engine_client.state.pipettes.get_by_mount( + MountType.from_hw_mount(mount) ) + if isinstance(pipette, LoadedPipette): + pipette_id = pipette.id + return self._engine_client.state.tips.get_next_tip( + pipette_id=pipette_id, + labware_id=self._labware_id, + num_tips=num_tips, + starting_tip_name=( + starting_tip.get_name() + if starting_tip and starting_tip.labware_id == self._labware_id + else None + ), + ) + else: + raise ValueError(f"No valid Pipette found for Mount {mount.name}") def get_well_columns(self) -> List[List[str]]: """Get the all well names, organized by column, from the labware's definition.""" diff --git a/api/src/opentrons/protocol_api/core/labware.py b/api/src/opentrons/protocol_api/core/labware.py index c26660adb7c..068f9de9168 100644 --- a/api/src/opentrons/protocol_api/core/labware.py +++ b/api/src/opentrons/protocol_api/core/labware.py @@ -10,7 +10,7 @@ LabwareDefinition as LabwareDefinitionDict, ) -from opentrons.types import DeckSlotName, Point +from opentrons.types import DeckSlotName, Point, Mount from .well import WellCoreType @@ -111,6 +111,7 @@ def reset_tips(self) -> None: @abstractmethod def get_next_tip( self, + mount: Mount, num_tips: int, starting_tip: Optional[WellCoreType], ) -> Optional[str]: diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_labware_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_labware_core.py index 4ef56bbce11..6edf4ee79be 100644 --- a/api/src/opentrons/protocol_api/core/legacy/legacy_labware_core.py +++ b/api/src/opentrons/protocol_api/core/legacy/legacy_labware_core.py @@ -4,7 +4,7 @@ from opentrons.protocols.geometry.labware_geometry import LabwareGeometry from opentrons.protocols.api_support.tip_tracker import TipTracker -from opentrons.types import DeckSlotName, Location, Point +from opentrons.types import DeckSlotName, Location, Point, Mount from opentrons_shared_data.labware.dev_types import LabwareParameters, LabwareDefinition from ..labware import AbstractLabware, LabwareLoadParams @@ -154,6 +154,7 @@ def reset_tips(self) -> None: def get_next_tip( self, + mount: Mount, num_tips: int, starting_tip: Optional[LegacyWellCore], ) -> Optional[str]: diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index e0762410d51..ce43a4e1f35 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/api/src/opentrons/protocol_api/instrument_context.py @@ -880,6 +880,7 @@ def pick_up_tip( # noqa: C901 " to pick up." ) tip_rack, well = labware.next_available_tip( + mount=self._core.get_mount(), starting_tip=self.starting_tip, tip_racks=self.tip_racks, channels=active_channels, @@ -891,6 +892,7 @@ def pick_up_tip( # noqa: C901 elif isinstance(location, labware.Labware): tip_rack, well = labware.next_available_tip( + mount=self._core.get_mount(), starting_tip=None, tip_racks=[location], channels=active_channels, @@ -906,6 +908,7 @@ def pick_up_tip( # noqa: C901 elif maybe_tip_rack is not None: tip_rack, well = labware.next_available_tip( + mount=self._core.get_mount(), starting_tip=None, tip_racks=[maybe_tip_rack], channels=active_channels, @@ -1321,6 +1324,7 @@ def transfer( # noqa: C901 if new_tip != types.TransferTipPolicy.NEVER: tr, next_tip = labware.next_available_tip( + self._core.get_mount(), self.starting_tip, self.tip_racks, active_channels, diff --git a/api/src/opentrons/protocol_api/labware.py b/api/src/opentrons/protocol_api/labware.py index 6f183b87e7b..481aa903a8c 100644 --- a/api/src/opentrons/protocol_api/labware.py +++ b/api/src/opentrons/protocol_api/labware.py @@ -17,7 +17,7 @@ from opentrons_shared_data.labware.dev_types import LabwareDefinition, LabwareParameters -from opentrons.types import Location, Point +from opentrons.types import Location, Point, Mount from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import requires_version, APIVersionError @@ -884,6 +884,7 @@ def tip_length(self, length: float) -> None: # TODO(mc, 2022-11-09): implementation detail; deprecate public method def next_tip( self, + mount: Mount, num_tips: int = 1, starting_tip: Optional[Well] = None, ) -> Optional[Well]: @@ -904,6 +905,7 @@ def next_tip( assert num_tips > 0, f"num_tips must be positive integer, but got {num_tips}" well_name = self._core.get_next_tip( + mount=mount, num_tips=num_tips, starting_tip=starting_tip._core if starting_tip else None, ) @@ -1063,6 +1065,7 @@ def split_tipracks(tip_racks: List[Labware]) -> Tuple[Labware, List[Labware]]: # TODO(mc, 2022-11-09): implementation detail, move to core def select_tiprack_from_list( + mount: Mount, tip_racks: List[Labware], num_channels: int, starting_point: Optional[Well] = None, @@ -1081,11 +1084,11 @@ def select_tiprack_from_list( else: first_well = first.wells()[0] - next_tip = first.next_tip(num_channels, first_well) + next_tip = first.next_tip(mount, num_channels, first_well) if next_tip: return first, next_tip else: - return select_tiprack_from_list(rest, num_channels) + return select_tiprack_from_list(mount, rest, num_channels) # TODO(mc, 2022-11-09): implementation detail, move to core @@ -1097,15 +1100,17 @@ def filter_tipracks_to_start( # TODO(mc, 2022-11-09): implementation detail, move to core def next_available_tip( + mount: Mount, starting_tip: Optional[Well], tip_racks: List[Labware], channels: int, ) -> Tuple[Labware, Well]: start = starting_tip if start is None: - return select_tiprack_from_list(tip_racks, channels) + return select_tiprack_from_list(mount, tip_racks, channels) else: return select_tiprack_from_list( + mount, filter_tipracks_to_start(start, tip_racks), channels, start, diff --git a/api/src/opentrons/protocol_engine/state/tips.py b/api/src/opentrons/protocol_engine/state/tips.py index 62697e52123..bc00e2e8998 100644 --- a/api/src/opentrons/protocol_engine/state/tips.py +++ b/api/src/opentrons/protocol_engine/state/tips.py @@ -43,7 +43,7 @@ class TipState: channels_by_pipette_id: Dict[str, int] length_by_pipette_id: Dict[str, float] active_channels_by_pipette_id: Dict[str, int] - nozzle_map: Optional[NozzleMap] + nozzle_map_by_pipette_id: Dict[str, NozzleMap] class TipStore(HasState[TipState], HandlesActions): @@ -59,7 +59,7 @@ def __init__(self) -> None: channels_by_pipette_id={}, length_by_pipette_id={}, active_channels_by_pipette_id={}, - nozzle_map=None, + nozzle_map_by_pipette_id={}, ) def handle_action(self, action: Action) -> None: @@ -70,6 +70,7 @@ def handle_action(self, action: Action) -> None: 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._handle_command(action.command) if isinstance(action.private_result, PipetteNozzleLayoutResultMixin): @@ -79,7 +80,7 @@ def handle_action(self, action: Action) -> None: self._state.active_channels_by_pipette_id[ pipette_id ] = nozzle_map.tip_count - self._state.nozzle_map = nozzle_map + self._state.nozzle_map_by_pipette_id[pipette_id] = nozzle_map else: self._state.active_channels_by_pipette_id[ pipette_id @@ -114,18 +115,24 @@ def _handle_command(self, command: Command) -> None: well_name = command.params.wellName pipette_id = command.params.pipetteId length = command.result.tipLength - self._set_used_tips(well_name=well_name, labware_id=labware_id) + self._set_used_tips( + pipette_id=pipette_id, well_name=well_name, labware_id=labware_id + ) self._state.length_by_pipette_id[pipette_id] = length elif isinstance(command.result, (DropTipResult, DropTipInPlaceResult)): pipette_id = command.params.pipetteId self._state.length_by_pipette_id.pop(pipette_id, None) - def _set_used_tips(self, well_name: str, labware_id: str) -> None: # noqa: C901 + def _set_used_tips( + self, pipette_id: str, well_name: str, labware_id: str + ) -> None: # noqa: C901 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 + nozzle_map = self._state.nozzle_map_by_pipette_id[pipette_id] + # fix this None logic? + # replace this with the whole nozzle map mask instead of the active nozzles? if nozzle_map is not None: num_nozzle_cols = len(nozzle_map.columns) num_nozzle_rows = len(nozzle_map.rows) @@ -152,8 +159,7 @@ def _set_used_tips(self, well_name: str, labware_id: str) -> None: # noqa: C901 well = columns[critical_column - i][critical_row - j] wells[well] = TipRackWellState.USED else: - # TODO: (cb, 2024-2-14): update/remove this case as soon as we gaurantee a nozzle map upon loading a pipette with (Jira RSS-441.) - wells[well_name] = TipRackWellState.USED + raise RuntimeError(f"No Nozzle Map found for Pipette-ID: {pipette_id}.") class TipView(HasState[TipState]): @@ -175,6 +181,7 @@ def __init__(self, state: TipState) -> None: # or backmost single nozzle configuration of an 8-channel. def get_next_tip( # noqa: C901 self, + pipette_id: str, labware_id: str, num_tips: int, starting_tip_name: Optional[str], @@ -182,10 +189,11 @@ def get_next_tip( # noqa: C901 """Get the next available clean tip.""" wells = self._state.tips_by_labware_id.get(labware_id, {}) columns = self._state.column_by_labware_id.get(labware_id, []) + nozzle_map = self._state.nozzle_map_by_pipette_id[pipette_id] - if self._state.nozzle_map is not None: - num_nozzle_cols = len(self._state.nozzle_map.columns) - num_nozzle_rows = len(self._state.nozzle_map.rows) + if nozzle_map is not None: + num_nozzle_cols = len(nozzle_map.columns) + num_nozzle_rows = len(nozzle_map.rows) def _identify_tip_cluster( critical_column: int, critical_row: int @@ -233,7 +241,7 @@ def _validate_tip_cluster(tip_cluster: List[str]) -> Optional[str]: f"Tiprack {labware_id} has no valid tip selection for current Nozzle Configuration." ) - if self._state.nozzle_map.starting_nozzle == "A1": + if nozzle_map.starting_nozzle == "A1": # Define the critical well by the position of the well relative to Tip Rack entry point H12 critical_column = len(columns) - num_nozzle_cols critical_row = len(columns[critical_column]) - num_nozzle_rows @@ -255,12 +263,12 @@ def _validate_tip_cluster(tip_cluster: List[str]) -> Optional[str]: ) return None - elif self._state.nozzle_map.starting_nozzle == "A12": + elif nozzle_map.starting_nozzle == "A12": # Define the critical well by the position of the well relative to Tip Rack entry point H1 critical_column = num_nozzle_cols - 1 critical_row = len(columns[critical_column]) - num_nozzle_rows - while critical_column <= len(columns): # change to max size of labware + while critical_column <= len(columns): # change to max size of labware tip_cluster = _identify_tip_cluster(critical_column, critical_row) result = _validate_tip_cluster(tip_cluster) if isinstance(result, str): @@ -277,7 +285,7 @@ def _validate_tip_cluster(tip_cluster: List[str]) -> Optional[str]: ) return None - elif self._state.nozzle_map.starting_nozzle == "H1": + elif nozzle_map.starting_nozzle == "H1": # Define the critical well by the position of the well relative to Tip Rack entry point A12 critical_column = len(columns) - num_nozzle_cols critical_row = num_nozzle_rows - 1 @@ -297,7 +305,7 @@ def _validate_tip_cluster(tip_cluster: List[str]) -> Optional[str]: critical_row = num_nozzle_rows - 1 return None - elif self._state.nozzle_map.starting_nozzle == "H12": + elif nozzle_map.starting_nozzle == "H12": # Define the critical well by the position of the well relative to Tip Rack entry point A1 critical_column = num_nozzle_cols - 1 critical_row = num_nozzle_rows - 1 @@ -319,10 +327,10 @@ def _validate_tip_cluster(tip_cluster: List[str]) -> Optional[str]: else: raise ValueError( - f"Nozzle {self._state.nozzle_map.starting_nozzle} is an invalid starting tip for automatic tip pickup." + f"Nozzle {nozzle_map.starting_nozzle} is an invalid starting tip for automatic tip pickup." ) else: - return None + raise RuntimeError(f"No Nozzle Map found for Pipette-ID: {pipette_id}.") def get_pipette_channels(self, pipette_id: str) -> int: """Return the given pipette's number of channels."""