feat(api): track volumes from multichannel configs (#16698)

This PR adds the capability to properly track volume changes made by multichannel pipettes (and partial tip loadings of multichannel pipettes) to the engine. It also adds a quick refactor of the visibility of nozzle map types. This is well separated by commit. ## multichannels This is in commit d49e90b There are two ways in which we need to handle multichannel nozzle configurations specially compared to single-channel configurations. First, and what EXEC-795 is about, is that pipettes with multiple active nozzles will aspirate out of or dispense into multiple wells in an aspirate/dispense/in_place command. Which wells the pipette touches is a matter of projecting the pipette nozzle map out over the layout of the labware and predicting which wells are interacted with. This is itself non-trivial because labware can have many formats. What we can do is make the math work correctly when possible - when the labware is laid out normally enough that we can do projections of this type - and fall back to pretending to be a single channel if we fail. Since we're computing the logical equivalent of actual physical state, and if the labware is irregular it's unlikely that a multiple nozzle layout will physically work with the labware, I think this is safe. Specifically the thing we need to do is generalize the logic used in the tip store to project which tips are picked up by a multichannel to labware of different formats. Our multichannel pipette nozzles are laid out to match SBS 96-well plates, and so that's our "default" labware. On labware that follows SBS patterns but is more dense - a 384 plate, for instance - then we have to subsample, picking a single well in each group of (well_count / 96) that occupies the same space as a 96-well well to interact with. On labware that follows SBS patterns but is less dense - a 12-column reservoir, for instance - then we have to supersample, letting a labware well be touched by multiple nozzles. The second thing we have to deal with is that if the labware is a reservoir or reservoir-like - it has fewer wells than we have nozzles - then the common case is that multiple nozzles are in a well, and in that case if we're keeping track of the volume taken out of or added into a well we have to multiply the operation volume by the number of nozzles per well, which we can get by just dividing sizes without taking into account pattern overlap. ## nozzle maps This is in commit d64b929 This came up as I was poking around with needing the nozzle map to be visible in new places; I found it pretty awful that it was just implicitly exposed including internals, so make a new interface protocol that is explicitly exposed in `opentrons.types` and hold all of the internals, well, internal, at least to the engine. Closes EXEC-795 ## to come out of draft - [x] tests. lots of tests - [x] general approval that this is the way forward ## testing `opentrons_cli analyze` is probably enough for these because it's all logical state manipulation. - [x] this works for a single pipette the way it has always works, ditto 1-channel configurations on 8 or 96 channel pipettes - [x] 8-channel pipettes properly handle 96-standard full-column operations - [x] 8-channel pipettes properly handle 96-standard offset operations - [x] 8-channel pipettes handle 384 plates, including A1 and B1 operations - [x] 8-channel pipettes handle 12-column reservoirs - [x] 96-channel pipettes handle 96-standard full operations - [x] 96-channel pipettes handle 384 plates, including A1, B1, A2, and B2 operations - [x] 96-channel pipettes handle 12-column and 1 column reservoirs --------- Co-authored-by: Ryan Howard <[email protected]>
Opentrons · Nov 8, 2024 · 1ae4b63 · 1ae4b63
1 parent f40bdcd
commit 1ae4b63
Show file tree

Hide file tree

Showing 43 changed files with 1,344 additions and 343 deletions.
diff --git a/api/src/opentrons/hardware_control/nozzle_manager.py b/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
             ),
         )

diff --git a/api/src/opentrons/hardware_control/ot3api.py b/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

diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/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
+        )
diff --git a/api/src/opentrons/protocol_api/core/engine/labware.py b/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,

diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/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])
diff --git a/api/src/opentrons/protocol_api/core/labware.py b/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."""
 

diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/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
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
@@ -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(

diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/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