Reforge Interface SDK
Implement your robot SDK, run robot calibration, and call the identification API.
robot.robot_base
DataRecorder class
Store time-series data collected during calibration and system ID.
def __init__(self) -> None:
Initialize empty data buffers for recording.
Args:
None.
Side Effects:
Allocates empty lists for all data fields.
Raises:
None.
Preconditions:
None.
def reset(self) -> None:
Clear all recorded data buffers.
Args:
None.
Returns:
`None`.
Side Effects:
Reinitializes internal data lists.
Raises:
None.
Preconditions:
None.
Robot class
Abstract base class defining the robot control interface.
def __init__(self, name: str) -> None:
Initialize the robot base class.
Args:
name: Human-readable robot name.
Side Effects:
Creates a `DataRecorder` and sets default joint/pose sizes.
Raises:
None.
Preconditions:
None.
Abstract properties
Properties that must be specified by instances of the Robot class.
@property in_sim_mode -> bool
Return whether the robot is in simulation mode.
Returns:
`bool` indicating simulation mode.
Side Effects:
None.
Raises:
None.
Preconditions:
None.
@property urdf_path -> str
Return the URDF path for the robot.
Returns:
`str` path to the URDF file.
Side Effects:
None.
Raises:
None.
Preconditions:
None.
Abstract methods
Methods that must be specified by instances of the Robot class.
def move_to_joint(self, target_joint: Tuple[float, ...]) -> None
Move the robot to the specified joint positions.
Args:
target_joint: Joint positions in radians.
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def move_to_pose(self, target_quat: List[float], target_xyz: List[float]) -> None
Move the robot to the specified Cartesian pose.
Args:
target_quat: Quaternion [qx, qy, qz, qw].
target_xyz: Position [x, y, z] in meters.
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def publish_and_record_joint_positions(self, time_data: List, position_stream: List, velocity_stream: List = [], acceleration_stream: List = [], Ts: float = 1 / DEFAULT_ROBOT_FREQ) -> Deque[Dict]
Publish joint trajectories and record data.
Args:
time_data: Time stamps for each command.
position_stream: Joint position commands (N x num_joints).
velocity_stream: Joint velocity commands (N x num_joints).
acceleration_stream: Joint acceleration commands (N x num_joints).
Ts: Sampling time [s].
Returns:
`collections.deque[dict]` log of recorded data entries.
Side Effects:
Commands robot motion and records data.
Raises:
None.
Preconditions:
Robot connection is active.
def calibrate_robot(self, Ts: float, axes_to_command: int, max_disp: float = DEFAULT_MAX_DISP, max_vel: float = DEFAULT_MAX_VEL, max_acc: float = DEFAULT_MAX_ACC, bcb_runtime: float = DEFAULT_BCB_RUNTIME, ctrl_config: str = DEFAULT_CONFIG, sysid_type: str = DEFAULT_SYSID_TYPE, nV: int = DEFAULT_SYSID_ANGLES, nR: int = DEFAULT_SYSID_RADII, min_sine_freq: float = DEFAULT_SINE_MIN_FREQ, max_sine_freq: float = DEFAULT_SINE_MAX_FREQ, sine_freq_spacing: float = DEFAULT_FREQ_SPACING, num_sine_cycles: int = DEFAULT_SINE_CYCLES, dwell_btw_sine: float = DEFAULT_DWELL_TIME, start_pose: int = 0, home_sign: int = 1, imu_to_tcp_x: float = DEFAULT_IMU_TO_TCP_X, imu_to_tcp_y: float = DEFAULT_IMU_TO_TCP_Y, imu_to_tcp_z: float = DEFAULT_IMU_TO_TCP_Z) -> str
Run the system identification trajectory and record calibration data.
Args:
Ts: Sampling time [s].
axes_to_command: Number of axes to command.
max_disp: Maximum displacement [rad].
max_vel: Maximum velocity [rad/s].
max_acc: Maximum acceleration [rad/s^2].
bcb_runtime: Runtime for bang-coast-bang system ID [s].
ctrl_config: Control configuration (`task` or `joint`).
sysid_type: System identification type (`bcb` or `sine`).
nV: Number of angle positions.
nR: Number of radius positions.
start_pose: Starting pose index.
home_sign: Sign of shoulder joint angle at home position.
imu_to_tcp_x: X component of IMU->TCP translation [m].
imu_to_tcp_y: Y component of IMU->TCP translation [m].
imu_to_tcp_z: Z component of IMU->TCP translation [m].
Returns:
`str` folder where calibration data is stored.
Side Effects:
Commands robot motion and writes calibration data to disk.
Raises:
None.
Preconditions:
Robot connection is active and data directory is writable.
def initialize_model_from_urdf(self, urdf_path: str) -> None
Load the robot kinematic and dynamic model from a URDF file.
Args:
urdf_path: Path to the URDF file.
Returns:
`None`.
Side Effects:
Loads a dynamics model and updates internal state.
Raises:
FileNotFoundError: If the URDF file does not exist.
RuntimeError: If the dynamics model cannot be constructed.
Preconditions:
URDF file is accessible.
def move_home(self, home_sign: int = 1, joint_move: bool = True) -> None
Move the robot to the home position.
Args:
home_sign: Sign of the shoulder joint angle.
joint_move: If True, move using joint angles; otherwise use pose.
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def move_home_joint(self, home_sign: int = 1) -> None
Move the robot to the home joint configuration.
Args:
home_sign: Sign of the shoulder joint angle.
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def move_home_pose(self) -> None
Move the robot to the configured home pose.
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def move_point_to_point_xyz(self, current_pose: List[float], target_xyz: List[float]) -> None
Move the robot to a Cartesian position while keeping orientation.
Args:
current_pose: Current pose [x, y, z, qx, qy, qz, qw].
target_xyz: Target position [x, y, z].
Returns:
`None`.
Side Effects:
Commands robot motion.
Raises:
None.
Preconditions:
Robot connection is active.
def process_motion_data(self, entry: Dict) -> None
Process a single motion data entry into the recorder.
Args:
entry: Dictionary containing motion data fields.
Returns:
`None`.
Side Effects:
Appends data to a recorder implementation.
Raises:
None.
Preconditions:
`entry` follows the expected data log schema.
def rt_periodic_task(self, Ts: float, trajectory: Trajectory) -> None
Publish a trajectory in real time and record data.
Args:
Ts: Sampling time [s].
trajectory: Trajectory containing position/velocity/acceleration data.
Returns:
`None`.
Side Effects:
Commands robot motion and records data.
Raises:
None.
Preconditions:
Robot connection is active and `trajectory` is populated.
def get_polar_coordinates(num_angles: int, num_radii: int, max_reach: float, min_angle: float = DEFAULT_MIN_CALIBRATION_ANGLE, max_angle: float = DEFAULT_MAX_CALIBRATION_ANGLE, min_radius_scale: float = DEFAULT_MIN_CALIBRATION_RADIUS_SCALE, max_radius_scale: float = DEFAULT_MAX_CALIBRATION_RADIUS_SCALE) -> Tuple[np.ndarray, np.ndarray]:
Generate polar coordinates (R and V) grid for calibration poses.
Args:
num_angles: Number of angle samples.
num_radii: Number of radius samples.
max_reach: Maximum reach of the robot [m].
min_angle: Minimum angle from horizontal [rad].
max_angle: Maximum angle from horizontal [rad].
min_radius_scale: Minimum radius scale (fraction of max reach).
max_radius_scale: Maximum radius scale (fraction of max reach).
Returns:
`tuple[np.ndarray, np.ndarray]` containing angle list `V` [rad] and
radius list `R` [m].
Side Effects:
None.
Raises:
None.
Preconditions:
`num_angles` and `num_radii` are positive.
def store_recorder_data_in_data_folder(recorder: DataRecorder, run_index: int, move_axis: int, data_folder: str) -> None:
Persist recorder data to motion and static CSV files.
Args:
recorder: DataRecorder containing recorded streams.
run_index: Pose index for file naming.
move_axis: Axis index for file naming.
data_folder: Output directory for CSV files.
Returns:
`None`.
Side Effects:
Creates directories and writes CSV files to disk.
Raises:
OSError: If files cannot be created or written.
Preconditions:
`recorder` contains at least one recorded sample.
def store_parameters_in_data_folder(traj_params: TrajParams, sysid_params: SystemIdParams, axes_commanded: int, num_joints: int, sample_time: float, start_pose: int, shoulder_len: float, base_height: float, robot_name: str, data_folder: str, tcp_payload: float = DEFAULT_TCP_PAYLOAD, tcp_payload_com_x: float = 0.0, tcp_payload_com_y: float = 0.0, tcp_payload_com_z: float = 0.0, imu_to_tcp_x: float = DEFAULT_IMU_TO_TCP_X, imu_to_tcp_y: float = DEFAULT_IMU_TO_TCP_Y, imu_to_tcp_z: float = DEFAULT_IMU_TO_TCP_Z) -> None:
Persist identification parameters to a CSV file.
Args:
traj_params: Trajectory parameters.
sysid_params: System identification parameters.
axes_commanded: Number of axes commanded.
num_joints: Number of robot joints.
sample_time: Sampling time [s].
start_pose: Starting pose index.
shoulder_len: Shoulder link length [m].
base_height: Base height [m].
robot_name: Robot name identifier.
data_folder: Output directory for the parameters file.
tcp_payload: TCP payload mass in URDF mass units (typically kg).
tcp_payload_com_x: X component of TCP payload center of mass [m].
tcp_payload_com_y: Y component of TCP payload center of mass [m].
tcp_payload_com_z: Z component of TCP payload center of mass [m].
imu_to_tcp_x: X component of IMU->TCP translation [m].
imu_to_tcp_y: Y component of IMU->TCP translation [m].
imu_to_tcp_z: Z component of IMU->TCP translation [m].
Returns:
`None`.
Side Effects:
Creates directories and writes `identification_parameters.csv` to disk.
Raises:
OSError: If the file cannot be created or written.
Preconditions:
`data_folder` is writable.
robot.robot_interface
RobotInterface(Robot) class
Provide a concrete robot implementation for system identification and calibration.
def __init__(robot_ip: str, tcp_payload: float = DEFAULT_TCP_PAYLOAD, tcp_payload_com: Sequence[float] | None = None, local_ip: str = "", sdk_token: str = "", robot_id: str = BOT_ID) -> None:
Initialize the robot interface and load the URDF model.
Args:
robot_ip: Robot IP address or `sim` for simulator mode.
tcp_payload: Payload of the tcp (default=0)
tcp_payload_com: Optional 3x1 center of mass of the tcp payload.
local_ip: Local machine IP address if required by the SDK.
sdk_token: SDK authentication token.
robot_id: Identifier used by the control stack.
Side Effects:
Loads the URDF model and may connect to robot hardware.
Raises:
RuntimeError: If the robot connection fails.
ValueError: If reported joint counts do not match the URDF.
Preconditions:
The URDF file is available and the SDK is installed.
def __get_joint_positions(self) -> List:
Get the current joint positions of the robot.
Returns:
`list[float]` joint positions in radians.
Side Effects:
Queries the robot hardware or simulator state.
Raises:
ValueError: If the number of joints does not match the URDF.
Preconditions:
The robot connection is active.
def __get_tcp_pose(self) -> List:
Get the current tool center point pose.
Returns:
`list[float]` pose as [x, y, z, qx, qy, qz, qw].
Side Effects:
Queries the robot hardware or simulator state.
Raises:
None.
Preconditions:
The robot connection is active.
def spin_thread(node):
Spin a ROS node in a background thread.
Args:
node: ROS node to spin.
Returns:
`None`.
Side Effects:
Runs a ROS event loop.
Raises:
None.
Preconditions:
ROS is initialized.
robot.ros_manager
ROS2 module for publishing joint trajectories and collecting feedback/IMU data.
JointStateSample class
Container for a single joint state sample with timestamps.
Args:
None.
Side Effects:
None.
Raises:
None.
Preconditions:
None.
ImuSample class
Container for a single IMU sample with timestamps.
Args:
None.
Side Effects:
None.
Raises:
None.
Preconditions:
None.
JointTrajectoryController(Node) class
Publish joint trajectories while recording aligned encoder and IMU data.
Args:
bot_id: Robot identifier used to build ROS topic names.
Ts: Sampling period for command publishing [s].
time_data: Command timestamps [s].
position_data: Joint position commands [rad].
velocity_data: Joint velocity commands [rad/s].
acceleration_data: Joint acceleration commands [rad/s^2].
publish_complete_event: Event signaled after publishing completes.
buffer_len_s: Buffer length for stored data [s].
Returns:
`None`.
Side Effects:
Creates ROS publishers/subscribers, timers, and data buffers.
Raises:
None.
Preconditions:
ROS is initialized and topic names are valid for the robot.
def __init__(self, bot_id: str, Ts: float = 0.005, time_data: list = [], position_data: list = [], velocity_data: list = [], acceleration_data: list = [], publish_complete_event: Optional[threading.Event] = None, buffer_len_s: float = 3600) -> None:
Initialize ROS publishers, subscribers, timers, and data buffers.
Args:
bot_id: Robot identifier used to build ROS topic names.
Ts: Sampling period for command publishing [s].
time_data: Command timestamps [s].
position_data: Joint position commands [rad].
velocity_data: Joint velocity commands [rad/s].
acceleration_data: Joint acceleration commands [rad/s^2].
publish_complete_event: Event signaled after publishing completes.
buffer_len_s: Buffer length for stored data [s].
Returns:
`None`.
Side Effects:
Creates ROS publishers/subscribers and allocates buffers.
Raises:
None.
Preconditions:
ROS is initialized and topic names are valid.
def joint_state_callback(msg: JointState) -> None:
Append a joint state sample with receive timing.
Args:
msg: ROS JointState message.
Returns:
`None`.
Side Effects:
Appends a `JointStateSample` to `self.state_data`.
Raises:
None.
Preconditions:
ROS node is active.
def encoder_is_ready() -> bool:
Check whether recent joint state data is available.
Returns:
`bool` True when encoder data is fresh.
Side Effects:
None.
Raises:
None.
Preconditions:
`self.state_data` is being populated by callbacks.
def imu_callback(msg: Imu) -> None:
Append an IMU sample with receive timing.
Args:
msg: ROS Imu message.
Returns:
`None`.
Side Effects:
Appends an `ImuSample` to `self.imu_data`.
Raises:
None.
Preconditions:
ROS node is active.
def imu_is_ready() -> bool:
Check whether recent IMU data is available.
Returns:
`bool` True when IMU data is fresh.
Side Effects:
None.
Raises:
None.
Preconditions:
`self.imu_data` is being populated by callbacks.
def publish_and_record() -> None:
Publish the next trajectory point and record aligned data.
Args:
None.
Returns:
`None`.
Side Effects:
Publishes ROS messages, updates buffers, and signals completion.
Raises:
None.
Preconditions:
`encoder_is_ready()` and `imu_is_ready()` return True before publishing.
def align_offline() -> None:
Align recorded encoder and IMU data to command timestamps.
Args:
None.
Returns:
`None`.
Side Effects:
Appends aligned records to `self.aligned_log`.
Raises:
RuntimeError: If `time_data` is not provided.
Preconditions:
`self.cmd_times` and sensor buffers are populated.
robot.run
run entry point
CLI entry point for robot setup, calibration, identification, vibration test, and fine-tuning.
def _run_model_generation_with_fine_tune_fallback(api_manager: ReforgeAPIManager, data_folder: str, fine_tune: bool) -> None:
Run cloud model generation with optional fine-tune fallback to identification.
Args:
api_manager: Cloud API manager with credentials.
data_folder: Folder containing calibration data.
fine_tune: If `True`, attempt fine-tuning first.
Returns:
`None`.
Side Effects:
May upload data, run remote jobs, and write local model files.
Raises:
Exception: Propagates API and filesystem errors from cloud model generation.
Preconditions:
`data_folder` exists and is readable.
def route_user_input(args: argparse.Namespace) -> None:
Route command-line arguments to the appropriate operation.
Args:
args: Parsed command-line arguments.
Side Effects:
May connect to hardware, run calibration, or write model files.
Raises:
None.
Preconditions:
`args.route` is a valid subcommand.
def main() -> None:
Entry point for robot command-line operations.
Returns:
`None`.
Side Effects:
Parses command-line arguments and dispatches actions.
Raises:
SystemExit: If argument parsing fails.
Preconditions:
None.
data folder
Destination for calibration CSVs (motion/static pairs per pose/axis) and run metadata; git-ignored so logs stay local. Created automatically by calibration.
models folder
Holds downloaded or generated control models; models/current is the active set used by the controller/identification pipeline.
urdf folder
Robot URDFs required for dynamics and kinematics; include OEM-provided URDF files (e.g., test_robot.urdf).
reforge_core.util
vibration_test
Utility to compare unshaped vs shaped point-to-point trajectories using the input shaper.
def run_vibration_test(robot_interface: RobotInterface, local_data_location: str, Ts: float, max_disp: float = DEFAULT_MAX_DISP) -> None:
Run vibration tests and store shaped/unshaped results.
Args:
robot_interface: Robot interface used to execute trajectories.
local_data_location: Folder containing calibration data.
Ts: Sampling time [s].
max_disp: Maximum displacement [rad].
plotting: Whether to plot comparisons.
Returns:
`None`.
Side Effects:
Commands robot motion and writes test data to disk.
Raises:
RuntimeError: If robot motion fails.
Preconditions:
Robot connection is active and calibration data exists.