feat(api): Tip tracking for all 96ch configurations (#14488)

Adds tip tracking for all 96ch and 8ch configurations as long as no starting tip is specified

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

@@ -33,6 +33,7 @@
 from opentrons_shared_data.pipette.dev_types import PipetteNameType
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 from . import deck_conflict
 
 from ..instrument import AbstractInstrument
@@ -675,6 +676,9 @@ def get_active_channels(self) -> int:
             self._pipette_id
         )
 
+    def get_nozzle_map(self) -> NozzleMap:
+        return self._engine_client.state.tips.get_pipette_nozzle_map(self._pipette_id)
+
     def has_tip(self) -> bool:
         return (
             self._engine_client.state.pipettes.get_attached_tip(self._pipette_id)
@@ -709,14 +713,9 @@ def is_tip_tracking_available(self) -> bool:
             return True
         else:
             if self.get_channels() == 96:
-                # SINGLE configuration with H12 nozzle is technically supported by the
-                # current tip tracking implementation but we don't do any deck conflict
-                # checks for it, so we won't provide full support for it yet.
-                return (
-                    self.get_nozzle_configuration() == NozzleConfigurationType.COLUMN
-                    and primary_nozzle == "A12"
-                )
+                return True
             if self.get_channels() == 8:
+                # TODO: (cb, 03/06/24): Enable automatic tip tracking on the 8 channel pipettes once PAPI support exists
                 return (
                     self.get_nozzle_configuration() == NozzleConfigurationType.SINGLE
                     and primary_nozzle == "H1"

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

@@ -11,6 +11,7 @@
 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.hardware_control.nozzle_manager import NozzleMap
 
 from ..labware import AbstractLabware, LabwareLoadParams
 from .well import WellCore
@@ -122,7 +123,10 @@ def reset_tips(self) -> None:
             raise TypeError(f"{self.get_display_name()} is not a tip rack.")
 
     def get_next_tip(
-        self, num_tips: int, starting_tip: Optional[WellCore]
+        self,
+        num_tips: int,
+        starting_tip: Optional[WellCore],
+        nozzle_map: Optional[NozzleMap],
     ) -> Optional[str]:
         return self._engine_client.state.tips.get_next_tip(
             labware_id=self._labware_id,
@@ -132,6 +136,7 @@ def get_next_tip(
                 if starting_tip and starting_tip.labware_id == self._labware_id
                 else None
             ),
+            nozzle_map=nozzle_map,
         )
 
     def get_well_columns(self) -> List[List[str]]:

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

@@ -9,6 +9,7 @@
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.protocols.api_support.util import FlowRates
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 
 from ..disposal_locations import TrashBin, WasteChute
 from .well import WellCoreType
@@ -218,6 +219,10 @@ def get_channels(self) -> int:
     def get_active_channels(self) -> int:
         ...
 
+    @abstractmethod
+    def get_nozzle_map(self) -> NozzleMap:
+        ...
+
     @abstractmethod
     def has_tip(self) -> bool:
         ...

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

@@ -11,6 +11,7 @@
 )
 
 from opentrons.types import DeckSlotName, Point
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 
 from .well import WellCoreType
 
@@ -110,7 +111,10 @@ def reset_tips(self) -> None:
 
     @abstractmethod
     def get_next_tip(
-        self, num_tips: int, starting_tip: Optional[WellCoreType]
+        self,
+        num_tips: int,
+        starting_tip: Optional[WellCoreType],
+        nozzle_map: Optional[NozzleMap],
     ) -> Optional[str]:
         """Get the name of the next available tip(s) in the rack, if available."""
 

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

@@ -18,6 +18,7 @@
 )
 from opentrons.protocols.geometry import planning
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 
 from ...disposal_locations import TrashBin, WasteChute
 from ..instrument import AbstractInstrument
@@ -550,6 +551,10 @@ 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"
 
+    def get_nozzle_map(self) -> NozzleMap:
+        """This will never be called because it was added in API 2.18."""
+        assert False, "get_nozzle_map only supported in API 2.18 & later"
+
     def is_tip_tracking_available(self) -> bool:
         # Tip tracking is always available in legacy context
         return True

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

@@ -5,6 +5,7 @@
 from opentrons.protocols.api_support.tip_tracker import TipTracker
 
 from opentrons.types import DeckSlotName, Location, Point
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 from opentrons_shared_data.labware.dev_types import LabwareParameters, LabwareDefinition
 
 from ..labware import AbstractLabware, LabwareLoadParams
@@ -153,8 +154,15 @@ def reset_tips(self) -> None:
                 well.set_has_tip(True)
 
     def get_next_tip(
-        self, num_tips: int, starting_tip: Optional[LegacyWellCore]
+        self,
+        num_tips: int,
+        starting_tip: Optional[LegacyWellCore],
+        nozzle_map: Optional[NozzleMap],
     ) -> Optional[str]:
+        if nozzle_map is not None:
+            raise ValueError(
+                "Nozzle Map cannot be provided to calls for next tip in legacy protocols."
+            )
         next_well = self._tip_tracker.next_tip(num_tips, starting_tip)
         return next_well.get_name() if next_well else None
 

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

@@ -23,6 +23,7 @@
 
 from ...disposal_locations import TrashBin, WasteChute
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 
 from ..instrument import AbstractInstrument
 
@@ -468,6 +469,10 @@ 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"
 
+    def get_nozzle_map(self) -> NozzleMap:
+        """This will never be called because it was added in API 2.18."""
+        assert False, "get_nozzle_map only supported in API 2.18 & later"
+
     def is_tip_tracking_available(self) -> bool:
         # Tip tracking is always available in legacy context
         return True

api/src/opentrons/protocol_api/instrument_context.py

@@ -27,6 +27,7 @@
     requires_version,
     APIVersionError,
 )
+from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 
 from .core.common import InstrumentCore, ProtocolCore
 from .core.engine import ENGINE_CORE_API_VERSION
@@ -56,6 +57,9 @@
 _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)
+"""The version after which a partial nozzle configuration became available for the 96 Channel Pipette."""
+_PARTIAL_NOZZLE_CONFIGURATION_AUTOMATIC_TIP_TRACKING_IN = APIVersion(2, 18)
+"""The version after which automatic tip tracking supported partially configured nozzle layouts."""
 
 
 class InstrumentContext(publisher.CommandPublisher):
@@ -877,20 +881,43 @@ def pick_up_tip(  # noqa: C901
             if self._api_version >= _PARTIAL_NOZZLE_CONFIGURATION_ADDED_IN
             else self.channels
         )
+        nozzle_map = (
+            self._core.get_nozzle_map()
+            if self._api_version
+            >= _PARTIAL_NOZZLE_CONFIGURATION_AUTOMATIC_TIP_TRACKING_IN
+            else None
+        )
 
         if location is None:
+            if (
+                nozzle_map is not None
+                and nozzle_map.configuration != NozzleConfigurationType.FULL
+                and self.starting_tip is not None
+            ):
+                # Disallowing this avoids concerning the system with the direction
+                # in which self.starting_tip consumes tips. It would currently vary
+                # depending on the configuration layout of a pipette at a given
+                # time, which means that some combination of starting tip and partial
+                # configuraiton are incompatible under the current understanding of
+                # starting tip behavior. Replacing starting_tip with an undeprecated
+                # Labware.has_tip may solve this.
+                raise CommandPreconditionViolated(
+                    "Automatic tip tracking is not available when using a partial pipette"
+                    " nozzle configuration and InstrumentContext.starting_tip."
+                    " Switch to a full configuration or set starting_tip to None."
+                )
             if not self._core.is_tip_tracking_available():
                 raise CommandPreconditionViolated(
                     "Automatic tip tracking is not available for the current pipette"
                     " nozzle configuration. We suggest switching to a configuration"
                     " that supports automatic tip tracking or specifying the exact tip"
                     " to pick up."
                 )
-
             tip_rack, well = labware.next_available_tip(
                 starting_tip=self.starting_tip,
                 tip_racks=self.tip_racks,
                 channels=active_channels,
+                nozzle_map=nozzle_map,
             )
 
         elif isinstance(location, labware.Well):
@@ -902,6 +929,7 @@ def pick_up_tip(  # noqa: C901
                 starting_tip=None,
                 tip_racks=[location],
                 channels=active_channels,
+                nozzle_map=nozzle_map,
             )
 
         elif isinstance(location, types.Location):
@@ -917,6 +945,7 @@ def pick_up_tip(  # noqa: C901
                     starting_tip=None,
                     tip_racks=[maybe_tip_rack],
                     channels=active_channels,
+                    nozzle_map=nozzle_map,
                 )
             else:
                 raise TypeError(
@@ -1323,6 +1352,12 @@ def transfer(  # noqa: C901
             if self._api_version >= _PARTIAL_NOZZLE_CONFIGURATION_ADDED_IN
             else self.channels
         )
+        nozzle_map = (
+            self._core.get_nozzle_map()
+            if self._api_version
+            >= _PARTIAL_NOZZLE_CONFIGURATION_AUTOMATIC_TIP_TRACKING_IN
+            else None
+        )
 
         if blow_out and not blowout_location:
             if self.current_volume:
@@ -1339,7 +1374,10 @@ def transfer(  # noqa: C901
 
         if new_tip != types.TransferTipPolicy.NEVER:
             tr, next_tip = labware.next_available_tip(
-                self.starting_tip, self.tip_racks, active_channels
+                self.starting_tip,
+                self.tip_racks,
+                active_channels,
+                nozzle_map=nozzle_map,
             )
             max_volume = min(next_tip.max_volume, self.max_volume)
         else:

api/src/opentrons/protocol_api/labware.py

@@ -20,6 +20,7 @@
 from opentrons.types import Location, Point
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocols.api_support.util import requires_version, APIVersionError
+from opentrons.hardware_control.nozzle_manager import NozzleMap
 
 # TODO(mc, 2022-09-02): re-exports provided for backwards compatibility
 # remove when their usage is no longer needed
@@ -883,7 +884,11 @@ def tip_length(self, length: float) -> None:
 
     # TODO(mc, 2022-11-09): implementation detail; deprecate public method
     def next_tip(
-        self, num_tips: int = 1, starting_tip: Optional[Well] = None
+        self,
+        num_tips: int = 1,
+        starting_tip: Optional[Well] = None,
+        *,
+        nozzle_map: Optional[NozzleMap] = None,
     ) -> Optional[Well]:
         """
         Find the next valid well for pick-up.
@@ -904,6 +909,7 @@ def next_tip(
         well_name = self._core.get_next_tip(
             num_tips=num_tips,
             starting_tip=starting_tip._core if starting_tip else None,
+            nozzle_map=nozzle_map,
         )
 
         return self._wells_by_name[well_name] if well_name is not None else None
@@ -1061,7 +1067,11 @@ 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(
-    tip_racks: List[Labware], num_channels: int, starting_point: Optional[Well] = None
+    tip_racks: List[Labware],
+    num_channels: int,
+    starting_point: Optional[Well] = None,
+    *,
+    nozzle_map: Optional[NozzleMap] = None,
 ) -> Tuple[Labware, Well]:
     try:
         first, rest = split_tipracks(tip_racks)
@@ -1074,14 +1084,16 @@ def select_tiprack_from_list(
         )
     elif starting_point:
         first_well = starting_point
+    elif nozzle_map:
+        first_well = None
     else:
         first_well = first.wells()[0]
 
-    next_tip = first.next_tip(num_channels, first_well)
+    next_tip = first.next_tip(num_channels, first_well, nozzle_map=nozzle_map)
     if next_tip:
         return first, next_tip
     else:
-        return select_tiprack_from_list(rest, num_channels)
+        return select_tiprack_from_list(rest, num_channels, None, nozzle_map=nozzle_map)
 
 
 # TODO(mc, 2022-11-09): implementation detail, move to core
@@ -1093,14 +1105,23 @@ def filter_tipracks_to_start(
 
 # TODO(mc, 2022-11-09): implementation detail, move to core
 def next_available_tip(
-    starting_tip: Optional[Well], tip_racks: List[Labware], channels: int
+    starting_tip: Optional[Well],
+    tip_racks: List[Labware],
+    channels: int,
+    *,
+    nozzle_map: Optional[NozzleMap] = None,
 ) -> Tuple[Labware, Well]:
     start = starting_tip
     if start is None:
-        return select_tiprack_from_list(tip_racks, channels)
+        return select_tiprack_from_list(
+            tip_racks, channels, None, nozzle_map=nozzle_map
+        )
     else:
         return select_tiprack_from_list(
-            filter_tipracks_to_start(start, tip_racks), channels, start
+            filter_tipracks_to_start(start, tip_racks),
+            channels,
+            start,
+            nozzle_map=nozzle_map,
         )