Added estop to hardware controller and robot server

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

@@ -67,6 +67,11 @@
     Move,
     Coordinates,
 )
+from opentrons_hardware.hardware_control.estop.detector import (
+    EstopDetector,
+)
+
+from opentrons.hardware_control.estop_state import EstopStateMachine
 
 from opentrons_hardware.hardware_control.motor_enable_disable import (
     set_enable_motor,
@@ -249,6 +254,8 @@ def __init__(
             network.NetworkInfo(self._messenger, self._usb_messenger),
             FirmwareUpdate(),
         )
+        self._estop_detector: Optional[EstopDetector] = None
+        self._estop_state_machine: Optional[EstopStateMachine] = None
         self._position = self._get_home_position()
         self._encoder_position = self._get_home_position()
         self._motor_status = {}
@@ -1162,3 +1169,18 @@ def _door_listener(msg: BinaryMessageDefinition) -> None:
 
     def status_bar_interface(self) -> status_bar.StatusBar:
         return self._status_bar
+
+    async def build_estop_state_machine(self) -> bool:
+        """Must be called to set up the estop detector & state machine."""
+        if self._drivers.usb_messenger is None:
+            return False
+        self._estop_detector = await EstopDetector.build(
+            usb_messenger=self._drivers.usb_messenger
+        )
+        self._estop_state_machine = EstopStateMachine(self._estop_detector)
+        return True
+
+    @property
+    def estop_state_machine(self) -> Optional[EstopStateMachine]:
+        """Accessor for the API to get the state machine, if it exists."""
+        return self._estop_state_machine

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

@@ -688,3 +688,8 @@ def subsystems(self) -> Dict[SubSystem, SubSystemState]:
             )
             for target in self._present_nodes
         }
+
+    @property
+    def estop_state_machine(self) -> None:
+        """Simulator will not return an estop state machine."""
+        return None

api/src/opentrons/hardware_control/ot3api.py

@@ -98,6 +98,10 @@
     StatusBarState,
     SubSystemState,
     TipStateType,
+    EstopOverallStatus,
+    EstopAttachLocation,
+    EstopState,
+    EstopPhysicalStatus,
 )
 from .errors import (
     MustHomeError,
@@ -317,6 +321,7 @@ async def build_hardware_controller(
             api_instance, board_revision=backend.board_revision
         )
         backend.module_controls = module_controls
+        await backend.build_estop_state_machine()
         door_state = await backend.door_state()
         api_instance._update_door_state(door_state)
         backend.add_door_state_listener(api_instance._update_door_state)
@@ -2248,3 +2253,29 @@ def _axis_map_from_ot3axis_map(
     def attached_subsystems(self) -> Dict[SubSystem, SubSystemState]:
         """Get a view of the state of the currently-attached subsystems."""
         return self._backend.subsystems
+
+    @property
+    def estop_status(self) -> EstopOverallStatus:
+        if self._backend.estop_state_machine is None:
+            return EstopOverallStatus(
+                state=EstopState.DISENGAGED,
+                left_physical_state=EstopPhysicalStatus.DISENGAGED,
+                right_physical_state=EstopPhysicalStatus.DISENGAGED,
+            )
+        return EstopOverallStatus(
+            state=self._backend.estop_state_machine.state,
+            left_physical_state=self._backend.estop_state_machine.get_physical_status(
+                EstopAttachLocation.LEFT
+            ),
+            right_physical_state=self._backend.estop_state_machine.get_physical_status(
+                EstopAttachLocation.RIGHT
+            ),
+        )
+
+    def estop_acknowledge_and_clear(self) -> EstopOverallStatus:
+        """Attempt to acknowledge an Estop event and clear the status.
+
+        Returns the estop status after clearing the status."""
+        if self._backend.estop_state_machine is not None:
+            self._backend.estop_state_machine.acknowledge_and_clear()
+        return self.estop_status

robot-server/robot_server/hardware.py

@@ -52,6 +52,8 @@
 from .subsystems.models import SubSystem
 from .service.task_runner import TaskRunner, get_task_runner
 
+from .robot.control.estop_handler import EstopHandler
+
 if TYPE_CHECKING:
     from opentrons.hardware_control.ot3api import OT3API
 
@@ -69,6 +71,7 @@
 _firmware_update_manager_accessor = AppStateAccessor[FirmwareUpdateManager](
     "firmware_update_manager"
 )
+_estop_handler_accessor = AppStateAccessor[EstopHandler]("estop_handler")
 
 
 class _ExcPassthrough(BaseException):
@@ -216,6 +219,20 @@ async def get_firmware_update_manager(
     return update_manager
 
 
+async def get_estop_handler(
+    app_state: AppState = Depends(get_app_state),
+    thread_manager: ThreadManagedHardware = Depends(get_thread_manager),
+) -> EstopHandler:
+    """Get an Estop Handler for working with the estop."""
+    hardware = get_ot3_hardware(thread_manager)
+    estop_handler = _estop_handler_accessor.get_from(app_state)
+
+    if estop_handler is None:
+        estop_handler = EstopHandler(hw_handle=hardware)
+        _estop_handler_accessor.set_on(app_state, estop_handler)
+    return estop_handler
+
+
 async def get_robot_type() -> RobotType:
     """Return what kind of robot this server is running on."""
     return "OT-3 Standard" if should_use_ot3() else "OT-2 Standard"

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

@@ -0,0 +1,42 @@
+"""Class to monitor estop status."""
+import logging
+from typing import TYPE_CHECKING
+from .models import EstopState, EstopPhysicalStatus
+
+if TYPE_CHECKING:
+    from opentrons.hardware_control.ot3api import OT3API
+
+log = logging.getLogger(__name__)
+
+
+class EstopHandler:
+    """Robot server interface for estop interactions."""
+
+    _hardware_handle: "OT3API"
+
+    def __init__(
+        self,
+        hw_handle: "OT3API",
+    ) -> None:
+        """Create a new EstopHandler."""
+        self._hardware_handle = hw_handle
+
+    def get_state(self) -> EstopState:
+        """Get the current estop state."""
+        return EstopState.from_hw_state(self._hardware_handle.estop_status.state)
+
+    def get_left_physical_status(self) -> EstopPhysicalStatus:
+        """Get the physical status of the left estop."""
+        return EstopPhysicalStatus.from_hw_physical_status(
+            self._hardware_handle.estop_status.left_physical_state
+        )
+
+    def get_right_physical_status(self) -> EstopPhysicalStatus:
+        """Get the physical status of the right estop."""
+        return EstopPhysicalStatus.from_hw_physical_status(
+            self._hardware_handle.estop_status.right_physical_state
+        )
+
+    def acknowledge_and_clear(self) -> None:
+        """Clear and acknowledge an Estop event."""
+        self._hardware_handle.estop_acknowledge_and_clear()

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

@@ -16,37 +16,42 @@ class EstopState(enum.Enum):
     LOGICALLY_ENGAGED = "logicallyEngaged"
     DISENGAGED = "disengaged"
 
-
     @classmethod
     def from_hw_state(cls, hw_state: HwEstopState) -> "EstopState":
         """Build from the hardware equivalent."""
         return _HW_STATE_TO_STATE[hw_state]
 
+
 _HW_STATE_TO_STATE = {
-    HwEstopState.NOT_PRESENT : EstopState.NOT_PRESENT,
-    HwEstopState.PHYSICALLY_ENGAGED : EstopState.PHYSICALLY_ENGAGED,
-    HwEstopState.LOGICALLY_ENGAGED : EstopState.LOGICALLY_ENGAGED,
-    HwEstopState.DISENGAGED : EstopState.DISENGAGED,
+    HwEstopState.NOT_PRESENT: EstopState.NOT_PRESENT,
+    HwEstopState.PHYSICALLY_ENGAGED: EstopState.PHYSICALLY_ENGAGED,
+    HwEstopState.LOGICALLY_ENGAGED: EstopState.LOGICALLY_ENGAGED,
+    HwEstopState.DISENGAGED: EstopState.DISENGAGED,
 }
 
+
 class EstopPhysicalStatus(enum.Enum):
     """Physical status of a specific estop."""
 
     ENGAGED = "engaged"
     DISENGAGED = "disengaged"
     NOT_PRESENT = "notPresent"
 
-    @classmethod 
-    def from_hw_physical_status(cls, hw_physical_status: HwEstopPhysicalStatus) -> "EstopPhysicalStatus":
+    @classmethod
+    def from_hw_physical_status(
+        cls, hw_physical_status: HwEstopPhysicalStatus
+    ) -> "EstopPhysicalStatus":
         """Build from the hardware equivalent."""
         return _HW_PHYSICAL_STATUS_TO_PHYSICAL_STATUS[hw_physical_status]
-
+
+
 _HW_PHYSICAL_STATUS_TO_PHYSICAL_STATUS = {
-    HwEstopPhysicalStatus.NOT_PRESENT : EstopState.NOT_PRESENT,
-    HwEstopPhysicalStatus.ENGAGED : EstopState.ENGAGED,
-    HwEstopPhysicalStatus.DISENGAGED : EstopState.DISENGAGED,
+    HwEstopPhysicalStatus.NOT_PRESENT: EstopPhysicalStatus.NOT_PRESENT,
+    HwEstopPhysicalStatus.ENGAGED: EstopPhysicalStatus.ENGAGED,
+    HwEstopPhysicalStatus.DISENGAGED: EstopPhysicalStatus.DISENGAGED,
 }
 
+
 class EstopStatusModel(BaseModel):
     """Model for the current estop status."""
 

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

@@ -8,35 +8,27 @@
     PydanticResponse,
     SimpleBody,
 )
-from robot_server.hardware import (
-    get_ot3_hardware,
-    get_thread_manager,
-)
-from opentrons.hardware_control import ThreadManagedHardware
 
 from .models import (
-    EstopState,
     EstopStatusModel,
-    EstopPhysicalStatus,
 )
+from .estop_handler import EstopHandler
+from robot_server.hardware import get_estop_handler
 
 if TYPE_CHECKING:
     from opentrons.hardware_control.ot3api import OT3API  # noqa: F401
 
 control_router = APIRouter()
 
 
-async def _get_estop_status(
-    thread_manager: ThreadManagedHardware,
+async def _get_estop_status_response(
+    estop_handler: EstopHandler,
 ) -> PydanticResponse[SimpleBody[EstopStatusModel]]:
     """Helper to generate the current Estop Status as a response model."""
-    get_ot3_hardware(thread_manager)
-
-    # TODO - unstub the response here
     data = EstopStatusModel.construct(
-        status=EstopState.DISENGAGED,
-        leftEstopPhysicalStatus=EstopPhysicalStatus.DISENGAGED,
-        rightEstopPhysicalStatus=EstopPhysicalStatus.DISENGAGED,
+        status=estop_handler.get_state(),
+        leftEstopPhysicalStatus=estop_handler.get_left_physical_status(),
+        rightEstopPhysicalStatus=estop_handler.get_right_physical_status(),
     )
     return await PydanticResponse.create(content=SimpleBody.construct(data=data))
 
@@ -51,10 +43,10 @@ async def _get_estop_status(
     },
 )
 async def get_estop_status(
-    thread_manager: ThreadManagedHardware = Depends(get_thread_manager),
+    estop_handler: EstopHandler = Depends(get_estop_handler),
 ) -> PydanticResponse[SimpleBody[EstopStatusModel]]:
     """Return the current status of the estop."""
-    return await _get_estop_status(thread_manager)
+    return await _get_estop_status_response(estop_handler)
 
 
 @control_router.put(
@@ -68,7 +60,8 @@ async def get_estop_status(
     },
 )
 async def put_acknowledge_estop_disengage(
-    thread_manager: ThreadManagedHardware = Depends(get_thread_manager),
+    estop_handler: EstopHandler = Depends(get_estop_handler),
 ) -> PydanticResponse[SimpleBody[EstopStatusModel]]:
     """Transition from the `logically_engaged` status if applicable."""
-    return await _get_estop_status(thread_manager)
+    estop_handler.acknowledge_and_clear()
+    return await _get_estop_status_response(estop_handler)