feat(api): block robot movement when Estop is engaged or missing (#13147

)

* Raise exceptions if the estop is not in the DISENGAGED state during any movements

* Added nicer error messages

* Added feature flag to enable movement without an estop

* Fixed broken test

* un-expose the estop setting over HTTP

* Add the estopNotRequired setting back to definitions but make it hidden, and set up Robot Server to actually hide settings that should be hidden

* Fix settings migration tests

* If the estopNotRequired setting is enabled, always return that the estop is present

api/src/opentrons/config/advanced_settings.py

@@ -191,6 +191,15 @@ class Setting(NamedTuple):
         title="Disable the LED status bar on the Flex.",
         description="This setting disables the LED status bar on the Flex.",
     ),
+    SettingDefinition(
+        _id="estopNotRequired",
+        title="If enabled, the Flex gantry can move with no estop attached.",
+        description="This setting allows the gantry on the Flex to move with no estop attached.",
+        show_if=(
+            "estopNotRequired",
+            True,
+        ),  # Configured so this only shows if it has been set by a user
+    ),
 ]
 
 if (
@@ -600,6 +609,16 @@ def _migrate27to28(previous: SettingsMap) -> SettingsMap:
     return newmap
 
 
+def _migrate28to29(previous: SettingsMap) -> SettingsMap:
+    """Migrate to version 29 of the feature flags file.
+
+    - Adds the estopNotRequired config element.
+    """
+    newmap = {k: v for k, v in previous.items()}
+    newmap["estopNotRequired"] = None
+    return newmap
+
+
 _MIGRATIONS = [
     _migrate0to1,
     _migrate1to2,
@@ -629,6 +648,7 @@ def _migrate27to28(previous: SettingsMap) -> SettingsMap:
     _migrate25to26,
     _migrate26to27,
     _migrate27to28,
+    _migrate28to29,
 ]
 """
 List of all migrations to apply, indexed by (version - 1). See _migrate below

api/src/opentrons/config/feature_flags.py

@@ -53,3 +53,8 @@ def status_bar_enabled() -> bool:
 def tip_presence_detection_enabled() -> bool:
     """Whether tip presence is enabled on the Flex"""
     return not advs.get_setting_with_env_overload("disableTipPresenceDetection")
+
+
+def require_estop() -> bool:
+    """Whether the OT3 should allow gantry movements with no Estop plugged in."""
+    return not advs.get_setting_with_env_overload("estopNotRequired")

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -126,6 +126,7 @@
     SubSystem,
     TipStateType,
     FailedTipStateCheck,
+    EstopState,
 )
 from opentrons.hardware_control.errors import (
     InvalidPipetteName,
@@ -160,6 +161,11 @@
 )
 from opentrons_shared_data.gripper.gripper_definition import GripForceProfile
 
+from opentrons_shared_data.errors.exceptions import (
+    EStopActivatedError,
+    EStopNotPresentError,
+)
+
 from .subsystem_manager import SubsystemManager
 
 if TYPE_CHECKING:
@@ -187,6 +193,29 @@ async def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
     return cast(Wrapped, wrapper)
 
 
+def requires_estop(func: Wrapped) -> Wrapped:
+    """Decorator that raises an exception if the Estop is engaged."""
+
+    @wraps(func)
+    async def wrapper(self: OT3Controller, *args: Any, **kwargs: Any) -> Any:
+        state = self._estop_state_machine.state
+        if state == EstopState.NOT_PRESENT and ff.require_estop():
+            raise EStopNotPresentError(
+                message="An Estop must be plugged in to move the robot."
+            )
+        if state == EstopState.LOGICALLY_ENGAGED:
+            raise EStopActivatedError(
+                message="Estop must be acknowledged and cleared to move the robot."
+            )
+        if state == EstopState.PHYSICALLY_ENGAGED:
+            raise EStopActivatedError(
+                message="Estop is currently engaged, robot cannot move."
+            )
+        return await func(self, *args, **kwargs)
+
+    return cast(Wrapped, wrapper)
+
+
 class OT3Controller:
     """OT3 Hardware Controller Backend."""
 
@@ -462,6 +491,7 @@ def _handle_motor_status_response(
             )
 
     @requires_update
+    @requires_estop
     async def move(
         self,
         origin: Coordinates[Axis, float],
@@ -567,6 +597,7 @@ def _build_home_gantry_z_runner(
         return None
 
     @requires_update
+    @requires_estop
     async def home(
         self, axes: Sequence[Axis], gantry_load: GantryLoad
     ) -> OT3AxisMap[float]:
@@ -656,6 +687,7 @@ async def tip_action(
             log.debug("no position returned from NodeId.pipette_left")
 
     @requires_update
+    @requires_estop
     async def gripper_grip_jaw(
         self,
         duty_cycle: float,
@@ -670,6 +702,7 @@ async def gripper_grip_jaw(
         self._handle_motor_status_response(positions)
 
     @requires_update
+    @requires_estop
     async def gripper_hold_jaw(
         self,
         encoder_position_um: int,
@@ -680,6 +713,7 @@ async def gripper_hold_jaw(
         self._handle_motor_status_response(positions)
 
     @requires_update
+    @requires_estop
     async def gripper_home_jaw(self, duty_cycle: float) -> None:
         move_group = create_gripper_jaw_home_group(duty_cycle)
         runner = MoveGroupRunner(move_groups=[move_group])

api/tests/opentrons/config/test_advanced_settings_migration.py

@@ -7,7 +7,7 @@
 
 @pytest.fixture
 def migrated_file_version() -> int:
-    return 28
+    return 29
 
 
 # make sure to set a boolean value in default_file_settings only if
@@ -28,6 +28,7 @@ def default_file_settings() -> Dict[str, Any]:
         "disableStatusBar": None,
         "disableOverpressureDetection": None,
         "disableTipPresenceDetection": None,
+        "estopNotRequired": None,
     }
 
 
@@ -346,6 +347,18 @@ def v28_config(v27_config: Dict[str, Any]) -> Dict[str, Any]:
     return r
 
 
+@pytest.fixture
+def v29_config(v28_config: Dict[str, Any]) -> Dict[str, Any]:
+    r = v28_config.copy()
+    r.update(
+        {
+            "_version": 29,
+            "estopNotRequired": None,
+        }
+    )
+    return r
+
+
 @pytest.fixture(
     scope="session",
     params=[
@@ -379,6 +392,7 @@ def v28_config(v27_config: Dict[str, Any]) -> Dict[str, Any]:
         lazy_fixture("v26_config"),
         lazy_fixture("v27_config"),
         lazy_fixture("v28_config"),
+        lazy_fixture("v29_config"),
     ],
 )
 def old_settings(request: pytest.FixtureRequest) -> Dict[str, Any]:
@@ -468,4 +482,5 @@ def test_ensures_config() -> None:
         "rearPanelIntegration": None,
         "disableStallDetection": None,
         "disableStatusBar": None,
+        "estopNotRequired": None,
     }

api/tests/opentrons/hardware_control/backends/test_ot3_controller.py

@@ -51,6 +51,7 @@
     UpdateState,
     TipStateType,
     FailedTipStateCheck,
+    EstopState,
 )
 from opentrons.hardware_control.errors import (
     FirmwareUpdateRequired,
@@ -73,6 +74,14 @@
     PipetteInformation,
     GripperInformation,
 )
+
+from opentrons.hardware_control.estop_state import EstopStateMachine
+
+from opentrons_shared_data.errors.exceptions import (
+    EStopActivatedError,
+    EStopNotPresentError,
+)
+
 from opentrons_hardware.hardware_control.move_group_runner import MoveGroupRunner
 
 
@@ -250,6 +259,16 @@ def mock_subsystem_manager(
         yield mock_subsystem
 
 
+@pytest.fixture
+def mock_estop_state_machine(
+    controller: OT3Controller, decoy: Decoy
+) -> Iterator[EstopStateMachine]:
+    with mock.patch.object(
+        controller, "_estop_state_machine", decoy.mock(cls=EstopStateMachine)
+    ) as mock_estop_state:
+        yield mock_estop_state
+
+
 @pytest.fixture
 def fw_update_info() -> Dict[NodeId, str]:
     return {
@@ -1070,3 +1089,26 @@ async def test_get_tip_present(
     ):
         with expectation:
             await controller.get_tip_present(mount, tip_state_type)
+
+
+@pytest.mark.parametrize(
+    "estop_state, expectation",
+    [
+        [EstopState.DISENGAGED, does_not_raise()],
+        [EstopState.NOT_PRESENT, pytest.raises(EStopNotPresentError)],
+        [EstopState.PHYSICALLY_ENGAGED, pytest.raises(EStopActivatedError)],
+        [EstopState.LOGICALLY_ENGAGED, pytest.raises(EStopActivatedError)],
+    ],
+)
+async def test_requires_estop(
+    controller: OT3Controller,
+    mock_estop_state_machine: EstopStateMachine,
+    decoy: Decoy,
+    estop_state: EstopState,
+    expectation: ContextManager[None],
+) -> None:
+    """Test that the estop state machine raises properly."""
+    decoy.when(mock_estop_state_machine.state).then_return(estop_state)
+
+    with expectation:
+        await controller.home([Axis.X, Axis.Y], gantry_load=GantryLoad.LOW_THROUGHPUT)

robot-server/robot_server/robot/control/estop_handler.py

@@ -2,6 +2,7 @@
 import logging
 from typing import TYPE_CHECKING
 from .models import EstopState, EstopPhysicalStatus
+from opentrons.config.feature_flags import require_estop
 
 if TYPE_CHECKING:
     from opentrons.hardware_control.ot3api import OT3API
@@ -23,7 +24,10 @@ def __init__(
 
     def get_state(self) -> EstopState:
         """Get the current estop state."""
-        return EstopState.from_hw_state(self._hardware_handle.estop_status.state)
+        state = EstopState.from_hw_state(self._hardware_handle.estop_status.state)
+        if state == EstopState.NOT_PRESENT and not require_estop():
+            return EstopState.DISENGAGED
+        return state
 
     def get_left_physical_status(self) -> EstopPhysicalStatus:
         """Get the physical status of the left estop."""

robot-server/robot_server/service/legacy/routers/settings.py

@@ -104,6 +104,7 @@ def _create_settings_response() -> AdvancedSettingsResponse:
                 value=s.value,
             )
             for s in data.values()
+            if s.definition.should_show()
         ],
     )