API reference

Everything a client program touches. Quaternions on this API are wxyz.

Client

class frankatwin.remote_client.FrankaTwinClient(cfg, *, connect_timeout_s=3.0, verbose=False)

Parameters:

• cfg (RobotConfig)

• connect_timeout_s (float)

• verbose (bool)

set_ee_target(target_pos, target_quat)

Publish a new EE setpoint (wxyz). Returns the daemon’s state_head right after the write – the target is in force from tick head + 1 (see frankatwin.ring_log); 0 from a daemon that predates ring logs.

Parameters:

• target_pos (ndarray)

• target_quat (ndarray)

Return type:

int

get_state(*, fresh=False)

Return the most recent cached state frame.

If fresh=True, force a synchronous round-trip to the daemon instead of using the SUB-stream cache. Use sparingly; the cache is sufficient for policy loops.

Parameters:

fresh (bool)

Return type:

RobotState | None

log_start(path=None, *, poll_hz=None)

Start a 1 kHz ring log (frankatwin.ring_log).

From now until log_stop the daemon copies every 1 kHz state frame out of the shm ring and stamps each set_ee_target with the tick it took effect. path is where log_stop / log_save write the CSV on this machine; its directory is created now, so that a path that cannot be written fails before the run and not after it. Without a path, read the rows with log_fetch. Returns {"path", "seq_start", "poll_hz", "log_id"}.

Parameters:

poll_hz (float | None)

Return type:

Dict[str, Any]

log_stop(*, save=True)

Stop the ring log and return the summary (num_frames, seq_first/seq_last, num_targets, gaps, dropped_frames, resets, …).

With save and a path from log_start the rows are fetched and written right away (log_save; path / targets_path in the summary). save=False returns as soon as the daemon has merged the session – call log_save when there is time for the transfer.

Parameters:

save (bool)

Return type:

Dict[str, Any]

log_fetch(table='merged', *, chunk_rows=10000)

The last finished ring log’s rows as a float64 array, pulled from the daemon chunk_rows at a time.

table is "merged" (one row per tick, ring_log.MERGED_COLUMNS) or "targets" (one row per setpoint, ring_log.TARGETS_COLUMNS). The daemon keeps the log until the next log_start.

Parameters:

• table (str)

• chunk_rows (int)

Return type:

ndarray

log_save(path=None)

Fetch the last finished ring log and write it on this machine: the merged CSV at path (default: log_start’s) and <stem>_targets.csv next to it. Returns {"path", "targets_path"}.

The daemon keeps the log until the next log_start, so a save that failed can be repeated, from a new client too.

Return type:

Dict[str, str]

move_to_pose(target_pos, target_quat, q_max_speed=None)

q_max_speed is approximate here – see LocalController.move_to_pose.

Parameters:

• target_pos (ndarray)

• target_quat (ndarray)

• q_max_speed (float | None)

Return type:

None

wait_for_state(timeout_s=3.0)

Block until the SUB stream delivers at least one frame.

Parameters:

timeout_s (float)

Return type:

RobotState

class frankatwin.local_controller.RobotState(timestamp_s, q, dq, ee_pos, ee_quat, tau, seq, ee_linvel=<factory>, ee_angvel=<factory>, tau_J=<factory>)

Plain-data snapshot of the latest robot state frame.

Parameters:

• timestamp_s (float)

• q (ndarray)

• dq (ndarray)

• ee_pos (ndarray)

• ee_quat (ndarray)

• tau (ndarray)

• seq (int)

• ee_linvel (ndarray)

• ee_angvel (ndarray)

• tau_J (ndarray)

NUC-side controller

class frankatwin.local_controller.LocalController(cfg, *, autostart=True, verbose=False)

Owns the shm segment and the C++ controller subprocess.

Typical usage:

cfg = load_config()
with LocalController(cfg) as robot:
    robot.set_gains(kp_pos=200, kp_ori=20)
    robot.set_ee_target(np.array([0.5, 0.0, 0.4]),
                        np.array([1.0, 0.0, 0.0, 0.0]))
    state = robot.get_state()

Parameters:

• cfg (RobotConfig)

• autostart (bool)

• verbose (bool)

is_controller_alive()

True iff the osc_shm subprocess exists and has not exited.

Return type:

bool

ensure_running()

Restart osc_shm if it has died unexpectedly.

Returns True if a restart was performed, False if the controller was already alive. Logs the dead process’ exit code + captured stdout/stderr before relaunching so the crash reason (libfranka reflex, RT overrun, etc.) is not lost. Intended to be driven by the daemon watchdog; callers must serialize this against move_to*/lifecycle ops (osc_shm needs the sole FCI session).

Return type:

bool

set_ee_target(target_pos, target_quat)

Publish a new EE setpoint. wxyz quaternion.

Returns the state_head read right after the write: the newest 1 kHz frame at that moment. osc_shm snapshots the command before it publishes a tick’s frame, so this target is in force from frame head + 1 at the latest (frankatwin.ring_log). An active ring log records it.

Parameters:

• target_pos (ndarray)

• target_quat (ndarray)

Return type:

int

set_gains(kp_pos=None, kp_ori=None, kd_pos=None, kd_ori=None, error_delta_pos=None)

Update controller gains. Any None argument keeps the current value.

Parameters:

• kp_pos (float | None)

• kp_ori (float | None)

• kd_pos (float | None)

• kd_ori (float | None)

• error_delta_pos (float | None)

Return type:

None

get_state(k=None)

Return the latest RobotState (or None if no frame yet).

If k is provided, returns a list of up to k most recent frames in chronological order.

Parameters:

k (int | None)

Return type:

RobotState | None

log_start(path=None, *, poll_hz=None)

Start a 1 kHz ring log (frankatwin.ring_log.RunRecorder).

A daemon thread copies every state frame out of the shm ring from now on, and every set_ee_target until log_stop is stamped with the tick it took effect. path is written on log_stop, on this machine. Without one the session stays in memory, to be read with log_fetch or written with log_save – what the daemon does, so that the files end up on the PC that drove the run and not here. One session at a time; starting one drops the last finished log.

Parameters:

poll_hz (float | None)

Return type:

Dict[str, Any]

log_stop(*, save=True)

Stop the ring log and return the summary.

The merged rows and the setpoint rows stay in memory until the next log_start (log_fetch, log_save). With save and a path from log_start the two CSVs are written here as well.

Parameters:

save (bool)

Return type:

Dict[str, Any]

log_fetch(table='merged', offset=0, count=None)

Rows [offset, offset + count) of the last finished ring log (a view).

table is "merged" (one row per tick, ring_log.MERGED_COLUMNS) or "targets" (one row per setpoint, ring_log.TARGETS_COLUMNS); count=None runs to the end.

Parameters:

• table (str)

• offset (int)

• count (int | None)

Return type:

ndarray

log_save(path=None)

Write the last finished ring log on this machine: the merged CSV at path (default: log_start’s) and <stem>_targets.csv next to it. Returns {"path", "targets_path"}.

Return type:

Dict[str, str]

move_to_q(q_target, q_max_speed=None)

Blocking joint-space reset using libfranka MotionGenerator.

Mutually exclusive with osc_shm: this method stops the controller, runs move_to, then restarts the controller.

Parameters:

• q_target (ndarray)

• q_max_speed (float | None)

Return type:

None

move_to_pose(target_pos, target_quat, q_max_speed=None)

Blocking task-space reset using libfranka CartesianPose motion type.

target_quat is wxyz. No external IK; libfranka solves internally.

q_max_speed is APPROXIMATE here: because the IK lives inside libfranka, move_to estimates the joint speed from the Jacobian at the start pose alone. It degrades over large reorientations and near singularities. See src/move_to.cpp.

Parameters:

• target_pos (ndarray)

• target_quat (ndarray)

• q_max_speed (float | None)

Return type:

None

Configuration

Load and validate frankatwin runtime config.

Config resolution order (first hit wins):

1. Explicit path= argument to load_config(path=…) / –config.

2. $FRANKATWIN_CONFIG.

3. <repo>/config/robot.yaml – the checkout this package was installed from (pip install -e .).

Relative paths.build_dir is resolved against the repo root.

frankatwin.config.resolve_config_path(path=None)

Pick the config file per the module docstring (existence is checked by load_config).

Parameters:

path (PathLike | None)

Return type:

Path

class frankatwin.config.NetworkConfig(nuc_host: 'str', cmd_port: 'int', state_port: 'int', state_cache: 'int' = 256)

Parameters:

• nuc_host (str)

• cmd_port (int)

• state_port (int)

• state_cache (int)

class frankatwin.config.RobotSpec(ip: 'str', init_q: 'List[float]')

Parameters:

• ip (str)

• init_q (List[float])

class frankatwin.config.ControlConfig(frequency_hz: 'int', kp_pos: 'float', kp_ori: 'float', kd_pos: 'Optional[float]', kd_ori: 'Optional[float]', error_delta_pos: 'float')

Parameters:

• frequency_hz (int)

• kp_pos (float)

• kp_ori (float)

• kd_pos (float | None)

• kd_ori (float | None)

• error_delta_pos (float)

class frankatwin.config.PathsConfig(build_dir: 'pathlib.Path', shm_name: 'str')

Parameters:

• build_dir (Path)

• shm_name (str)

class frankatwin.config.ResetConfig(q_max_speed: 'float' = 0.5)

Parameters:

q_max_speed (float)

class frankatwin.config.GripperConfig(enabled=True, move_speed=0.1, grasp_speed=0.1, grasp_force=70.0, grasp_width=-0.01, epsilon_inner=0.08, epsilon_outer=0.08, max_width=0.08)

Franka Hand defaults for the gripper_* commands (libfranka Gripper).

Closing is a libfranka grasp: the fingers drive towards grasp_width and squeeze with grasp_force once they stall on the object, so the OBJECT sets the resting width and grasp_force sets how hard it is held. The default grasp_width = -0.01 (past full closure) means “close as far as you can and hold”. epsilon_inner/outer is the band around grasp_width inside which libfranka reports is_grasped; 0.08 (the whole stroke) counts every stall as a grasp, matching the deoxys stack this replaces.

Parameters:

• enabled (bool)

• move_speed (float)

• grasp_speed (float)

• grasp_force (float)

• grasp_width (float)

• epsilon_inner (float)

• epsilon_outer (float)

• max_width (float)

class frankatwin.config.LoadConfig(mass=0.0, com=<factory>, inertia=<factory>)

End-effector payload (e.g. a mounted camera) for gravity compensation.

mass <= 0 means “no extra load” -> osc_shm leaves the Desk-configured load untouched (calls no setLoad). com is the flange->load COM vector [m]; inertia is the 3x3 about the COM, row-major [kg m^2].

Parameters:

• mass (float)

• com (List[float])

• inertia (List[float])

class frankatwin.config.CollisionConfig(torque_threshold=100.0, cartesian_threshold=100.0)

libfranka collision-reflex thresholds passed to setCollisionBehavior.

Each scalar fills ALL entries of the torque (7) / Cartesian (6) lower & upper, nominal & acceleration arrays. The task-impedance controller’s own push is bounded by kp*error_delta (~25 N / ~9 Nm at kp_pos=500/kp_ori=30), so thresholds below that range trip “cartesian_reflex” on insertion contact. Defaults match the deoxys stack (100), which never reflexed during insertion.

Parameters:

• torque_threshold (float)

• cartesian_threshold (float)

class frankatwin.config.RobotConfig(network, robot, control, paths, reset=<factory>, gripper=<factory>, load=<factory>, collision=<factory>, source_path=None)

Top-level config bundle. Pass-through for clients/daemon/local.

Parameters:

• network (NetworkConfig)

• robot (RobotSpec)

• control (ControlConfig)

• paths (PathsConfig)

• reset (ResetConfig)

• gripper (GripperConfig)

• load (LoadConfig)

• collision (CollisionConfig)

• source_path (Path | None)

frankatwin.config.load_config(path=None)

Load and validate robot.yaml. See module docstring for resolution rules.

Parameters:

path (PathLike | None)

Return type:

RobotConfig

Quaternions

Minimal quaternion helpers in the API’s wxyz convention (numpy only).

Everything on the wire – RobotState.ee_quat, set_ee_target, move_to_pose – is wxyz. Sidecars / CSVs / IsaacLab are xyzw; convert at the boundary.

frankatwin.quat.mul_wxyz(a, b)

Hamilton product a ⊗ b (apply b first, then a).

Parameters:

• a (ndarray)

• b (ndarray)

Return type:

ndarray

frankatwin.quat.from_rotvec_wxyz(rotvec)

Axis-angle vector (rad) -> unit quaternion; exact for small angles.

Parameters:

rotvec (ndarray)

Return type:

ndarray

frankatwin.quat.to_rotvec_wxyz(q)

Unit quaternion -> axis-angle vector (rad), shortest path.

Parameters:

q (ndarray)

Return type:

ndarray

frankatwin.quat.error_rotvec_wxyz(q_des, q_cur)

Orientation error as osc_shm computes it: 2·vec(q_des ⊗ q_cur⁻¹), shortest path.

Parameters:

• q_des (ndarray)

• q_cur (ndarray)

Return type:

ndarray

Excitation references

frankatwin.excitation.build_multiband_trajectory(t_s, x_anchor, q_anchor_xyzw, amp_x=0.06, amp_y=0.06, amp_z=0.06, amp_yaw=0.25, amp_roll=0.0, amp_ramp_s=2.0, *, amp_rx=0.2, amp_ry=0.15, ramp_down_s=2.0, profile='heldout')

Build the multi-band excitation trajectory on an arbitrary time grid.

Returns:

• x_des ((N, 3) Cartesian target positions, in the same frame as x_anchor.)

• dx_des ((N, 3) target Cartesian velocities (analytic derivative).)

• quat_des ((N, 4) target orientation quaternions, **xyzw* order, unit-norm.*)

• yaw ((N,) yaw angle (about world-z) applied to the anchor.)

• roll ((N,) roll angle (about EE local-z) applied to the anchor.)

• tilt ((N, 2) world-frame axis-angle components (rx, ry) applied to the anchor.)

• The defaults are MULTIBAND_PROFILES["heldout"] (0.06 m on x / y / z,

• yaw 0.25 rad, tilt 0.20 / 0.15 rad, roll off, 2 s fade-in and fade-out).

• profile picks the tone table (multiband_tones) and nothing else.

• Setting the four rotation amplitudes to 0 gives the position-only variant

• (orientation held at q_anchor_xyzw).

Parameters:

• t_s (ndarray)

• x_anchor (ndarray)

• q_anchor_xyzw (ndarray)

• amp_x (float)

• amp_y (float)

• amp_z (float)

• amp_yaw (float)

• amp_roll (float)

• amp_ramp_s (float)

• amp_rx (float)

• amp_ry (float)

• ramp_down_s (float)

• profile (str)

Return type:

tuple[ndarray, ndarray, ndarray, ndarray, ndarray, ndarray]

frankatwin.excitation.build_chirp_trajectory(t_s, x_anchor, q_anchor_xyzw, *, f0=0.1, f1=0.7, amp_x=0.1, amp_y=0.1, amp_z=0.15, amp_rx=0.5, amp_ry=0.25, amp_rz=0.5, ramp_up_s=2.0, ramp_down_s=3.0, amp_taper_exp=0.0)

Build a 6-DOF chirp reference trajectory around the given anchor pose.

amp_taper_exp scales every amplitude by (f0 / f_inst)^p along the sweep (0 = constant amplitude, the low band; 0.5, the high band; 1 = constant reference velocity; 2 = constant reference acceleration – see CHIRP_BANDS).

Returns:

(x_des, dx_des, quat_des_xyzw, rot_offsets) – x_des (T, 3) Cartesian target positions; dx_des (T, 3) analytic target velocities; quat_des_xyzw (T, 4) target orientation (xyzw), the anchor pre-rotated by the world-frame axis-angle offset; rot_offsets (T, 3) that axis-angle offset (for diagnostics and the sidecar).

Parameters:

• t_s (ndarray)

• x_anchor (ndarray)

• q_anchor_xyzw (ndarray)

• f0 (float)

• f1 (float)

• amp_x (float)

• amp_y (float)

• amp_z (float)

• amp_rx (float)

• amp_ry (float)

• amp_rz (float)

• ramp_up_s (float)

• ramp_down_s (float)

• amp_taper_exp (float)

Return type:

tuple[ndarray, ndarray, ndarray, ndarray]

Shared memory

POSIX shared-memory layout for frankatwin.

This module is a faithful numpy.dtype mirror of src/shm_layout.h. The C++ side static-asserts the field offsets; this side enforces them via a unit test (tests/test_shm_layout.py) using ctypes to cross-check. Keep both in sync.

Concurrency model: - Command (Python writer, C++ reader): seqlock with 64-bit seq. Even = stable. - State (C++ writer, Python reader): SPSC ring buffer indexed by state_head.

All atomic operations are performed with explicit memory barriers (numpy.atomic-like) via _atomic_* helpers below; numpy itself does not provide atomics, so we rely on the GIL + write ordering of single-uint64 stores being atomic on x86_64 (they are when 8-byte aligned, which our layout guarantees).

class frankatwin.shm_layout.ShmView(raw, header, command, states)

Read/write views over an open POSIX shared-memory segment.

Lifetime: caller owns the underlying shared_memory.SharedMemory object (see SharedMemoryAccess). All numpy views are zero-copy aliases of the same buffer.

Parameters:

• raw (memoryview)

• header (ndarray)

• command (ndarray)

• states (ndarray)

write_command(target_pos, target_quat, kp_pos, kp_ori, kd_pos, kd_ori, error_delta_pos, enabled)

Atomically publish a new command. Uses the seqlock protocol.

Caller must ensure target_pos has shape (3,) and target_quat has shape (4,) in wxyz convention with unit norm.

Parameters:

• target_pos (ndarray)

• target_quat (ndarray)

• kp_pos (float)

• kp_ori (float)

• kd_pos (float)

• kd_ori (float)

• error_delta_pos (float)

• enabled (bool)

Return type:

None

read_command()

Read the current command (seqlock-safe). Returns a copy.

Return type:

ndarray

latest_state()

Return a copy of the most recently published state frame.

Returns None if the producer has not written any frame yet.

Return type:

ndarray | None

last_k_states(k)

Return up to k most recent state frames, oldest first. Best-effort: if the producer wrapped during the copy, only the consistent suffix is returned. Returns an empty array if no frames exist.

Parameters:

k (int)

Return type:

ndarray

class frankatwin.shm_layout.SharedMemoryAccess(name='/frankatwin_osc', create=False)

RAII wrapper over multiprocessing.shared_memory.SharedMemory.

On Linux this maps to /dev/shm/<name> (the leading ‘/’ in the C-style POSIX shm name is stripped by SharedMemory; pass either /frankatwin_osc or frankatwin_osc and we normalize).

Parameters:

• name (str)

• create (bool)

frankatwin.shm_layout.layout_offsets()

Return (header_size, command_size, state_frame_size, total_size).

Return type:

Tuple[int, int, int, int]