feat(api): Expose the GET_ESTOP_ENGAGED (M112) G-Code for the Flex Stacker to support testing. (#19135)

Author: vegano1

api/src/opentrons/drivers/flex_stacker/driver.py

@@ -128,6 +128,15 @@ def parse_installation_detected(cls, response: str) -> bool:
             )
         return bool(int(match.group(1)))
 
+    @classmethod
+    def parse_estop_engaged(cls, response: str) -> bool:
+        """Parse estop enagaged."""
+        _RE = re.compile(rf"^{GCODE.GET_ESTOP_ENGAGED} E:(\d)$")
+        match = _RE.match(response)
+        if not match:
+            raise ValueError(f"Incorrect Response for Estop engaged: {response}")
+        return bool(int(match.group(1)))
+
     @classmethod
     def parse_move_params(cls, response: str) -> MoveParams:
         """Parse move params."""
@@ -652,6 +661,16 @@ async def get_installation_detected(self) -> bool:
         )
         return self.parse_installation_detected(response)
 
+    async def get_estop_engaged(self) -> bool:
+        """Get whether or not the estop was detected by this stacker.
+
+        :return: True if the estop is trigguered, False otherwise
+        """
+        response = await self._connection.send_command(
+            GCODE.GET_ESTOP_ENGAGED.build_command()
+        )
+        return self.parse_estop_engaged(response)
+
     async def move_in_mm(
         self, axis: StackerAxis, distance: float, params: MoveParams | None = None
     ) -> MoveResult:

api/src/opentrons/drivers/flex_stacker/types.py

@@ -12,6 +12,7 @@ class GCODE(str, Enum):
     HOME_AXIS = "G28"
     STOP_MOTORS = "M0"
     ENABLE_MOTORS = "M17"
+    GET_ESTOP_ENGAGED = "M112"
     GET_RESET_REASON = "M114"
     DEVICE_INFO = "M115"
     GET_LIMIT_SWITCH = "M119"

api/tests/opentrons/drivers/flex_stacker/test_driver.py

@@ -750,3 +750,29 @@ async def test_set_tof_configuration(
     )
     connection.send_command.assert_any_call(set_config)
     connection.reset_mock()
+
+
+async def test_get_estop_engaged(
+    subject: FlexStackerDriver, connection: AsyncMock
+) -> None:
+    """It should send a get estop command return boolean."""
+    connection.send_command.return_value = "M112 E:1"
+    response = await subject.get_estop_engaged()
+    assert response
+
+    estop = types.GCODE.GET_ESTOP_ENGAGED.build_command()
+    connection.send_command.assert_any_call(estop)
+    connection.reset_mock()
+
+
+async def test_get_install_detected(
+    subject: FlexStackerDriver, connection: AsyncMock
+) -> None:
+    """It should send a get install detected command and return boolean."""
+    connection.send_command.return_value = "M123 I:1"
+    response = await subject.get_installation_detected()
+    assert response
+
+    install_detect = types.GCODE.GET_INSTALL_DETECTED.build_command()
+    connection.send_command.assert_any_call(install_detect)
+    connection.reset_mock()