Add a gripper#
Prerequisites: Python and the gripper contract. Read ROS actions before implementing a ROS adapter. Outcome: a fake goal-based gripper with completion and cancellation tests.
1. Select the gripper independently#
GRIPPER_TYPE=my_gripper and GTA_BACKEND=ros2 select
scene/gripper/my_gripper/ros2.py, exporting Gripper(BaseGripper).
Provide the package’s __init__.py too. ROBOT_TYPE is independent.
There is no fallback to Robotiq and no separate gripper-backend setting.
2. Implement the lifecycle#
"""A borrowed-client gripper adapter with a deterministic memory-only client."""
import math
from types import SimpleNamespace
from scene.gripper.base import BaseGripper
class Future:
def __init__(self, value):
self.value = value
self.ready = False
def done(self):
return self.ready
def result(self):
if not self.ready:
raise RuntimeError("Future not complete")
return self.value
class Goal:
def __init__(self, accepted=True, succeeded=True):
self.accepted = accepted
self.succeeded = succeeded
self.cancelled = False
def result_future(self):
return Future(SimpleNamespace(succeeded=self.succeeded))
def cancel_future(self):
self.cancelled = True
return Future(True)
class FakeClient:
"""The example's positions are synthetic; they are not Robotiq values."""
def __init__(self, available=True, accepted=True, succeeded=True):
self.available = available
self.goal = Goal(accepted, succeeded)
self.positions = []
self.destroyed = False
def wait_for_server(self, timeout_s):
return self.available
def send(self, position):
self.positions.append(position)
return Future(self.goal)
def destroy(self):
self.destroyed = True
def complete_future(future, timeout_s):
"""A fake executor: complete immediately, without threads or ROS."""
future.ready = True
class Gripper(BaseGripper):
def __init__(self, *, node=None, spin_until_complete=complete_future,
check_cancel=None, client=None):
self.client = client if client is not None else FakeClient()
self.spin = spin_until_complete # Borrowed; never own ROS lifecycle.
self.check_cancel = check_cancel or (lambda: None)
self.pending = self.active = None
@staticmethod
def _timeout(value):
value = 2.0 if value is None else float(value)
if not math.isfinite(value) or value <= 0:
raise ValueError("Timeout must be finite and positive")
return value
def _wait(self, future, seconds, *, check_cancel=True):
if check_cancel:
self.check_cancel()
self.spin(future, seconds)
if check_cancel:
self.check_cancel()
if not future.done():
raise TimeoutError("Goal did not complete")
return future.result()
def wait_for_server(self, timeout_s=None):
self.check_cancel()
if not self.client.wait_for_server(self._timeout(timeout_s)):
raise TimeoutError("Gripper unavailable")
def _send(self, position, timeout_s):
self.check_cancel()
seconds = self._timeout(timeout_s)
self.pending = self.client.send(position)
goal = self._wait(self.pending, seconds)
self.pending = None
if not goal.accepted:
raise RuntimeError("Goal rejected")
self.active = goal
result = self._wait(goal.result_future(), seconds)
if not result.succeeded:
raise RuntimeError("Goal failed")
self.active = None
return result
def open(self, timeout_s=None):
return self._send(1.0, timeout_s)
def close(self, timeout_s=None):
return self._send(0.0, timeout_s)
def cancel(self, *, ack_timeout_s=0.5, timeout_s=0.5):
errors = []
try:
if self.pending is not None:
goal = self._wait(self.pending, ack_timeout_s, check_cancel=False)
self.pending = None
if goal.accepted:
self.active = goal
if self.active is not None:
confirmed = self._wait(
self.active.cancel_future(), timeout_s, check_cancel=False
)
if not confirmed:
raise RuntimeError("Cancellation not confirmed")
self.active = None
except Exception as exc:
errors.append(str(exc))
return errors
def shutdown(self):
try:
if self.client.destroy() is False:
return ["Client destruction failed"]
except Exception as exc:
return [str(exc)]
return []
The example uses 1.0 for open and 0.0 for closed. The bundled Robotiq
adapter’s positions are listed in the ROS reference.
from docs.examples.fake_gripper import FakeClient, Gripper
client = FakeClient()
gripper = Gripper(client=client)
gripper.wait_for_server() # Raises if unavailable.
gripper.open()
gripper.close()
assert client.positions == [1.0, 0.0]
assert gripper.cancel() == []
assert gripper.shutdown() == []
assert client.destroyed
close() closes the jaws; shutdown() releases client resources.
Optional half_open() inherits a clear unsupported-operation
error unless implemented.
3. Keep ROS ownership with the caller#
A production ROS 2 client accepts the caller’s node,
spin_until_complete(future, timeout_s), and optional check_cancel.
It must not initialize/shutdown the shared ROS context, create robot command
subscriptions, or open an RTDE connection. The supervised gripper-only
operation creates its own node and executor outside the device implementation.
Keep availability and goal/result waits bounded where the caller requests a deadline. Rejected goals, failed action status, and timeouts are failures. Cancellation can arrive before the goal handle: retain the pending acknowledgement, then cancel a late accepted goal. Cancellation/cleanup must remain callable after the normal operation cancellation flag has been set. Return cleanup-error lists instead of hiding failed cleanup.
The fake makes future completion explicit so tests can inject rejection, failure, timeout, and late acknowledgement. A real action implementation must also handle ROS future exceptions and cancelled futures.
4. Follow the existing caller policies#
The implementations in scene/gripper/robotiq/ show the current integration:
ROS 2 uses
control_msgs/action/GripperCommand, tracks pending and active goals, and supports cancellation deadlines. A successful action status withreached_goal=Falseproduces a warning rather than a failure.ROS 1 uses
robotiq_mm_ros. Its server retains unbounded result waiting without strict success validation; the standalone gripper utilities request bounded waits and success checks.The robot servers keep GTA’s open/close topic convention and suppress repeated identical commands. The gripper backend owns action details.
Use the goal units and open/closed positions defined by your gripper driver.
Verify: use fake futures to cover accepted/rejected/failed/timed-out goals, unavailable servers, cancellation before acknowledgement, and cleanup errors. Confirm a gripper-only operation creates no robot-command subscriptions or arm connections. For action startup errors see execution troubleshooting.