Scenarios — what to run

The system spans two machines: the robot machine (YAM arms on CAN, runs the portal robot server) and the workstation (RealSense cameras + LeRobot, connects over portal / plain TCP). Two launchers run the right env for you — robot/yam on the robot, workstation/yam-data on the workstation. A single config.yaml at the repo root holds shared settings (robot host, camera serials, gains…), auto-discovered by every tool. Run each scenario top to bottom.

One-time setup

robot — persistent CAN names (no env to activate; robot/yam resolves it via uv run):

bash robot/setup_robot_env.sh    # optional: pre-create the uv env + install
bash robot/setup_can_ids.sh      # plug 4 adapters one-by-one → fixed names

workstation — conda env + uv (so you can also pip install other policy repos into it):

bash workstation/setup_workstation_env.sh # conda create yam_ws + uv pip install + udev rules
conda activate yam_ws
workstation/yam-data cams             # list RealSense serials

Then edit config.yaml at the repo root — at least the camera serials and robot.host:

robot:   {host: 192.168.0.42, port: 11331}
cameras: {agentview: "<D455>", wrist_left: "<D405>", wrist_right: "<D405>"}

A · Bimanual teleop only (no recording)

robot/yam canup                          # bring up the 4 CAN interfaces (after boot)
robot/yam teleop
# lift both gellos to engage; bring both home to stop & auto-return.

B · Data collection (teleop + LeRobot recorder) — main flow

1. robot — start the teleop server (serves state / action / gate over portal):

robot/yam canup
robot/yam teleop

2. workstation — start the recorder GUI (host + serials come from config.yaml):

workstation/yam-data record --repo-id user/yam_pick

3. The recorder opens on a Setup page:

1. Confirm repo_id / root / task and the source (teleop / dagger / eval). The dataset is written to <root>/<name>, where name is the last segment of repo_id (e.g. ~/lerobot_data + hello/pick_and_place → ~/lerobot_data/pick_and_place). The status line shows cameras detected and whether that dataset already exists.
2. To add to an existing dataset, tick Continue collecting (resume/append). Otherwise START creates it fresh — and if the folder already exists it asks twice before overwriting.
3. START connects the robot, opens cameras + dataset, and (with auto_arm) arms collection immediately.

Then teleoperate — lift both gellos to start recording, bring both home to end the episode:

• With review_before_save: true the episode is held in the review panel for Keep (S/F) or Delete (D). With review_before_save: false it auto-saves on each engage→idle.
• Leader handle buttons end and label in one press (configurable in config.yaml under recorder.buttons, keyed <side>.<index>). Default: left lower → success, right lower → fail, either upper → discard (force-home without saving).
• Close the window when done (calls finalize() so the dataset is complete). The cameras run on their own capture thread, so the live view and saving never stall on a slow frame.

C · Deployment / DAgger (policy + human takeover)

1. robot dagger server · 2. policy a websocket policy server · 3. workstation the bridge that joins them:

robot/yam deploy                                         # robot
python -m yam_policy.serve                               # policy host (:8000)
workstation/yam-data deploy --prompt "pick up the cube"  # workstation (reads config.yaml)
# press a handle button (or RobotClient.set_intervention) to take over; release to hand back.

D · Replay a dataset onto the robot

1. robot — run the wrapper server so it tracks the streamed targets:

robot/yam canup
robot/yam wrapper

2. workstation — open the replay GUI, then Load → pick episode → (tick "Overlay live" to match the scene) → tick "Send to robot" → Play (it ramps to the first frame, then streams each frame's action over portal):

workstation/yam-data replay --repo-id user/yam_pick

Below: the full codebase reference — architecture, motor/CAN protocol, robot layer, the portal/websocket serving stack, and the data tools.

I2RT (i2rt.com) builds affordable, capable torque-controlled robot hardware. This repository is the Python client library that drives that hardware. It supports three main product families:

Key Features

• Plug-and-play Python interface — get a robot object in one line, then read and command joint positions.
• Real-time CAN bus control — drives Damiao (DM) series motors in MIT impedance-control mode at 250 Hz.
• MuJoCo gravity compensation — inverse dynamics cancels the arm's weight to provide a "zero-gravity" mode.
• Simulation & visualization — MuJoCo desktop viewer, a Viser web interface, and bundled URDF/MJCF models.
• Gripper force control & auto-calibration — clog detection and force limiting.
• Bimanual teleoperation + trajectory record/replay — ready to plug into standard robot-learning pipelines.

Core Concepts (read this first)

A handful of ideas recur throughout the code. Skim these before diving in.

Architecture — Layered Design

The system is split into clear layers. The higher you go, the more abstract; the lower you go, the closer to hardware.

Control-loop data flow

Here is what happens in one cycle when you position-control a YAM arm.

Directory layout

Install & Quick Start

The repo recommends uv (a fast Python package manager) and Python 3.11.

git clone https://github.com/i2rt-robotics/i2rt.git && cd i2rt
curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.local/bin/env
uv venv --python 3.11
source .venv/bin/activate

sudo apt update
sudo apt install build-essential python3-dev linux-headers-$(uname -r)
uv pip install -e .

The key dependencies are python-can (CAN comms), mujoco + mink (simulation/IK), viser (web UI), ruckig (trajectory generation), portal (RPC), qpsolvers (QP solver for IK), and tyro (CLI parsing). (Source: pyproject.toml.)

First program — read and command joints

from i2rt.robots.get_robot import get_yam_robot
from i2rt.robots.utils import GripperType
import numpy as np

robot = get_yam_robot(channel="can0", gripper_type=GripperType.LINEAR_4310)

# Read joint positions (radians) — 6 arm joints
q = robot.get_joint_pos()        # shape: (6,) or (7,) including gripper

# Command a target configuration (PD control under the hood)
robot.command_joint_pos(np.zeros(6))

Zero-gravity mode (move it by hand)

This command floats the arm with gravity compensation only, so you can move it freely by hand.

python i2rt/robots/motor_chain_robot.py --channel can0 --gripper linear_4310

CAN Bus Setup

# Check detected CAN devices
ls -l /sys/class/net/can*

# Bring up the interface at 1 Mbit/s
sudo ip link set can0 up type can bitrate 1000000

# Auto-enable on boot (install udev rules)
sudo sh devices/install_devices.sh

# Reset an unresponsive adapter
bash robot/reset_all_can.sh

robot/reset_all_can.sh brings every can* interface down and back up at 1 Mbit/s. With multiple adapters, device names can shuffle depending on plug order; to pin stable names, follow the serial-based udev setup in docs/guides/set-persistent-can-ids.md (e.g. can_follower_l, can_leader_r). CAN interface names are at most 13 characters and must start with can.

Motors & CAN Drivers

The lowest layer lives in i2rt/motor_drivers/. Three files form its core.

Control modes & control frequency

The control loop runs at 250 Hz (4 ms period). If a step exceeds 7 ms it logs a performance warning (constants at the top of dm_driver.py).

Supported motor types

MotorType defines each motor's position/velocity/torque limits (MotorConstants). Those limits double as the scale for the float ↔ integer quantization.

DMChainCanInterface — threading model

Multiple motors are bundled into a single chain. When created with start_thread=True, a background control thread starts and loops over:

• Sending the current command to all motors under command_lock (an RLock).
• On a detected motor error, _try_recover_motors() performs clear error → re-enable up to 3 times.
• Updating feedback state (a list of FeedbackFrameInfo) under state_lock.
• Reading any same-bus encoder state, if a get_same_bus_device_driver was provided.
• Tracking the actual loop rate with a RateRecorder and reporting when a step exceeds 7 ms.

On construction it also runs a CAN bandwidth check: if num_motors × 2 frames × 130 bits × 250 Hz exceeds bitrate / 1.1, it warns.

Coordinate transform (motor ↔ control)

Each motor has a motor_offset and motor_direction that convert between the motor's raw angle and the control coordinate.

pos_sim  = (pos_real - offset) * direction      # when reading
pos_real =  pos_sim  * direction  + offset      # when commanding

_update_absolute_positions() also detects multi-turn wrap-around past the ± position limits so the position never jumps discontinuously.

To bundle several CAN buses (e.g. a bimanual robot) into one object, MultiDMChainCanInterface aggregates each bus's DMChainCanInterface and dispatches commands by index. The reported comm frequency is the slowest bus (the bottleneck).

DM Motor CAN Protocol (in depth)

This is the part closest to the hardware. Every float is quantized to an integer over the motor's limit range, then packed into bit fields.

def float_to_uint(x, x_min, x_max, bits):
    span = x_max - x_min
    x = min(max(x, x_min), x_max)            # clip to range
    return int((x - x_min) * ((1 << bits) - 1) / span)

MIT control frame (8-byte send)

Position 16 bits, velocity 12 bits, kp 12 bits, kd 12 bits, torque 12 bits — densely packed into 8 bytes.

data[0] = (pos >> 8) & 0xFF                  # pos[15:8]
data[1] =  pos       & 0xFF                  # pos[7:0]
data[2] = (vel >> 4) & 0xFF                  # vel[11:4]
data[3] = ((vel & 0xF) << 4) | (kp >> 8)     # vel[3:0] | kp[11:8]
data[4] =  kp        & 0xFF                  # kp[7:0]
data[5] = (kd >> 4) & 0xFF                   # kd[11:4]
data[6] = ((kd & 0xF) << 4) | (tor >> 8)     # kd[3:0] | tor[11:8]
data[7] =  tor       & 0xFF                  # tor[7:0]

Feedback frame (8-byte receive)

The receive ID is determined by ReceiveMode. The default p16 means receive_id = send_id + 16. (Other options: same, zero, plus_one.)

Motor command bytes

Beyond the normal control frame, special commands are distinguished by the last data byte.

Register configuration protocol (persistent settings)

Persistent parameters like motor ID, timeout, and gear ratio are read/written/saved to EEPROM via special messages on the broadcast ID 0x7FF (motor_config_tool/utils.py).

Key registers: id (8), master_id (7), timeout (9), gear_ratio (20), KT_value (torque constant, 1), sw_ver (14), and more.

Error codes

Motor Config Tools

The CLIs in i2rt/motor_config_tool/ handle initial motor setup.

Ping motors — which ones are alive

python i2rt/motor_config_tool/ping_motors.py --channel can0

It iterates IDs 1–7, attempts motor_on, prints the motors that respond, then turns them off again.

Set zero offset

python i2rt/motor_config_tool/set_zero.py --channel can0 --motor_id 1

Enables the motor, stores the current position as zero (save_zero_position), then re-reads to verify it is near zero (<0.01 rad). For a standard YAM, run it for each of IDs 1–6.

Set safety timeout

The factory default is 400 ms. To disable it you must run twice.

# Disable timeout (advanced users — run twice)
python i2rt/motor_config_tool/set_timeout.py --channel can0
python i2rt/motor_config_tool/set_timeout.py --channel can0
# Re-enable timeout
python i2rt/motor_config_tool/set_timeout.py --channel can0 --timeout

Internally it writes 0 (off) or 8000 ms to register 9 (timeout) and saves to EEPROM.

Robot Interface (the protocol)

Robot in i2rt/robots/robot.py is a @runtime_checkable Protocol. That means any object implementing the same methods counts as a "robot," without inheriting from an ABC. Both real hardware (MotorChainRobot) and simulation (SimRobot) satisfy this contract, so the same code can swap between them.

get_observations() return example

{
  "joint_pos":   ndarray,   # [rad]  arm joints (excludes gripper)
  "joint_vel":   ndarray,   # [rad/s]
  "joint_eff":   ndarray,   # [Nm]   torques
  "gripper_pos": ndarray,   # [0-1]  normalized gripper position
  "gripper_vel": ndarray,
  "gripper_eff": ndarray,
  "temp_mos":    ndarray,   # [°C]   MOSFET temperature (optional)
  "temp_rotor":  ndarray,   # [°C]   rotor temperature (optional)
}

MotorChainRobot — real hardware

i2rt/robots/motor_chain_robot.py is the implementation for a physical robot. Given a DMChainCanInterface (motor chain) and a MuJoCo model, a background server thread continuously performs the following.

Gravity-compensation pipeline

1. Compute gravity torque — MuJoCo inverse dynamics (MuJoCoKDL.compute_inverse_dynamics(q, 0, 0)) gives the torque g that holds the current pose.
2. Scale — multiply by per-joint gravity_comp_factor to correct for friction and gearing losses.
3. Friction compensation — add coulomb_friction · sign(q̇).
4. Sum & clip — τ = τ_cmd + g·factor + friction, clipped to torque limits.

friction_comp = coulomb_friction * np.sign(joint_state.vel)
motor_torques = joint_commands.torques + g * gravity_comp_factor + friction_comp

Zero-gravity idle mode

With zero_gravity_mode=True, the position target is held at zero and only kd is set (to grav_comp_kd), so the PD term does not drag the arm — only gravity compensation acts. A human can move the arm freely. enter_gravity_comp_idle() returns to this safe state at any time.

Gripper normalization (JointMapper)

Externally the gripper is exposed as [0,1] (0 = closed, 1 = open), but the internal motor angle is a physical range (e.g. [-0.048, 0] rad). JointMapper (robots/utils.py) converts between the two transparently.

pos = self._motor_state_to_joint_state(motor_state)
pos = self.remapper.to_command_joint_pos_space(pos)   # physical angle → [0,1]

Gripper force limiting & clog detection

When limit_gripper_force > 0, a GripperForceLimiter watches the gripper torque history in a lock-free ring buffer. If torque exceeds clog_force_threshold while speed is below clog_speed_threshold, it treats this as a "clog" and reduces the target position to cap the applied force. The target-force → motor-torque mapping uses a linear or crank (zero-linkage) curve per gripper type.

Joint limits & safety

At init and at runtime it checks that joints stay within limits and, on a violation, raises a RuntimeError with the offending position/limit details. command_joint_pos() clips commands to the limits before sending.

SimRobot — MuJoCo simulation

i2rt/robots/sim_robot.py is a drop-in replacement for MotorChainRobot, used for development and testing without hardware (sim=True). It runs physics in MuJoCo and satisfies the same Robot protocol.

• A background physics loop (200 Hz by default) applies gravity-compensation torques and updates qpos/qvel.
• Maps the gripper [0,1] ↔ MuJoCo physical range (_cmd_to_mj_qpos / _mj_to_cmd_qpos).
• Even simulates motor temperatures (temp_mos, temp_rotor with a little noise) for realism.

This lets you move the same higher-level code between sim and real.

get_robot Factory

get_yam_robot() in i2rt/robots/get_robot.py hides the whole assembly behind one function. It is the recommended entry point for the YAM family.

get_yam_robot(
    channel="can0",
    arm_type=ArmType.YAM,
    gripper_type=GripperType.LINEAR_4310,
    zero_gravity_mode=True,
    ee_mass=None, ee_inertia=None,          # override end-effector inertia (tools)
    gravity_comp_factor=None,
    sim=False,                              # True returns a SimRobot
)

What it does internally

1. _load_arm_config(arm_type) — reads the motor list, gains, and gravity-comp factors from config/yam.yml etc.
2. combine_arm_and_gripper_xml() — merges the arm XML and gripper XML (with the mount transform: pos/quat/axis) into a temporary MJCF.
3. Parses joint limits from the combined XML.
4. Appends the gripper motor (CAN ID 0x07) to the end of the motor list.
5. Creates the DMChainCanInterface motor chain, auto-calibrating gripper limits if needed.
6. Corrects motor angles that wrap around ±π using motor_offset.
7. Instantiates MotorChainRobot (or SimRobot) with the full configuration.

An armless gripper-only mode (ArmType.NO_ARM) is also supported via _get_gripper_only_robot().

Grippers

Gripper variants are expressed as the GripperType enum, each defining its motor, limits, force-torque curve, and mount transform in YAML.

Force → torque mapping

• linear: target_torque = gripper_force · gripper_stroke / motor_stroke
• crank: a nonlinear map from crank radius and angle — target_torque = gripper_force · r · sin(θ)

Kinematics (FK / IK)

The Kinematics class in i2rt/robots/kinematics.py uses mink, a differential-IK library built on MuJoCo.

Forward kinematics (FK)

def fk(self, q, site_name=None) -> np.ndarray:
    """Return the 4x4 SE(3) transform of the site (in world) at joints q."""
    self._configuration.update(q)
    return self._configuration.get_transform_frame_to_world(site_name, "site").as_matrix()

Inverse kinematics (IK)

It sets a FrameTask toward a target 4×4 pose, solves for velocity with a QP solver (quadprog), and integrates iteratively. When the position and orientation errors fall below a threshold (1e-4), it reports convergence and returns (success, q).

ok, q_solution = kin.ik(target_pose, site_name="grasp_site", init_q=current_q)

The test (tests/test_kinematics.py) verifies an FK→IK→FK round trip recovers the original pose.

Robot Models & YAML Config

A robot is defined by two kinds of assets: a physical model (MJCF/URDF + STL meshes) and a hardware profile (YAML).

Model files (i2rt/robot_models/)

• arm/{yam, yam_pro, yam_ultra, big_yam}/ — each arm's .xml (MuJoCo), some .urdf, and assets/*.stl meshes.
• gripper/{linear_4310, linear_3507, crank_4310, flexible_4310, ...}/ — gripper models.
• At runtime the arm + gripper XML are merged dynamically into a combined model in /tmp.

Arm config schema (e.g. config/yam.yml)

motor_list:                       # [CAN_ID, motor_type] — 6 arm joints
  - [0x01, "DM4340"]              # shoulder
  - [0x02, "DM4340"]
  - [0x03, "DM4340"]
  - [0x04, "DM4310"]              # elbow
  - [0x05, "DM4310"]              # wrist
  - [0x06, "DM4310"]
directions:          [1, 1, 1, 1, 1, 1]      # motor polarity (+1/−1)
kp:                  [80, 80, 80, 40, 10, 10] # position gains
kd:                  [5, 5, 5, 1.5, 1.5, 1.5] # damping gains
gravity_comp_factor: [1.0, 1.1, 1.1, 1.2, 1.0, 1.0]  # gravity torque scale
grav_comp_kd:        [0.1, 0.1, 0.1, 0.3, 0.05, 0.05] # zero-grav idle damping
coulomb_friction:    [0.3, 0.3, 0.3, 0.06, 0.06, 0.06] # friction comp [Nm]

BIG_YAM uses DM6248 (±120 Nm) at the shoulder and reverses some motors (directions: [-1,-1,1,1,-1,1]).

Gripper config schema (e.g. config/linear_4310.yml)

last_joint_mount:            # per-arm wrist mount transform
  yam:
    pos:  "2.4e-07 -0.0419 0.0405"
    quat: "0.5 -0.5 -0.5 -0.5"
    axis: "0 0 -1"
motor_type: "DM4310"
motor_kp: 20.0
motor_kd: 0.5
gripper_limits: null         # null = needs calibration
needs_calibration: true
limiter:                     # clog detection / force limiting
  clog_force_threshold: 0.5  # [Nm]
  clog_speed_threshold: 0.3  # [rad/s]
  force_torque_map: "linear" # "linear" or "crank"
  motor_stroke: 6.57         # [rad]
  gripper_stroke: 0.096      # [m]

Tests

i2rt/robots/tests/ verifies: XML assembly (test_assembly), gravity-comp torque ranges (test_gravity_comp, BIG_YAM ≤ 25 Nm and others ≤ 20 Nm), FK/IK round trips (test_kinematics), simulated loading of every arm/gripper combo (test_robot_variants), and control-interface integration (test_control_interface).

Control Interfaces (MuJoCo & Viser)

There are two interactive control UIs. Both mirror robot state in real time, let you drag the end-effector with IK, and check for self-collisions before commanding.

MuJoCo desktop viewer

i2rt/utils/mujoco_control_interface.py. It opens the native MuJoCo viewer and toggles two modes with SPACE.

• VIS mode (read-only): mirrors the real robot into the simulation. A mocap target follows the end-effector via FK. No commands are sent.
• CONTROL mode: move per-joint sliders, or drag the mocap marker, and an IK solution is commanded to the robot. The marker turns red.

The viewer runs on the main thread (a MuJoCo requirement) while control logic runs in a background thread (control_dt defaults to 5 ms = 200 Hz). Teaching-handle button states show as colored spheres, and joint state / gravity torque / temperatures stream to the terminal as a live table.

Viser web interface

i2rt/utils/viser_control_interface.py. View and control the robot in 3D in a browser. It extracts the MuJoCo meshes into the Viser scene and updates them each frame via FK.

• Safety gate: you must tick "Alignment Confirmed" → the "Enable Robot" button activates before control is possible (prevents accidents).
• Modes: VIS (mirror), IK control (drag a 6-DOF transform frame), and Joint sliders (per-joint angles).
• Also supports a gripper slider (0–1), teaching-handle button display, and live PD-gain tuning (update_kp_kd).
• The control loop runs at 50 Hz. On a collision it blocks the command and prints a message.

Run: python examples/control_with_viser/control_with_viser.py --sim (or --channel can0 --port 8080).

Gravity-comp util (MuJoCoKDL)

MuJoCoKDL in i2rt/utils/mujoco_utils.py calls mujoco.mj_inverse() on a model with collisions and joint limits disabled, isolating pure gravity torques. With q̇=0, q̈=0 the result is exactly "the torque needed to hold the current pose against gravity."

Leader-Follower Teleoperation (GELLO style)

The core demo is examples/minimum_gello/minimum_gello.py. A leader arm (a human-held teaching handle) drags a follower arm across the network, with bilateral force feedback.

Server / client structure

• ServerRobot — exposes robot methods (get_joint_pos, command_joint_pos, …) as Portal RPC endpoints. Runs on the follower side.
• ClientRobot — a Portal client that connects to the remote follower.
• YAMLeaderRobot — a wrapper around the leader arm + teaching handle (passive encoder / buttons).

How to run

# Follower arm
python examples/minimum_gello/minimum_gello.py \
    --gripper linear_4310 --mode follower --can-channel can0 --bilateral-kp 0.2

# Leader arm (teaching handle)
python examples/minimum_gello/minimum_gello.py \
    --gripper yam_teaching_handle --mode leader --can-channel can1 --bilateral-kp 0.2

• Top button (press once): enable synchronization — the follower tracks the leader.
• Press again: disable synchronization.
• --bilateral-kp: the resistance felt on the leader (0.1–0.2 recommended).

How bilateral force feedback works

When sync turns on, the leader's kp is scaled by bilateral_kp. The follower's position error feeds back to the leader as torque (τ = kp·(q_follower − q_leader)), so the human can feel the resistance the follower encounters. At the moment sync starts, the follower is slowly interpolated to the leader's position (slow_move) to avoid a sudden jump.

Bimanual teleoperation

examples/bimanual_lead_follower/ launches minimum_gello.py as four processes: right-arm leader/follower (port 1234) and left-arm leader/follower (port 1235). It validates that each CAN channel (can_follower_r, can_leader_r, …) exists, polls process health and restarts on failure, and shuts down in stages (terminate → SIGINT → SIGKILL).

Trajectory record & replay

examples/record_replay_trajectory/ is a curses-based TUI.

A trajectory is saved to .npy as a dict: {"trajectory": (N, DOF), "timestamps": (N,), "frequency": f}. On replay it first interpolates to the first waypoint over 1.5 s, then calls command_joint_pos in sequence.

Single-motor PD control

examples/single_motor_position_pd_control/ raises and lowers a single motor's target position with the arrow keys, while showing position, velocity, torque, temperature, and error code live. Handy for motor debugging.

Passive Encoders & Teaching Handles

A teaching handle's joint angle and buttons are read from a passive encoder. The key trick is that this encoder shares the same CAN bus as the motors — no second CAN connection is opened.

• i2rt/utils/encoder_manager.py — the encoder CAN driver (PassiveJointEncoder). Request/response protocol, zero/frequency config, firmware version & EEPROM, and firmware validation (validate_encoders). CAN IDs: request 0x50E, report 0x50F, event 0x510.
• i2rt/utils/encoder_utils.py — the get_encoder_chain() factory. It takes a DMChainCanInterface's CAN interface and returns an EncoderChain that polls the encoder on the same bus.

If you pass get_same_bus_device_driver=get_encoder_chain when creating the motor chain, the control thread reads encoder state alongside motor state and exposes it via motor_chain.get_same_bus_device_states(). The returned PassiveEncoderInfo holds position, velocity, and io_inputs (button) bits.

can_interface = CanInterface(channel=args.channel)
encoder_chain = get_encoder_chain(can_interface)
while True:
    s = encoder_chain.read_states()[0]
    print(s.position, s.io_inputs)     # position + buttons
    time.sleep(0.01)

Gamepad input

The Gamepad class in i2rt/utils/gamepad_utils.py initializes pygame headless (the SDL dummy driver) to read a joystick. From the analog sticks it returns an [x, y, θ] command (with a 0.05 dead zone) and button states. It is used mainly for Flow Base teleoperation.

Flow Base — Omnidirectional Mobile Base

i2rt/flow_base/. A 4-caster holonomic (omnidirectional) base inspired by TidyBot++, moving in x, y, and θ independently. A control server runs on a Raspberry Pi, and external machines connect via Portal RPC.

Hardware & frequencies

• 4 caster wheel modules = 4 steering motors (DM4310V) + 4 drive motors (DM_FLOW_WHEEL) = 8 motors.
• An optional 9th motor = the linear-rail lift (DM8009/DM4310).
• Control loop 200 Hz (FIFO real-time scheduler); policy commands 10 Hz.
• Ruckig OTG (online trajectory generation) for acceleration-limited smooth motion.
• Default port 11323, static IP 172.6.2.20.

Kinematics & coordinate frames

The Vehicle class maintains the Jacobians (C, C_p, C_pinv) that convert operational-space velocity [dx, dy, dθ] into steering/drive motor velocities. Velocity commands can be given in two frames:

• LOCAL: relative to the robot's current heading (multiplied by the rotation matrix R.T).
• GLOBAL: in a fixed world frame (accumulated by odometry).

Velocity/acceleration are bounded by Ruckig (e.g. max_vel = (0.5, 0.5, 1.57)). If commands stall for 2.5 × policy period, it holds position; if the steering speed exceeds 720 deg/s, it slows down to prevent caster flips.

Network client API

from i2rt.flow_base.flow_base_client import FlowBaseClient

client = FlowBaseClient(host="172.6.2.20")
client.set_target_velocity([0.1, 0.0, 0.0], frame="local")  # 0.1 m/s in x
odom = client.get_odometry()      # {translation [x,y], rotation rad}
client.reset_odometry()

FlowBaseClient keeps sending commands at 50 Hz from a background thread to maintain a heartbeat. With the linear rail, it uses a 4D command [x, y, θ, rail_vel].

Linear rail controller

linear_rail_controller.py drives the 9th motor's vertical lift together with a GPIO brake and upper/lower limit switches. At startup it homes to the lower limit at 50% speed, and protects with a 0.25 s command timeout and limit checks. A SingleMotorControlInterface drives only the rail motor while preserving the velocities of the 8 base motors.

Joystick demo

python i2rt/flow_base/flow_base_controller.py

Left stick = translation (x/y), right stick X = rotation (θ), right stick Y = rail up/down. A mode button switches between local/global frames; other buttons reset odometry and override API commands (safety).

Networking & Serving

The YAM rig is exposed to the workstation over portal (plain TCP). The robot machine runs a portal server (i2rt/serving/) that owns the real-time control loop; the workstation connects with RobotClient. Policy inference is a separate websocket + msgpack link (openpi-compatible). Bimanual by default (2 leaders + 2 followers).

Robot server modes — i2rt.serving.run_robot_server

The robot runs one of three modes (launch with robot/yam <mode>, which uses uv run — nothing to activate; add --sim for no hardware). The real-time 250 Hz motor loop stays inside MotorChainRobot; each mode reads the rig and serves a snapshot over portal.

robot/yam teleop                           # auto-reads config.yaml
robot/yam deploy                           # a policy drives; button = takeover
robot/yam wrapper                          # replay target

Teleop auto-gate (both arms together, no button): HOMING → robot & gellos ramp home; IDLE → at home, gellos free; lift both past --engage-thr → ENGAGED (followers track directly via their own PD gains, no lag); bring both home for --dwells → back to HOMING. The exact rate-limited command sent is the recorded action, so episodes reproduce precisely. Only the one-time engage approach + the (gentler) homing return are ramped (--ramp-speed / --home-speed).

The snapshot (get_observation()) carries, per side, follower pos/vel/eff, leader_pos, applied (the action), human (while intervening), buttons, plus the gate state and estop. Follower gains live in one place — i2rt/serving/control_config.py — and apply to teleop, dagger, and the replay wrapper, so collection and replay match.

Policy serving & the deploy client

Policies run in their own unconstrained env (any Python / CUDA, even a remote GPU box) behind an openpi-compatible websocket server; the workstation deploy client joins the two links — it reads the robot snapshot + cameras, builds an openpi obs dict, queries the policy, and sends the result to the robot. One chunk is executed per inference (there is no prefetching variant: one existed, and measured, its freshness came out a wash).

python -m yam_policy.serve --policy <module>:<Class> --config k=v   # policy host (:8000)
robot/yam deploy                                                # robot
workstation/yam-data deploy --prompt "pick up the cube"          # workstation

The client auto-configures the chunk length and image keys from the server's metadata, so a real openpi checkpoint drops in unchanged — and so does a LeRobot checkpoint, through yam_policy.policies.lerobot_policy:LeRobotPolicy, with no conversion step. Deploy names which stack answered, since all of them speak this wire but read different observations. Add your own policy by subclassing BasePolicy.infer(obs)→{"actions": (H, D)} (see policy_serving/ with dummy / openpi / lerobot templates).

Safety model

• Network E-STOP — a button in both GUIs (and RobotClient.set_estop) holds the followers until released; re-applied automatically on reconnect.
• Link-loss watchdog — if no fresh command arrives within command_timeout (≈0.5 s), the follower holds instead of replaying a stale target (covers a workstation crash or network drop).
• Limits — every commanded target is clamped to optional per-joint FOLLOWER_JOINT_LIMITS; an optional FOLLOWER_EFFORT_LIMIT trips an automatic e-stop on a collision/overload.
• Always-on gravity comp + smoothing — added inside MotorChainRobot every tick; transitions are rate-limited (TargetSmoother), and a missing/invalid reading holds position. The motor 400 ms watchdog is the hardware backstop.

Details + the snapshot contract: i2rt/serving/README.md.

Data tools & config (LeRobot)

The workstation app (workstation/lerobot_recorder/) records LeRobot v3.0 datasets from the robot snapshot over portal + three RealSense cameras (2× D405 wrist + 1× D455 agentview). Episodes auto-start/stop from the source gate; a finished episode is queued to a background writer so encoding never blocks the next collection. The GUI shows a color status banner, a health strip (robot / cameras / disk / queue), live success-rate stats, audio cues, and labeling by mouse / keyboard / leader buttons. --source dagger records complete policy rollouts with per-frame policy/intervention labels; --source eval records manually bounded policy rollouts. The replay GUI plays an episode back onto the wrapper (with a live-camera overlay to match the scene first).

Launchers: workstation/yam-data record / replay / bridge / cams / doctor (dataset stats + success rate). Full runbook: workstation/lerobot_recorder/README.md.

Examples

Each example folder has its own README.md with detailed CLI args and usage. In robot/ you'll find run_yam_leader.py (run the leader standalone), run_gello_with_passive_encoder.py (encoder integration demo), and read_encoder.py.

Devices & Hardware Setup

devices/ auto-configures CAN hardware on Linux / Raspberry Pi.

• install_devices.sh — copies devices/rules/*.rules into /etc/udev/rules.d/, adds the user to the plugdev and video groups, and reloads udev.
• rules/flow_base.rules — detects the USB-CAN adapter (vendor 1d50, product 606f, gs_usb/candle), loads the gs_usb driver, and brings the CAN interface up at 1 Mbit/s automatically.
• pi_setup.md — how to back up / restore the Raspberry Pi SD card with dd + PiShrink.
• docs/guides/set-persistent-can-ids.md — udev rules that assign stable names (e.g. can_leader_r) from each USB-CAN adapter's serial.

API Cheatsheet

from i2rt.robots.get_robot import get_yam_robot
from i2rt.robots.utils import GripperType, ArmType
import numpy as np

# 1) Create a robot (real or sim)
robot = get_yam_robot(channel="can0",
                      arm_type=ArmType.YAM,
                      gripper_type=GripperType.LINEAR_4310,
                      zero_gravity_mode=True,
                      sim=False)

# 2) Read state
q   = robot.get_joint_pos()           # (DOF,) rad
obs = robot.get_observations()        # dict: joint_pos/vel/eff, gripper, temp
st  = robot.get_joint_state()         # {"pos", "vel"}

# 3) Command
robot.command_joint_pos(np.zeros(6))                         # position (PD)
robot.command_joint_state({"pos": q, "vel": np.zeros(6)})    # explicit
robot.enter_gravity_comp_idle()                              # back to safe idle

# 4) Shut down
robot.close()

from i2rt.motor_drivers.dm_driver import DMChainCanInterface, ControlMode
from i2rt.motor_drivers.utils import ReceiveMode
import numpy as np

chain = DMChainCanInterface(
    motor_list=[[0x01, "DM4340"], [0x02, "DM4340"]],
    motor_offset=np.zeros(2), motor_direction=np.ones(2),
    channel="can0", control_mode=ControlMode.MIT,
    receive_mode=ReceiveMode.p16,
)
chain.set_commands(torques=np.zeros(2), pos=np.zeros(2),
                   vel=np.zeros(2), kp=np.full(2, 30.0), kd=np.full(2, 1.0))
states = chain.read_states()

from i2rt.flow_base.flow_base_client import FlowBaseClient
client = FlowBaseClient(host="172.6.2.20")
client.set_target_velocity([0.1, 0.0, 0.0], frame="local")
print(client.get_odometry())

Glossary

Acknowledgments

• TidyBot++ — inspiration for the Flow Base hardware and control.
• GELLO — inspiration for the teleoperation design.

Support & License

Email support@i2rt.com · Sales sales@i2rt.com · License MIT.