add protocol engine stacker home command

Author: ahiuchingau

api/src/opentrons/hardware_control/modules/flex_stacker.py

@@ -416,6 +416,21 @@ async def _prepare_for_action(self) -> bool:
         await self.close_latch()
         return True
 
+    async def home_all(self) -> None:
+        """Home all axes based on current state, assuming normal operation."""
+        await self._reader.read()
+        # we should always be able to home the X axis first
+        await self.home_axis(StackerAxis.X, Direction.RETRACT)
+        # If latch is open, we must first close it
+        if self.limit_switch_status[StackerAxis.L] == StackerAxisState.EXTENDED:
+            if self.limit_switch_status[StackerAxis.Z] != StackerAxisState.RETRACTED:
+                # it was likely in the middle of a dispense/store command
+                # z should be moved up before we can safely close the latch
+                await self.home_axis(StackerAxis.Z, Direction.EXTEND)
+            await self.close_latch()
+        await self.home_axis(StackerAxis.Z, Direction.RETRACT)
+        await self.home_axis(StackerAxis.X, Direction.EXTEND)
+
 
 class FlexStackerReader(Reader):
     error: Optional[str]

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

@@ -482,6 +482,7 @@
         flex_stacker.SetStoredLabware,
         flex_stacker.Fill,
         flex_stacker.Empty,
+        flex_stacker.Home,
         calibration.CalibrateGripper,
         calibration.CalibratePipette,
         calibration.CalibrateModule,
@@ -580,6 +581,7 @@
     flex_stacker.SetStoredLabwareParams,
     flex_stacker.FillParams,
     flex_stacker.EmptyParams,
+    flex_stacker.HomeParams,
     calibration.CalibrateGripperParams,
     calibration.CalibratePipetteParams,
     calibration.CalibrateModuleParams,
@@ -676,6 +678,7 @@
     flex_stacker.SetStoredLabwareCommandType,
     flex_stacker.FillCommandType,
     flex_stacker.EmptyCommandType,
+    flex_stacker.HomeCommandType,
     calibration.CalibrateGripperCommandType,
     calibration.CalibratePipetteCommandType,
     calibration.CalibrateModuleCommandType,
@@ -773,6 +776,7 @@
         flex_stacker.SetStoredLabwareCreate,
         flex_stacker.FillCreate,
         flex_stacker.EmptyCreate,
+        flex_stacker.HomeCreate,
         calibration.CalibrateGripperCreate,
         calibration.CalibratePipetteCreate,
         calibration.CalibrateModuleCreate,
@@ -878,6 +882,7 @@
     flex_stacker.SetStoredLabwareResult,
     flex_stacker.FillResult,
     flex_stacker.EmptyResult,
+    flex_stacker.HomeResult,
     calibration.CalibrateGripperResult,
     calibration.CalibratePipetteResult,
     calibration.CalibrateModuleResult,

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

@@ -29,6 +29,8 @@
 
 from .empty import EmptyCommandType, EmptyParams, EmptyResult, Empty, EmptyCreate
 
+from .home import HomeCommandType, HomeParams, HomeResult, Home, HomeCreate
+
 
 __all__ = [
     # flexStacker/store
@@ -62,4 +64,10 @@
     "EmptyResult",
     "Empty",
     "EmptyCreate",
+    # flexStacker/home
+    "HomeCommandType",
+    "HomeParams",
+    "HomeResult",
+    "Home",
+    "HomeCreate",
 ]

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

@@ -0,0 +1,105 @@
+"""Command models to home the stacker."""
+
+from __future__ import annotations
+
+from __future__ import annotations
+from typing import Literal, Union, TYPE_CHECKING
+from typing_extensions import Type
+
+from pydantic import BaseModel, Field
+
+from ..flex_stacker.common import FlexStackerStallOrCollisionError
+from opentrons_shared_data.errors.exceptions import FlexStackerStallError
+
+from ..command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    SuccessData,
+    DefinedErrorData,
+)
+from ...errors import ErrorOccurrence
+from ...resources import ModelUtils
+
+if TYPE_CHECKING:
+    from ...state.state import StateView
+    from ...execution import EquipmentHandler
+
+HomeCommandType = Literal["flexStacker/home"]
+
+
+class HomeParams(BaseModel):
+    """The parameters defining how a stacker should be emptied."""
+
+    moduleId: str = Field(..., description="Unique ID of the Flex Stacker")
+
+
+class HomeResult(BaseModel):
+    """Result data from a stacker empty command."""
+
+
+_ExecuteReturn = Union[
+    SuccessData[HomeResult], DefinedErrorData[FlexStackerStallOrCollisionError]
+]
+
+
+class HomeImpl(AbstractCommandImpl[HomeParams, _ExecuteReturn]):
+    """Implementation of a stacker empty command."""
+
+    def __init__(
+        self,
+        state_view: StateView,
+        equipment: EquipmentHandler,
+        model_utils: ModelUtils,
+        **kwargs: object,
+    ) -> None:
+        self._state_view = state_view
+        self._equipment = equipment
+        self._model_utils = model_utils
+
+    async def execute(self, params: HomeParams) -> _ExecuteReturn:
+        """Execute the stacker empty command."""
+        stacker_state = self._state_view.modules.get_flex_stacker_substate(
+            params.moduleId
+        )
+        # Allow propagation of ModuleNotAttachedError.
+        stacker_hw = self._equipment.get_module_hardware_api(stacker_state.module_id)
+
+        try:
+            if stacker_hw is not None:
+                await stacker_hw.home_all()
+        except FlexStackerStallError as e:
+            return DefinedErrorData(
+                public=FlexStackerStallOrCollisionError(
+                    id=self._model_utils.generate_id(),
+                    createdAt=self._model_utils.get_timestamp(),
+                    wrappedErrors=[
+                        ErrorOccurrence.from_failed(
+                            id=self._model_utils.generate_id(),
+                            createdAt=self._model_utils.get_timestamp(),
+                            error=e,
+                        )
+                    ],
+                ),
+            )
+
+        return SuccessData(public=HomeResult())
+
+
+class Home(BaseCommand[HomeParams, HomeResult, ErrorOccurrence]):
+    """A command to home a Flex Stacker."""
+
+    commandType: HomeCommandType = "flexStacker/home"
+    params: HomeParams
+    result: HomeResult | None = None
+
+    _ImplementationCls: Type[HomeImpl] = HomeImpl
+
+
+class HomeCreate(BaseCommandCreate[HomeParams]):
+    """A request to execute a Flex Stacker home command."""
+
+    commandType: HomeCommandType = "flexStacker/home"
+    params: HomeParams
+
+    _CommandCls: Type[Home] = Home

api/tests/opentrons/protocol_engine/commands/flex_stacker/test_home.py

@@ -0,0 +1,106 @@
+"""Test Flex Stacker home command implementation."""
+
+from datetime import datetime
+
+import pytest
+from decoy import Decoy, matchers
+
+from opentrons.drivers.flex_stacker.types import StackerAxis
+from opentrons.hardware_control.modules import FlexStacker
+from opentrons.protocol_engine.commands.flex_stacker.common import (
+    FlexStackerStallOrCollisionError,
+)
+from opentrons.protocol_engine.resources import ModelUtils
+from opentrons.protocol_engine.state.state import StateView
+from opentrons.protocol_engine.state.module_substates import (
+    FlexStackerSubState,
+    FlexStackerId,
+)
+from opentrons.protocol_engine.execution import EquipmentHandler
+from opentrons.protocol_engine.commands import flex_stacker
+from opentrons.protocol_engine.commands.command import SuccessData, DefinedErrorData
+from opentrons.protocol_engine.commands.flex_stacker.home import HomeImpl
+
+from opentrons_shared_data.errors.exceptions import FlexStackerStallError
+
+
+@pytest.fixture
+def subject(
+    state_view: StateView, equipment: EquipmentHandler, model_utils: ModelUtils
+) -> HomeImpl:
+    """Get a home command to test."""
+    return HomeImpl(state_view=state_view, equipment=equipment, model_utils=model_utils)
+
+
+async def test_home_command(
+    decoy: Decoy,
+    state_view: StateView,
+    equipment: EquipmentHandler,
+    subject: HomeImpl,
+    stacker_id: FlexStackerId,
+    stacker_hardware: FlexStacker,
+) -> None:
+    """It should return a success data."""
+    data = flex_stacker.HomeParams(moduleId=stacker_id)
+
+    fs_module_substate = FlexStackerSubState(
+        module_id=stacker_id,
+        pool_primary_definition=None,
+        pool_adapter_definition=None,
+        pool_lid_definition=None,
+        pool_count=0,
+        max_pool_count=0,
+    )
+    decoy.when(
+        state_view.modules.get_flex_stacker_substate(module_id=stacker_id)
+    ).then_return(fs_module_substate)
+
+    result = await subject.execute(data)
+
+    decoy.verify(await stacker_hardware.home_all(), times=1)
+
+    assert result == SuccessData(public=flex_stacker.HomeResult())
+
+
+async def test_home_command_with_stall_detected(
+    decoy: Decoy,
+    state_view: StateView,
+    equipment: EquipmentHandler,
+    subject: HomeImpl,
+    model_utils: ModelUtils,
+    stacker_id: FlexStackerId,
+    stacker_hardware: FlexStacker,
+) -> None:
+    """It should return a success data."""
+    err_id = "error-id"
+    err_timestamp = datetime(year=2025, month=3, day=19)
+
+    data = flex_stacker.HomeParams(moduleId=stacker_id)
+
+    fs_module_substate = FlexStackerSubState(
+        module_id=stacker_id,
+        pool_primary_definition=None,
+        pool_adapter_definition=None,
+        pool_lid_definition=None,
+        pool_count=0,
+        max_pool_count=0,
+    )
+    decoy.when(
+        state_view.modules.get_flex_stacker_substate(module_id=stacker_id)
+    ).then_return(fs_module_substate)
+    decoy.when(model_utils.generate_id()).then_return(err_id)
+    decoy.when(model_utils.get_timestamp()).then_return(err_timestamp)
+
+    decoy.when(await stacker_hardware.home_all()).then_raise(
+        FlexStackerStallError(serial="123", axis=StackerAxis.Z)
+    )
+
+    result = await subject.execute(data)
+
+    assert result == DefinedErrorData(
+        public=FlexStackerStallOrCollisionError.model_construct(
+            id=err_id,
+            createdAt=err_timestamp,
+            wrappedErrors=[matchers.Anything()],
+        )
+    )