ROS 2 adapters

Most people only need the ROS 2 bridge. Use these classes when you want to build your own ROS 2 node instead.

from zeroruntime.teleops.adapters.ros2 import (JointRanges, ROS2FollowerAdapter,
                                               ROS2LeaderAdapter)

ROS2FollowerAdapter

Use this on the robot machine.

ROS2FollowerAdapter(meeting_id=None, *, ranges, sink, latest=None,
                    robot_id="ros2", state_timeout_s=0.5, control_hz=200, **kw)

JointRanges

required

How far each joint can move. See JointRanges.

callable

required

Called with each command, as {joint: position} in your topic’s units. Send it to your robot from here.

object

Holds the robot’s latest joint positions, filled by your subscription. See the example.

any

Any argument of Follower, such as safety or cameras.

ROS2LeaderAdapter

Use this on the operator machine. It takes the same ranges and latest, but no sink.

ROS2LeaderAdapter(meeting_id=None, *, ranges, latest=None, robot_id="ros2",
                  state_timeout_s=0.5, deadband=0.0, **kw)

any

Any argument of Leader.

JointRanges

How far each joint can move on this machine’s arm, in your topic’s units.

ranges = JointRanges.from_lists(("j1", "j2", "j3"),
                                mins=[-3.14, -1.57, -2.0],
                                maxs=[3.14, 1.57, 2.0])

Use the same joint names, in the same order, on both machines.

Example

A follower node. The subscription stores the robot’s position in latest, and the adapter sends commands through send. Export ZERORUNTIME_AUTH_TOKEN and ZERORUNTIME_MEETING_ID first, or pass meeting_id= and token= to the adapter.

follower_node.py

import threading
import time

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState
from std_msgs.msg import Float64MultiArray

from zeroruntime.teleops.adapters.ros2 import JointRanges, ROS2FollowerAdapter

JOINTS = ("j1", "j2", "j3")
RANGES = JointRanges.from_lists(JOINTS, mins=[-3.14, -1.57, -2.0],
                                maxs=[3.14, 1.57, 2.0])


class Latest:
    """The latest joint positions, and how old they are."""

    def __init__(self):
        self._lock = threading.Lock()
        self._values, self._stamp = None, 0.0

    def put(self, values):
        with self._lock:
            self._values, self._stamp = dict(values), time.monotonic()

    def get(self):
        with self._lock:
            if self._values is None:
                return None, float("inf")
            return dict(self._values), time.monotonic() - self._stamp


class ArmBridge(Node):
    def __init__(self):
        super().__init__("arm_bridge")
        self.latest = Latest()
        self.pub = self.create_publisher(
            Float64MultiArray, "/follower/forward_controller/commands", 10)
        self.create_subscription(
            JointState, "/follower/joint_states", self.on_state, 10)

        self.adapter = ROS2FollowerAdapter(
            ranges=RANGES, latest=self.latest, sink=self.send)
        self.adapter.start()
        self.create_timer(1 / 200, self.adapter.tick)

    def on_state(self, msg):
        self.latest.put(dict(zip(msg.name, msg.position)))

    def send(self, joints):
        self.pub.publish(Float64MultiArray(data=[float(joints[n]) for n in JOINTS]))


rclpy.init()
node = ArmBridge()
try:
    rclpy.spin(node)
finally:
    node.adapter.stop()
    rclpy.try_shutdown()

start() joins the room, and the timer calls tick() to run the control loop.

What’s next

Follower

All the arguments the follower takes.

ROS 2 bridge

The ready-made bridge built on these classes.

Python SDK

Adapters

Classes

ROS2FollowerAdapter

ROS2LeaderAdapter

JointRanges

Example

What’s next

Follower

ROS 2 bridge

​ROS2FollowerAdapter

​ROS2LeaderAdapter

​JointRanges

​Example

​What’s next

Follower

ROS 2 bridge

ROS2FollowerAdapter

ROS2LeaderAdapter

JointRanges

Example

What’s next