From 4aec1fcc5ba57ef2936f19f350d535b913b70911 Mon Sep 17 00:00:00 2001 From: Seth Foster Date: Wed, 6 Nov 2024 10:57:06 -0500 Subject: [PATCH] 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). --- .../hardware_control/nozzle_manager.py | 81 +++++----- api/src/opentrons/hardware_control/ot3api.py | 3 +- .../protocol_api/core/engine/instrument.py | 13 +- .../protocol_api/core/engine/labware.py | 5 +- .../opentrons/protocol_api/core/instrument.py | 7 +- .../opentrons/protocol_api/core/labware.py | 5 +- .../core/legacy/legacy_instrument_core.py | 7 +- .../core/legacy/legacy_labware_core.py | 6 +- .../legacy_instrument_core.py | 7 +- .../protocol_api/instrument_context.py | 21 +-- api/src/opentrons/protocol_api/labware.py | 10 +- .../protocol_engine/commands/liquid_probe.py | 6 + .../protocol_engine/execution/tip_handler.py | 6 +- .../protocol_engine/state/_well_math.py | 1 - .../protocol_engine/state/pipettes.py | 16 +- .../opentrons/protocol_engine/state/tips.py | 11 +- .../protocol_runner/run_orchestrator.py | 7 +- .../protocols/advanced_control/transfers.py | 5 +- api/src/opentrons/types.py | 92 +++++++++++- .../instruments/test_nozzle_manager.py | 117 ++++++--------- .../hardware_control/test_ot3_api.py | 3 +- .../core/engine/test_deck_conflict.py | 9 +- .../core/engine/test_instrument_core.py | 3 +- .../protocol_api/test_instrument_context.py | 62 +------- .../commands/test_liquid_probe.py | 40 ++++- .../state/test_pipette_view.py | 139 +++++++++++++++++- .../robot_server/runs/router/base_router.py | 2 +- .../robot_server/runs/run_data_manager.py | 8 +- .../runs/run_orchestrator_store.py | 7 +- .../tests/runs/router/test_base_router.py | 4 +- 30 files changed, 442 insertions(+), 261 deletions(-) diff --git a/api/src/opentrons/hardware_control/nozzle_manager.py b/api/src/opentrons/hardware_control/nozzle_manager.py index bf42476f7ee..a80683c6a3b 100644 --- a/api/src/opentrons/hardware_control/nozzle_manager.py +++ b/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 ), ) diff --git a/api/src/opentrons/hardware_control/ot3api.py b/api/src/opentrons/hardware_control/ot3api.py index f90a0a539dc..f652ea05308 100644 --- 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 index dc174988069..2f172c8cda2 100644 --- 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 index f09a51ef181..cf92a5ad3b8 100644 --- 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 index f88633a7a6d..d17ab43dd4f 100644 --- 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 index 67b452cca6d..c82dc7f1b06 100644 --- 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 index c112fc32abc..90a8a05c6da 100644 --- 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 index 575fd7a8cc6..241d8b932df 100644 --- 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 index f02d1e66fd1..66c33aae511 100644 --- 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 diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py index 93c485f8087..eed78c93813 100644 --- a/api/src/opentrons/protocol_api/instrument_context.py +++ b/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 diff --git a/api/src/opentrons/protocol_api/labware.py b/api/src/opentrons/protocol_api/labware.py index 0e8a17d07d3..9ba7a32c5e8 100644 --- a/api/src/opentrons/protocol_api/labware.py +++ b/api/src/opentrons/protocol_api/labware.py @@ -17,14 +17,14 @@ from opentrons_shared_data.labware.types import LabwareDefinition, LabwareParameters -from opentrons.types import Location, Point +from opentrons.types import Location, Point, NozzleMapInterface from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import ( requires_version, APIVersionError, UnsupportedAPIError, ) -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 @@ -932,7 +932,7 @@ def next_tip( num_tips: int = 1, starting_tip: Optional[Well] = None, *, - nozzle_map: Optional[NozzleMap] = None, + nozzle_map: Optional[NozzleMapInterface] = None, ) -> Optional[Well]: """ Find the next valid well for pick-up. @@ -1121,7 +1121,7 @@ def select_tiprack_from_list( num_channels: int, starting_point: Optional[Well] = None, *, - nozzle_map: Optional[NozzleMap] = None, + nozzle_map: Optional[NozzleMapInterface] = None, ) -> Tuple[Labware, Well]: try: first, rest = split_tipracks(tip_racks) @@ -1159,7 +1159,7 @@ def next_available_tip( tip_racks: List[Labware], channels: int, *, - nozzle_map: Optional[NozzleMap] = None, + nozzle_map: Optional[NozzleMapInterface] = None, ) -> Tuple[Labware, Well]: start = starting_tip if start is None: diff --git a/api/src/opentrons/protocol_engine/commands/liquid_probe.py b/api/src/opentrons/protocol_engine/commands/liquid_probe.py index f78cd5bb55c..b1d41fc4c50 100644 --- a/api/src/opentrons/protocol_engine/commands/liquid_probe.py +++ b/api/src/opentrons/protocol_engine/commands/liquid_probe.py @@ -12,6 +12,7 @@ PipetteNotReadyToAspirateError, TipNotEmptyError, IncompleteLabwareDefinitionError, + TipNotAttachedError, ) from opentrons.types import MountType from opentrons_shared_data.errors.exceptions import ( @@ -111,6 +112,10 @@ async def _execute_common( pipette_id = params.pipetteId labware_id = params.labwareId well_name = params.wellName + if not state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id): + raise TipNotAttachedError( + "Either the front right or back left nozzle must have a tip attached to probe liquid height." + ) state_update = update_types.StateUpdate() @@ -202,6 +207,7 @@ async def execute(self, params: _CommonParams) -> _LiquidProbeExecuteReturn: MustHomeError: as an undefined error, if the plunger is not in a valid position. """ + z_pos_or_error, state_update, deck_point = await _execute_common( self._state_view, self._movement, self._pipetting, params ) diff --git a/api/src/opentrons/protocol_engine/execution/tip_handler.py b/api/src/opentrons/protocol_engine/execution/tip_handler.py index dde67ece007..2a6816dcfdd 100644 --- a/api/src/opentrons/protocol_engine/execution/tip_handler.py +++ b/api/src/opentrons/protocol_engine/execution/tip_handler.py @@ -1,11 +1,12 @@ """Tip pickup and drop procedures.""" + from typing import Optional, Dict from typing_extensions import Protocol as TypingProtocol from opentrons.hardware_control import HardwareControlAPI from opentrons.hardware_control.types import FailedTipStateCheck, InstrumentProbeType from opentrons.protocol_engine.errors.exceptions import PickUpTipTipNotAttachedError -from opentrons.types import Mount +from opentrons.types import Mount, NozzleConfigurationType from opentrons_shared_data.errors.exceptions import ( CommandPreconditionViolated, @@ -23,9 +24,6 @@ ProtocolEngineError, ) -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType - - PRIMARY_NOZZLE_TO_ENDING_NOZZLE_MAP = { "A1": {"COLUMN": "H1", "ROW": "A12"}, "H1": {"COLUMN": "A1", "ROW": "H12"}, diff --git a/api/src/opentrons/protocol_engine/state/_well_math.py b/api/src/opentrons/protocol_engine/state/_well_math.py index 5f4b9bcca19..f50b88be122 100644 --- a/api/src/opentrons/protocol_engine/state/_well_math.py +++ b/api/src/opentrons/protocol_engine/state/_well_math.py @@ -1,7 +1,6 @@ """Utilities for doing coverage math on wells.""" from typing import Iterator -from typing_extensions import assert_never from opentrons_shared_data.errors.exceptions import ( InvalidStoredData, InvalidProtocolData, diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py index 9f90a89ffa7..e0f2cef1155 100644 --- a/api/src/opentrons/protocol_engine/state/pipettes.py +++ b/api/src/opentrons/protocol_engine/state/pipettes.py @@ -1,4 +1,5 @@ """Basic pipette data state and store.""" + from __future__ import annotations import dataclasses @@ -14,15 +15,13 @@ from typing_extensions import assert_never from opentrons_shared_data.pipette import pipette_definition -from opentrons_shared_data.labware.utils import well_ordinals_from_well_name from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE from opentrons.hardware_control.dev_types import PipetteDict from opentrons.hardware_control import CriticalPoint from opentrons.hardware_control.nozzle_manager import ( - NozzleConfigurationType, NozzleMap, ) -from opentrons.types import MountType, Mount as HwMount, Point +from opentrons.types import MountType, Mount as HwMount, Point, NozzleConfigurationType from . import update_types, fluid_stack from .. import errors @@ -762,3 +761,14 @@ def get_liquid_presence_detection(self, pipette_id: str) -> bool: raise errors.PipetteNotLoadedError( f"Pipette {pipette_id} not found; unable to determine if pipette liquid presence detection enabled." ) from e + + def get_nozzle_configuration_supports_lld(self, pipette_id: str) -> bool: + """Determine if the current partial tip configuration supports LLD.""" + nozzle_map = self.get_nozzle_configuration(pipette_id) + if ( + nozzle_map.physical_nozzle_count == 96 + and nozzle_map.back_left != nozzle_map.full_instrument_back_left + and nozzle_map.front_right != nozzle_map.full_instrument_front_right + ): + return False + return True diff --git a/api/src/opentrons/protocol_engine/state/tips.py b/api/src/opentrons/protocol_engine/state/tips.py index 507ad67bab4..c5c097ec693 100644 --- a/api/src/opentrons/protocol_engine/state/tips.py +++ b/api/src/opentrons/protocol_engine/state/tips.py @@ -1,8 +1,10 @@ """Tip state tracking.""" + from dataclasses import dataclass from enum import Enum from typing import Dict, Optional, List, Union +from opentrons.types import NozzleMapInterface from opentrons.protocol_engine.state import update_types from ._abstract_store import HasState, HandlesActions @@ -135,7 +137,7 @@ def get_next_tip( # noqa: C901 labware_id: str, num_tips: int, starting_tip_name: Optional[str], - nozzle_map: Optional[NozzleMap], + nozzle_map: Optional[NozzleMapInterface], ) -> Optional[str]: """Get the next available clean tip. Does not support use of a starting tip if the pipette used is in a partial configuration.""" wells = self._state.tips_by_labware_id.get(labware_id, {}) @@ -192,10 +194,7 @@ def _validate_tip_cluster( return None else: # In the case of an 8ch pipette where a column has mixed state tips we may simply progress to the next column in our search - if ( - nozzle_map is not None - and len(nozzle_map.full_instrument_map_store) == 8 - ): + if nozzle_map is not None and nozzle_map.physical_nozzle_count == 8: return None # In the case of a 96ch we can attempt to index in by singular rows and columns assuming that indexed direction is safe @@ -325,7 +324,7 @@ def _cluster_search_H12(active_columns: int, active_rows: int) -> Optional[str]: return None if starting_tip_name is None and nozzle_map is not None and columns: - num_channels = len(nozzle_map.full_instrument_map_store) + num_channels = nozzle_map.physical_nozzle_count num_nozzle_cols = len(nozzle_map.columns) num_nozzle_rows = len(nozzle_map.rows) # Each pipette's cluster search is determined by the point of entry for a given pipette/configuration: diff --git a/api/src/opentrons/protocol_runner/run_orchestrator.py b/api/src/opentrons/protocol_runner/run_orchestrator.py index dfa66e6a55a..8339b00f930 100644 --- a/api/src/opentrons/protocol_runner/run_orchestrator.py +++ b/api/src/opentrons/protocol_runner/run_orchestrator.py @@ -1,11 +1,13 @@ """Engine/Runner provider.""" + from __future__ import annotations import enum -from typing import Optional, Union, List, Dict, AsyncGenerator +from typing import Optional, Union, List, Dict, AsyncGenerator, Mapping from anyio import move_on_after +from opentrons.types import NozzleMapInterface from opentrons_shared_data.labware.types import LabwareUri from opentrons_shared_data.labware.labware_definition import LabwareDefinition from opentrons_shared_data.errors import GeneralError @@ -14,7 +16,6 @@ from . import protocol_runner, RunResult, JsonRunner, PythonAndLegacyRunner from ..hardware_control import HardwareControlAPI from ..hardware_control.modules import AbstractModule as HardwareModuleAPI -from ..hardware_control.nozzle_manager import NozzleMap from ..protocol_engine import ( ProtocolEngine, CommandCreate, @@ -414,7 +415,7 @@ def get_deck_type(self) -> DeckType: """Get engine deck type.""" return self._protocol_engine.state_view.config.deck_type - def get_nozzle_maps(self) -> Dict[str, NozzleMap]: + def get_nozzle_maps(self) -> Mapping[str, NozzleMapInterface]: """Get current nozzle maps keyed by pipette id.""" return self._protocol_engine.state_view.tips.get_pipette_nozzle_maps() diff --git a/api/src/opentrons/protocols/advanced_control/transfers.py b/api/src/opentrons/protocols/advanced_control/transfers.py index 5ad9dd64d24..77e0de81c8b 100644 --- a/api/src/opentrons/protocols/advanced_control/transfers.py +++ b/api/src/opentrons/protocols/advanced_control/transfers.py @@ -20,7 +20,6 @@ from opentrons.protocol_api.labware import Labware, Well from opentrons import types from opentrons.protocols.api_support.types import APIVersion -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType AdvancedLiquidHandling = Union[ @@ -435,7 +434,7 @@ def __init__( # then avoid iterating through its Wells. # ii. if using single channel pipettes, flatten a multi-dimensional # list of Wells into a 1 dimensional list of Wells - pipette_configuration_type = NozzleConfigurationType.FULL + pipette_configuration_type = types.NozzleConfigurationType.FULL normalized_sources: List[Union[Well, types.Location]] normalized_dests: List[Union[Well, types.Location]] if self._api_version >= _PARTIAL_TIP_SUPPORT_ADDED: @@ -444,7 +443,7 @@ def __init__( ) if ( self._instr.channels > 1 - and pipette_configuration_type == NozzleConfigurationType.FULL + and pipette_configuration_type == types.NozzleConfigurationType.FULL ): normalized_sources, normalized_dests = self._multichannel_transfer( srcs, dsts diff --git a/api/src/opentrons/types.py b/api/src/opentrons/types.py index 22611393f40..e231ab9bd48 100644 --- a/api/src/opentrons/types.py +++ b/api/src/opentrons/types.py @@ -1,7 +1,16 @@ from __future__ import annotations import enum from math import sqrt, isclose -from typing import TYPE_CHECKING, Any, NamedTuple, Iterator, Union, List, Optional +from typing import ( + TYPE_CHECKING, + Any, + NamedTuple, + Iterator, + Union, + List, + Optional, + Protocol, +) from opentrons_shared_data.robot.types import RobotType @@ -426,3 +435,84 @@ class TransferTipPolicy(enum.Enum): DeckLocation = Union[int, str] ALLOWED_PRIMARY_NOZZLES = ["A1", "H1", "A12", "H12"] + + +class NozzleConfigurationType(enum.Enum): + """Short names for types of nozzle configurations. + + Represents the current nozzle configuration stored in a NozzleMap. + """ + + COLUMN = "COLUMN" + ROW = "ROW" + SINGLE = "SINGLE" + FULL = "FULL" + SUBRECT = "SUBRECT" + + +class NozzleMapInterface(Protocol): + """ + A NozzleMap instance represents a specific configuration of active nozzles on a pipette. + + It exposes properties of the configuration like the configuration's front-right, front-left, + back-left and starting nozzles as well as a map of all the nozzles active in the configuration. + + Because NozzleMaps represent configurations directly, the properties of the NozzleMap may not + match the properties of the physical pipette. For instance, a NozzleMap for a single channel + configuration of an 8-channel pipette - say, A1 only - will have its front left, front right, + and active channels all be A1, while the physical configuration would have the front right + channel be H1. + """ + + @property + def starting_nozzle(self) -> str: + """The nozzle that automated operations that count nozzles should start at.""" + ... + + @property + def rows(self) -> dict[str, list[str]]: + """A map of all the rows active in this configuration.""" + ... + + @property + def columns(self) -> dict[str, list[str]]: + """A map of all the columns active in this configuration.""" + ... + + @property + def back_left(self) -> str: + """The backest, leftest (i.e. back if it's a column, left if it's a row) nozzle of the configuration. + + Note: This is the value relevant for this particular configuration, and it may not represent the back left nozzle + of the underlying physical pipette. For instance, the back-left nozzle of a configuration representing nozzles + D7 to H12 of a 96-channel pipette is D7, which is not the back-left nozzle of the physical pipette (A1). + """ + ... + + @property + def configuration(self) -> NozzleConfigurationType: + """The kind of configuration represented by this nozzle map.""" + ... + + @property + def front_right(self) -> str: + """The frontest, rightest (i.e. front if it's a column, right if it's a row) nozzle of the configuration. + + Note: This is the value relevant for this configuration, not the physical pipette. See the note on back_left. + """ + ... + + @property + 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.""" + ... + + @property + def physical_nozzle_count(self) -> int: + """The number of actual physical nozzles on the pipette, regardless of configuration.""" + ... + + @property + def active_nozzles(self) -> list[str]: + """An unstructured list of all nozzles active in the configuration.""" + ... diff --git a/api/tests/opentrons/hardware_control/instruments/test_nozzle_manager.py b/api/tests/opentrons/hardware_control/instruments/test_nozzle_manager.py index 5030bec31fe..ba1f10aaaef 100644 --- a/api/tests/opentrons/hardware_control/instruments/test_nozzle_manager.py +++ b/api/tests/opentrons/hardware_control/instruments/test_nozzle_manager.py @@ -3,7 +3,7 @@ from opentrons.hardware_control import nozzle_manager -from opentrons.types import Point +from opentrons.types import Point, NozzleConfigurationType from opentrons_shared_data.pipette.load_data import load_definition from opentrons_shared_data.pipette.types import ( @@ -258,7 +258,7 @@ ], ) def test_single_pipettes_always_full( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.SINGLE_CHANNEL, pipette_details[1] @@ -266,22 +266,13 @@ def test_single_pipettes_always_full( subject = nozzle_manager.NozzleConfigurationManager.build_from_config( config, ValidNozzleMaps(maps=A1) ) - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.update_nozzle_configuration("A1", "A1", "A1") - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.reset_to_default_configuration() - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL @pytest.mark.parametrize( @@ -295,7 +286,7 @@ def test_single_pipettes_always_full( ], ) def test_single_pipette_map_entries( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.SINGLE_CHANNEL, pipette_details[1] @@ -332,7 +323,7 @@ def test_map_entries(nozzlemap: nozzle_manager.NozzleMap) -> None: ], ) def test_single_pipette_map_geometry( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.SINGLE_CHANNEL, pipette_details[1] @@ -365,7 +356,7 @@ def test_map_geometry(nozzlemap: nozzle_manager.NozzleMap) -> None: ], ) def test_multi_config_identification( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.EIGHT_CHANNEL, pipette_details[1] @@ -375,55 +366,43 @@ def test_multi_config_identification( ValidNozzleMaps(maps=EIGHT_CHANNEL_FULL | A1_D1 | A1 | H1), ) - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.update_nozzle_configuration("A1", "H1", "A1") - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.reset_to_default_configuration() - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.update_nozzle_configuration("A1", "D1", "A1") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.COLUMN + == NozzleConfigurationType.COLUMN ) subject.update_nozzle_configuration("A1", "A1", "A1") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SINGLE + == NozzleConfigurationType.SINGLE ) subject.update_nozzle_configuration("H1", "H1", "H1") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SINGLE + == NozzleConfigurationType.SINGLE ) subject.reset_to_default_configuration() - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL @pytest.mark.parametrize( @@ -437,7 +416,7 @@ def test_multi_config_identification( ], ) def test_multi_config_map_entries( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.EIGHT_CHANNEL, pipette_details[1] @@ -503,7 +482,7 @@ def assert_offset_in_center_of( ], ) def test_multi_config_geometry( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.EIGHT_CHANNEL, pipette_details[1] @@ -554,7 +533,7 @@ def test_map_geometry( "pipette_details", [(PipetteModelType.p1000, PipetteVersionType(major=3, minor=5))] ) def test_96_config_identification( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.NINETY_SIX_CHANNEL, pipette_details[1] @@ -577,97 +556,91 @@ def test_96_config_identification( ), ) - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.update_nozzle_configuration("A1", "H12") - assert ( - subject.current_configuration.configuration - == nozzle_manager.NozzleConfigurationType.FULL - ) + assert subject.current_configuration.configuration == NozzleConfigurationType.FULL subject.update_nozzle_configuration("A1", "H1") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.COLUMN + == NozzleConfigurationType.COLUMN ) subject.update_nozzle_configuration("A12", "H12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.COLUMN + == NozzleConfigurationType.COLUMN ) subject.update_nozzle_configuration("A1", "A12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.ROW + == NozzleConfigurationType.ROW ) subject.update_nozzle_configuration("H1", "H12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.ROW + == NozzleConfigurationType.ROW ) subject.update_nozzle_configuration("E1", "H6") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) subject.update_nozzle_configuration("E7", "H12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) subject.update_nozzle_configuration("A1", "B12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) subject.update_nozzle_configuration("G1", "H12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) subject.update_nozzle_configuration("A1", "H3") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) subject.update_nozzle_configuration("A10", "H12") assert ( cast( - nozzle_manager.NozzleConfigurationType, + NozzleConfigurationType, subject.current_configuration.configuration, ) - == nozzle_manager.NozzleConfigurationType.SUBRECT + == NozzleConfigurationType.SUBRECT ) @@ -675,7 +648,7 @@ def test_96_config_identification( "pipette_details", [(PipetteModelType.p1000, PipetteVersionType(major=3, minor=5))] ) def test_96_config_map_entries( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.NINETY_SIX_CHANNEL, pipette_details[1] @@ -1012,7 +985,7 @@ def _nozzles() -> Iterator[str]: "pipette_details", [(PipetteModelType.p1000, PipetteVersionType(major=3, minor=5))] ) def test_96_config_geometry( - pipette_details: Tuple[PipetteModelType, PipetteVersionType] + pipette_details: Tuple[PipetteModelType, PipetteVersionType], ) -> None: config = load_definition( pipette_details[0], PipetteChannelType.NINETY_SIX_CHANNEL, pipette_details[1] diff --git a/api/tests/opentrons/hardware_control/test_ot3_api.py b/api/tests/opentrons/hardware_control/test_ot3_api.py index 064ea087c6b..2a553ea35bd 100644 --- a/api/tests/opentrons/hardware_control/test_ot3_api.py +++ b/api/tests/opentrons/hardware_control/test_ot3_api.py @@ -59,14 +59,13 @@ EstopStateNotification, TipStateType, ) -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType from opentrons.hardware_control.errors import InvalidCriticalPoint from opentrons.hardware_control.ot3api import OT3API from opentrons.hardware_control import ThreadManager from opentrons.hardware_control.backends.ot3simulator import OT3Simulator from opentrons_hardware.firmware_bindings.constants import NodeId -from opentrons.types import Point, Mount +from opentrons.types import Point, Mount, NozzleConfigurationType from opentrons_hardware.hardware_control.motion_planning.types import Move diff --git a/api/tests/opentrons/protocol_api/core/engine/test_deck_conflict.py b/api/tests/opentrons/protocol_api/core/engine/test_deck_conflict.py index 42e17983018..a6f981733e5 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_deck_conflict.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_deck_conflict.py @@ -7,7 +7,6 @@ from opentrons_shared_data.robot.types import RobotType from opentrons.hardware_control import CriticalPoint -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType from opentrons.motion_planning import deck_conflict as wrapped_deck_conflict from opentrons.motion_planning import adjacent_slots_getters from opentrons.motion_planning.adjacent_slots_getters import _MixedTypeSlots @@ -31,7 +30,13 @@ from opentrons.protocol_engine.clients import SyncClient from opentrons.protocol_engine.errors import LabwareNotLoadedOnModuleError -from opentrons.types import DeckSlotName, Point, StagingSlotName, MountType +from opentrons.types import ( + DeckSlotName, + Point, + StagingSlotName, + MountType, + NozzleConfigurationType, +) from opentrons.protocol_engine.types import ( DeckType, diff --git a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py index 0ab9ac9da73..b2dff4a7254 100644 --- a/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py +++ b/api/tests/opentrons/protocol_api/core/engine/test_instrument_core.py @@ -10,7 +10,6 @@ from opentrons.hardware_control import SyncHardwareAPI from opentrons.hardware_control.dev_types import PipetteDict -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType from opentrons.protocol_engine import ( DeckPoint, LoadedPipette, @@ -51,7 +50,7 @@ ) from opentrons.protocols.api_support.definitions import MAX_SUPPORTED_VERSION from opentrons.protocols.api_support.types import APIVersion -from opentrons.types import Location, Mount, MountType, Point +from opentrons.types import Location, Mount, MountType, Point, NozzleConfigurationType from ... import versions_below, versions_at_or_above diff --git a/api/tests/opentrons/protocol_api/test_instrument_context.py b/api/tests/opentrons/protocol_api/test_instrument_context.py index 27d2f6ebb33..6523ef20fd0 100644 --- a/api/tests/opentrons/protocol_api/test_instrument_context.py +++ b/api/tests/opentrons/protocol_api/test_instrument_context.py @@ -1,4 +1,5 @@ """Tests for the InstrumentContext public interface.""" + import inspect import pytest from collections import OrderedDict @@ -27,11 +28,6 @@ INSTRUMENT_CORE_NOZZLE_LAYOUT_TEST_SPECS, ExpectedCoreArgs, ) -from tests.opentrons.protocol_engine.pipette_fixtures import ( - NINETY_SIX_COLS, - NINETY_SIX_MAP, - NINETY_SIX_ROWS, -) from opentrons.protocols.api_support import instrument as mock_instrument_support from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import ( @@ -1490,59 +1486,6 @@ def test_measure_liquid_height( assert pcfe.value is errorToRaise -def test_96_tip_config_valid( - decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext -) -> None: - """It should error when there's no tips on the correct corner nozzles.""" - nozzle_map = NozzleMap.build( - physical_nozzles=NINETY_SIX_MAP, - physical_rows=NINETY_SIX_ROWS, - physical_columns=NINETY_SIX_COLS, - starting_nozzle="A5", - back_left_nozzle="A5", - front_right_nozzle="H5", - valid_nozzle_maps=ValidNozzleMaps(maps={"Column12": NINETY_SIX_COLS["5"]}), - ) - decoy.when(mock_instrument_core.get_nozzle_map()).then_return(nozzle_map) - decoy.when(mock_instrument_core.get_active_channels()).then_return(96) - with pytest.raises(TipNotAttachedError): - subject._96_tip_config_valid() - - -def test_96_tip_config_invalid( - decoy: Decoy, mock_instrument_core: InstrumentCore, subject: InstrumentContext -) -> None: - """It should return True when there are tips on the correct corner nozzles.""" - nozzle_map = NozzleMap.build( - physical_nozzles=NINETY_SIX_MAP, - physical_rows=NINETY_SIX_ROWS, - physical_columns=NINETY_SIX_COLS, - starting_nozzle="A1", - back_left_nozzle="A1", - front_right_nozzle="H12", - valid_nozzle_maps=ValidNozzleMaps( - maps={ - "Full": sum( - [ - NINETY_SIX_ROWS["A"], - NINETY_SIX_ROWS["B"], - NINETY_SIX_ROWS["C"], - NINETY_SIX_ROWS["D"], - NINETY_SIX_ROWS["E"], - NINETY_SIX_ROWS["F"], - NINETY_SIX_ROWS["G"], - NINETY_SIX_ROWS["H"], - ], - [], - ) - } - ), - ) - decoy.when(mock_instrument_core.get_nozzle_map()).then_return(nozzle_map) - decoy.when(mock_instrument_core.get_active_channels()).then_return(96) - assert subject._96_tip_config_valid() is True - - @pytest.mark.parametrize( "api_version", versions_between( @@ -1628,6 +1571,9 @@ def test_mix_with_lpd( decoy.when(mock_instrument_core.get_dispense_flow_rate(1.23)).then_return(5.67) decoy.when(mock_instrument_core.has_tip()).then_return(True) decoy.when(mock_instrument_core.get_current_volume()).then_return(0.0) + decoy.when(mock_instrument_core.nozzle_configuration_valid_for_lld()).then_return( + True + ) subject.liquid_presence_detection = True subject.mix(repetitions=10, volume=10.0, location=input_location, rate=1.23) diff --git a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py index 2cada4f3e24..1624da03fd6 100644 --- a/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py +++ b/api/tests/opentrons/protocol_engine/commands/test_liquid_probe.py @@ -1,7 +1,11 @@ """Test LiquidProbe commands.""" + from datetime import datetime from typing import Type, Union +from decoy import matchers, Decoy +import pytest + from opentrons.protocol_engine.errors.exceptions import ( MustHomeError, PipetteNotReadyToAspirateError, @@ -11,8 +15,9 @@ from opentrons_shared_data.errors.exceptions import ( PipetteLiquidNotFoundError, ) -from decoy import matchers, Decoy -import pytest +from opentrons_shared_data.pipette.pipette_definition import ValidNozzleMaps + +from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_engine.commands.pipetting_common import LiquidNotFoundError from opentrons.protocol_engine.state.state import StateView @@ -37,6 +42,12 @@ ) from opentrons.protocol_engine.resources.model_utils import ModelUtils +from ..pipette_fixtures import ( + NINETY_SIX_COLS, + NINETY_SIX_MAP, + NINETY_SIX_ROWS, +) + EitherImplementationType = Union[ Type[LiquidProbeImplementation], Type[TryLiquidProbeImplementation] @@ -61,7 +72,7 @@ def types( @pytest.fixture def implementation_type( - types: tuple[EitherImplementationType, object, object] + types: tuple[EitherImplementationType, object, object], ) -> EitherImplementationType: """Return an implementation type. Kept in sync with the params and result types.""" return types[0] @@ -145,6 +156,9 @@ async def test_liquid_probe_implementation( height=15.0, ), ).then_return(30.0) + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld("abc") + ).then_return(True) timestamp = datetime(year=2020, month=1, day=2) decoy.when(model_utils.get_timestamp()).then_return(timestamp) @@ -219,7 +233,9 @@ async def test_liquid_not_found_error( well_location=well_location, ), ).then_raise(PipetteLiquidNotFoundError()) - + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) decoy.when(model_utils.generate_id()).then_return(error_id) decoy.when(model_utils.get_timestamp()).then_return(error_timestamp) @@ -278,7 +294,9 @@ async def test_liquid_probe_tip_checking( wellName=well_name, wellLocation=well_location, ) - + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_raise( TipNotAttachedError() ) @@ -306,7 +324,9 @@ async def test_liquid_probe_plunger_preparedness_checking( wellName=well_name, wellLocation=well_location, ) - + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) decoy.when(state_view.pipettes.get_aspirated_volume(pipette_id)).then_return(None) with pytest.raises(PipetteNotReadyToAspirateError): await subject.execute(data) @@ -336,12 +356,17 @@ async def test_liquid_probe_volume_checking( decoy.when( state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id), ).then_return(123) + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) + with pytest.raises(TipNotEmptyError): await subject.execute(data) decoy.when( state_view.pipettes.get_aspirated_volume(pipette_id=pipette_id), ).then_return(None) + with pytest.raises(PipetteNotReadyToAspirateError): await subject.execute(data) @@ -373,5 +398,8 @@ async def test_liquid_probe_location_checking( mount=MountType.LEFT, ), ).then_return(False) + decoy.when( + state_view.pipettes.get_nozzle_configuration_supports_lld(pipette_id) + ).then_return(True) with pytest.raises(MustHomeError): await subject.execute(data) diff --git a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py index 60bb528ba85..64e663a24e5 100644 --- a/api/tests/opentrons/protocol_engine/state/test_pipette_view.py +++ b/api/tests/opentrons/protocol_engine/state/test_pipette_view.py @@ -1,4 +1,5 @@ """Tests for pipette state accessors in the protocol_engine state store.""" + from collections import OrderedDict from typing import cast, Dict, List, Optional, Tuple, NamedTuple @@ -12,7 +13,7 @@ from opentrons.config.defaults_ot2 import Z_RETRACT_DISTANCE from opentrons.hardware_control import CriticalPoint -from opentrons.types import MountType, Mount as HwMount, Point +from opentrons.types import MountType, Mount as HwMount, Point, NozzleConfigurationType from opentrons.hardware_control.dev_types import PipetteDict from opentrons.protocol_engine import errors from opentrons.protocol_engine.types import ( @@ -33,7 +34,7 @@ PipetteBoundingBoxOffsets, ) from opentrons.protocol_engine.state import fluid_stack -from opentrons.hardware_control.nozzle_manager import NozzleMap, NozzleConfigurationType +from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.protocol_engine.errors import TipNotAttachedError, PipetteNotLoadedError from ..pipette_fixtures import ( @@ -977,3 +978,137 @@ def test_get_pipette_bounds_at_location( ) == pipette_bounds_result ) + + +@pytest.mark.parametrize( + "nozzle_map,allowed", + [ + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="H12", + valid_nozzle_maps=ValidNozzleMaps( + maps={ + "Full": sum( + [ + NINETY_SIX_ROWS["A"], + NINETY_SIX_ROWS["B"], + NINETY_SIX_ROWS["C"], + NINETY_SIX_ROWS["D"], + NINETY_SIX_ROWS["E"], + NINETY_SIX_ROWS["F"], + NINETY_SIX_ROWS["G"], + NINETY_SIX_ROWS["H"], + ], + [], + ) + } + ), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="H1", + valid_nozzle_maps=ValidNozzleMaps( + maps={"Column1": NINETY_SIX_COLS["1"]} + ), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A12", + back_left_nozzle="A12", + front_right_nozzle="H12", + valid_nozzle_maps=ValidNozzleMaps( + maps={"Column12": NINETY_SIX_COLS["12"]} + ), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="A1", + valid_nozzle_maps=ValidNozzleMaps(maps={"Single": ["A1"]}), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=OrderedDict((("A1", Point(0.0, 1.0, 2.0)),)), + physical_rows=OrderedDict((("1", ["A1"]),)), + physical_columns=OrderedDict((("A", ["A1"]),)), + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="A1", + valid_nozzle_maps=ValidNozzleMaps(maps={"Single": ["A1"]}), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=EIGHT_CHANNEL_MAP, + physical_rows=EIGHT_CHANNEL_ROWS, + physical_columns=EIGHT_CHANNEL_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="H1", + valid_nozzle_maps=ValidNozzleMaps( + maps={"Full": EIGHT_CHANNEL_COLS["1"]} + ), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=EIGHT_CHANNEL_MAP, + physical_rows=EIGHT_CHANNEL_ROWS, + physical_columns=EIGHT_CHANNEL_COLS, + starting_nozzle="A1", + back_left_nozzle="A1", + front_right_nozzle="A1", + valid_nozzle_maps=ValidNozzleMaps(maps={"Full": ["A1"]}), + ), + True, + ), + ( + NozzleMap.build( + physical_nozzles=NINETY_SIX_MAP, + physical_rows=NINETY_SIX_ROWS, + physical_columns=NINETY_SIX_COLS, + starting_nozzle="A5", + back_left_nozzle="A5", + front_right_nozzle="H5", + valid_nozzle_maps=ValidNozzleMaps( + maps={"Column12": NINETY_SIX_COLS["5"]} + ), + ), + False, + ), + ], +) +def test_lld_config_validation(nozzle_map: NozzleMap, allowed: bool) -> None: + """It should validate partial tip configurations for LLD.""" + pipette_id = "pipette-id" + subject = get_pipette_view( + nozzle_layout_by_id={pipette_id: nozzle_map}, + ) + assert subject.get_nozzle_configuration_supports_lld(pipette_id) == allowed diff --git a/robot-server/robot_server/runs/router/base_router.py b/robot-server/robot_server/runs/router/base_router.py index b7df09f8992..e6aefa03b98 100644 --- a/robot-server/robot_server/runs/router/base_router.py +++ b/robot-server/robot_server/runs/router/base_router.py @@ -596,7 +596,7 @@ async def get_current_state( # noqa: C901 nozzle_layouts = { pipetteId: ActiveNozzleLayout.construct( startingNozzle=nozzle_map.starting_nozzle, - activeNozzles=list(nozzle_map.map_store.keys()), + activeNozzles=nozzle_map.active_nozzles, config=NozzleLayoutConfig(nozzle_map.configuration.value.lower()), ) for pipetteId, nozzle_map in active_nozzle_maps.items() diff --git a/robot-server/robot_server/runs/run_data_manager.py b/robot-server/robot_server/runs/run_data_manager.py index d30f5c33979..47fe28232d1 100644 --- a/robot-server/robot_server/runs/run_data_manager.py +++ b/robot-server/robot_server/runs/run_data_manager.py @@ -1,12 +1,12 @@ """Manage current and historical run data.""" + from datetime import datetime -from typing import List, Optional, Callable, Union, Dict +from typing import List, Optional, Callable, Union, Mapping from opentrons_shared_data.labware.labware_definition import LabwareDefinition from opentrons_shared_data.errors.exceptions import InvalidStoredData, EnumeratedError -from opentrons.hardware_control.nozzle_manager import NozzleMap - +from opentrons.types import NozzleMapInterface from opentrons.protocol_engine import ( EngineStatus, LabwareOffsetCreate, @@ -509,7 +509,7 @@ def get_command_errors(self, run_id: str) -> list[ErrorOccurrence]: # TODO(tz, 8-5-2024): Change this to return the error list from the DB when we implement https://opentrons.atlassian.net/browse/EXEC-655. raise RunNotCurrentError() - def get_nozzle_maps(self, run_id: str) -> Dict[str, NozzleMap]: + def get_nozzle_maps(self, run_id: str) -> Mapping[str, NozzleMapInterface]: """Get current nozzle maps keyed by pipette id.""" if run_id == self._run_orchestrator_store.current_run_id: return self._run_orchestrator_store.get_nozzle_maps() diff --git a/robot-server/robot_server/runs/run_orchestrator_store.py b/robot-server/robot_server/runs/run_orchestrator_store.py index ee82ea034ac..250f5bea966 100644 --- a/robot-server/robot_server/runs/run_orchestrator_store.py +++ b/robot-server/robot_server/runs/run_orchestrator_store.py @@ -1,8 +1,10 @@ """In-memory storage of ProtocolEngine instances.""" + import asyncio import logging -from typing import List, Optional, Callable, Dict +from typing import List, Optional, Callable, Mapping +from opentrons.types import NozzleMapInterface from opentrons.protocol_engine.errors.exceptions import EStopActivatedError from opentrons.protocol_engine.types import ( PostRunHardwareState, @@ -16,7 +18,6 @@ from opentrons.config import feature_flags from opentrons.hardware_control import HardwareControlAPI -from opentrons.hardware_control.nozzle_manager import NozzleMap from opentrons.hardware_control.types import ( EstopState, HardwareEvent, @@ -327,7 +328,7 @@ def get_loaded_labware_definitions(self) -> List[LabwareDefinition]: """Get loaded labware definitions.""" return self.run_orchestrator.get_loaded_labware_definitions() - def get_nozzle_maps(self) -> Dict[str, NozzleMap]: + def get_nozzle_maps(self) -> Mapping[str, NozzleMapInterface]: """Get the current nozzle map keyed by pipette id.""" return self.run_orchestrator.get_nozzle_maps() diff --git a/robot-server/tests/runs/router/test_base_router.py b/robot-server/tests/runs/router/test_base_router.py index 9052b588bc9..e43027b3bf1 100644 --- a/robot-server/tests/runs/router/test_base_router.py +++ b/robot-server/tests/runs/router/test_base_router.py @@ -8,7 +8,7 @@ from decoy import Decoy from pathlib import Path -from opentrons.types import DeckSlotName, Point +from opentrons.types import DeckSlotName, Point, NozzleConfigurationType from opentrons.protocol_engine import ( LabwareOffsetCreate, types as pe_types, @@ -18,7 +18,7 @@ ) from opentrons.protocol_reader import ProtocolSource, JsonProtocolConfig -from opentrons.hardware_control.nozzle_manager import NozzleConfigurationType, NozzleMap +from opentrons.hardware_control.nozzle_manager import NozzleMap from robot_server.data_files.data_files_store import ( DataFilesStore,