API reference¶

The public Python API is generated from the package docstrings.

Robot¶

zero_franky.Robot is an alias for zero_franky.zmq_client.RobotProxy.

class zero_franky.zmq_client.RobotProxy(fci_hostname, *, client=None, **kwargs)¶

Client-side proxy for a robot controlled by a zero-franky server.

The fci_hostname is resolved by the server.

Parameters:

fci_hostname (str)
client (ZmqRpcClient | None)

get_last_teleop_state()¶

The last state the server’s motion callback recorded.

Costs an RPC round trip. Use latest_state to get the one cached from the state stream.

join_motion(timeout=None)¶

Wait for the active motion to finish, returning whether it did.

Costs an RPC round trip, and the server holds it open for up to timeout seconds.

Parameters:: timeout (float | None)
Return type:: bool

move(motion, asynchronous=False)¶

Run a motion, blocking until it finishes unless asynchronous.

Costs an RPC round trip to enqueue. A synchronous move always enqueues asynchronously on the server and then polls join_motion, so that a long motion cannot outlive the RPC timeout; expect a round trip per poll.

Parameters:: asynchronous (bool)

poll_motion()¶

Whether the active motion has finished, without waiting for it.

Costs an RPC round trip.

Return type:: bool

recover_from_errors()¶

Clear the robot’s error state so it will accept motions again.

Costs an RPC round trip.

start_cartesian_impedance_tracker(policy=None, *, policy_transport='import', period=0.001, stop_on_policy_error=True, **motion_kwargs)¶

Start Cartesian impedance tracking, the networked counterpart to franky.CartesianImpedanceTracker.

Starting costs an RPC round trip.

Parameters:

policy_transport (Literal['import', 'cloudpickle'])
period (float | None)
stop_on_policy_error (bool)

Return type:

CartesianImpedanceTrackerProxy

start_joint_impedance_tracker(policy=None, *, policy_transport='import', period=0.001, stop_on_policy_error=True, **motion_kwargs)¶

Start joint impedance tracking, the networked counterpart to franky.JointImpedanceTracker.

Without a policy, the returned proxy is a passthrough: set references from client code with set_target. With one, the server runs the policy loop at period next to the robot.

Starting costs an RPC round trip.

Parameters:

policy_transport (Literal['import', 'cloudpickle'])
period (float | None)
stop_on_policy_error (bool)

Return type:

JointImpedanceTrackerProxy

start_state_stream(topic='robot.state', timeout_ms=250)¶

Subscribe to the server’s state PUB socket on a background thread.

Costs no round trip: the server publishes each control cycle whether or not anyone is listening. Idempotent while the thread is alive, so it is cheap to call from a loop. Errors are stored rather than raised here, and surface from wait_for_state and tick.

Parameters:

topic (str)
timeout_ms (int)

Return type:

None

start_torque_tracker(period=None, **motion_kwargs)¶

Start direct joint-torque control.

Call set_torque more frequently than signal_timeout (50 ms by default). The server-side franky watchdog terminates stale streams.

Starting costs an RPC round trip.

Parameters:: period (float | None)
Return type:: TorqueTrackerProxy

state_subscriber(topic='robot.state', timeout_ms=1000)¶

A raw subscriber onto the state PUB socket, filtered to this robot.

For reading the stream yourself instead of through start_state_stream. Connecting costs no round trip.

Parameters:

topic (str)
timeout_ms (int)

stop()¶

Stop the active motion, as franky’s Robot.stop does.

Costs an RPC round trip. This is the hard stop; a tracker’s own stop ramps the arm down instead.

stop_state_stream(join_timeout=1.0)¶

Stop the background subscriber and join its thread.

Leaves latest_state holding the last snapshot received. Anything that reads the stream restarts it on next use.

Parameters:: join_timeout (float | None)
Return type:: None

wait_for_state(timeout=None)¶

Block until the state stream delivers a snapshot, and return it.

Waits on the stream rather than asking the server, so it costs no round trip, but it needs start_state_stream to have been called. Raises whatever the stream raised, or TimeoutError if timeout elapses first.

Parameters:: timeout (float | None)
Return type:: dict[str, Any]

property latest_state: dict[str, Any] | None¶

The most recent state snapshot, or None if none has landed yet.

A plain read of what the background stream last received, so it costs no round trip and never blocks. It does not start the stream, and it outlives the tracker that produced it — wait_for_state blocks for a first snapshot instead.

Connection setup¶

zero_franky.setup_zero_franky(ip, port)¶

Configure the default server connection used by zero_franky.Robot.

port is the RPC port; state and tracker transports use port + 1 and port + 2.

Return type:: None

Tracker proxies¶

Joint impedance¶

class zero_franky.zmq_client.JointImpedanceTrackerProxy(client, session_id, *, push_socket=None, robot=None, period=None)¶

Parameters:

client (ZmqRpcClient)
session_id (str)
period (float | None)

set_cartesian_gains(gains=None, *, translational_stiffness=None, rotational_stiffness=None, damping=None)¶

Update the hybrid Cartesian gain shaping added on top of the joint-space stiffness.

The controller only uses these gains when the tracker was started with cartesian_stiffness. This accepts the same arguments as CartesianImpedanceTrackerProxy.set_gains() minus the nullspace.

Parameters:

gains (CartesianImpedanceGains | None) – Full Cartesian gains; exclusive with the scalar stiffness and damping arguments.
translational_stiffness (float | None) – Scalar translational stiffness for the three translational axes.
rotational_stiffness (float | None) – Scalar rotational stiffness for the three rotational axes.
damping (sequence[float] | None) – Six damping values in [x, y, z, rx, ry, rz] order, or franky.CRITICAL to explicitly unpin damping.

Unlike set_target, this costs an RPC round trip. Naming no gain at all is a no-op.

set_gains(gains=None, *, stiffness=None, damping=None)¶

Update joint impedance gains, smoothed in the RT loop.

stiffness and damping are orthogonal 7-vectors. Passing damping pins it; omitting it means critical damping, while franky.CRITICAL explicitly unpins it. A stiffness change therefore re-tracks critical damping unless damping is passed too.

Accepts a franky.JointImpedanceGains positionally or stiffness/damping keywords. Omitted stiffness is unchanged; naming neither gain is a no-op.

Parameters:

gains (JointImpedanceGains | None) – Full joint gains; exclusive with stiffness and damping.
stiffness (sequence[float] | None) – Seven joint stiffness values.
damping (sequence[float] | None) – Seven joint damping values, or franky.CRITICAL to explicitly unpin damping.

Unlike set_target, this costs an RPC round trip, so it is not free to call every cycle.

set_target(q, dq=None, tau_ff=None)¶

Update the joint target position, optional velocity, and optional feedforward torque.

Goes over the conflating tracker socket, so it costs no round trip and a slower update may be dropped in favour of a newer one; last_reference reports what the controller actually picked up.

status()¶

The server’s view of this session: running, iterations, error.

Costs an RPC round trip.

Return type:: dict[str, Any]

stop(stop_motion=None, join_timeout=None, rpc_timeout_ms=30000)¶

Gracefully stop the tracker and wait for the arm to come to rest.

Matches the corresponding Franky tracker’s franky.JointImpedanceTracker.stop() / franky.CartesianImpedanceTracker.stop() behavior: the server enqueues a franky.TorqueStopMotion ramp and joins it, so this returns once the arm is at rest.

Since the join blocks on the robot, this call gets its own longer RPC timeout. Calling it twice is harmless.

Parameters:

stop_motion (Any)
join_timeout (float | None)
rpc_timeout_ms (int)

tick()¶

Sleep to maintain the requested period and return the measured timestep.

Returns the wall-clock seconds elapsed since the previous tick, or None once the tracker is no longer active, so a loop can pace itself and get a timestep to integrate with:

while dt := tracker.tick():
tracker.set_target(integrate(dt))

The returned timestep is always non-zero positive.

On the first call, returns immediately (no sleep) and reports period. On subsequent calls, sleeps the remaining time until the next tick boundary so that loop body time is compensated for; a body that overran its period shows up as a timestep larger than period. If no period was set, measures and returns without sleeping.

Pacing re-anchors on _t_next += period ever tick. An overrun resets the schedule to the moment the tick returns, so a slow body re-phases the loop instead of making the next few ticks fire back to back to catch up like the franky tracker.

Where franky reads robot.is_in_control in-process, here we get notified from the state stream, which it starts on first use. Until a snapshot confirms the controller is in control, it falls back to a once-a-second status() RPC. That also means tick() raises whatever the state stream raises.

Return type:: float | None

property current_joint_positions: list[float] | None¶

Measured joint positions from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_joint_state: JointState | None¶

The current joint state as a JointState (shorthand for robot.current_joint_state).

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_joint_velocities: list[float] | None¶

Measured joint velocities from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_pose: RobotPose | None¶

The end-effector pose, as franky’s current_pose is.

Read from the state stream (no network round trip). None until the first snapshot lands.

property dt: float¶: The timestep most recently returned by tick() (0.0 before the first tick).

property elapsed_time: float¶: Seconds since this proxy was created.

property is_running: bool¶: Whether the tracker is still active. Costs an RPC round trip.

property iterations: int¶: Server-side policy loop iteration count. Costs an RPC round trip.

property last_reference: dict[str, Any] | None¶

The reference the controller last picked up, from the state stream.

References are sent over a CONFLATE socket, which drops stale undelivered messages. Clients only know what the server actually recieved when the serer tells them.

None before the first snapshot, or before any reference is set.

property period: float | None¶: The configured loop period in seconds, or None if the tracker is unpaced.

property state: dict[str, Any] | None¶

Latest robot state, read from the state PUB stream rather than by RPC.

Unlike franky’s tracker.state this is the msgpack snapshot dict, not a franky.RobotState. The background stream starts on first access, so this is None until the first snapshot lands; use robot.wait_for_state() to block for it.

The snapshot contains q/dq (7-vectors), O_T_EE (4x4), O_F_ext_hat_K, K_F_ext_hat_K, tau_ext_hat_filtered, and O_dP_EE_est (a 6-vector or empty), plus time_step, rel_time, abs_time, control_signal, robot_id, in_control, and reference. The control_signal and reference payloads are motion-specific dictionaries.

property tick_count: int¶: Number of ticks that have run, i.e. returned a timestep rather than None.

Cartesian impedance¶

class zero_franky.zmq_client.CartesianImpedanceTrackerProxy(client, session_id, *, push_socket=None, robot=None, period=None)¶

Parameters:

client (ZmqRpcClient)
session_id (str)
period (float | None)

set_gains(gains=None, *, translational_stiffness=None, rotational_stiffness=None, damping=None, posture_stiffness=None, nullspace_gains=None)¶

Update Cartesian impedance gains, smoothed in the RT loop.

translational_stiffness and rotational_stiffness each set one scalar stiffness block. damping is a 6-vector in [x, y, z, rx, ry, rz] order; omitting it means critical damping, while franky.CRITICAL explicitly unpins it.

gains total-replaces the full franky.CartesianImpedanceGains and is exclusive with the stiffness/damping keywords. posture_stiffness (a scalar or 7-vector) nudges the configured posture task; nullspace_gains replaces the configured posture and manipulability gains. They are mutually exclusive and only affect tasks configured at start.

Parameters:

gains (CartesianImpedanceGains | None) – Full Cartesian gains; exclusive with the scalar stiffness and damping arguments.
translational_stiffness (float | None) – Scalar translational stiffness for the three translational axes.
rotational_stiffness (float | None) – Scalar rotational stiffness for the three rotational axes.
damping (sequence[float] | None) – Six damping values in [x, y, z, rx, ry, rz] order, or franky.CRITICAL to explicitly unpin damping.
posture_stiffness (float | sequence[float] | None) – Scalar or seven-vector that nudges the configured posture task.
nullspace_gains (NullspaceGains | None) – Full replacement for configured posture and manipulability gains.

A stiffness change re-tracks critical damping unless damping is passed. Unlike set_target, this costs an RPC round trip. Naming no gain at all is a no-op.

set_nullspace_gains(gains=None, *, posture_stiffness=0.0, posture_damping=None, posture_max_torque=None, manipulability_gain=0.0, manipulability_damping=0.0, manipulability_max_torque=None)¶

Update the nullspace gains acting on the redundant joint.

Only retunes tasks configured at start. Unlike set_target, this costs an RPC round trip.

Parameters:

posture_stiffness (Any)
posture_damping (Any | None)
posture_max_torque (float | None)
manipulability_gain (float)
manipulability_damping (float)
manipulability_max_torque (float | None)

set_target(pose, twist=None, acceleration=None)¶

Update the Cartesian target pose and optional twist/acceleration feedforward.

Goes over the conflating tracker socket, so it costs no round trip and a slower update may be dropped in favour of a newer one; last_reference reports what the controller actually picked up.

status()¶

The server’s view of this session: running, iterations, error.

Costs an RPC round trip.

Return type:: dict[str, Any]

stop(stop_motion=None, join_timeout=None, rpc_timeout_ms=30000)¶

Gracefully stop the tracker and wait for the arm to come to rest.

Since the join blocks on the robot, this call gets its own longer RPC timeout. Calling it twice is harmless.

Parameters:

stop_motion (Any)
join_timeout (float | None)
rpc_timeout_ms (int)

tick()¶

Sleep to maintain the requested period and return the measured timestep.

Returns the wall-clock seconds elapsed since the previous tick, or None once the tracker is no longer active, so a loop can pace itself and get a timestep to integrate with:

while dt := tracker.tick():
tracker.set_target(integrate(dt))

The returned timestep is always non-zero positive.

Return type:: float | None

property current_joint_positions: list[float] | None¶

Measured joint positions from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_joint_velocities: list[float] | None¶

Measured joint velocities from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_pose: RobotPose | None¶

The end-effector pose, as franky’s current_pose is.

Read from the state stream (no network round trip). None until the first snapshot lands.

property dt: float¶: The timestep most recently returned by tick() (0.0 before the first tick).

property elapsed_time: float¶: Seconds since this proxy was created.

property is_running: bool¶: Whether the tracker is still active. Costs an RPC round trip.

property iterations: int¶: Server-side policy loop iteration count. Costs an RPC round trip.

property last_reference: dict[str, Any] | None¶

The reference the controller last picked up, from the state stream.

References are sent over a CONFLATE socket, which drops stale undelivered messages. Clients only know what the server actually recieved when the serer tells them.

None before the first snapshot, or before any reference is set.

property period: float | None¶: The configured loop period in seconds, or None if the tracker is unpaced.

property state: dict[str, Any] | None¶

Latest robot state, read from the state PUB stream rather than by RPC.

property tick_count: int¶: Number of ticks that have run, i.e. returned a timestep rather than None.

Torque¶

class zero_franky.zmq_client.TorqueTrackerProxy(client, session_id, *, push_socket=None, robot=None, period=None)¶

Parameters:

client (ZmqRpcClient)
session_id (str)
period (float | None)

get_torque()¶

The torque the server most recently handed to the controller.

Costs an RPC round trip; the same value rides the state stream as last_reference.

set_torque(torque)¶

Command raw joint torques. Call faster than the motion’s signal_timeout.

Goes over the conflating tracker socket, so it costs no round trip and a slower update may be dropped in favour of a newer one; last_reference reports what the controller actually picked up.

status()¶

The server’s view of this session: running, iterations, error.

Costs an RPC round trip.

Return type:: dict[str, Any]

stop(stop_motion=None, join_timeout=None, rpc_timeout_ms=30000)¶

Gracefully stop the tracker and wait for the arm to come to rest.

Since the join blocks on the robot, this call gets its own longer RPC timeout. Calling it twice is harmless.

Parameters:

stop_motion (Any)
join_timeout (float | None)
rpc_timeout_ms (int)

tick()¶

Sleep to maintain the requested period and return the measured timestep.

Returns the wall-clock seconds elapsed since the previous tick, or None once the tracker is no longer active, so a loop can pace itself and get a timestep to integrate with:

while dt := tracker.tick():
tracker.set_target(integrate(dt))

The returned timestep is always non-zero positive.

Return type:: float | None

property current_joint_positions: list[float] | None¶

Measured joint positions from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_joint_velocities: list[float] | None¶

Measured joint velocities from the latest state snapshot.

Read from the state stream (no network round trip). None until the first snapshot lands.

property current_pose: RobotPose | None¶

The end-effector pose, as franky’s current_pose is.

Read from the state stream (no network round trip). None until the first snapshot lands.

property dt: float¶: The timestep most recently returned by tick() (0.0 before the first tick).

property elapsed_time: float¶: Seconds since this proxy was created.

property is_running: bool¶: Whether the tracker is still active. Costs an RPC round trip.

property iterations: int¶: Server-side policy loop iteration count. Costs an RPC round trip.

property last_reference: dict[str, Any] | None¶

The reference the controller last picked up, from the state stream.

References are sent over a CONFLATE socket, which drops stale undelivered messages. Clients only know what the server actually recieved when the serer tells them.

None before the first snapshot, or before any reference is set.

property period: float | None¶: The configured loop period in seconds, or None if the tracker is unpaced.

property state: dict[str, Any] | None¶

Latest robot state, read from the state PUB stream rather than by RPC.

property tick_count: int¶: Number of ticks that have run, i.e. returned a timestep rather than None.

Value types mirror the franky API where applicable. Use the local stand-ins from franky Python API for type semantics.