feat(api): add tip tracking for partial tip configuration

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -548,6 +548,11 @@ def get_hardware_state(self) -> PipetteDict:
     def get_channels(self) -> int:
         return self._engine_client.state.tips.get_pipette_channels(self._pipette_id)
 
+    def get_active_channels(self) -> int:
+        return self._engine_client.state.tips.get_pipette_active_channels(
+            self._pipette_id
+        )
+
     def has_tip(self) -> bool:
         return (
             self._engine_client.state.pipettes.get_attached_tip(self._pipette_id)

api/src/opentrons/protocol_api/core/instrument.py

@@ -211,6 +211,10 @@ def get_hardware_state(self) -> PipetteDict:
     def get_channels(self) -> int:
         ...
 
+    @abstractmethod
+    def get_active_channels(self) -> int:
+        ...
+
     @abstractmethod
     def has_tip(self) -> bool:
         ...

api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py

@@ -528,5 +528,9 @@ def configure_nozzle_layout(
         primary_nozzle: Optional[str],
         front_right_nozzle: Optional[str],
     ) -> None:
-        """This will never be called because it was added in API 2.15."""
+        """This will never be called because it was added in API 2.16."""
         pass
+
+    def get_active_channels(self) -> int:
+        """This will never be called because it was added in API 2.16."""
+        assert False, "get_active_channels only supported in API 2.16 & later"

api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py

@@ -448,3 +448,7 @@ def configure_nozzle_layout(
     ) -> None:
         """This will never be called because it was added in API 2.15."""
         pass
+
+    def get_active_channels(self) -> int:
+        """This will never be called because it was added in API 2.16."""
+        assert False, "get_active_channels only supported in API 2.16 & later"

api/src/opentrons/protocol_api/instrument_context.py

@@ -56,6 +56,7 @@
 """The version after which the pick-up tip procedure deprecates presses and increment arguments."""
 _DROP_TIP_LOCATION_ALTERNATING_ADDED_IN = APIVersion(2, 15)
 """The version after which a drop-tip-into-trash procedure drops tips in different alternating locations within the trash well."""
+_PARTIAL_NOZZLE_CONFIGURATION_ADDED_IN = APIVersion(2, 16)
 
 
 class InstrumentContext(publisher.CommandPublisher):
@@ -101,11 +102,9 @@ def __init__(
             default_aspirate=_DEFAULT_ASPIRATE_CLEARANCE,
             default_dispense=_DEFAULT_DISPENSE_CLEARANCE,
         )
-
         self._user_specified_trash: Union[
             labware.Labware, TrashBin, WasteChute, None
         ] = trash
-
         self.requested_as = requested_as
 
     @property  # type: ignore
@@ -817,12 +816,17 @@ def pick_up_tip(
         well: labware.Well
         tip_rack: labware.Labware
         move_to_location: Optional[types.Location] = None
+        active_channels = (
+            self.active_channels
+            if self._api_version >= _PARTIAL_NOZZLE_CONFIGURATION_ADDED_IN
+            else self.channels
+        )
 
         if location is None:
             tip_rack, well = labware.next_available_tip(
                 starting_tip=self.starting_tip,
                 tip_racks=self.tip_racks,
-                channels=self.channels,
+                channels=active_channels,
             )
 
         elif isinstance(location, labware.Well):
@@ -833,7 +837,7 @@ def pick_up_tip(
             tip_rack, well = labware.next_available_tip(
                 starting_tip=None,
                 tip_racks=[location],
-                channels=self.channels,
+                channels=active_channels,
             )
 
         elif isinstance(location, types.Location):
@@ -848,7 +852,7 @@ def pick_up_tip(
                 tip_rack, well = labware.next_available_tip(
                     starting_tip=None,
                     tip_racks=[maybe_tip_rack],
-                    channels=self.channels,
+                    channels=active_channels,
                 )
             else:
                 raise TypeError(
@@ -1233,6 +1237,11 @@ def transfer(
 
         blow_out = kwargs.get("blow_out")
         blow_out_strategy = None
+        active_channels = (
+            self.active_channels
+            if self._api_version >= _PARTIAL_NOZZLE_CONFIGURATION_ADDED_IN
+            else self.channels
+        )
 
         if blow_out and not blowout_location:
             if self.current_volume:
@@ -1249,7 +1258,7 @@ def transfer(
 
         if new_tip != types.TransferTipPolicy.NEVER:
             tr, next_tip = labware.next_available_tip(
-                self.starting_tip, self.tip_racks, self.channels
+                self.starting_tip, self.tip_racks, active_channels
             )
             max_volume = min(next_tip.max_volume, self.max_volume)
         else:
@@ -1581,6 +1590,12 @@ def channels(self) -> int:
         Possible values are 1, 8, or 96."""
         return self._core.get_channels()
 
+    @property  # type: ignore
+    @requires_version(2, 16)
+    def active_channels(self) -> int:
+        """The number of channels configured for active use using configure_nozzle_layout()."""
+        return self._core.get_active_channels()
+
     @property  # type: ignore
     @requires_version(2, 2)
     def return_height(self) -> float:
@@ -1728,6 +1743,7 @@ def configure_nozzle_layout(
         style: NozzleLayout,
         start: Optional[str] = None,
         front_right: Optional[str] = None,
+        tip_racks: Optional[List[labware.Labware]] = None,
     ) -> None:
         """Configure a pipette to pick up less than the maximum tip capacity. The pipette
         will remain in its partial state until this function is called again without any inputs. All subsequent
@@ -1743,6 +1759,7 @@ def configure_nozzle_layout(
         :param front_right: Signifies the ending nozzle in your partial configuration. It is not required for NozzleLayout.COLUMN, NozzleLayout.ROW, or NozzleLayout.SINGLE
         configurations.
         :type front_right: string or None.
+        :type tip_racks: List of tipracks to use during this configuration
 
         .. note::
             Your `start` and `front_right` strings should be formatted similarly to a well, so in the format of <LETTER><NUMBER>.
@@ -1774,5 +1791,9 @@ def configure_nozzle_layout(
                     "Cannot configure a QUADRANT layout without a front right nozzle."
                 )
         self._core.configure_nozzle_layout(
-            style, primary_nozzle=start, front_right_nozzle=front_right
+            style,
+            primary_nozzle=start,
+            front_right_nozzle=front_right,
         )
+        # TODO (spp, 2023-12-05): verify that tipracks are on adapters for only full 96 channel config
+        self._tip_racks = tip_racks or []

api/src/opentrons/protocol_api/labware.py

@@ -1022,7 +1022,7 @@ def select_tiprack_from_list(
 
     if starting_point and starting_point.parent != first:
         raise TipSelectionError(
-            "The starting tip you selected " f"does not exist in {first}"
+            f"The starting tip you selected does not exist in {first}"
         )
     elif starting_point:
         first_well = starting_point

api/src/opentrons/protocol_engine/commands/__init__.py

@@ -287,6 +287,7 @@
     ConfigureNozzleLayoutCreate,
     ConfigureNozzleLayoutParams,
     ConfigureNozzleLayoutResult,
+    ConfigureNozzleLayoutPrivateResult,
     ConfigureNozzleLayoutCommandType,
 )
 
@@ -521,6 +522,7 @@
     "ConfigureNozzleLayoutParams",
     "ConfigureNozzleLayoutResult",
     "ConfigureNozzleLayoutCommandType",
+    "ConfigureNozzleLayoutPrivateResult",
     # get pipette tip presence bundle
     "GetTipPresence",
     "GetTipPresenceCreate",

api/src/opentrons/protocol_engine/state/tips.py

@@ -16,7 +16,10 @@
     DropTipResult,
     DropTipInPlaceResult,
 )
-from ..commands.configuring_common import PipetteConfigUpdateResultMixin
+from ..commands.configuring_common import (
+    PipetteConfigUpdateResultMixin,
+    PipetteNozzleLayoutResultMixin,
+)
 
 
 class TipRackWellState(Enum):
@@ -37,6 +40,7 @@ class TipState:
     column_by_labware_id: Dict[str, List[List[str]]]
     channels_by_pipette_id: Dict[str, int]
     length_by_pipette_id: Dict[str, float]
+    active_channels_by_pipette_id: Dict[str, int]
 
 
 class TipStore(HasState[TipState], HandlesActions):
@@ -51,18 +55,31 @@ def __init__(self) -> None:
             column_by_labware_id={},
             channels_by_pipette_id={},
             length_by_pipette_id={},
+            active_channels_by_pipette_id={},
         )
 
     def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
         if isinstance(action, UpdateCommandAction):
             if isinstance(action.private_result, PipetteConfigUpdateResultMixin):
+                pipette_id = action.private_result.pipette_id
                 config = action.private_result.config
-                self._state.channels_by_pipette_id[
-                    action.private_result.pipette_id
-                ] = config.channels
+                self._state.channels_by_pipette_id[pipette_id] = config.channels
+                self._state.active_channels_by_pipette_id[pipette_id] = config.channels
             self._handle_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
+                else:
+                    self._state.active_channels_by_pipette_id[
+                        pipette_id
+                    ] = self._state.channels_by_pipette_id[pipette_id]
+
         elif isinstance(action, ResetTipsAction):
             labware_id = action.labware_id
 
@@ -92,7 +109,6 @@ 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(
                 pipette_id=pipette_id, well_name=well_name, labware_id=labware_id
             )
@@ -103,7 +119,7 @@ def _handle_command(self, command: Command) -> None:
             self._state.length_by_pipette_id.pop(pipette_id, None)
 
     def _set_used_tips(self, pipette_id: str, well_name: str, labware_id: str) -> None:
-        pipette_channels = self._state.channels_by_pipette_id.get(pipette_id)
+        pipette_channels = self._state.active_channels_by_pipette_id.get(pipette_id)
         columns = self._state.column_by_labware_id.get(labware_id, [])
         wells = self._state.tips_by_labware_id.get(labware_id, {})
 
@@ -135,14 +151,18 @@ def __init__(self, state: TipState) -> None:
         """
         self._state = state
 
+    # TODO (spp, 2023-12-05): update this logic once we support partial nozzle configurations
+    #  that require the tip tracking to move right to left or front to back;
+    #  for example when using leftmost column config of 96-channel
+    #  or backmost single nozzle configuration of an 8-channel.
     def get_next_tip(  # noqa: C901
         self, labware_id: str, num_tips: int, starting_tip_name: Optional[str]
     ) -> Optional[str]:
         """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, [])
 
-        if columns and num_tips == len(columns[0]):
+        if columns and num_tips == len(columns[0]):  # Get next tips for 8-channel
             column_head = [column[0] for column in columns]
             starting_column_index = 0
 
@@ -158,7 +178,7 @@ def get_next_tip(  # noqa: C901
                 if not any(wells[well] == TipRackWellState.USED for well in column):
                     return column[0]
 
-        elif num_tips == len(wells.keys()):
+        elif num_tips == len(wells.keys()):  # Get next tips for 96 channel
             if starting_tip_name and starting_tip_name != columns[0][0]:
                 return None
 
@@ -167,7 +187,7 @@ def get_next_tip(  # noqa: C901
             ):
                 return next(iter(wells))
 
-        else:
+        else:  # Get next tips for single channel
             if starting_tip_name is not None:
                 wells = _drop_wells_before_starting_tip(wells, starting_tip_name)
 
@@ -181,6 +201,10 @@ 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]
 
+    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]
+
     def has_clean_tip(self, labware_id: str, well_name: str) -> bool:
         """Get whether a well in a labware has a clean tip.
 

api/tests/opentrons/protocol_api/test_instrument_context.py

@@ -480,7 +480,7 @@ def test_pick_up_tip_from_labware(
     mock_well = decoy.mock(cls=Well)
     top_location = Location(point=Point(1, 2, 3), labware=mock_well)
 
-    decoy.when(mock_instrument_core.get_channels()).then_return(123)
+    decoy.when(mock_instrument_core.get_active_channels()).then_return(123)
     decoy.when(
         labware.next_available_tip(
             starting_tip=None,
@@ -534,7 +534,7 @@ def test_pick_up_tip_from_labware_location(
     location = Location(point=Point(1, 2, 3), labware=mock_tip_rack)
     top_location = Location(point=Point(1, 2, 3), labware=mock_well)
 
-    decoy.when(mock_instrument_core.get_channels()).then_return(123)
+    decoy.when(mock_instrument_core.get_active_channels()).then_return(123)
     decoy.when(
         labware.next_available_tip(
             starting_tip=None,
@@ -568,7 +568,7 @@ def test_pick_up_from_associated_tip_racks(
     mock_well = decoy.mock(cls=Well)
     top_location = Location(point=Point(1, 2, 3), labware=mock_well)
 
-    decoy.when(mock_instrument_core.get_channels()).then_return(123)
+    decoy.when(mock_instrument_core.get_active_channels()).then_return(123)
     decoy.when(
         labware.next_available_tip(
             starting_tip=mock_starting_tip,