feat(api): Flex Stacker Module Support for EVT (#17300)

Covers EXEC-967, EXEC-965, EXEC-946, EXEC-1078

This PR introduces the .store(), .retrieve(), and
.load_labware_to_hopper(...) commands. These commands allow the
declaration of labware inside the hopper, retrieval of a handleable
labware core from the stacker, and storage of labware into the stacker.

Author: ahiuchingau

analyses-snapshot-testing/tests/__snapshots__/analyses_snapshot_test/test_analysis_snapshot[6126498df7][Flex_X_v2_16_NO_PIPETTES_TM_ModuleInStagingAreaCol4].json

@@ -25,7 +25,7 @@
   "errors": [
     {
       "createdAt": "TIMESTAMP",
-      "detail": "ValueError [line 15]: Cannot load a module onto a staging slot.",
+      "detail": "ValueError [line 15]: Cannot load temperature module gen2 onto a staging slot.",
       "errorCode": "4000",
       "errorInfo": {},
       "errorType": "ExceptionInProtocolError",
@@ -34,12 +34,12 @@
       "wrappedErrors": [
         {
           "createdAt": "TIMESTAMP",
-          "detail": "ValueError: Cannot load a module onto a staging slot.",
+          "detail": "ValueError: Cannot load temperature module gen2 onto a staging slot.",
           "errorCode": "4000",
           "errorInfo": {
-            "args": "('Cannot load a module onto a staging slot.',)",
+            "args": "('Cannot load temperature module gen2 onto a staging slot.',)",
             "class": "ValueError",
-            "traceback": "  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_run\n    exec(\"run(__context)\", new_globs)\n\n  File \"<string>\", line N, in <module>\n\n  File \"Flex_X_v2_16_NO_PIPETTES_TM_ModuleInStagingAreaCol4.py\", line N, in run\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/api_support/util.py\", line N, in _check_version_wrapper\n    return decorated_obj(*args, **kwargs)\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_api/protocol_context.py\", line N, in load_module\n    raise ValueError(\"Cannot load a module onto a staging slot.\")\n"
+            "traceback": "  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_run\n    exec(\"run(__context)\", new_globs)\n\n  File \"<string>\", line N, in <module>\n\n  File \"Flex_X_v2_16_NO_PIPETTES_TM_ModuleInStagingAreaCol4.py\", line N, in run\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/api_support/util.py\", line N, in _check_version_wrapper\n    return decorated_obj(*args, **kwargs)\n\n  File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_api/protocol_context.py\", line N, in load_module\n    raise ValueError(f\"Cannot load {module_name} onto a staging slot.\")\n"
           },
           "errorType": "PythonException",
           "id": "UUID",

api/src/opentrons/protocol_api/__init__.py

@@ -27,6 +27,7 @@
     HeaterShakerContext,
     MagneticBlockContext,
     AbsorbanceReaderContext,
+    FlexStackerContext,
 )
 from .disposal_locations import TrashBin, WasteChute
 from ._liquid import Liquid, LiquidClass
@@ -70,6 +71,7 @@
     "HeaterShakerContext",
     "MagneticBlockContext",
     "AbsorbanceReaderContext",
+    "FlexStackerContext",
     "ParameterContext",
     "Labware",
     "TrashBin",

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

@@ -300,6 +300,9 @@ def _map_module(
                 is_semi_configuration=False,
             ),
         )
+    elif module_type == ModuleType.FLEX_STACKER:
+        # TODO: This is a placeholder. We need to implement this.
+        return None
     else:
         return (
             mapped_location,

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

@@ -700,16 +700,30 @@ class FlexStackerCore(ModuleCore, AbstractFlexStackerCore):
 
     _sync_module_hardware: SynchronousAdapter[hw_modules.FlexStacker]
 
+    def set_static_mode(self, static: bool) -> None:
+        """Set the Flex Stacker's static mode.
+
+        The Flex Stacker cannot retrieve and or store when in static mode.
+        This allows the Flex Stacker carriage to be used as a staging slot,
+        and allowed the labware to be loaded onto it.
+        """
+        self._engine_client.execute_command(
+            cmd.flex_stacker.ConfigureParams(
+                moduleId=self.module_id,
+                static=static,
+            )
+        )
+
     def retrieve(self) -> None:
-        """Retrieve a labware from the bottom of the Flex Stacker's stack."""
+        """Retrieve a labware from the Flex Stacker's hopper."""
         self._engine_client.execute_command(
             cmd.flex_stacker.RetrieveParams(
                 moduleId=self.module_id,
             )
         )
 
     def store(self) -> None:
-        """Store a labware at the bottom of the Flex Stacker's stack."""
+        """Store a labware into Flex Stacker's hopper."""
         self._engine_client.execute_command(
             cmd.flex_stacker.StoreParams(
                 moduleId=self.module_id,

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

@@ -75,6 +75,7 @@
     NonConnectedModuleCore,
     MagneticBlockCore,
     AbsorbanceReaderCore,
+    FlexStackerCore,
 )
 from .exceptions import InvalidModuleLocationError
 from . import load_labware_params, deck_conflict, overlap_versions
@@ -373,6 +374,34 @@ def load_lid(
         self._labware_cores_by_id[labware_core.labware_id] = labware_core
         return labware_core
 
+    def load_labware_to_flex_stacker_hopper(
+        self,
+        module_core: Union[ModuleCore, NonConnectedModuleCore],
+        load_name: str,
+        quantity: int,
+        label: Optional[str],
+        namespace: Optional[str],
+        version: Optional[int],
+        lid: Optional[str],
+    ) -> None:
+        """Load one or more labware with or without a lid to the flex stacker hopper."""
+        assert isinstance(module_core, FlexStackerCore)
+        for _ in range(quantity):
+            labware_core = self.load_labware(
+                load_name=load_name,
+                location=module_core,
+                label=label,
+                namespace=namespace,
+                version=version,
+            )
+            if lid is not None:
+                self.load_lid(
+                    load_name=lid,
+                    location=labware_core,
+                    namespace=namespace,
+                    version=version,
+                )
+
     def move_labware(
         self,
         labware_core: LabwareCore,
@@ -726,6 +755,7 @@ def _create_module_core(
             ModuleType.THERMOCYCLER: ThermocyclerModuleCore,
             ModuleType.HEATER_SHAKER: HeaterShakerModuleCore,
             ModuleType.ABSORBANCE_READER: AbsorbanceReaderCore,
+            ModuleType.FLEX_STACKER: FlexStackerCore,
         }
 
         module_type = load_module_result.model.as_type()

api/src/opentrons/protocol_api/core/legacy/legacy_protocol_core.py

@@ -524,6 +524,19 @@ def load_lid_stack(
         """Load a Stack of Lids to a given location, creating a Lid Stack."""
         raise APIVersionError(api_element="Lid stack")
 
+    def load_labware_to_flex_stacker_hopper(
+        self,
+        module_core: legacy_module_core.LegacyModuleCore,
+        load_name: str,
+        quantity: int,
+        label: Optional[str],
+        namespace: Optional[str],
+        version: Optional[int],
+        lid: Optional[str],
+    ) -> None:
+        """Load labware to a Flex stacker hopper."""
+        raise APIVersionError(api_element="Flex stacker")
+
     def get_module_cores(self) -> List[legacy_module_core.LegacyModuleCore]:
         """Get loaded module cores."""
         return self._module_cores

api/src/opentrons/protocol_api/core/module.py

@@ -390,11 +390,14 @@ class AbstractFlexStackerCore(AbstractModuleCore):
     def get_serial_number(self) -> str:
         """Get the module's unique hardware serial number."""
 
+    @abstractmethod
+    def set_static_mode(self, static: bool) -> None:
+        """Set the Flex Stacker's static mode."""
+
     @abstractmethod
     def retrieve(self) -> None:
-        """Release and return a labware at the bottom of the labware stack."""
+        """Release a labware from the hopper to the staging slot."""
 
     @abstractmethod
     def store(self) -> None:
-        """Store a labware at the bottom of the labware stack."""
-        pass
+        """Store a labware in the stacker hopper."""

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

@@ -111,6 +111,20 @@ def load_lid(
         """Load an individual lid labware using its identifying parameters. Must be loaded on a labware."""
         ...
 
+    @abstractmethod
+    def load_labware_to_flex_stacker_hopper(
+        self,
+        module_core: ModuleCoreType,
+        load_name: str,
+        quantity: int,
+        label: Optional[str],
+        namespace: Optional[str],
+        version: Optional[int],
+        lid: Optional[str],
+    ) -> None:
+        """Load one or more labware with or without a lid to the flex stacker hopper."""
+        ...
+
     @abstractmethod
     def move_labware(
         self,

api/src/opentrons/protocol_api/module_contexts.py

@@ -1112,21 +1112,79 @@ class FlexStackerContext(ModuleContext):
 
     _core: FlexStackerCore
 
+    @requires_version(2, 23)
+    def load_labware_to_hopper(
+        self,
+        load_name: str,
+        quantity: int,
+        label: Optional[str] = None,
+        namespace: Optional[str] = None,
+        version: Optional[int] = None,
+        lid: Optional[str] = None,
+    ) -> None:
+        """Load one or more labware onto the flex stacker."""
+        self._protocol_core.load_labware_to_flex_stacker_hopper(
+            module_core=self._core,
+            load_name=load_name,
+            quantity=quantity,
+            label=label,
+            namespace=namespace,
+            version=version,
+            lid=lid,
+        )
+
+    @requires_version(2, 23)
+    def enter_static_mode(self) -> None:
+        """Enter static mode.
+
+        In static mode, the Flex Stacker will not move labware between the hopper and
+        the deck, and can be used as a staging slot area.
+        """
+        self._core.set_static_mode(static=True)
+
+    @requires_version(2, 23)
+    def exit_static_mode(self) -> None:
+        """End static mode.
+
+        In static mode, the Flex Stacker will not move labware between the hopper and
+        the deck, and can be used as a staging slot area.
+        """
+        self._core.set_static_mode(static=False)
+
     @property
     @requires_version(2, 23)
     def serial_number(self) -> str:
         """Get the module's unique hardware serial number."""
         return self._core.get_serial_number()
 
     @requires_version(2, 23)
-    def retrieve(self) -> None:
+    def retrieve(self) -> Labware:
         """Release and return a labware at the bottom of the labware stack."""
         self._core.retrieve()
+        labware_core = self._protocol_core.get_labware_on_module(self._core)
+        # the core retrieve command should have already raised the error
+        # if labware_core is None, this is just to satisfy the type checker
+        assert labware_core is not None, "Retrieve failed to return labware"
+        # check core map first
+        try:
+            labware = self._core_map.get(labware_core)
+        except KeyError:
+            # If the labware is not already in the core map,
+            # create a new Labware object
+            labware = Labware(
+                core=labware_core,
+                api_version=self._api_version,
+                protocol_core=self._protocol_core,
+                core_map=self._core_map,
+            )
+            self._core_map.add(labware_core, labware)
+        return labware
 
     @requires_version(2, 23)
     def store(self, labware: Labware) -> None:
         """Store a labware at the bottom of the labware stack.
 
         :param labware: The labware object to store.
         """
+        assert labware._core is not None
         self._core.store()

api/src/opentrons/protocol_api/protocol_context.py

@@ -22,6 +22,7 @@
 from opentrons.hardware_control.modules.types import (
     MagneticBlockModel,
     AbsorbanceReaderModel,
+    FlexStackerModuleModel,
 )
 from opentrons.legacy_commands import protocol_commands as cmds, types as cmd_types
 from opentrons.legacy_commands.helpers import (
@@ -61,6 +62,7 @@
     AbstractHeaterShakerCore,
     AbstractMagneticBlockCore,
     AbstractAbsorbanceReaderCore,
+    AbstractFlexStackerCore,
 )
 from .robot_context import RobotContext, HardwareManager
 from .core.engine import ENGINE_CORE_API_VERSION
@@ -79,6 +81,7 @@
     HeaterShakerContext,
     MagneticBlockContext,
     AbsorbanceReaderContext,
+    FlexStackerContext,
     ModuleContext,
 )
 from ._parameters import Parameters
@@ -94,6 +97,7 @@
     HeaterShakerContext,
     MagneticBlockContext,
     AbsorbanceReaderContext,
+    FlexStackerContext,
 ]
 
 
@@ -862,6 +866,9 @@ def load_module(
 
                   .. versionchanged:: 2.15
                     Added ``MagneticBlockContext`` return value.
+
+                  .. versionchanged:: 2.23
+                    Added ``FlexStackerModuleContext`` return value.
         """
         if configuration:
             if self._api_version < APIVersion(2, 4):
@@ -890,7 +897,18 @@ def load_module(
             requested_model, AbsorbanceReaderModel
         ) and self._api_version < APIVersion(2, 21):
             raise APIVersionError(
-                f"Module of type {module_name} is only available in versions 2.21 and above."
+                api_element=f"Module of type {module_name}",
+                until_version="2.21",
+                current_version=f"{self._api_version}",
+            )
+        if (
+            isinstance(requested_model, FlexStackerModuleModel)
+            and self._api_version < validation.FLEX_STACKER_VERSION_GATE
+        ):
+            raise APIVersionError(
+                api_element=f"Module of type {module_name}",
+                until_version=str(validation.FLEX_STACKER_VERSION_GATE),
+                current_version=f"{self._api_version}",
             )
 
         deck_slot = (
@@ -901,7 +919,11 @@ def load_module(
             )
         )
         if isinstance(deck_slot, StagingSlotName):
-            raise ValueError("Cannot load a module onto a staging slot.")
+            # flex stacker modules can only be loaded into staging slot inside a protocol
+            if isinstance(requested_model, FlexStackerModuleModel):
+                deck_slot = validation.convert_flex_stacker_load_slot(deck_slot)
+            else:
+                raise ValueError(f"Cannot load {module_name} onto a staging slot.")
 
         module_core = self._core.load_module(
             model=requested_model,
@@ -1572,6 +1594,8 @@ def _create_module_context(
         module_cls = MagneticBlockContext
     elif isinstance(module_core, AbstractAbsorbanceReaderCore):
         module_cls = AbsorbanceReaderContext
+    elif isinstance(module_core, AbstractFlexStackerCore):
+        module_cls = FlexStackerContext
     else:
         assert False, "Unsupported module type"