feat(shared-data,protocol-engine): Use the tallest fixture, not the tallest addressable area, for motion planning (#14082)

Closes RSS-409.

* Give every `cutoutFixture` a height. This new field represents the overall height of the physical thing mounted to the deck. This is different from the existing heights of the `addressableArea`s—those are logical "interaction points" that aren't necessarily tied to the physical geometry.
* During a protocol run, make sure to move the pipette above the highest `cutoutFixture` height. Formerly, we were moving the pipette above the highest `addressableArea` height, which was dangerous because the physical thing could be taller than that.
* Various small refactors.

Author: SyntaxColoring

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

@@ -612,7 +612,7 @@ def get_slot_center(self, slot_name: DeckSlotName) -> Point:
 
     def get_highest_z(self) -> float:
         """Get the highest Z point of all deck items."""
-        return self._engine_client.state.geometry.get_all_labware_highest_z()
+        return self._engine_client.state.geometry.get_all_obstacle_highest_z()
 
     def get_labware_cores(self) -> List[LabwareCore]:
         """Get all loaded labware cores."""

api/src/opentrons/protocol_engine/state/addressable_areas.py

@@ -1,6 +1,6 @@
 """Basic addressable area data state and store."""
 from dataclasses import dataclass
-from typing import Dict, Set, Union, List
+from typing import Dict, List, Optional, Set, Union
 
 from opentrons_shared_data.deck.dev_types import DeckDefinitionV4, SlotDefV3
 
@@ -36,9 +36,30 @@ class AddressableAreaState:
     """State of all loaded addressable area resources."""
 
     loaded_addressable_areas_by_name: Dict[str, AddressableArea]
+    """The addressable areas that have been loaded so far.
+
+    When `use_simulated_deck_config` is `False`, these are the addressable areas that the
+    deck configuration provided.
+
+    When `use_simulated_deck_config` is `True`, these are the addressable areas that have been
+    referenced by the protocol so far.
+    """
+
     potential_cutout_fixtures_by_cutout_id: Dict[str, Set[PotentialCutoutFixture]]
+
     deck_definition: DeckDefinitionV4
+
+    deck_configuration: Optional[DeckConfigurationType]
+    """The host robot's full deck configuration.
+
+    If `use_simulated_deck_config` is `True`, this is meaningless and this value is undefined.
+    In practice it will probably be `None` or `[]`.
+
+    If `use_simulated_deck_config` is `False`, this will be non-`None`.
+    """
+
     use_simulated_deck_config: bool
+    """See `Config.use_simulated_deck_config`."""
 
 
 def _get_conflicting_addressable_areas(
@@ -115,6 +136,7 @@ def __init__(
                 )
             )
         self._state = AddressableAreaState(
+            deck_configuration=deck_configuration,
             loaded_addressable_areas_by_name=loaded_addressable_areas_by_name,
             potential_cutout_fixtures_by_cutout_id={},
             deck_definition=deck_definition,
@@ -127,7 +149,11 @@ def handle_action(self, action: Action) -> None:
             self._handle_command(action.command)
         if isinstance(action, PlayAction):
             current_state = self._state
-            if action.deck_configuration is not None:
+            if (
+                action.deck_configuration is not None
+                and not self._state.use_simulated_deck_config
+            ):
+                self._state.deck_configuration = action.deck_configuration
                 self._state.loaded_addressable_areas_by_name = (
                     self._get_addressable_areas_from_deck_configuration(
                         deck_config=action.deck_configuration,
@@ -158,7 +184,7 @@ def _handle_command(self, command: Command) -> None:
     def _get_addressable_areas_from_deck_configuration(
         deck_config: DeckConfigurationType, deck_definition: DeckDefinitionV4
     ) -> Dict[str, AddressableArea]:
-        """Load all provided addressable areas with a valid deck configuration."""
+        """Return all addressable areas provided by the given deck configuration."""
         # TODO uncomment once execute is hooked up with this properly
         # assert (
         #     len(deck_config) == 12
@@ -196,11 +222,10 @@ def _check_location_is_addressable_area(
             addressable_area_name = location
 
         if addressable_area_name not in self._state.loaded_addressable_areas_by_name:
-            # TODO Uncomment this out once robot server side stuff is hooked up
-            # if not self._state.use_simulated_deck_config:
-            #     raise AreaNotInDeckConfigurationError(
-            #         f"{addressable_area_name} not provided by deck configuration."
-            #     )
+            # TODO Validate that during an actual run, the deck configuration provides the requested
+            # addressable area. If it does not, MoveToAddressableArea.execute() needs to raise;
+            # this store class cannot raise because Protocol Engine stores are not allowed to.
+
             cutout_id = self._validate_addressable_area_for_simulation(
                 addressable_area_name
             )
@@ -244,6 +269,9 @@ def _validate_addressable_area_for_simulation(
                     set(self.state.loaded_addressable_areas_by_name),
                     self._state.deck_definition,
                 )
+                # FIXME(mm, 2023-12-01): This needs to be raised from within
+                # MoveToAddressableAreaImplementation.execute(). Protocol Engine stores are not
+                # allowed to raise.
                 raise IncompatibleAddressableAreaError(
                     f"Cannot load {addressable_area_name}, not compatible with one or more of"
                     f" the following areas: {loaded_areas_on_cutout}"
@@ -282,6 +310,21 @@ def get_all(self) -> List[str]:
         """Get a list of all loaded addressable area names."""
         return list(self._state.loaded_addressable_areas_by_name)
 
+    def get_all_cutout_fixtures(self) -> Optional[List[str]]:
+        """Get the names of all fixtures present in the host robot's deck configuration.
+
+        If `use_simulated_deck_config` is `True` (see `Config`), we don't have a
+        meaningful concrete layout of fixtures, so this will return `None`.
+        """
+        if self._state.use_simulated_deck_config:
+            return None
+        else:
+            assert self._state.deck_configuration is not None
+            return [
+                cutout_fixture_id
+                for _, cutout_fixture_id in self._state.deck_configuration
+            ]
+
     def _get_loaded_addressable_area(
         self, addressable_area_name: str
     ) -> AddressableArea:
@@ -377,10 +420,12 @@ def get_addressable_area_center(self, addressable_area_name: str) -> Point:
             z=position.z,
         )
 
-    def get_addressable_area_height(self, addressable_area_name: str) -> float:
-        """Get the z height of an addressable area."""
-        addressable_area = self.get_addressable_area(addressable_area_name)
-        return addressable_area.bounding_box.z
+    def get_fixture_height(self, cutout_fixture_name: str) -> float:
+        """Get the z height of a cutout fixture."""
+        cutout_fixture = deck_configuration_provider.get_cutout_fixture(
+            cutout_fixture_name, self._state.deck_definition
+        )
+        return cutout_fixture["height"]
 
     def get_slot_definition(self, slot: DeckSlotName) -> SlotDefV3:
         """Get the definition of a slot in the deck."""

api/src/opentrons/protocol_engine/state/geometry.py

@@ -89,9 +89,8 @@ def get_labware_highest_z(self, labware_id: str) -> float:
 
         return self._get_highest_z_from_labware_data(labware_data)
 
-    # TODO(mc, 2022-06-24): rename this method
-    def get_all_labware_highest_z(self) -> float:
-        """Get the highest Z-point across all labware."""
+    def get_all_obstacle_highest_z(self) -> float:
+        """Get the highest Z-point across all obstacles that the instruments need to fly over."""
         highest_labware_z = max(
             (
                 self._get_highest_z_from_labware_data(lw_data)
@@ -109,15 +108,33 @@ def get_all_labware_highest_z(self) -> float:
             default=0.0,
         )
 
-        highest_addressable_area_z = max(
-            (
-                self._addressable_areas.get_addressable_area_height(area_name)
-                for area_name in self._addressable_areas.get_all()
-            ),
-            default=0.0,
-        )
+        cutout_fixture_names = self._addressable_areas.get_all_cutout_fixtures()
+        if cutout_fixture_names is None:
+            # We're using a simulated deck config (see `Config.use_simulated_deck_config`).
+            # We only know the addressable areas referenced by the protocol, not the fixtures
+            # providing them. And there is more than one possible configuration of fixtures
+            # to provide them. So, we can't know what the highest fixture is. Default to 0.
+            #
+            # Defaulting to 0 may not be the right thing to do here.
+            # For example, suppose a protocol references an addressable area that implies a tall
+            # fixture must be on the deck, and then it uses long tips that wouldn't be able to
+            # clear the top of that fixture. We should perhaps raise an analysis error for that,
+            # but defaulting to 0 here means we won't.
+            highest_fixture_z = 0.0
+        else:
+            highest_fixture_z = max(
+                (
+                    self._addressable_areas.get_fixture_height(cutout_fixture_name)
+                    for cutout_fixture_name in cutout_fixture_names
+                ),
+                default=0.0,
+            )
 
-        return max(highest_labware_z, highest_module_z, highest_addressable_area_z)
+        return max(
+            highest_labware_z,
+            highest_module_z,
+            highest_fixture_z,
+        )
 
     def get_min_travel_z(
         self,
@@ -134,7 +151,7 @@ def get_min_travel_z(
         ):
             min_travel_z = self.get_labware_highest_z(labware_id)
         else:
-            min_travel_z = self.get_all_labware_highest_z()
+            min_travel_z = self.get_all_obstacle_highest_z()
         if minimum_z_height:
             min_travel_z = max(min_travel_z, minimum_z_height)
         return min_travel_z

api/src/opentrons/protocol_engine/state/motion.py

@@ -160,7 +160,7 @@ def get_movement_waypoints_to_addressable_area(
         # TODO(jbl 11-28-2023) This may need to change for partial tip configurations on a 96
         destination_cp = CriticalPoint.XY_CENTER
 
-        all_labware_highest_z = self._geometry.get_all_labware_highest_z()
+        all_labware_highest_z = self._geometry.get_all_obstacle_highest_z()
         if minimum_z_height is None:
             minimum_z_height = float("-inf")
         min_travel_z = max(all_labware_highest_z, minimum_z_height)
@@ -215,7 +215,7 @@ def get_movement_waypoints_to_coords(
                 Ignored if `direct` is True. If lower than the default height,
                 the default is used; this can only increase the height, not decrease it.
         """
-        all_labware_highest_z = self._geometry.get_all_labware_highest_z()
+        all_labware_highest_z = self._geometry.get_all_obstacle_highest_z()
         if additional_min_travel_z is None:
             additional_min_travel_z = float("-inf")
         min_travel_z = max(all_labware_highest_z, additional_min_travel_z)

api/tests/opentrons/protocol_api/core/engine/test_protocol_core.py

@@ -1552,7 +1552,7 @@ def test_get_highest_z(
 ) -> None:
     """It should return a slot center from engine state."""
     decoy.when(
-        mock_engine_client.state.geometry.get_all_labware_highest_z()
+        mock_engine_client.state.geometry.get_all_obstacle_highest_z()
     ).then_return(9001)
 
     result = subject.get_highest_z()

api/tests/opentrons/protocol_engine/state/test_addressable_area_store.py

@@ -91,6 +91,7 @@ def test_initial_state_simulated(
         loaded_addressable_areas_by_name={},
         potential_cutout_fixtures_by_cutout_id={},
         deck_definition=ot3_standard_deck_def,
+        deck_configuration=[],
         use_simulated_deck_config=True,
     )
 
@@ -103,6 +104,7 @@ def test_initial_state(
     assert subject.state.potential_cutout_fixtures_by_cutout_id == {}
     assert not subject.state.use_simulated_deck_config
     assert subject.state.deck_definition == ot3_standard_deck_def
+    assert subject.state.deck_configuration == _make_deck_config()
     # Loading 9 regular slots, 1 trash, 2 Staging Area slots and 3 waste chute types
     assert len(subject.state.loaded_addressable_areas_by_name) == 15
 

api/tests/opentrons/protocol_engine/state/test_addressable_area_view.py

@@ -21,6 +21,7 @@
 from opentrons.protocol_engine.types import (
     AddressableArea,
     AreaType,
+    DeckConfigurationType,
     PotentialCutoutFixture,
     Dimensions,
     DeckPoint,
@@ -29,8 +30,10 @@
 
 
 @pytest.fixture(autouse=True)
-def patch_mock_move_types(decoy: Decoy, monkeypatch: pytest.MonkeyPatch) -> None:
-    """Mock out move_types.py functions."""
+def patch_mock_deck_configuration_provider(
+    decoy: Decoy, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Mock out deck_configuration_provider.py functions."""
     for name, func in inspect.getmembers(
         deck_configuration_provider, inspect.isfunction
     ):
@@ -43,6 +46,7 @@ def get_addressable_area_view(
         Dict[str, Set[PotentialCutoutFixture]]
     ] = None,
     deck_definition: Optional[DeckDefinitionV4] = None,
+    deck_configuration: Optional[DeckConfigurationType] = None,
     use_simulated_deck_config: bool = False,
 ) -> AddressableAreaView:
     """Get a labware view test subject."""
@@ -51,12 +55,37 @@ def get_addressable_area_view(
         potential_cutout_fixtures_by_cutout_id=potential_cutout_fixtures_by_cutout_id
         or {},
         deck_definition=deck_definition or cast(DeckDefinitionV4, {"otId": "fake"}),
+        deck_configuration=deck_configuration or [],
         use_simulated_deck_config=use_simulated_deck_config,
     )
 
     return AddressableAreaView(state=state)
 
 
+def test_get_all_cutout_fixtures_simulated_deck_config() -> None:
+    """It should return no cutout fixtures when the deck config is simulated."""
+    subject = get_addressable_area_view(
+        deck_configuration=None,
+        use_simulated_deck_config=True,
+    )
+    assert subject.get_all_cutout_fixtures() is None
+
+
+def test_get_all_cutout_fixtures_non_simulated_deck_config() -> None:
+    """It should return the cutout fixtures from the deck config, if it's not simulated."""
+    subject = get_addressable_area_view(
+        deck_configuration=[
+            ("cutout-id-1", "cutout-fixture-id-1"),
+            ("cutout-id-2", "cutout-fixture-id-2"),
+        ],
+        use_simulated_deck_config=False,
+    )
+    assert subject.get_all_cutout_fixtures() == [
+        "cutout-fixture-id-1",
+        "cutout-fixture-id-2",
+    ]
+
+
 def test_get_loaded_addressable_area() -> None:
     """It should get the loaded addressable area."""
     addressable_area = AddressableArea(
@@ -251,6 +280,43 @@ def test_get_addressable_area_center() -> None:
     assert result == Point(6, 12, 3)
 
 
+def test_get_fixture_height(decoy: Decoy) -> None:
+    """It should return the height of the requested fixture."""
+    subject = get_addressable_area_view()
+    decoy.when(
+        deck_configuration_provider.get_cutout_fixture(
+            "someShortCutoutFixture", subject.state.deck_definition
+        )
+    ).then_return(
+        {
+            "height": 10,
+            # These values don't matter:
+            "id": "id",
+            "mayMountTo": [],
+            "displayName": "",
+            "providesAddressableAreas": {},
+        }
+    )
+
+    decoy.when(
+        deck_configuration_provider.get_cutout_fixture(
+            "someTallCutoutFixture", subject.state.deck_definition
+        )
+    ).then_return(
+        {
+            "height": 9000.1,
+            # These values don't matter:
+            "id": "id",
+            "mayMountTo": [],
+            "displayName": "",
+            "providesAddressableAreas": {},
+        }
+    )
+
+    assert subject.get_fixture_height("someShortCutoutFixture") == 10
+    assert subject.get_fixture_height("someTallCutoutFixture") == 9000.1
+
+
 def test_get_slot_definition() -> None:
     """It should return a deck slot's definition."""
     subject = get_addressable_area_view(