Skip to content

RTIClient

A class in the client from robotclient import RTIClient

Real-time streaming control of one robot, beside its RobotClient.

It carries the real-time link: commands from your PC to the robot's UDP port 30001 and state packets from the robot to your PC's UDP port 30002, or both over one TCP stream to the robot's port 30003. The RobotClient it is given carries the handshake: pairing, claiming, releasing. Joint values are in radians here, unlike the rest of the client.

In RTI mode (hard real time, connect() without a rate) your PC sends one target every 4 ms tick over UDP, each with its send time, and closes the loop itself: the robot replays your position targets one tick behind and refuses a stream the arm cannot follow. In SOFT_RTI (connect() with rate_hz) you send joint position targets at your own rate, over UDP or TCP, and the robot fills in the ticks between them.

When commands stop coming or a stream breaks a limit, the arm brakes to rest within its limits, then the robot latches an error and disables: RTI_LINK_LOST, RTI_START_POSE (a first command too far from the arm), RTI_NO_SEND_TIME, or per joint RTI_POSITION_LIMIT, RTI_VELOCITY_LIMIT, RTI_ACCELERATION_LIMIT or RTI_JERK_LIMIT. Its state stream then ends, and control() raises ControlLost.

Create one

Create the real-time link for a connected RobotClient.

It opens two UDP sockets and, for UDP, binds the state socket to listen_port on every interface at once. Nothing is sent until connect(). Call close() when done.

rti = RTIClient(robot, clock='robot', listen_port=30002, transport='udp', timestamps=True)

Arguments

Name Type Default Description
robot required A connected RobotClient; it carries the handshake.
clock 'robot' "robot": control() runs your callback on each state packet the robot sends. "pc": you send when you like, with send_joint_position() and the other send calls; control() is not available. The robot behaves the same either way.
listen_port 30002 The UDP port state packets arrive on. The robot always sends to 30002, so keep the default. Not used with TCP.
transport 'udp' "udp" or "tcp". TCP carries both directions on one stream that this PC dials, in SOFT_RTI only: a lost packet then stalls everything behind it until it is sent again, while a closed stream is noticed at once.
timestamps True Whether every UDP command carries its send time, so that the robot follows your real timing instead of the packets' arrival. RTI mode needs it; False is for SOFT_RTI only. TCP never carries it.

Raises

  • ValueError: clock or transport is not one of its values.
  • OSError: The UDP port listen_port is not free.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.close()

Methods and properties

Name What it does
connect() Pair with the robot and switch it to a real-time mode.
claim() Claim real-time control: the robot starts applying your commands.
release() Release real-time control over the robot connection, staying paired.
disconnect() Unpair from the robot and switch it to IDLE mode.
close() Close the link's sockets.
control() Run your control callback on every state packet the robot sends.
send_joint_position() Send one joint position target to the robot.
send_joint_velocity() Send one joint velocity command to the robot.
send_joint_torque() Send one joint torque command to the robot.
stop() Ask the robot to end the stream: brake to rest and hold.
read_state() Return the newest state packet that has arrived, or None.
read_once() Return the current state packet, waiting for one if none is queued.
capture() Record every state packet that read_state() and read_once() read.
latency_stats() Return the loop delay this client measured, in milliseconds.

RTIClient.connect

Pair with the robot and switch it to a real-time mode.

It checks that the robot speaks this client's real-time packet and telemetry formats, then calls robot.rti_connect() with this PC's address, and over TCP dials the robot's stream port. It returns once the robot has answered: unless the RobotClient runs in sync mode, the robot switches a few milliseconds later, so wait for the mode (robot.wait_mode()) before claim().

rate_hz picks the contract: none for RTI mode, your command rate for SOFT_RTI (see RobotClient.rti_connect()).

rti.connect(pc_ip=None, rate_hz=None, transport=None, *, allow_mismatch=False)

Arguments

Name Type Default Description
pc_ip str | None None The address the robot sends state to and accepts commands from. None uses this PC's address on its connection to the robot.
rate_hz float | None None Your command rate in Hz for SOFT_RTI; None for RTI mode. Prefer one that divides 250 Hz evenly: 125, 50, 25 or 10 Hz.
transport str | None None "udp" or "tcp", replacing the one given at creation; None keeps it. RTI mode is UDP only.
allow_mismatch bool False Pair even when the robot reports another real-time packet or telemetry format version.

Returns RTIClient: This RTIClient, so that calls can be chained.

Raises

  • CommandRejected: The robot refused the pairing: it is not enabled or not homed, or another client holds control.
  • RobotTimeout: The robot did not answer within 5 s, or, in sync mode, did not report the mode within 10 s.
  • RobotConnectionError: The robot connection was lost.
  • VersionMismatch: The robot reports another real-time packet or telemetry format version, and allow_mismatch is false. Nothing is sent.
  • ValueError: Without rate_hz, transport is "tcp" or the client sends no send times; or transport or rate_hz is not valid. Nothing is sent.
  • TypeError: rate_hz is not a number. Nothing is sent.
  • RuntimeError: robot is not connected. Nothing is sent.
  • OSError: Over TCP, the robot's stream port did not accept the connection within 5 s.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect(rate_hz=50)
robot.wait_mode("SOFT_RTI")
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.claim

Claim real-time control: the robot starts applying your commands.

It calls robot.rti_claim(): see there. Send the first command at once, because the watchdog runs from the claim: 40 ms in RTI mode, three of your intervals in SOFT_RTI.

rti.claim()

Arguments

None.

Returns RTIClient: This RTIClient, so that calls can be chained.

Raises

  • CommandRejected: The robot refused it: it is not in a real-time mode (for example right after connect(), before the switch), not enabled or not homed, or another client holds control.
  • RobotTimeout: The robot did not answer within 5 s.
  • RobotConnectionError: The robot connection is not open.

Example

from robotclient import RTIClient, MotionFinished
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
rti.claim()
rti.control(lambda state: MotionFinished())
robot.rti_disconnect(wait=True)
rti.close()

Warning

From the claim on, the arm follows your PC's commands.

RTIClient.release

Release real-time control over the robot connection, staying paired.

It calls robot.rti_release(): the arm brakes to rest within its limits and holds, and after 0.5 s the session is paired and idle again. It works in both real-time modes, also when the stream itself is down. Send it within the command timeout after your last command, or the robot latches RTI_LINK_LOST first.

rti.release()

Arguments

None.

Raises

  • CommandRejected: Another client holds control.
  • RobotTimeout: The robot did not answer within 5 s.
  • RobotConnectionError: The robot connection is not open.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot, clock="pc")
rti.connect(rate_hz=50)
robot.wait_mode("SOFT_RTI")
q = rti.read_once()["j_p"]
rti.claim()
for i in range(50):
    rti.send_joint_position(q)
    time.sleep(0.02)

rti.release()
time.sleep(1)
rti.disconnect()
rti.close()

RTIClient.disconnect

Unpair from the robot and switch it to IDLE mode.

It ends a running control() loop and calls robot.rti_disconnect(), which switches to IDLE mode at once, without braking: release first and let the arm come to rest. It waits for IDLE mode only when the RobotClient runs in sync mode. The sockets stay open: close() them, or connect() again.

rti.disconnect()

Arguments

None.

Raises

  • CommandRejected: Another client holds control.
  • RobotTimeout: The robot did not answer within 5 s, or, in sync mode, did not report IDLE mode within 10 s.
  • RobotConnectionError: The robot connection is not open.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
rti.disconnect()
robot.wait_mode("IDLE")
rti.close()

Warning

The arm does not brake: sent while it moves, it coasts in freedrive until friction stops it (see robot.rti_disconnect()).

RTIClient.close

Close the link's sockets.

It closes the TCP stream and both UDP sockets; the robot is not told. Unpair first with disconnect(). The client cannot be used afterwards.

rti.close()

Arguments

None.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.close()

RTIClient.control

Run your control callback on every state packet the robot sends.

It blocks. For each state packet it calls callback(state) with a RobotState and sends the command the callback returns at once, stamped with state.elapsed when that is robot time. Return None to send nothing that tick: the robot carries on alone (a position stream along the same curve, a velocity or a torque unchanged) until its command timeout. Return MotionFinished to send the stop and return.

It needs an active claim (claim()): without one the robot ignores the commands, and nothing reports it. When the recipe carries rti_state, it returns quietly once the robot ends the claim cleanly, as rti_release() does.

rti.control(callback, first_packet_timeout=2.0, timeout=0.1)

Arguments

Name Type Default Description
callback Callable[[RobotState], _Command | MotionFinished | None] required Called with each RobotState. It returns a JointPositionCommand, JointVelocityCommand or JointTorqueCommand to send, MotionFinished to end, or None.
first_packet_timeout float 2.0 Seconds to wait for the first state packet.
timeout float 0.1 Seconds to wait for each later packet. A longer silence means the robot has left real-time control.

Raises

  • ControlLost: The robot revoked the claim (rti_state left 2 other than by a clean release), the state stream stopped after it had started, or the robot closed the TCP stream.
  • RuntimeError: clock is not "robot", or no state packet came within first_packet_timeout, for example before connect().

Example

import math
from robotclient import RTIClient
from robotclient import JointPositionCommand, MotionFinished
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
q0 = rti.read_once()["j_p"]
rti.claim()
def follow(state):
    t = state.elapsed
    if t > 2.0:
        return MotionFinished()
    q = list(q0)
    q[5] += 0.0125 * (1 - math.cos(math.pi * t / 2.0))
    return JointPositionCommand(q)

rti.control(follow)
robot.rti_disconnect(wait=True)
rti.close()

Note

Over UDP the packets that arrive while your callback runs queue up, and the next one handed to it is the oldest of them. Keep the callback shorter than a tick, or it falls further behind with every packet.

Warning

Your callback moves the arm. Start from the measured pose (state["j_p"]) and keep the first motions small.

RTIClient.send_joint_position

Send one joint position target to the robot.

It sends the target at once and does not wait. The robot applies it only while control is claimed: in RTI mode send one every 4 ms tick, each with its send time; in SOFT_RTI at your own rate. A target beyond a joint's soft limit, or heading for it faster than the joint can still stop, ends the stream with RTI_POSITION_LIMIT after the arm has braked to rest. Over TCP it first reads the state that is waiting, so that the stream does not back up.

rti.send_joint_position(positions, feedforward_speed=None, timestamp=None)

Arguments

Name Type Default Description
positions Sequence[float] required The six joint targets in radians. Fewer values are padded with 0 rad, which commands those joints to zero; more are cut.
feedforward_speed Sequence[float] | None None Each joint's speed at the target, in rad/s; None sends zeros. "ruckig" and "ruckig_advancing" pass through the target at this speed, and the dynamics feedforward reads your stream's speed from it.
timestamp float | None None The time this target is for, in seconds of a monotonic clock, the same clock for every target; used over UDP with timestamps on only. None stamps the moment of sending. Better: read the clock once, compute the target from that reading and pass it, so that a pause between computing and sending does not look like a jump.

Raises

  • ControlLost: Over TCP, the robot closed the stream.
  • RuntimeError: connect() was not called.
  • OSError: The packet could not be sent.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot, clock="pc")
rti.connect(rate_hz=50)
robot.wait_mode("SOFT_RTI")
q = rti.read_once()["j_p"]
rti.claim()
for i in range(50):
    q[5] += 0.0005
    rti.send_joint_position(q)
    time.sleep(0.02)

rti.release()
time.sleep(1)
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.send_joint_velocity

Send one joint velocity command to the robot.

RTI mode only: SOFT_RTI refuses velocity commands, and a session that sends only those ends with RTI_LINK_LOST. The robot applies the velocity every tick until the next command, for at most the command timeout (40 ms), so send one every tick. With the rate limiter on it is limited in speed, acceleration and jerk; the tool speed cap and the soft joint limits always apply (a joint brakes before its limit). The first one after a claim must be small (rti_set_start_pose_check()).

rti.send_joint_velocity(velocities, feedforward_torque=None)

Arguments

Name Type Default Description
velocities Sequence[float] required The six joint velocities in rad/s. Fewer values are padded with 0 rad/s; more are cut.
feedforward_torque Sequence[float] | None None Torque in Nm added to each joint's command, limited in its rate of change like a torque command; None sends zeros.

Raises

  • ControlLost: Over TCP, the robot closed the stream.
  • RuntimeError: connect() was not called.
  • OSError: The packet could not be sent.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot, clock="pc")
rti.connect()
robot.wait_mode("RTI")
rti.claim()
for i in range(250):
    rti.send_joint_velocity([0, 0, 0, 0, 0, 0.02])
    time.sleep(0.004)

rti.stop()
time.sleep(1)
robot.rti_disconnect(wait=True)
rti.close()

Warning

A velocity command keeps the arm moving until the next command or the command timeout. Send zeros, or stop(), to stop.

RTIClient.send_joint_torque

Send one joint torque command to the robot.

RTI mode only, like velocity commands. The robot adds the torque that holds the arm up against gravity (with both gravity switches on), limits the rate of change while the rate limiter is on, and cuts torque that pushes a joint further past its soft limit; the drives limit the current. It applies the torque every tick until the next command, for at most the command timeout (40 ms). The first one after a claim must be small (rti_set_start_pose_check()).

rti.send_joint_torque(torques)

Arguments

Name Type Default Description
torques Sequence[float] required The six joint torques in Nm, without gravity. Fewer values are padded with 0 Nm; more are cut.

Raises

  • ControlLost: Over TCP, the robot closed the stream.
  • RuntimeError: connect() was not called.
  • OSError: The packet could not be sent.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot, clock="pc")
rti.connect()
robot.wait_mode("RTI")
rti.claim()
for i in range(50):
    rti.send_joint_torque([0, 0, 0, 0, 0, 0])
    time.sleep(0.004)

rti.stop()
time.sleep(1)
robot.rti_disconnect(wait=True)
rti.close()

Warning

Under torque control nothing holds a position: with zero torque the arm floats as in freedrive, and an error in the gravity model lets it drift. Keep a hand near the E-stop.

RTIClient.stop

Ask the robot to end the stream: brake to rest and hold.

It sends the stop command inside the stream and does not wait; it also ends a control() loop running in another thread. The robot then does what rti_release() does, in RTI and SOFT_RTI mode alike: the arm brakes to rest within its limits and holds, and after 0.5 s the session is paired and idle again.

rti.stop()

Arguments

None.

Raises

  • ControlLost: Over TCP, the robot closed the stream.
  • RuntimeError: connect() was not called.
  • OSError: The packet could not be sent.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot, clock="pc")
rti.connect()
robot.wait_mode("RTI")
rti.claim()
for i in range(250):
    rti.send_joint_velocity([0, 0, 0, 0, 0, 0.02])
    time.sleep(0.004)

rti.stop()
time.sleep(1)
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.read_state

Return the newest state packet that has arrived, or None.

It never blocks: it reads every packet waiting, keeps the newest and discards the others (capture() records them). It works while merely paired, without a claim.

rti.read_state()

Arguments

None.

Returns RobotState | None: The newest RobotState, or None when no packet arrived since the last read. Its elapsed is 0: only control() sets it.

Raises

  • ControlLost: Over TCP, the robot closed the stream.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
state = rti.read_state()
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.read_once

Return the current state packet, waiting for one if none is queued.

It reads every packet waiting and returns the newest; only when none is waiting does it block, up to timeout. It works while merely paired, without a claim: read the arm's pose from it before streaming. The oldest packet would show the pose of up to the whole wait before.

rti.read_once(timeout=1.0)

Arguments

Name Type Default Description
timeout float 1.0 Seconds to wait when no packet is waiting.

Returns RobotState: The newest RobotState. Its elapsed is 0: only control() sets it.

Raises

  • ControlLost: Over TCP, the robot closed the stream.
  • RuntimeError: No packet arrived within timeout.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect(rate_hz=50)
robot.wait_mode("SOFT_RTI")
rti.read_once()["j_p"]
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.capture

Record every state packet that read_state() and read_once() read.

Normally a read keeps only the newest packet. With capture on, each packet they read is appended to captured as an (arrival time, packet) pair: the time in seconds of this PC's time.perf_counter(), the packet as the raw bytes the robot sent. Packets that control() receives are not recorded. Switching it on empties captured; past limit entries the oldest quarter is dropped.

rti.capture(on=True, limit=200000)

Arguments

Name Type Default Description
on bool True True to record, False to stop; what was recorded stays.
limit int 200000 The most entries kept.

Returns RTIClient: This RTIClient, so that calls can be chained.

Example

import time
from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
rti.capture(True)
time.sleep(0.1)
state = rti.read_state()
len(rti.captured) > 1
robot.rti_disconnect(wait=True)
rti.close()

RTIClient.latency_stats

Return the loop delay this client measured, in milliseconds.

The robot echoes the sequence number of the last command it accepted (rti_seq in the state packet). The time from sending a command until a state packet echoes it is one loop delay: how long until your command's effect is visible to you, including up to one robot tick. It needs rti_seq in the active recipe; the standard recipe has it.

rti.latency_stats()

Arguments

None.

Returns dict | None: None before any delay was measured. Otherwise n, the samples; min, med (the median), p95, p99 and max in ms; and lost, the commands the robot never echoed.

Example

from robotclient import RTIClient
from robotclient import JointPositionCommand, MotionFinished
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
q0 = rti.read_once()["j_p"]
rti.claim()
rti.control(lambda state: MotionFinished() if state.elapsed > 1
            else JointPositionCommand(q0))
sorted(rti.latency_stats())
robot.rti_disconnect(wait=True)
rti.close()

RobotState

One state packet from the robot, as telemetry tag to value.

Each key is a tag of the active telemetry recipe (set_recipe()). A tag with one value maps to a number, a per-joint or pose tag to a list, mode to its text; values are rounded to the recipe's precision. The standard recipe, the default, carries:

  • j_p: the joint positions, in rad.
  • j_s: the joint speeds, unfiltered, in rad/s.
  • j_err: each drive's error byte; 0 when the drive reports no fault.
  • tick: the robot's loop tick counter.
  • TCP_p: the tool pose [x, y, z, rx, ry, rz]: metres, and radians with R = Rx(rx) Ry(ry) Rz(rz).
  • rti_seq: the sequence number of the last command the robot accepted; 0 before the first.
  • rti_state: 0 not paired, 1 paired without a claim, 2 control active, 3 stopping after a release or a stop, 4 stopping after an error.
  • mode: the robot's mode, such as "RTI".
  • period_ms: the measured length of the tick that sent the packet, in ms.

The diagnostics recipe adds rti_ok, the share of the last 100 ticks that brought a command (0 to 1), and rti_miss, the ticks in a row without one. latest_status()["recipes"] lists the recipes on the robot.

Field Type Meaning
elapsed Seconds since control() began: robot time (its tick count times its tick length) when robot_time is true, else this PC's clock. 0 outside control().
robot_time Whether elapsed is robot time: the recipe carries tick and the robot reported its tick length.

Example

from robotclient import RTIClient
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
state = rti.read_once()
state["mode"], state["rti_state"]
robot.rti_disconnect(wait=True)
rti.close()

JointPositionCommand

Joint position targets for one real-time command.

Return it from a control() callback. It works in both real-time modes.

Field Type Meaning
payload The targets, as given.
feedforward The speeds, as given, or six zeros.

Example

from robotclient import JointPositionCommand
cmd = JointPositionCommand([1.571, -1.833, 2.88, 0.0, -0.524, 3.142])

JointVelocityCommand

Joint velocities for one real-time command, RTI mode only.

Return it from a control() callback. SOFT_RTI refuses it. The robot applies it every tick until the next command, for at most the command timeout.

Field Type Meaning
payload The velocities, as given.
feedforward The torques, as given, or six zeros.

Example

from robotclient import JointVelocityCommand
cmd = JointVelocityCommand([0, 0, 0, 0, 0, 0.02])

JointTorqueCommand

Joint torques for one real-time command, RTI mode only.

Return it from a control() callback. SOFT_RTI refuses it. Leave gravity out: the robot adds it. It applies the torque every tick until the next command, for at most the command timeout.

Field Type Meaning
payload The torques, as given.
feedforward Six zeros.

Example

from robotclient import JointTorqueCommand
cmd = JointTorqueCommand([0, 0, 0, 0, 0, 0])

ControlLost

Real-time control that the robot ended while this PC streamed.

control() raises it when a state packet shows the claim revoked (rti_state neither 2 nor a clean release), and when the state stream stops after it had started: a latched error takes the robot out of its real-time mode, which ends the stream. Any read or send raises it when the robot closes the TCP stream. The message names the robot's errors and the link figures this client has; robot.errors() gives the robot's errors in full.

Example

from robotclient import RTIClient, ControlLost, MotionFinished
rti = RTIClient(robot)
rti.connect()
robot.wait_mode("RTI")
rti.claim()
try:
    rti.control(lambda state: MotionFinished())
except ControlLost:
    print(robot.errors())

robot.rti_disconnect(wait=True)
rti.close()

MotionFinished

Signal from a control() callback that the motion is done.

control() then sends the stop command and returns. The arm brakes to rest within its limits and holds, and after 0.5 s the session is paired and idle again, in RTI and SOFT_RTI mode alike.

Field Type Meaning
cmd As given.

Example

from robotclient import MotionFinished
def callback(state):
    return MotionFinished()