feat(api): track volumes from multichannel configs

api/src/opentrons/hardware_control/nozzle_manager.py

@@ -1,11 +1,13 @@
 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,
+)
 from opentrons_shared_data.pipette.pipette_definition import (
     PipetteGeometryDefinition,
     PipetteRowDefinition,
@@ -41,43 +43,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 +78,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 +203,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 +271,7 @@ def build(  # noqa: C901
         )
 
         if (
-            NozzleConfigurationType.determine_nozzle_configuration(
+            cls.determine_nozzle_configuration(
                 physical_rows, rows, physical_columns, columns
             )
             != NozzleConfigurationType.FULL
@@ -289,6 +286,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 +300,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