refactor(api): control nozzlemap type better

There's this hardware controller NozzleMap type that is mostly internal
but actually exposed upstream in a couple weird uncontrolled ways.
Refactor this so that
- There's a controlled interface that is in opentrons.types and that is
the only thing that is exposed above the engine
- The engine and lower are allowed to see the actual type

This had some knock-on consequences because some functionality had to
move to lower layers. Specifically, "can this layout support LLD" is a
question that now only the engine can answer because the physical
configuration type is not in the interface, so push it down (which feels
better anyway because now you get this checked if you use the commands api).

api/src/opentrons/hardware_control/nozzle_manager.py

@@ -1,11 +1,14 @@
 from typing import Dict, List, Optional, Any, Sequence, Iterator, Tuple, cast
 from dataclasses import dataclass
 from collections import OrderedDict
-from enum import Enum
 from itertools import chain
 
 from opentrons.hardware_control.types import CriticalPoint
-from opentrons.types import Point
+from opentrons.types import (
+    Point,
+    NozzleConfigurationType,
+    NozzleMapInterface,
+)
 from opentrons_shared_data.pipette.pipette_definition import (
     PipetteGeometryDefinition,
     PipetteRowDefinition,
@@ -41,43 +44,6 @@ def _row_col_indices_for_nozzle(
     )
 
 
-class NozzleConfigurationType(Enum):
-    """
-    Nozzle Configuration Type.
-
-    Represents the current nozzle
-    configuration stored in NozzleMap
-    """
-
-    COLUMN = "COLUMN"
-    ROW = "ROW"
-    SINGLE = "SINGLE"
-    FULL = "FULL"
-    SUBRECT = "SUBRECT"
-
-    @classmethod
-    def determine_nozzle_configuration(
-        cls,
-        physical_rows: "OrderedDict[str, List[str]]",
-        current_rows: "OrderedDict[str, List[str]]",
-        physical_cols: "OrderedDict[str, List[str]]",
-        current_cols: "OrderedDict[str, List[str]]",
-    ) -> "NozzleConfigurationType":
-        """
-        Determine the nozzle configuration based on the starting and
-        ending nozzle.
-        """
-        if physical_rows == current_rows and physical_cols == current_cols:
-            return NozzleConfigurationType.FULL
-        if len(current_rows) == 1 and len(current_cols) == 1:
-            return NozzleConfigurationType.SINGLE
-        if len(current_rows) == 1:
-            return NozzleConfigurationType.ROW
-        if len(current_cols) == 1:
-            return NozzleConfigurationType.COLUMN
-        return NozzleConfigurationType.SUBRECT
-
-
 @dataclass
 class NozzleMap:
     """
@@ -113,6 +79,28 @@ class NozzleMap:
     full_instrument_rows: Dict[str, List[str]]
     #: A map of all the rows of an instrument
 
+    @classmethod
+    def determine_nozzle_configuration(
+        cls,
+        physical_rows: "OrderedDict[str, List[str]]",
+        current_rows: "OrderedDict[str, List[str]]",
+        physical_cols: "OrderedDict[str, List[str]]",
+        current_cols: "OrderedDict[str, List[str]]",
+    ) -> "NozzleConfigurationType":
+        """
+        Determine the nozzle configuration based on the starting and
+        ending nozzle.
+        """
+        if physical_rows == current_rows and physical_cols == current_cols:
+            return NozzleConfigurationType.FULL
+        if len(current_rows) == 1 and len(current_cols) == 1:
+            return NozzleConfigurationType.SINGLE
+        if len(current_rows) == 1:
+            return NozzleConfigurationType.ROW
+        if len(current_cols) == 1:
+            return NozzleConfigurationType.COLUMN
+        return NozzleConfigurationType.SUBRECT
+
     def __str__(self) -> str:
         return f"back_left_nozzle: {self.back_left} front_right_nozzle: {self.front_right} configuration: {self.configuration}"
 
@@ -216,6 +204,16 @@ def tip_count(self) -> int:
         """The total number of active nozzles in the configuration, and thus the number of tips that will be picked up."""
         return len(self.map_store)
 
+    @property
+    def physical_nozzle_count(self) -> int:
+        """The number of physical nozzles, regardless of configuration."""
+        return len(self.full_instrument_map_store)
+
+    @property
+    def active_nozzles(self) -> list[str]:
+        """An unstructured list of all nozzles active in the configuration."""
+        return list(self.map_store.keys())
+
     @classmethod
     def build(  # noqa: C901
         cls,
@@ -274,7 +272,7 @@ def build(  # noqa: C901
         )
 
         if (
-            NozzleConfigurationType.determine_nozzle_configuration(
+            cls.determine_nozzle_configuration(
                 physical_rows, rows, physical_columns, columns
             )
             != NozzleConfigurationType.FULL
@@ -289,6 +287,7 @@ def build(  # noqa: C901
             if valid_nozzle_maps.maps[map_key] == list(map_store.keys()):
                 validated_map_key = map_key
                 break
+
         if validated_map_key is None:
             raise IncompatibleNozzleConfiguration(
                 "Attempted Nozzle Configuration does not match any approved map layout for the current pipette."
@@ -302,7 +301,7 @@ def build(  # noqa: C901
             full_instrument_map_store=physical_nozzles,
             full_instrument_rows=physical_rows,
             columns=columns,
-            configuration=NozzleConfigurationType.determine_nozzle_configuration(
+            configuration=cls.determine_nozzle_configuration(
                 physical_rows, rows, physical_columns, columns
             ),
         )

api/src/opentrons/hardware_control/ot3api.py

@@ -45,7 +45,6 @@
     LiquidProbeSettings,
 )
 from opentrons.drivers.rpi_drivers.types import USBPort, PortGroup
-from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 from opentrons_shared_data.errors.exceptions import (
     EnumeratedError,
     PythonException,
@@ -1826,7 +1825,7 @@ async def tip_pickup_moves(
         if (
             self.gantry_load == GantryLoad.HIGH_THROUGHPUT
             and instrument.nozzle_manager.current_configuration.configuration
-            == NozzleConfigurationType.FULL
+            == top_types.NozzleConfigurationType.FULL
         ):
             spec = self._pipette_handler.plan_ht_pick_up_tip(
                 instrument.nozzle_manager.current_configuration.tip_count

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

@@ -1,10 +1,11 @@
 """ProtocolEngine-based InstrumentContext core implementation."""
+
 from __future__ import annotations
 
 from typing import Optional, TYPE_CHECKING, cast, Union
 from opentrons.protocols.api_support.types import APIVersion
 
-from opentrons.types import Location, Mount
+from opentrons.types import Location, Mount, NozzleConfigurationType, NozzleMapInterface
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.dev_types import PipetteDict
 from opentrons.protocols.api_support.util import FlowRates, find_value_for_api_version
@@ -32,8 +33,6 @@
 from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION
 from opentrons_shared_data.pipette.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 overlap_versions, pipette_movement_conflict
 
 from ..instrument import AbstractInstrument
@@ -737,7 +736,7 @@ def get_active_channels(self) -> int:
             self._pipette_id
         )
 
-    def get_nozzle_map(self) -> NozzleMap:
+    def get_nozzle_map(self) -> NozzleMapInterface:
         return self._engine_client.state.tips.get_pipette_nozzle_map(self._pipette_id)
 
     def has_tip(self) -> bool:
@@ -935,3 +934,9 @@ def liquid_probe_without_recovery(
         self._protocol_core.set_last_location(location=loc, mount=self.get_mount())
 
         return result.z_position
+
+    def nozzle_configuration_valid_for_lld(self) -> bool:
+        """Check if the nozzle configuration currently supports LLD."""
+        return self._engine_client.state.pipettes.get_nozzle_configuration_supports_lld(
+            self.pipette_id
+        )

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

@@ -19,8 +19,7 @@
     LabwareOffsetCreate,
     LabwareOffsetVector,
 )
-from opentrons.types import DeckSlotName, Point
-from opentrons.hardware_control.nozzle_manager import NozzleMap
+from opentrons.types import DeckSlotName, Point, NozzleMapInterface
 
 
 from ..labware import AbstractLabware, LabwareLoadParams
@@ -158,7 +157,7 @@ def get_next_tip(
         self,
         num_tips: int,
         starting_tip: Optional[WellCore],
-        nozzle_map: Optional[NozzleMap],
+        nozzle_map: Optional[NozzleMapInterface],
     ) -> Optional[str]:
         return self._engine_client.state.tips.get_next_tip(
             labware_id=self._labware_id,

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

@@ -9,7 +9,6 @@
 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
@@ -230,7 +229,7 @@ def get_active_channels(self) -> int:
         ...
 
     @abstractmethod
-    def get_nozzle_map(self) -> NozzleMap:
+    def get_nozzle_map(self) -> types.NozzleMapInterface:
         ...
 
     @abstractmethod
@@ -335,5 +334,9 @@ def liquid_probe_without_recovery(
         """Do a liquid probe to find the level of the liquid in the well."""
         ...
 
+    @abstractmethod
+    def nozzle_configuration_valid_for_lld(self) -> bool:
+        """Check if the nozzle configuration currently supports LLD."""
+
 
 InstrumentCoreType = TypeVar("InstrumentCoreType", bound=AbstractInstrument[Any])

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

@@ -10,8 +10,7 @@
     LabwareDefinition as LabwareDefinitionDict,
 )
 
-from opentrons.types import DeckSlotName, Point
-from opentrons.hardware_control.nozzle_manager import NozzleMap
+from opentrons.types import DeckSlotName, Point, NozzleMapInterface
 
 from .well import WellCoreType
 
@@ -114,7 +113,7 @@ def get_next_tip(
         self,
         num_tips: int,
         starting_tip: Optional[WellCoreType],
-        nozzle_map: Optional[NozzleMap],
+        nozzle_map: Optional[NozzleMapInterface],
     ) -> 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

@@ -19,7 +19,6 @@
 )
 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
@@ -559,7 +558,7 @@ 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:
+    def get_nozzle_map(self) -> types.NozzleMapInterface:
         """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"
 
@@ -586,3 +585,7 @@ def liquid_probe_without_recovery(
     ) -> float:
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
+
+    def nozzle_configuration_valid_for_lld(self) -> bool:
+        """Check if the nozzle configuration currently supports LLD."""
+        return False

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

@@ -4,8 +4,8 @@
 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.hardware_control.nozzle_manager import NozzleMap
+from opentrons.types import DeckSlotName, Location, Point, NozzleMapInterface
+
 from opentrons_shared_data.labware.types import LabwareParameters, LabwareDefinition
 
 from ..labware import AbstractLabware, LabwareLoadParams
@@ -157,7 +157,7 @@ def get_next_tip(
         self,
         num_tips: int,
         starting_tip: Optional[LegacyWellCore],
-        nozzle_map: Optional[NozzleMap],
+        nozzle_map: Optional[NozzleMapInterface],
     ) -> Optional[str]:
         if nozzle_map is not None:
             raise ValueError(

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

@@ -24,7 +24,6 @@
 
 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
 
@@ -477,7 +476,7 @@ 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:
+    def get_nozzle_map(self) -> types.NozzleMapInterface:
         """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"
 
@@ -504,3 +503,7 @@ def liquid_probe_without_recovery(
     ) -> float:
         """This will never be called because it was added in API 2.20."""
         assert False, "liquid_probe_without_recovery only supported in API 2.20 & later"
+
+    def nozzle_configuration_valid_for_lld(self) -> bool:
+        """Check if the nozzle configuration currently supports LLD."""
+        return False

api/src/opentrons/protocol_api/instrument_context.py

@@ -28,7 +28,6 @@
     APIVersionError,
     UnsupportedAPIError,
 )
-from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType
 
 from .core.common import InstrumentCore, ProtocolCore
 from .core.engine import ENGINE_CORE_API_VERSION
@@ -259,7 +258,7 @@ def aspirate(
             self.api_version >= APIVersion(2, 20)
             and well is not None
             and self.liquid_presence_detection
-            and self._96_tip_config_valid()
+            and self._core.nozzle_configuration_valid_for_lld()
             and self._core.get_current_volume() == 0
         ):
             self.require_liquid_presence(well=well)
@@ -946,7 +945,7 @@ def pick_up_tip(  # noqa: C901
         if location is None:
             if (
                 nozzle_map is not None
-                and nozzle_map.configuration != NozzleConfigurationType.FULL
+                and nozzle_map.configuration != types.NozzleConfigurationType.FULL
                 and self.starting_tip is not None
             ):
                 # Disallowing this avoids concerning the system with the direction
@@ -1882,19 +1881,6 @@ def _get_last_location_by_api_version(self) -> Optional[types.Location]:
         else:
             return self._protocol_core.get_last_location()
 
-    def _96_tip_config_valid(self) -> bool:
-        n_map = self._core.get_nozzle_map()
-        channels = self._core.get_active_channels()
-        if channels == 96:
-            if (
-                n_map.back_left != n_map.full_instrument_back_left
-                and n_map.front_right != n_map.full_instrument_front_right
-            ):
-                raise TipNotAttachedError(
-                    "Either the front right or the back left nozzle must have a tip attached to do LLD."
-                )
-        return True
-
     def __repr__(self) -> str:
         return "<{}: {} in {}>".format(
             self.__class__.__name__,
@@ -2156,7 +2142,6 @@ def detect_liquid_presence(self, well: labware.Well) -> bool:
             The pressure sensors for the Flex 8-channel pipette are on channels 1 and 8 (positions A1 and H1). For the Flex 96-channel pipette, the pressure sensors are on channels 1 and 96 (positions A1 and H12). Other channels on multi-channel pipettes do not have sensors and cannot detect liquid.
         """
         loc = well.top()
-        self._96_tip_config_valid()
         return self._core.detect_liquid_presence(well._core, loc)
 
     @requires_version(2, 20)
@@ -2169,7 +2154,6 @@ def require_liquid_presence(self, well: labware.Well) -> None:
             The pressure sensors for the Flex 8-channel pipette are on channels 1 and 8 (positions A1 and H1). For the Flex 96-channel pipette, the pressure sensors are on channels 1 and 96 (positions A1 and H12). Other channels on multi-channel pipettes do not have sensors and cannot detect liquid.
         """
         loc = well.top()
-        self._96_tip_config_valid()
         self._core.liquid_probe_with_recovery(well._core, loc)
 
     @requires_version(2, 20)
@@ -2184,7 +2168,6 @@ def measure_liquid_height(self, well: labware.Well) -> float:
         """
 
         loc = well.top()
-        self._96_tip_config_valid()
         height = self._core.liquid_probe_without_recovery(well._core, loc)
         return height