Clean rewrite of the driver in Python

.flake8

@@ -3,5 +3,5 @@
 # Use black's line length (default 88) instead of the flake8 default of 79:
 max-line-length = 88
 
-per-file-ignores =
-    schunk_egu_egk_gripper_dummy/*.py:E203
+ignore = E203, W503
+# W503: Change in best practices, see here: https://www.flake8rules.com/rules/W503.html

.github/script/install_dependencies.sh

@@ -1,9 +1,9 @@
 #/usr/bin/bash
 cd $HOME
-apt-get install -y curl libcurl4-openssl-dev
+apt-get install -y socat
 
 # Python dependencies
-python_deps="fastapi uvicorn httpx requests coverage python-multipart"
+python_deps="fastapi uvicorn httpx requests coverage python-multipart pymodbus pyserial"
 os_name=$(lsb_release -cs)
 
 case $os_name in

schunk_egu_egk_gripper_driver/launch/schunk.launch.py

@@ -49,7 +49,7 @@ def generate_launch_description():
                 package="schunk_egu_egk_gripper_driver",
                 plugin="SchunkGripperNode",
                 name="schunk_gripper_driver",
-                namespace="",
+                namespace="schunk",
                 parameters=[
                     {"IP": LaunchConfiguration("IP")},
                     {"port": LaunchConfiguration("port")},

schunk_gripper_driver/README.md

@@ -0,0 +1,22 @@
+# SCHUNK Gripper Driver
+
+
+## Lifecycle behavior
+
+The driver implements a [lifecycle node](https://github.com/ros2/demos/tree/humble/lifecycle) with defined state transitions.
+This gives us advanced control over configuration and resetting processes.
+
+### Command line
+With built-in features
+
+```bash
+ros2 lifecycle get /schunk/driver
+ros2 lifecycle set /schunk/driver configure  # activate | deactivate | cleanup | shutdown
+```
+
+And with ROS2 services
+
+```bash
+ros2 service call /schunk/driver/get_state lifecycle_msgs/srv/GetState '{}'
+ros2 service call /schunk/driver/change_state lifecycle_msgs/srv/ChangeState '{transition: {label: configure}}'  # activate | deactivate etc.
+```

schunk_gripper_driver/launch/driver.launch.py

@@ -0,0 +1,52 @@
+# Copyright 2025 SCHUNK SE & Co. KG
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <https://www.gnu.org/licenses/>.
+# --------------------------------------------------------------------------------
+
+from launch import LaunchDescription
+from launch_ros.actions import Node
+from launch.actions import DeclareLaunchArgument
+from launch.substitutions import LaunchConfiguration
+
+
+port = DeclareLaunchArgument(
+    "port",
+    default_value="/dev/ttyUSB0",
+    description="The gripper's serial port",
+)
+device_id = DeclareLaunchArgument(
+    "device_id",
+    default_value="12",
+    description="The gripper's Modbus device id",
+)
+args = [port, device_id]
+
+
+def generate_launch_description():
+    return LaunchDescription(
+        args
+        + [
+            Node(
+                package="schunk_gripper_driver",
+                namespace="schunk",
+                executable="driver.py",
+                name="driver",
+                parameters=[
+                    {"port": LaunchConfiguration("port")},
+                    {"device_id": LaunchConfiguration("device_id")},
+                ],
+                output="both",
+            )
+        ]
+    )

schunk_gripper_driver/package.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>schunk_gripper_driver</name>
+  <version>0.0.0</version>
+  <description>ROS2 driver for SCHUNK`s EGU, EGK, and EZU mechatronic grippers</description>
+  <maintainer email="[email protected]">stefan</maintainer>
+  <license>GPL-3.0-or-later</license>
+
+  <depend>rclpy</depend>
+  <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>
+  <test_depend>python3-pytest</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

schunk_gripper_driver/schunk_gripper_driver/driver.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+# Copyright 2025 SCHUNK SE & Co. KG
+#
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+# more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program. If not, see <https://www.gnu.org/licenses/>.
+# --------------------------------------------------------------------------------
+
+import rclpy
+
+from rclpy.lifecycle import Node
+from rclpy.lifecycle import State
+from rclpy.lifecycle import TransitionCallbackReturn
+from schunk_gripper_library.driver import Driver as GripperDriver
+from std_srvs.srv import Trigger
+import asyncio
+
+
+class Driver(Node):
+
+    def __init__(self, node_name: str, **kwargs):
+        super().__init__(node_name, **kwargs)
+        self.declare_parameter("port", rclpy.Parameter.Type.STRING)
+        self.declare_parameter("device_id", rclpy.Parameter.Type.INTEGER)
+        self.port = self.get_parameter_or("port", "/dev/ttyUSB0").value
+        self.gripper = GripperDriver()
+
+    def on_configure(self, state: State) -> TransitionCallbackReturn:
+        self.get_logger().info("on_configure() is called.")
+        self.get_logger().info(f"Connecting on port {self.port}")
+        if not self.gripper.connect(
+            protocol="modbus",
+            port=self.get_parameter_or("port", "/dev/ttyUSB0").value,
+            device_id=self.get_parameter_or("device_id", 12).value,
+        ):
+            self.get_logger().warn("Gripper connect failed")
+            return TransitionCallbackReturn.FAILURE
+
+        # Services
+        self.acknowledge_srv = self.create_service(
+            Trigger, "~/acknowledge", self._acknowledge_cb
+        )
+        self.fast_stop_srv = self.create_service(
+            Trigger, "~/fast_stop", self._fast_stop_cb
+        )
+
+        # State update
+        self.timer = self.create_timer(0.5, self.status_update)
+
+        # Clear control bits
+        self.gripper.clear_plc_output()
+        self.gripper.set_control_bit(bit=0, value=True)
+        self.gripper.send_plc_output()
+        return TransitionCallbackReturn.SUCCESS
+
+    def on_activate(self, state: State) -> TransitionCallbackReturn:
+        self.get_logger().info("on_activate() is called.")
+        return super().on_activate(state)
+
+    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
+        self.get_logger().info("on_deactivate() is called.")
+        return super().on_deactivate(state)
+
+    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
+        self.get_logger().info("on_cleanup() is called.")
+        self.gripper.disconnect()
+
+        # Release services
+        if not self.destroy_service(self.acknowledge_srv):
+            return TransitionCallbackReturn.FAILURE
+        if not self.destroy_service(self.fast_stop_srv):
+            return TransitionCallbackReturn.FAILURE
+
+        return TransitionCallbackReturn.SUCCESS
+
+    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
+        self.get_logger().info("on_shutdown() is called.")
+        return TransitionCallbackReturn.SUCCESS
+
+    def status_update(self):
+        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")
+        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")
+        response.success = asyncio.run(self.gripper.fast_stop())
+        response.message = self.gripper.get_status_diagnostics()
+        return response
+
+
+def main():
+    rclpy.init()
+    executor = rclpy.executors.SingleThreadedExecutor()
+    driver = Driver("driver")
+    executor.add_node(driver)
+    try:
+        executor.spin()
+    except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
+        driver.destroy_node()
+
+
+if __name__ == "__main__":
+    main()

schunk_gripper_driver/setup.cfg

@@ -0,0 +1,4 @@
+[develop]
+script_dir=$base/lib/schunk_gripper_driver
+[install]
+install_scripts=$base/lib/schunk_gripper_driver

schunk_gripper_driver/setup.py

@@ -0,0 +1,27 @@
+from setuptools import find_packages, setup
+from glob import glob
+import os
+
+package_name = "schunk_gripper_driver"
+
+setup(
+    name=package_name,
+    version="0.0.0",
+    packages=find_packages(exclude=["test"]),
+    data_files=[
+        ("share/ament_index/resource_index/packages", ["resource/" + package_name]),
+        ("share/" + package_name, ["package.xml"]),
+        (os.path.join("lib", package_name), [package_name + "/driver.py"]),
+        (os.path.join("share", package_name, "launch"), glob("launch/*.launch.py")),
+    ],
+    install_requires=["setuptools", "pymodbus"],
+    zip_safe=True,
+    maintainer="stefan",
+    maintainer_email="[email protected]",
+    description="ROS2 driver for SCHUNK`s EGU, EGK, and EZU mechatronic grippers",
+    license="GPL-3.0-or-later",
+    tests_require=["pytest"],
+    entry_points={
+        "console_scripts": [],
+    },
+)

schunk_gripper_driver/test/README.md

@@ -0,0 +1,11 @@
+## Run tests locally
+Inside the driver's package
+
+```bash
+pip install --user pytest coverage
+```
+
+```bash
+coverage run -m pytest test/
+coverage report
+```