Use the library's coroutines in the ROS2 driver

This is now as thin as we want it.
Also drop receiving the module's state periodically in the driver.
The library does that now.

Author: stefanscherzinger

schunk_gripper_driver/launch/driver.launch.py

@@ -22,7 +22,7 @@
 
 port = DeclareLaunchArgument(
     "port",
-    default_value="/dev/ur-ttylink/ttyTool",
+    default_value="/dev/ttyUSB0",
     description="The gripper's serial port",
 )
 device_id = DeclareLaunchArgument(

schunk_gripper_driver/package.xml

@@ -11,6 +11,7 @@
   <depend>launch</depend>
   <depend>launch_ros</depend>
   <depend>schunk_gripper_library</depend>
+  <depend>std_srvs</depend>
 
   <test_depend>launch_pytest</test_depend>
   <test_depend>ament_copyright</test_depend>

schunk_gripper_driver/schunk_gripper_driver/driver.py

@@ -22,7 +22,7 @@
 from rclpy.lifecycle import TransitionCallbackReturn
 from schunk_gripper_library.driver import Driver as GripperDriver
 from std_srvs.srv import Trigger
-import time
+import asyncio
 
 
 class Driver(Node):
@@ -87,62 +87,18 @@ def on_shutdown(self, state: State) -> TransitionCallbackReturn:
         return TransitionCallbackReturn.SUCCESS
 
     def status_update(self):
-        self.gripper.receive_plc_input()
         self.get_logger().info(f"---> Status update: {self.gripper.get_plc_input()}")
 
     # Service callbacks
     def _acknowledge_cb(self, request: Trigger.Request, response: Trigger.Response):
         self.get_logger().info("---> Acknowledge")
-        # Reset
-        self.gripper.clear_plc_output()
-        self.gripper.set_control_bit(bit=0, value=True)
-        self.gripper.send_plc_output()
-        time.sleep(0.5)
-        # Acknowledge
-        self.gripper.set_control_bit(bit=2, value=True)
-        for bit in self.gripper.valid_control_bits:
-            self.get_logger().info(
-                f"---> Control bit {bit}: {self.gripper.get_control_bit(bit=bit)}"
-            )
-        cmd_before = self.gripper.get_status_bit(bit=5)
-        self.gripper.send_plc_output()
-        time.sleep(0.5)
-        self.gripper.receive_plc_input()
-        cmd_after = self.gripper.get_status_bit(bit=5)
-        self.get_logger().info(f"---> PLC input: {self.gripper.get_plc_input()}")
-        for bit in self.gripper.valid_status_bits:
-            self.get_logger().info(
-                f"---> Status bit {bit}: {self.gripper.get_status_bit(bit=bit)}"
-            )
-
-        response.success = self.gripper.get_status_bit(bit=0) == 1 and (
-            cmd_after != cmd_before
-        )
+        response.success = asyncio.run(self.gripper.acknowledge())
         response.message = self.gripper.get_status_diagnostics()
         return response
 
     def _fast_stop_cb(self, request: Trigger.Request, response: Trigger.Response):
         self.get_logger().info("---> Fast stop")
-        self.gripper.clear_plc_output()
-        self.gripper.set_control_bit(bit=0, value=False)
-        for bit in self.gripper.valid_control_bits:
-            self.get_logger().info(
-                f"---> Control bit {bit}: {self.gripper.get_control_bit(bit=bit)}"
-            )
-        cmd_before = self.gripper.get_status_bit(bit=5)
-        self.gripper.send_plc_output()
-        time.sleep(0.5)
-        self.gripper.receive_plc_input()
-        cmd_after = self.gripper.get_status_bit(bit=5)
-        self.get_logger().info(f"---> PLC input: {self.gripper.get_plc_input()}")
-        for bit in self.gripper.valid_status_bits:
-            self.get_logger().info(
-                f"---> Status bit {bit}: {self.gripper.get_status_bit(bit=bit)}"
-            )
-
-        response.success = self.gripper.get_status_bit(bit=4) == 1 and (
-            cmd_after != cmd_before
-        )
+        response.success = asyncio.run(self.gripper.fast_stop())
         response.message = self.gripper.get_status_diagnostics()
         return response
 

schunk_gripper_driver/test/test_services.py

@@ -49,11 +49,13 @@ def test_driver_implements_acknowledge(lifecycle_interface):
     client.wait_for_service(timeout_sec=2)
     future = client.call_async(Trigger.Request())
     rclpy.spin_until_future_complete(node, future)
+    lifecycle_interface.change_state(Transition.TRANSITION_CLEANUP)
 
     assert future.result().success
-    expected_msg = "error_code: 0, warning_code: 0, additional_code: 0"  # everything ok
+    expected_msg = (
+        "error_code: 0x0, warning_code: 0x0, additional_code: 0x0"  # everything ok
+    )
     assert future.result().message == expected_msg
-    lifecycle_interface.change_state(Transition.TRANSITION_CLEANUP)
 
 
 @skip_without_gripper
@@ -65,8 +67,8 @@ def test_driver_implements_fast_stop(lifecycle_interface):
     client.wait_for_service(timeout_sec=2)
     future = client.call_async(Trigger.Request())
     rclpy.spin_until_future_complete(node, future)
+    lifecycle_interface.change_state(Transition.TRANSITION_CLEANUP)
 
     assert future.result().success
-    expected_msg = "error_code: 0, warning_code: 0, additional_code: 0"  # everything ok
+    expected_msg = "error_code: 0xD9, warning_code: 0x0, additional_code: 0x0"
     assert future.result().message == expected_msg
-    lifecycle_interface.change_state(Transition.TRANSITION_CLEANUP)