feat(api): add and use inStackerHopperLocation (#17535)

Labware objects that are in the stacker hopper have the new location
InStackerHopperLocation, which also identifies the labware Id of the
stacker in question.

In this PR, we set the location of labware that get loaded onto the
stacker module while static=False to InStackerHopperLocation. However,
after the rest of the work in the epic, the only way to get a labware
into `InStackerHopperLocation` is to pass it to `stacker/store`. It's
not a location you can specify in any parameters, which is already true
in this PR.

Because we're adding it fresh and thus it isn't already serialized
anywhere, we can give the location object a `kind` member and therefore
don't have to have a separate location sequence component.

## review requests
does this look like the right data? are we happy with it?

## testing
none required, this is well-covered by internal tests and the data that
it exposes that changed isn't actually used by anything yet.

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

@@ -1,4 +1,5 @@
 """A Protocol-Engine-friendly wrapper for opentrons.motion_planning.deck_conflict."""
+
 from __future__ import annotations
 import itertools
 import logging
@@ -24,6 +25,7 @@
     ModuleLocation,
     OnLabwareLocation,
     AddressableAreaLocation,
+    InStackerHopperLocation,
     OFF_DECK_LOCATION,
     SYSTEM_LOCATION,
 )
@@ -249,6 +251,7 @@ def _map_labware(
     elif (
         location_from_engine == OFF_DECK_LOCATION
         or location_from_engine == SYSTEM_LOCATION
+        or isinstance(location_from_engine, InStackerHopperLocation)
     ):
         # This labware is off-deck. Exclude it from conflict checking.
         # todo(mm, 2023-02-23): Move this logic into wrapped_deck_conflict.

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

@@ -48,7 +48,7 @@
     ModuleModel as ProtocolEngineModuleModel,
     OFF_DECK_LOCATION,
     SYSTEM_LOCATION,
-    LabwareLocation,
+    LoadableLabwareLocation,
     NonStackedLocation,
 )
 from opentrons.protocol_engine.clients import SyncClient as ProtocolEngineClient
@@ -1135,7 +1135,7 @@ def _convert_labware_location(
             WasteChute,
             TrashBin,
         ],
-    ) -> LabwareLocation:
+    ) -> LoadableLabwareLocation:
         if isinstance(location, LabwareCore):
             return OnLabwareLocation(labwareId=location.labware_id)
         else:

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

@@ -5,6 +5,7 @@
     ModuleLocation,
     OnLabwareLocation,
     AddressableAreaLocation,
+    InStackerHopperLocation,
 )
 
 
@@ -17,6 +18,15 @@ def well(engine_client: SyncClient, well_name: str, labware_id: str) -> str:
     return f"{well_name} of {_labware_location_string(engine_client, labware_location)}"
 
 
+def _module_in_location_string(module_id: str, engine_client: SyncClient) -> str:
+    module_name = engine_client.state.modules.get_definition(
+        module_id=module_id
+    ).displayName
+    module_on = engine_client.state.modules.get_location(module_id=module_id)
+    module_on_string = _labware_location_string(engine_client, module_on)
+    return f"{module_name} on {module_on_string}"
+
+
 def _labware_location_string(
     engine_client: SyncClient, location: LabwareLocation
 ) -> str:
@@ -26,14 +36,7 @@ def _labware_location_string(
         return f"slot {location.slotName.id}"
 
     elif isinstance(location, ModuleLocation):
-        module_name = engine_client.state.modules.get_definition(
-            module_id=location.moduleId
-        ).displayName
-        module_on = engine_client.state.modules.get_location(
-            module_id=location.moduleId
-        )
-        module_on_string = _labware_location_string(engine_client, module_on)
-        return f"{module_name} on {module_on_string}"
+        return _module_in_location_string(location.moduleId, engine_client)
 
     elif isinstance(location, OnLabwareLocation):
         labware_name = _labware_name(engine_client, location.labwareId)
@@ -53,6 +56,11 @@ def _labware_location_string(
     elif location == "systemLocation":
         return "[systemLocation]"
 
+    elif isinstance(location, InStackerHopperLocation):
+        return (
+            f"stored in {_module_in_location_string(location.moduleId, engine_client)}"
+        )
+
 
 def _labware_name(engine_client: SyncClient, labware_id: str) -> str:
     """Return the user-specified labware label, or fall back to the display name from the def."""

api/src/opentrons/protocol_engine/__init__.py

@@ -40,6 +40,7 @@
     DeckPoint,
     DeckType,
     DeckSlotLocation,
+    InStackerHopperLocation,
     ModuleLocation,
     OnLabwareLocation,
     AddressableAreaLocation,
@@ -48,6 +49,7 @@
     Dimensions,
     EngineStatus,
     LabwareLocation,
+    LoadableLabwareLocation,
     NonStackedLocation,
     LoadedLabware,
     LoadedModule,
@@ -118,11 +120,13 @@
     "ModuleLocation",
     "OnLabwareLocation",
     "AddressableAreaLocation",
+    "InStackerHopperLocation",
     "OFF_DECK_LOCATION",
     "SYSTEM_LOCATION",
     "Dimensions",
     "EngineStatus",
     "LabwareLocation",
+    "LoadableLabwareLocation",
     "NonStackedLocation",
     "LoadedLabware",
     "LoadedModule",

api/src/opentrons/protocol_engine/commands/flex_stacker/retrieve.py

@@ -15,8 +15,6 @@
 from ...state import update_types
 from ...types import (
     ModuleLocation,
-    OnAddressableAreaLocationSequenceComponent,
-    OnModuleLocationSequenceComponent,
     LabwareLocationSequence,
 )
 
@@ -98,6 +96,11 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
         original_location_sequence = self._state_view.geometry.get_location_sequence(
             lw_id
         )
+        destination_location_sequence = (
+            self._state_view.geometry.get_predicted_location_sequence(
+                ModuleLocation(moduleId=params.moduleId)
+            )
+        )
         labware = self._state_view.labware.get(lw_id)
         labware_height = self._state_view.labware.get_dimensions(labware_id=lw_id).z
         if labware.lid_id is not None:
@@ -110,9 +113,6 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
         if stacker_hw is not None:
             await stacker_hw.dispense_labware(labware_height=labware_height)
 
-        own_addressable_area = self._state_view.modules.get_provided_addressable_area(
-            params.moduleId
-        )
         # update the state to reflect the labware is now in the flex stacker slot
         state_update.set_labware_location(
             labware_id=lw_id,
@@ -126,12 +126,7 @@ async def execute(self, params: RetrieveParams) -> SuccessData[RetrieveResult]:
             public=RetrieveResult(
                 labware_id=lw_id,
                 originLocationSequence=original_location_sequence,
-                eventualDestinationLocationSequence=[
-                    OnModuleLocationSequenceComponent(moduleId=params.moduleId),
-                    OnAddressableAreaLocationSequenceComponent(
-                        addressableAreaName=own_addressable_area,
-                    ),
-                ],
+                eventualDestinationLocationSequence=destination_location_sequence,
             ),
             state_update=state_update,
         )

api/src/opentrons/protocol_engine/commands/flex_stacker/store.py

@@ -14,10 +14,8 @@
 )
 from ...state import update_types
 from ...types import (
-    OFF_DECK_LOCATION,
-    NotOnDeckLocationSequenceComponent,
-    OnModuleLocationSequenceComponent,
     LabwareLocationSequence,
+    InStackerHopperLocation,
 )
 
 
@@ -85,6 +83,11 @@ async def execute(self, params: StoreParams) -> SuccessData[StoreResult]:
         original_location_sequence = self._state_view.geometry.get_location_sequence(
             lw_id
         )
+        eventual_target_location_sequence = (
+            self._state_view.geometry.get_predicted_location_sequence(
+                InStackerHopperLocation(moduleId=params.moduleId)
+            )
+        )
         labware = self._state_view.labware.get(lw_id)
         labware_height = self._state_view.labware.get_dimensions(labware_id=lw_id).z
         if labware.lid_id is not None:
@@ -102,7 +105,7 @@ async def execute(self, params: StoreParams) -> SuccessData[StoreResult]:
         # update the state to reflect the labware is store in the stack
         state_update.set_labware_location(
             labware_id=lw_id,
-            new_location=OFF_DECK_LOCATION,
+            new_location=InStackerHopperLocation(moduleId=params.moduleId),
             new_offset_id=None,
         )
         state_update.store_flex_stacker_labware(
@@ -112,12 +115,7 @@ async def execute(self, params: StoreParams) -> SuccessData[StoreResult]:
         return SuccessData(
             public=StoreResult(
                 originLocationSequence=original_location_sequence,
-                eventualDestinationLocationSequence=[
-                    OnModuleLocationSequenceComponent(moduleId=params.moduleId),
-                    NotOnDeckLocationSequenceComponent(
-                        logicalLocationName=OFF_DECK_LOCATION
-                    ),
-                ],
+                eventualDestinationLocationSequence=eventual_target_location_sequence,
             ),
             state_update=state_update,
         )

api/src/opentrons/protocol_engine/commands/load_labware.py

@@ -12,14 +12,15 @@
 from ..errors import LabwareIsNotAllowedInLocationError
 from ..resources import labware_validation, fixture_validation
 from ..types import (
-    LabwareLocation,
+    LoadableLabwareLocation,
     ModuleLocation,
     ModuleModel,
     OnLabwareLocation,
     DeckSlotLocation,
     AddressableAreaLocation,
     LoadedModule,
-    OFF_DECK_LOCATION,
+    InStackerHopperLocation,
+    LabwareLocation,
 )
 
 from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
@@ -42,7 +43,7 @@ def _remove_default(s: dict[str, Any]) -> None:
 class LoadLabwareParams(BaseModel):
     """Payload required to load a labware into a slot."""
 
-    location: LabwareLocation = Field(
+    location: LoadableLabwareLocation = Field(
         ...,
         description="Location the labware should be loaded into.",
     )
@@ -96,7 +97,7 @@ def __init__(
         self._state_view = state_view
 
     def _is_loading_to_module(
-        self, location: LabwareLocation, module_model: ModuleModel
+        self, location: LoadableLabwareLocation, module_model: ModuleModel
     ) -> TypeGuard[ModuleLocation]:
         if not isinstance(location, ModuleLocation):
             return False
@@ -141,6 +142,9 @@ async def execute(  # noqa: C901
             )
             state_update.set_addressable_area_used(params.location.slotName.id)
 
+        # TODO: make this LoadableLabwareLocation again once we add the rest of the commands
+        # for stacker labware pool configuration. Until then, this is the only way to put a
+        # labware in the stacker hopper at the time the protocol starts.
         verified_location: LabwareLocation
         if (
             self._is_loading_to_module(
@@ -150,10 +154,9 @@ async def execute(  # noqa: C901
                 params.location.moduleId
             ).in_static_mode
         ):
-            # labware loaded to the flex stacker hopper is considered offdeck. This is
-            # a temporary solution until the hopper can be represented as non-addressable
-            # addressable area in the deck configuration.
-            verified_location = OFF_DECK_LOCATION
+            verified_location = InStackerHopperLocation(
+                moduleId=params.location.moduleId
+            )
         else:
             verified_location = self._state_view.geometry.ensure_location_not_occupied(
                 params.location
@@ -204,11 +207,9 @@ async def execute(  # noqa: C901
                 loaded_labware.definition
             )
 
-        if self._is_loading_to_module(
-            params.location, ModuleModel.FLEX_STACKER_MODULE_V1
-        ):
+        if isinstance(verified_location, InStackerHopperLocation):
             state_update.load_flex_stacker_hopper_labware(
-                module_id=params.location.moduleId,
+                module_id=verified_location.moduleId,
                 labware_id=loaded_labware.labware_id,
             )
 
@@ -222,7 +223,7 @@ async def execute(  # noqa: C901
                 definition=loaded_labware.definition,
                 offsetId=loaded_labware.offsetId,
                 locationSequence=self._state_view.geometry.get_predicted_location_sequence(
-                    params.location
+                    verified_location,
                 ),
             ),
             state_update=state_update,

api/src/opentrons/protocol_engine/commands/load_lid.py

@@ -10,7 +10,7 @@
 from ..errors import LabwareCannotBeStackedError, LabwareIsNotAllowedInLocationError
 from ..resources import labware_validation
 from ..types import (
-    LabwareLocation,
+    LoadableLabwareLocation,
     OnLabwareLocation,
     OnLabwareLocationSequenceComponent,
 )
@@ -31,7 +31,7 @@
 class LoadLidParams(BaseModel):
     """Payload required to load a lid onto a labware."""
 
-    location: LabwareLocation = Field(
+    location: LoadableLabwareLocation = Field(
         ...,
         description="Labware the lid should be loaded onto.",
     )

api/src/opentrons/protocol_engine/commands/load_lid_stack.py

@@ -11,7 +11,7 @@
 from ..errors import LabwareIsNotAllowedInLocationError, ProtocolEngineError
 from ..resources import fixture_validation, labware_validation
 from ..types import (
-    LabwareLocation,
+    LoadableLabwareLocation,
     SYSTEM_LOCATION,
     OnLabwareLocation,
     DeckSlotLocation,
@@ -44,7 +44,7 @@ def _remove_default(s: dict[str, Any]) -> None:
 class LoadLidStackParams(BaseModel):
     """Payload required to load a lid stack onto a location."""
 
-    location: LabwareLocation = Field(
+    location: LoadableLabwareLocation = Field(
         ...,
         description="Location the lid stack should be loaded into.",
     )
@@ -93,7 +93,7 @@ class LoadLidStackResult(BaseModel):
         ...,
         description="The full definition data for this lid labware.",
     )
-    location: LabwareLocation = Field(
+    location: LoadableLabwareLocation = Field(
         ..., description="The Location that the stack of lid labware has been loaded."
     )
     stackLocationSequence: LabwareLocationSequence | None = Field(
@@ -117,7 +117,7 @@ def __init__(
         self._equipment = equipment
         self._state_view = state_view
 
-    def _validate_location(self, params: LoadLidStackParams) -> LabwareLocation:
+    def _validate_location(self, params: LoadLidStackParams) -> LoadableLabwareLocation:
         if isinstance(params.location, AddressableAreaLocation):
             area_name = params.location.addressableAreaName
             if not (
@@ -142,7 +142,7 @@ def _validate_location(self, params: LoadLidStackParams) -> LabwareLocation:
 
     def _format_results(
         self,
-        verified_location: LabwareLocation,
+        verified_location: LoadableLabwareLocation,
         lid_stack_object: LoadedLabwareData,
         loaded_lid_labwares: list[LoadedLabwareData],
         lid_labware_definition: LabwareDefinition | None,

api/src/opentrons/protocol_engine/commands/move_labware.py

@@ -19,7 +19,7 @@
 from ..types import (
     ModuleModel,
     CurrentWell,
-    LabwareLocation,
+    LoadableLabwareLocation,
     DeckSlotLocation,
     ModuleLocation,
     OnLabwareLocation,
@@ -68,7 +68,9 @@ class MoveLabwareParams(BaseModel):
     """Input parameters for a ``moveLabware`` command."""
 
     labwareId: str = Field(..., description="The ID of the labware to move.")
-    newLocation: LabwareLocation = Field(..., description="Where to move the labware.")
+    newLocation: LoadableLabwareLocation = Field(
+        ..., description="Where to move the labware."
+    )
     strategy: LabwareMovementStrategy = Field(
         ...,
         description="Whether to use the gripper to perform the labware movement"