Get Publisher Hz

SUMMARY

Pass name=... when constructing any Synapse robot and it automatically streams joint state, TCP pose, and link transforms over Zenoh using BabyROS - no manual publisher setup required. get_publisher_hz reads the measured publish rate of that background publisher.

The Skill

robot.get_publisher_hz()

Topics Published

Topic	Contents
synapse/robots/<ClassName>/<name>/state	Joint state (positions, velocities) and TCP pose.
synapse/robots/<ClassName>/<name>/tf	Namespaced link transforms for the full kinematic tree.

<ClassName> is the robot's Python class name (e.g. UniversalRobotsUR10E) and <name> is the string passed to the constructor. get_publisher_hz reports how fast these topics are actually being published.

The Code

"""
Logs the measured state/TF publish rate of a named robot.

Supports Universal Robots (UR), Epson, virtual, and Isaac Sim.

Usage:
    python get_publisher_hz.py [--ip <ROBOT_IP>] [--prim_path <PRIM_PATH>]
"""

import argparse
import time

from loguru import logger

from telekinesis.synapse.robots.manipulators import universal_robots


def main(ip: str | None, prim_path: str | None) -> None:
    """Log the measured state/TF publish rate of a named robot."""

    #===================== Create Robot ==========================================
    robot = universal_robots.UniversalRobotsUR10E(name='UR10e')

    try:
        #===================== Connect Robot ==========================================
        if ip:
            robot.connect(ip=ip)
        elif prim_path:
            robot.connect(simulation_prim_path=prim_path)

        # ==================== Run Skill ============================================
        time.sleep(1.0)  # let the publisher run for a moment before sampling
        logger.success(f"publisher_hz: {robot.get_publisher_hz()}")
    except (ConnectionError, OSError) as e:
        logger.error(f"Error occurred: {e}")
    finally:
        robot.disconnect()
        robot.shutdown()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Read publisher rate Synapse example")
    parser.add_argument("--ip", type=str, default=None,
                         help="UR robot IP address for real hardware, e.g. 192.168.1.100")
    parser.add_argument("--prim_path", type=str, default=None,
                         help='Isaac Sim articulation prim path, e.g. "/World/ur10e"')
    args = parser.parse_args()

    main(ip=args.ip, prim_path=args.prim_path)

Running the Example

Virtual

python get_publisher_hz.py

Isaac Sim

python get_publisher_hz.py --prim_path /World/ur10e

Real

python get_publisher_hz.py --ip 192.168.1.100

For all options:

python get_publisher_hz.py --help

Parameter Configuration

This skill takes no input parameters.

Returns

Type	Description
Optional[float]	The measured state/TF publish rate in Hz (a smoothed estimate of actual throughput, not a configured target). Returns None if the robot was constructed without a name (no publisher exists), and 0.0 if the publisher exists but has not published yet.

How to Tune the Parameters

get_publisher_hz takes no parameters, but its reading depends on timing: the example sleeps briefly after connect() before sampling because the background publisher needs a moment to emit a few messages before the smoothed rate estimate stabilizes. Sampling immediately after connecting can return 0.0 even on a healthy publisher.

Where to Use the Skill

• Health checks - Confirm the automatic publisher is actually running and keeping up with the expected control-loop rate.
• Debugging dropped messages - A publisher rate far below the expected control frequency can point to network or Zenoh session issues.
• Tuning downstream consumers - Know the true publish rate before deciding how a BabyROS subscriber or Rerun visualization should sample it.

When Not to Use the Skill

• You need the topic names/types, not the rate - use Get Publisher Names and Types instead.
• Single-process scripts - If everything runs in one process, call Get Joint Positions, Get Cartesian Pose, or Get Visual Mesh Transforms directly instead of publishing over the network.