Add EE keyboard teleop/sim for AX arm with robust IK

- teleoperate_ee_keyboard.py: velocity IK (dual-quaternion resolved-rate
  ported from realtime_servoing, plus a position-Jacobian solver) integrated
  to position commands; local/global frame toggle where global uses the
  position Jacobian with damped least-squares; constant Cartesian-speed
  scaling with a joint-step cap; ramped homing to the reference pose.
- simulate_ee_keyboard.py: hardware-free 3D sim reusing the same IK.
- urdf_mapping.py: widen shoulder_pan (+/-90) and elbow_flex (0-180) limits.
- ax_arm.py: guard send_action against non-.pos / non-string keys.

lerobot_robot_ax_arm/README.md

@@ -0,0 +1,49 @@
+# lerobot_robot_ax_arm
+
+A third-party [LeRobot](https://github.com/huggingface/lerobot) robot: a 4-DoF arm driven by Dynamixel
+AX-series servos (e.g. AX-12A) over **Protocol 1.0**.
+
+| Motor ID | Joint           | Normalization |
+| -------- | --------------- | ------------- |
+| 1        | `shoulder_pan`  | [-100, 100]   |
+| 2        | `shoulder_lift` | [-100, 100]   |
+| 3        | `elbow_flex`    | [-100, 100]   |
+| 4        | `gripper`       | [0, 100]      |
+
+## Protocol 1.0 notes
+
+AX-series motors differ from the X-series (Protocol 2.0) used elsewhere in LeRobot:
+
+- **No Sync Read** — positions are read sequentially (`get_observation` / calibration). Sync Write is still
+  used for `Goal_Position`.
+- **No `Operating_Mode` / PID registers** — `configure()` only lowers the return delay time.
+- **No homing offset register** — calibration records the range of motion only (`homing_offset = 0`) and is
+  stored via the CW/CCW angle limits.
+
+## Install
+
+```bash
+pip install -e .
+```
+
+This is discovered automatically by LeRobot thanks to the `lerobot_robot_` package prefix.
+
+## Usage
+
+```bash
+lerobot-record \
+  --robot.type=ax_arm \
+  --robot.port=/dev/tty.usbserial-AL02L1E0 \
+  # ... other arguments
+```
+
+Or from Python:
+
+```python
+from lerobot_robot_ax_arm import AXArm, AXArmConfig
+
+robot = AXArm(AXArmConfig(port="/dev/tty.usbserial-AL02L1E0"))
+robot.connect()
+obs = robot.get_observation()
+robot.disconnect()
+```

lerobot_robot_ax_arm/examples/simulate_ee_keyboard.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python
+
+"""Hardware-free 3D simulation of the AX-arm keyboard EE teleop.
+
+Reuses the real IK (``_build_kinematics`` / ``_joint_velocity`` from ``teleoperate_ee_keyboard``)
+and a synthetic calibration, driving a simulated (ideal) servo bus instead of a real one. Lets you
+sanity-check the solver/frame behaviour and joint limits in a matplotlib window.
+
+Controls (focus the plot window):
+    - w / s : +X / -X       - a / d : +Y / -Y       - r / f : +Z / -Z
+    - o / c : open / close gripper
+    - t     : toggle global <-> local frame
+    - m     : toggle dq <-> pos solver
+    - q / esc : quit
+
+Run:
+    python examples/simulate_ee_keyboard.py [--solver pos] [--frame global] [--speed 0.06]
+"""
+
+import argparse
+import importlib.util
+import os
+import tempfile
+from pathlib import Path
+
+os.environ.setdefault("MPLCONFIGDIR", tempfile.gettempdir())
+
+import numpy as np
+
+import matplotlib.pyplot as plt
+from matplotlib.animation import FuncAnimation
+
+from lerobot.motors import MotorCalibration
+from lerobot_robot_ax_arm.urdf_mapping import (
+    ARM_JOINTS,
+    REFERENCE_URDF_DEG,
+    SCALE,
+    URDF_LIMITS_DEG,
+    ticks_to_urdf_vector,
+    urdf_vector_to_ticks,
+)
+
+# Reuse the real teleop IK helpers without duplicating them.
+_EE = importlib.util.spec_from_file_location(
+    "_ee_teleop", str(Path(__file__).with_name("teleoperate_ee_keyboard.py"))
+)
+ee = importlib.util.module_from_spec(_EE)
+_EE.loader.exec_module(ee)
+
+TICK_REF = 512  # tick chosen to sit at each joint's URDF reference angle
+GRIP_RANGE = (350, 600)
+
+
+def _synthetic_calibration() -> dict[str, MotorCalibration]:
+    """Calibration consistent with urdf_mapping: reference tick + travel limits per joint."""
+    calib: dict[str, MotorCalibration] = {}
+    for i, j in enumerate(ARM_JOINTS):
+        lo_deg, hi_deg = URDF_LIMITS_DEG[j]
+        ticks = sorted(int(round(TICK_REF + (d - REFERENCE_URDF_DEG[j]) / SCALE)) for d in (lo_deg, hi_deg))
+        calib[j] = MotorCalibration(id=i + 1, drive_mode=0, homing_offset=TICK_REF,
+                                    range_min=ticks[0], range_max=ticks[1])
+    calib["gripper"] = MotorCalibration(id=4, drive_mode=0, homing_offset=0,
+                                        range_min=GRIP_RANGE[0], range_max=GRIP_RANGE[1])
+    return calib
+
+
+class _SimBus:
+    """Ideal servo bus: Present_Position instantly follows the last commanded Goal_Position."""
+
+    def __init__(self, ticks: dict[str, float]):
+        self.ticks = ticks
+
+    def read(self, _reg, motor, normalize=False):
+        return self.ticks[motor]
+
+    def write(self, _reg, motor, value, normalize=False):
+        self.ticks[motor] = float(value)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--speed", type=float, default=ee.CART_STEP_M)
+    parser.add_argument("--frame", choices=("local", "global"), default="local")
+    parser.add_argument("--solver", choices=("dq", "pos"), default="dq")
+    args = parser.parse_args()
+
+    from importlib.resources import files
+    urdf_path = str(files("lerobot_robot_ax_arm") / "urdf" / "ax_arm.urdf")
+    kin = ee._build_kinematics(urdf_path)
+    link_fks = ee.build_link_fks(urdf_path)
+    calib = _synthetic_calibration()
+
+    q_start_deg = np.array([REFERENCE_URDF_DEG[j] for j in ARM_JOINTS])  # reference "zero" pose (0, 45, 90)
+    start_ticks = urdf_vector_to_ticks(q_start_deg, calib)
+    ticks = {j: float(start_ticks[j]) for j in ARM_JOINTS}
+    ticks["gripper"] = float(sum(GRIP_RANGE) / 2)
+    bus = _SimBus(ticks)
+
+    pending = {"x": 0.0, "y": 0.0, "z": 0.0, "g": 0.0}
+    state = {"frame": args.frame, "solver": args.solver}
+    keymap = {"w": ("x", 1), "s": ("x", -1), "a": ("y", 1), "d": ("y", -1), "r": ("z", 1), "f": ("z", -1),
+              "o": ("g", 1), "c": ("g", -1)}
+
+    fig = plt.figure(figsize=(7, 6))
+    ax = fig.add_subplot(projection="3d")
+    (chain_line,) = ax.plot([], [], [], "-o", lw=3, color="tab:blue")
+    (ee_pt,) = ax.plot([], [], [], "o", ms=10, color="tab:red")
+    reach = 0.32
+    ax.set_xlim(-reach, reach); ax.set_ylim(-reach, reach); ax.set_zlim(-0.05, reach)
+    ax.set_xlabel("X"); ax.set_ylabel("Y"); ax.set_zlabel("Z")
+
+    def on_key(event):
+        k = (event.key or "").lower()
+        if k in ("q", "escape"):
+            plt.close(fig)
+        elif k == "t":
+            state["frame"] = "global" if state["frame"] == "local" else "local"
+        elif k == "m":
+            state["solver"] = "pos" if state["solver"] == "dq" else "dq"
+        elif k in keymap:
+            axis, direction = keymap[k]
+            pending[axis] += direction
+
+    fig.canvas.mpl_connect("key_press_event", on_key)
+
+    def update(_frame):
+        q_rad = np.deg2rad(ticks_to_urdf_vector({j: ticks[j] for j in ARM_JOINTS}, calib))
+        cmd = np.array([np.sign(pending["x"]), np.sign(pending["y"]), np.sign(pending["z"])], dtype=float)
+        pending["x"] = pending["y"] = pending["z"] = 0.0
+
+        if np.any(cmd):
+            q_dot = ee._joint_velocity(kin, q_rad, cmd, state["frame"], state["solver"])
+            ee_speed = float(np.linalg.norm(np.array(kin["pos_jac"](q_rad)) @ q_dot))
+            if ee_speed > 1e-6:
+                q_step = q_dot * (args.speed / ee_speed)
+                step_norm = np.linalg.norm(q_step)
+                if step_norm > ee.MAX_JOINT_STEP_RAD:
+                    q_step *= ee.MAX_JOINT_STEP_RAD / step_norm
+                q_target = np.clip(q_rad + q_step, kin["lower"], kin["upper"])
+                target_ticks = urdf_vector_to_ticks(np.rad2deg(q_target), calib)
+                for j in ARM_JOINTS:
+                    c = calib[j]
+                    bus.write("Goal_Position", j, int(np.clip(target_ticks[j], c.range_min, c.range_max)))
+
+        g = np.sign(pending["g"]); pending["g"] = 0.0
+        if g:
+            gc = calib["gripper"]
+            bus.write("Goal_Position", "gripper",
+                      int(np.clip(ticks["gripper"] + g * ee.GRIP_STEP_TICK, gc.range_min, gc.range_max)))
+
+        pts = ee.chain_points(link_fks, np.deg2rad(ticks_to_urdf_vector({j: ticks[j] for j in ARM_JOINTS}, calib)))
+        chain_line.set_data(pts[:, 0], pts[:, 1]); chain_line.set_3d_properties(pts[:, 2])
+        ee_pt.set_data(pts[-1:, 0], pts[-1:, 1]); ee_pt.set_3d_properties(pts[-1:, 2])
+        grip_frac = (ticks["gripper"] - GRIP_RANGE[0]) / (GRIP_RANGE[1] - GRIP_RANGE[0])
+        ax.set_title(f"frame={state['frame']}  solver={state['solver']}  gripper={grip_frac:.0%}\n"
+                     f"w/s/a/d/r/f=move  o/c=grip  t=frame  m=solver  q=quit")
+        return chain_line, ee_pt
+
+    anim = FuncAnimation(fig, update, interval=int(1000 / ee.FPS), blit=False, cache_frame_data=False)
+    fig._anim = anim  # keep a reference so it isn't garbage-collected
+    plt.show()
+
+
+if __name__ == "__main__":
+    main()

lerobot_robot_ax_arm/examples/teleoperate_ee_keyboard.py

@@ -0,0 +1,321 @@
+#!/usr/bin/env python
+
+"""Keyboard end-effector teleoperation for the 4-DoF AX arm (velocity IK, position output).
+
+Adapted from a resolved-rate (twist) servoing controller: instead of commanding joint
+velocities, the per-tick joint velocity is integrated into a joint *position* target and sent as
+``Goal_Position`` (the AX arm runs in position mode).
+
+Each frame we read the raw motor ticks, map them to URDF joint angles, solve for a joint velocity
+that produces the requested Cartesian motion, scale it to a fixed end-effector Cartesian step
+``CART_STEP_M`` per tick (capped in joint space near singularities), and command ``Goal_Position``.
+
+Two IK solvers are selectable at runtime (``--solver`` / ``m`` key):
+  - "dq"  : dual-quaternion resolved-rate (matches the full pose velocity via ``scipy.least_squares``;
+            faithful to the source controller, but rotation/translation coupling on a 3-DoF arm
+            causes axis leakage),
+  - "pos" : position-only Jacobian ``dEE_pos/dq`` solved with least-squares (crisp axis-aligned
+            Cartesian motion).
+
+The motion frame is also selectable (``--frame`` / ``t`` key): "local" moves along the
+end-effector's own axes, "global" along the fixed world axes. Global always uses the position
+Jacobian (the dq solver's held-orientation constraint leaks axes on this underactuated arm).
+
+The tick<->URDF mapping is established once by ``lerobot-calibrate`` (reference pose + travel
+limits), so no separate alignment step is needed here.
+
+Controls (letter keys; hold to keep moving via terminal key-repeat):
+    - w / s : +X / -X   (forward / back)
+    - a / d : +Y / -Y   (left / right)
+    - r / f : +Z / -Z   (up / down)
+    - o / c : open / close gripper
+    - t     : toggle global <-> local frame
+    - m     : toggle dq <-> pos solver
+    - ESC / q : stop
+
+Run:
+    python examples/teleoperate_ee_keyboard.py --port /dev/tty.usbserial-XXXX --id my_ax_arm
+"""
+
+import argparse
+import time
+from importlib.resources import files
+
+import casadi as cs
+import numpy as np
+import scipy as sp
+from urdf2casadi import urdfparser as u2c
+from urdf2casadi.geometry import dual_quaternion, quaternion
+
+from lerobot.utils.keyboard_input import create_key_listener
+from lerobot.utils.robot_utils import precise_sleep
+
+from lerobot_robot_ax_arm import AXArm, AXArmConfig
+from lerobot_robot_ax_arm.urdf_mapping import (
+    ARM_JOINTS,
+    REFERENCE_URDF_DEG,
+    ticks_to_urdf_vector,
+    urdf_vector_to_ticks,
+)
+
+FPS = 30
+HOME_TIME_S = 2.0  # duration of the ramped move to the reference pose at startup
+CART_STEP_M = 0.008  # default end-effector Cartesian motion per tick, meters (override with --speed)
+MAX_JOINT_STEP_RAD = 0.15  # safety cap on joint motion per tick (keeps motion bounded near singularities)
+DLS_LAMBDA = 0.02  # damping factor for the position IK, well-behaved near singularities
+GRIP_STEP_TICK = 15  # gripper ticks per press
+FIT_THRESHOLD = 0.1  # only fit dual-quaternion derivative components above this magnitude
+
+ROOT_LINK = "base_link"
+TIP_LINK = "gripper_link"
+CHAIN_LINKS = ("robot_link_1", "robot_link_2", "robot_link_3", "gripper_link")
+
+
+def _skew(x: np.ndarray) -> np.ndarray:
+    return np.array([[0, -x[2], x[1]], [x[2], 0, -x[0]], [-x[1], x[0], 0]])
+
+
+def _build_kinematics(urdf_path: str) -> dict:
+    """FK + Jacobians (dual-quaternion and position-only) and joint limits from the URDF."""
+    parser = u2c.URDFparser()
+    parser.from_file(urdf_path)
+    fk = parser.get_forward_kinematics(ROOT_LINK, TIP_LINK)
+    q_sym = fk["q"]
+    fk_dq = fk["dual_quaternion_fk"]
+    fk_T = fk["T_fk"]
+    return {
+        "fk_dq": fk_dq,
+        "fk_T": fk_T,
+        "dq_jac": cs.Function("dq_jac", [q_sym], [cs.jacobian(fk_dq(q_sym), q_sym)]),
+        "pos_jac": cs.Function("pos_jac", [q_sym], [cs.jacobian(fk_T(q_sym)[:3, 3], q_sym)]),
+        "lower": np.array(fk["lower"], dtype=float).flatten(),
+        "upper": np.array(fk["upper"], dtype=float).flatten(),
+    }
+
+
+def build_link_fks(urdf_path: str):
+    """Casadi T_fk functions base->each link in CHAIN_LINKS, for drawing the arm."""
+    fks = []
+    for tip in CHAIN_LINKS:
+        parser = u2c.URDFparser()
+        parser.from_file(urdf_path)
+        fk = parser.get_forward_kinematics(ROOT_LINK, tip)
+        fks.append((fk["T_fk"], fk["q"].shape[0]))
+    return fks
+
+
+def chain_points(link_fks, q_rad: np.ndarray) -> np.ndarray:
+    """3D positions of the base and each link origin along the kinematic chain."""
+    pts = [np.zeros(3)]
+    for T, n in link_fks:
+        pts.append(np.array(T(q_rad[:n]))[:3, 3])
+    return np.array(pts)
+
+
+def open_live_view(link_fks, reach: float = 0.42):
+    """Open a non-blocking 3D plot; returns an update(q_rad, title) callback (False once closed)."""
+    import os
+    import tempfile
+
+    os.environ.setdefault("MPLCONFIGDIR", tempfile.gettempdir())
+    import matplotlib.pyplot as plt
+
+    fig = plt.figure(figsize=(7, 6))
+    ax = fig.add_subplot(projection="3d")
+    (chain_line,) = ax.plot([], [], [], "-o", lw=3, color="tab:blue")
+    (ee_pt,) = ax.plot([], [], [], "o", ms=10, color="tab:red")
+    ax.set_xlim(-reach, reach); ax.set_ylim(-reach, reach); ax.set_zlim(-0.05, reach)
+    ax.set_xlabel("X"); ax.set_ylabel("Y"); ax.set_zlabel("Z")
+    plt.ion(); plt.show(block=False)
+
+    def update(q_rad: np.ndarray, title: str) -> bool:
+        if not plt.fignum_exists(fig.number):
+            return False
+        pts = chain_points(link_fks, q_rad)
+        chain_line.set_data(pts[:, 0], pts[:, 1]); chain_line.set_3d_properties(pts[:, 2])
+        ee_pt.set_data(pts[-1:, 0], pts[-1:, 1]); ee_pt.set_3d_properties(pts[-1:, 2])
+        ax.set_title(title)
+        fig.canvas.draw_idle(); fig.canvas.flush_events()
+        return True
+
+    return update
+
+
+def _world_twist_to_dq_dot(pose: np.ndarray, twist_world: np.ndarray) -> np.ndarray:
+    """World-frame twist [v, w] -> dual-quaternion derivative x_dot at the current pose."""
+    T = dual_quaternion.to_numpy_transformation_matrix(pose)
+    adj = np.zeros((6, 6))
+    adj[:3, :3] = T[:3, :3]
+    adj[3:, 3:] = T[:3, :3]
+    adj[:3, 3:] = _skew(T[:3, -1]) @ T[:3, :3]
+    twist = adj @ twist_world
+
+    v = np.append(twist[:3], 0)
+    w = np.append(twist[3:], 0)
+    primal = pose[:4]
+    dual = pose[4:]
+    primal_conj = np.append(-primal[:3], primal[3])
+    p = 2 * quaternion.numpy_product(dual, primal_conj)
+
+    xi = np.zeros(8)
+    xi[:4] = np.append(0, twist[3:])
+    xi[4:] = v + (quaternion.numpy_product(p, w) - quaternion.numpy_product(w, p)) / 2
+    return 0.5 * dual_quaternion.numpy_product(xi, pose)
+
+
+def _fitness(q_dot, jacobian, x_dot):
+    index = np.where(np.abs(x_dot) > FIT_THRESHOLD)
+    delta = (np.dot(jacobian, q_dot) - x_dot)[index]
+    return np.dot(delta.T, delta)
+
+
+def _fitness_jacobian(q_dot, jacobian, x_dot):
+    return 2 * np.dot(jacobian.T, np.dot(jacobian, q_dot) - x_dot)
+
+
+def _joint_velocity(kin: dict, q_rad: np.ndarray, cmd: np.ndarray, frame: str, solver: str) -> np.ndarray:
+    """Joint velocity producing the requested unit Cartesian motion ``cmd`` (x, y, z)."""
+    rot = np.array(kin["fk_T"](q_rad))[:3, :3]
+    # World-frame ("global") translation on this 3-DoF (no-wrist) arm is only reliable via the position
+    # Jacobian: the dq resolved-rate tries to hold EE orientation fixed, which an underactuated arm
+    # cannot do, leaking the motion onto other axes as the arm reorients. So global always uses pos.
+    if solver == "pos" or frame == "global":
+        v_world = cmd.astype(float) if frame == "global" else rot @ cmd
+        J = np.array(kin["pos_jac"](q_rad))
+        # Damped least squares: bounded, well-conditioned joint velocity even near singularities.
+        return J.T @ np.linalg.solve(J @ J.T + DLS_LAMBDA**2 * np.eye(3), v_world)
+
+    # Dual-quaternion resolved-rate (local frame only): move along the end-effector's own axes.
+    lin = cmd.astype(float)
+    pose = np.array(kin["fk_dq"](q_rad)).flatten()
+    x_dot = _world_twist_to_dq_dot(pose, np.concatenate([lin, np.zeros(3)]))
+    jacobian = np.array(kin["dq_jac"](q_rad))
+    sol = sp.optimize.least_squares(
+        _fitness, np.zeros(len(ARM_JOINTS)), args=(jacobian, x_dot), xtol=1e-4, jac=_fitness_jacobian,
+    )
+    return sol.x if sol.success else np.zeros(len(ARM_JOINTS))
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", required=True, help="Serial port of the AX arm")
+    parser.add_argument("--id", default="my_ax_arm", help="Robot id used for calibration files")
+    parser.add_argument("--speed", type=float, default=CART_STEP_M,
+                        help="End-effector Cartesian motion per tick in meters (higher = faster)")
+    parser.add_argument("--frame", choices=("local", "global"), default="local",
+                        help="Motion frame: 'local' = end-effector axes, 'global' = world axes")
+    parser.add_argument("--solver", choices=("dq", "pos"), default="dq",
+                        help="IK solver: 'dq' = dual-quaternion resolved-rate, 'pos' = position Jacobian")
+    parser.add_argument("--view", action="store_true",
+                        help="Show a live 3D plot of the real arm (digital twin) alongside teleop")
+    args = parser.parse_args()
+    cart_step = args.speed
+
+    robot = AXArm(AXArmConfig(port=args.port, id=args.id, use_degrees=True))
+    robot.connect(calibrate=False)
+    if not robot.calibration:
+        raise RuntimeError(f"No calibration found for id '{args.id}'. Run lerobot-calibrate first.")
+
+    urdf_path = str(files("lerobot_robot_ax_arm") / "urdf" / "ax_arm.urdf")
+    kin = _build_kinematics(urdf_path)
+    q_lower, q_upper = kin["lower"], kin["upper"]
+    view_update = open_live_view(build_link_fks(urdf_path)) if args.view else None
+
+    grip_calib = robot.calibration["gripper"]
+    grip_tick = int(robot.bus.read("Present_Position", "gripper", normalize=False))
+
+    # Slowly ramp to the reference ("zero") pose (0, 45, 90) before starting teleop.
+    home_deg = np.array([REFERENCE_URDF_DEG[j] for j in ARM_JOINTS])
+    home_ticks = urdf_vector_to_ticks(home_deg, robot.calibration)
+    start_ticks = {j: float(robot.bus.read("Present_Position", j, normalize=False)) for j in ARM_JOINTS}
+    print("Homing to zero pose (0, 45, 90)...")
+    steps = max(1, int(HOME_TIME_S * FPS))
+    for i in range(1, steps + 1):
+        alpha = i / steps
+        for j in ARM_JOINTS:
+            c = robot.calibration[j]
+            tick = (1 - alpha) * start_ticks[j] + alpha * home_ticks[j]
+            robot.bus.write("Goal_Position", j, int(np.clip(tick, c.range_min, c.range_max)), normalize=False)
+        precise_sleep(1.0 / FPS)
+
+    pending = {"x": 0.0, "y": 0.0, "z": 0.0, "g": 0.0}
+    state = {"quit": False, "frame": args.frame, "solver": args.solver}
+    keymap = {"w": ("x", 1), "s": ("x", -1), "a": ("y", 1), "d": ("y", -1), "r": ("z", 1), "f": ("z", -1),
+              "o": ("g", 1), "c": ("g", -1)}
+
+    def on_key(name: str) -> None:
+        k = name.lower()
+        if k in ("esc", "q"):
+            state["quit"] = True
+        elif k == "t":
+            state["frame"] = "global" if state["frame"] == "local" else "local"
+        elif k == "m":
+            state["solver"] = "pos" if state["solver"] == "dq" else "dq"
+        elif k in keymap:
+            axis, direction = keymap[k]
+            pending[axis] += direction
+
+    listener = create_key_listener(
+        on_key, controls_help="w/s a/d r/f = XYZ, o/c = gripper, t = frame, m = solver, esc = stop"
+    )
+    if listener is None:
+        raise RuntimeError("Needs an interactive terminal with a usable key listener.")
+
+    print("Keyboard EE teleop (velocity IK -> position). w/s=X a/d=Y r/f=Z o/c=gripper, t=frame, m=solver, ESC=stop.")
+    try:
+        while not state["quit"]:
+            t0 = time.perf_counter()
+
+            ticks = {j: float(robot.bus.read("Present_Position", j, normalize=False)) for j in ARM_JOINTS}
+            q_deg = ticks_to_urdf_vector(ticks, robot.calibration)
+            q_rad = np.deg2rad(q_deg)
+
+            lin = np.array([np.sign(pending["x"]), np.sign(pending["y"]), np.sign(pending["z"])])
+            cmd_disp = lin.copy()
+            pending["x"] = pending["y"] = pending["z"] = 0.0
+
+            q_target_rad = q_rad
+            if np.any(lin):
+                q_dot = _joint_velocity(kin, q_rad, lin.astype(float), state["frame"], state["solver"])
+                # Scale for a consistent end-effector Cartesian speed (uniform across axes/poses),
+                # then cap the joint step so motion stays bounded near singularities.
+                ee_speed = float(np.linalg.norm(np.array(kin["pos_jac"](q_rad)) @ q_dot))
+                if ee_speed > 1e-6:
+                    q_step = q_dot * (cart_step / ee_speed)
+                    step_norm = np.linalg.norm(q_step)
+                    if step_norm > MAX_JOINT_STEP_RAD:
+                        q_step *= MAX_JOINT_STEP_RAD / step_norm
+                    q_target_rad = np.clip(q_rad + q_step, q_lower, q_upper)
+
+            target_ticks = urdf_vector_to_ticks(np.rad2deg(q_target_rad), robot.calibration)
+            for j in ARM_JOINTS:
+                c = robot.calibration[j]
+                tick = int(np.clip(target_ticks[j], c.range_min, c.range_max))
+                robot.bus.write("Goal_Position", j, tick, normalize=False)
+
+            g = np.sign(pending["g"])
+            pending["g"] = 0.0
+            if g:
+                grip_tick = int(np.clip(grip_tick + g * GRIP_STEP_TICK, grip_calib.range_min, grip_calib.range_max))
+                robot.bus.write("Goal_Position", "gripper", grip_tick, normalize=False)
+
+            urdf_str = " ".join(f"{j}={v:+6.1f}" for j, v in zip(ARM_JOINTS, np.rad2deg(q_target_rad)))
+            print(f"[{state['frame']:>6}|{state['solver']}] cmd[x={cmd_disp[0]:+.0f} y={cmd_disp[1]:+.0f}"
+                  f" z={cmd_disp[2]:+.0f} g={g:+.0f}] -> urdf[{urdf_str}] gripper={grip_tick}",
+                  end="\r", flush=True)
+
+            if view_update is not None:
+                # Draw the arm from the measured joint angles (real state, not the command).
+                view_update(q_rad, f"REAL arm | frame={state['frame']} solver={state['solver']}")
+
+            precise_sleep(max(1.0 / FPS - (time.perf_counter() - t0), 0.0))
+    except KeyboardInterrupt:
+        pass
+    finally:
+        listener.stop()
+        print()
+        robot.disconnect()
+
+
+if __name__ == "__main__":
+    main()

lerobot_robot_ax_arm/lerobot_robot_ax_arm/__init__.py

@@ -0,0 +1,4 @@
+from .ax_arm import AXArm
+from .config_ax_arm import AXArmConfig
+
+__all__ = ["AXArm", "AXArmConfig"]

lerobot_robot_ax_arm/lerobot_robot_ax_arm/ax_arm.py

@@ -0,0 +1,266 @@
+import logging
+import time
+from functools import cached_property
+
+from lerobot.cameras import make_cameras_from_configs
+from lerobot.motors import Motor, MotorCalibration, MotorNormMode
+from lerobot.motors.dynamixel import DynamixelMotorsBus
+from lerobot.robots import Robot
+from lerobot.robots.utils import ensure_safe_goal_position
+from lerobot.types import RobotAction, RobotObservation
+from lerobot.utils.decorators import check_if_already_connected, check_if_not_connected
+from lerobot.utils.keyboard_input import create_key_listener
+
+from .config_ax_arm import AXArmConfig
+from .urdf_mapping import ARM_JOINTS, REFERENCE_URDF_DEG, URDF_LIMITS_DEG
+
+logger = logging.getLogger(__name__)
+
+
+class AXArm(Robot):
+    """A 4-DoF arm driven by Dynamixel AX-series servos over Protocol 1.0.
+
+    Protocol 1.0 has no Sync Read, no Operating_Mode register and no homing offset, so this robot reads
+    positions sequentially and calibrates by recording the range of motion only.
+    """
+
+    config_class = AXArmConfig
+    name = "ax_arm"
+
+    def __init__(self, config: AXArmConfig):
+        super().__init__(config)
+        self.config = config
+        norm_mode_body = MotorNormMode.DEGREES if config.use_degrees else MotorNormMode.RANGE_M100_100
+        self.bus = DynamixelMotorsBus(
+            port=self.config.port,
+            # NOTE: dict order (pan, lift, elbow, gripper) must stay aligned with the URDF joints
+            # (robot_joint_1/2/3) and keep the gripper last for the kinematics pipeline. Only the
+            # motor IDs below reflect the physical bus wiring.
+            motors={
+                "shoulder_pan": Motor(3, "ax-12a", norm_mode_body),
+                "shoulder_lift": Motor(4, "ax-12a", norm_mode_body),
+                "elbow_flex": Motor(2, "ax-12a", norm_mode_body),
+                "gripper": Motor(1, "ax-12a", MotorNormMode.RANGE_0_100),
+            },
+            calibration=self.calibration,
+            protocol_version=1,
+        )
+        self.cameras = make_cameras_from_configs(config.cameras)
+
+    @property
+    def _motors_ft(self) -> dict[str, type]:
+        return {f"{motor}.pos": float for motor in self.bus.motors}
+
+    @property
+    def _cameras_ft(self) -> dict[str, tuple]:
+        return {cam: (self.cameras[cam].height, self.cameras[cam].width, 3) for cam in self.cameras}
+
+    @cached_property
+    def observation_features(self) -> dict[str, type | tuple]:
+        return {**self._motors_ft, **self._cameras_ft}
+
+    @cached_property
+    def action_features(self) -> dict[str, type]:
+        return self._motors_ft
+
+    @property
+    def is_connected(self) -> bool:
+        return self.bus.is_connected and all(cam.is_connected for cam in self.cameras.values())
+
+    @check_if_already_connected
+    def connect(self, calibrate: bool = True) -> None:
+        self.bus.connect()
+        if not self.is_calibrated and calibrate:
+            logger.info("No matching calibration found, running calibration.")
+            self.calibrate()
+
+        for cam in self.cameras.values():
+            cam.connect()
+
+        self.configure()
+        logger.info(f"{self} connected.")
+
+    @property
+    def is_calibrated(self) -> bool:
+        # Protocol 1.0 cannot read back homing_offset/drive_mode (we repurpose them to store the
+        # URDF mapping: tick_ref and sign), so only the range limits are actually stored on the
+        # servos. Compare just those to decide whether the on-file calibration matches the hardware.
+        if not self.calibration:
+            return False
+        hw = self.bus.read_calibration()
+        return all(
+            self.calibration[m].range_min == hw[m].range_min
+            and self.calibration[m].range_max == hw[m].range_max
+            for m in self.bus.motors
+        )
+
+    def calibrate(self) -> None:
+        if self.calibration:
+            user_input = input(
+                f"Press ENTER to use provided calibration file associated with the id {self.id}, or type 'c' and press ENTER to run calibration: "
+            )
+            if user_input.strip().lower() != "c":
+                logger.info(f"Writing calibration file associated with the id {self.id} to the motors")
+                self.bus.write_calibration(self.calibration)
+                return
+
+        logger.info(f"\nRunning calibration of {self}")
+
+        # This arm cannot be backdriven with torque off, so each joint is jogged with the keyboard
+        # (torque on). Arm joints capture a reference pose (known URDF angle) plus the lower/upper
+        # travel limits; the reference tick and mounting sign encode the URDF mapping (see urdf_mapping).
+        captured = self._record_calibration()
+
+        self.calibration = {}
+        for motor, m in self.bus.motors.items():
+            c = captured[motor]
+            if motor in ARM_JOINTS:
+                lower, upper = c["lower limit"], c["upper limit"]
+                # sign: +1 if jogging toward the URDF-upper limit increased ticks, else -1.
+                drive_mode = 0 if upper >= lower else 1
+                self.calibration[motor] = MotorCalibration(
+                    id=m.id,
+                    drive_mode=drive_mode,
+                    homing_offset=c["reference"],  # tick at the URDF reference pose
+                    range_min=min(lower, upper),
+                    range_max=max(lower, upper),
+                )
+            else:  # gripper: plain range, no URDF mapping
+                lo, hi = c["min"], c["max"]
+                self.calibration[motor] = MotorCalibration(
+                    id=m.id, drive_mode=0, homing_offset=0, range_min=min(lo, hi), range_max=max(lo, hi)
+                )
+
+        self.bus.write_calibration(self.calibration)
+        self._save_calibration()
+        print("Calibration saved to", self.calibration_fpath)
+
+    # Raw ticks the selected joint moves per keypress (~3° on an AX-12A).
+    _JOG_STEP = 10
+
+    def _capture_plan(self) -> list[tuple[str, str, float | None]]:
+        """Ordered (motor, label, target_urdf_deg) captures. Arm joints: reference + lower/upper
+        URDF limits; gripper: plain min/max. ``target_urdf_deg`` is None when there is no URDF angle."""
+        plan: list[tuple[str, str, float | None]] = []
+        for motor in self.bus.motors:
+            if motor in ARM_JOINTS:
+                lower, upper = URDF_LIMITS_DEG[motor]
+                plan.append((motor, "reference", REFERENCE_URDF_DEG[motor]))
+                plan.append((motor, "lower limit", lower))
+                plan.append((motor, "upper limit", upper))
+            else:
+                plan += [(motor, "min", None), (motor, "max", None)]
+        return plan
+
+    def _record_calibration(self) -> dict[str, dict[str, int]]:
+        # Protocol 1.0 has no Sync Read, so positions are read sequentially, one motor at a time.
+        motor_names = list(self.bus.motors)
+        max_pos = min(self.bus.model_resolution_table[m.model] for m in self.bus.motors.values()) - 1
+
+        # Temporarily widen the angle limits to full travel so pre-existing (possibly narrow) limits do
+        # not cap the reachable range; write_calibration() sets them to the recorded min/max afterwards.
+        for motor in motor_names:
+            self.bus.write("CW_Angle_Limit", motor, 0)
+            self.bus.write("CCW_Angle_Limit", motor, max_pos)
+        self.bus.enable_torque()
+
+        plan = self._capture_plan()
+        captured: dict[str, dict[str, int]] = {motor: {} for motor in motor_names}
+        state = {"i": 0, "target": 0, "capture": False, "done": False}
+        state["target"] = int(self.bus.read("Present_Position", plan[0][0], normalize=False))
+
+        def on_key(name: str) -> None:
+            key = name.lower()
+            if key == "up":
+                state["target"] = min(max_pos, state["target"] + self._JOG_STEP)
+            elif key == "down":
+                state["target"] = max(0, state["target"] - self._JOG_STEP)
+            elif key == "enter":
+                state["capture"] = True
+
+        listener = create_key_listener(on_key, controls_help="up/down=jog, enter=capture")
+        if listener is None:
+            raise RuntimeError(
+                "Keyboard calibration requires an interactive terminal with a usable key listener."
+            )
+
+        print(
+            "Jog each joint to the requested pose and press ENTER to capture it.\n"
+            "  up/down = jog | enter = capture"
+        )
+        try:
+            while not state["done"]:
+                motor, label, target_deg = plan[state["i"]]
+                self.bus.write("Goal_Position", motor, state["target"], normalize=False)
+                pos = int(self.bus.read("Present_Position", motor, normalize=False))
+                hint = f" (URDF {target_deg:+.0f} deg)" if target_deg is not None else ""
+                print(
+                    f"  [{state['i'] + 1}/{len(plan)}] jog '{motor}' to {label}{hint}: {pos:4d} tick   ",
+                    end="\r",
+                    flush=True,
+                )
+
+                if state["capture"]:
+                    state["capture"] = False
+                    captured[motor][label] = pos
+                    print(f"  [{state['i'] + 1}/{len(plan)}] {motor} {label}{hint} = {pos} tick")
+                    state["i"] += 1
+                    state["done"] = state["i"] >= len(plan)
+                    if not state["done"]:
+                        state["target"] = int(
+                            self.bus.read("Present_Position", plan[state["i"]][0], normalize=False)
+                        )
+                time.sleep(0.02)
+        finally:
+            listener.stop()
+            print()
+
+        for motor, c in captured.items():
+            if len(set(c.values())) < len(c):
+                raise ValueError(f"Motor '{motor}' has duplicate captured ticks: {c}")
+        return captured
+
+    def configure(self) -> None:
+        # AX-series has no Operating_Mode/PID registers; configure_motors only lowers the return delay time.
+        with self.bus.torque_disabled():
+            self.bus.configure_motors()
+
+    @check_if_not_connected
+    def get_observation(self) -> RobotObservation:
+        start = time.perf_counter()
+        # Protocol 1.0 has no Sync Read, so read each motor sequentially.
+        obs_dict = {f"{motor}.pos": self.bus.read("Present_Position", motor) for motor in self.bus.motors}
+        dt_ms = (time.perf_counter() - start) * 1e3
+        logger.debug(f"{self} read state: {dt_ms:.1f}ms")
+
+        for cam_key, cam in self.cameras.items():
+            obs_dict[cam_key] = cam.async_read()
+
+        return obs_dict
+
+    @check_if_not_connected
+    def send_action(self, action: RobotAction) -> RobotAction:
+        goal_pos = {
+            key.removesuffix(".pos"): val
+            for key, val in action.items()
+            if isinstance(key, str) and key.endswith(".pos")
+        }
+        if not goal_pos:
+            return {}
+
+        if self.config.max_relative_target is not None:
+            present_pos = {motor: self.bus.read("Present_Position", motor) for motor in goal_pos}
+            goal_present_pos = {key: (g_pos, present_pos[key]) for key, g_pos in goal_pos.items()}
+            goal_pos = ensure_safe_goal_position(goal_present_pos, self.config.max_relative_target)
+
+        # Sync Write is available on Protocol 1.0 (only Sync Read and Broadcast Ping are not).
+        self.bus.sync_write("Goal_Position", goal_pos)
+        return {f"{motor}.pos": val for motor, val in goal_pos.items()}
+
+    @check_if_not_connected
+    def disconnect(self) -> None:
+        self.bus.disconnect(self.config.disable_torque_on_disconnect)
+        for cam in self.cameras.values():
+            cam.disconnect()
+
+        logger.info(f"{self} disconnected.")

lerobot_robot_ax_arm/lerobot_robot_ax_arm/config_ax_arm.py

@@ -0,0 +1,24 @@
+from dataclasses import dataclass, field
+
+from lerobot.cameras import CameraConfig
+from lerobot.robots import RobotConfig
+
+
+@RobotConfig.register_subclass("ax_arm")
+@dataclass
+class AXArmConfig(RobotConfig):
+    """Configuration for a 4-DoF Dynamixel AX-series (Protocol 1.0) arm."""
+
+    # Serial port the arm is connected to (e.g. "/dev/tty.usbserial-AL02L1E0").
+    port: str
+
+    disable_torque_on_disconnect: bool = True
+
+    # Limits the magnitude of relative positional targets for safety. Scalar (same for all motors) or a
+    # dict mapping motor names to their own cap.
+    max_relative_target: float | dict[str, float] | None = None
+
+    cameras: dict[str, CameraConfig] = field(default_factory=dict)
+
+    # Normalize body joints to [-100, 100] (and gripper to [0, 100]) when False, or to degrees when True.
+    use_degrees: bool = False

lerobot_robot_ax_arm/lerobot_robot_ax_arm/urdf_mapping.py

@@ -0,0 +1,61 @@
+"""Mapping between AX motor ticks and the URDF joint convention.
+
+The AX-12A reports raw ticks (0-1023) over ~300 deg of travel, with a zero and direction
+unrelated to the URDF. We map ticks <-> URDF joint degrees with a per-joint affine transform
+anchored at a reference pose captured during calibration::
+
+    q_urdf_deg = q_ref + sign * (tick - tick_ref) * SCALE
+
+- ``SCALE``    : AX-12A mechanical travel per tick (300 deg / 1023 ticks).
+- ``q_ref``    : URDF angle of the reference pose (:data:`REFERENCE_URDF_DEG`).
+- ``tick_ref`` : tick captured at that reference pose, stored as ``MotorCalibration.homing_offset``.
+- ``sign``     : mounting direction (+1/-1), stored as ``MotorCalibration.drive_mode`` (0 -> +1, 1 -> -1).
+
+The ``homing_offset``/``drive_mode`` fields are unused by the Dynamixel Protocol-1.0 normalization
+path, so repurposing them here does not affect anything else.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+
+from lerobot.motors import MotorCalibration
+
+# Arm joints in URDF order (base yaw, shoulder pitch, elbow pitch). The gripper is not remapped.
+ARM_JOINTS = ("shoulder_pan", "shoulder_lift", "elbow_flex")
+URDF_JOINT_NAMES = ["robot_joint_1", "robot_joint_2", "robot_joint_3"]
+
+AX_TRAVEL_DEG = 300.0  # AX-12A mechanical travel over the full 0-1023 tick range
+AX_MAX_TICK = 1023.0
+SCALE = AX_TRAVEL_DEG / AX_MAX_TICK  # URDF degrees per motor tick
+
+# URDF joint angle (deg) of each arm joint at the calibration reference pose.
+REFERENCE_URDF_DEG = {"shoulder_pan": 0.0, "shoulder_lift": 45.0, "elbow_flex": 90.0}
+# URDF joint limits (deg) from ax_arm.urdf, used to guide the lower/upper jog during calibration.
+URDF_LIMITS_DEG = {
+    "shoulder_pan": (-90.0, 90.0),
+    "shoulder_lift": (0.0, 90.0),
+    "elbow_flex": (0.0, 180.0),
+}
+
+
+def _sign(calib: MotorCalibration) -> float:
+    return -1.0 if calib.drive_mode else 1.0
+
+
+def tick_to_urdf_deg(joint: str, tick: float, calib: MotorCalibration) -> float:
+    return REFERENCE_URDF_DEG[joint] + _sign(calib) * (tick - calib.homing_offset) * SCALE
+
+
+def urdf_deg_to_tick(joint: str, q_deg: float, calib: MotorCalibration) -> int:
+    return int(round(calib.homing_offset + _sign(calib) * (q_deg - REFERENCE_URDF_DEG[joint]) / SCALE))
+
+
+def ticks_to_urdf_vector(ticks: dict[str, float], calibration: dict[str, MotorCalibration]) -> np.ndarray:
+    """Arm joint ticks -> URDF joint angles (deg), in :data:`ARM_JOINTS` order."""
+    return np.array([tick_to_urdf_deg(j, ticks[j], calibration[j]) for j in ARM_JOINTS], dtype=float)
+
+
+def urdf_vector_to_ticks(q_deg: np.ndarray, calibration: dict[str, MotorCalibration]) -> dict[str, int]:
+    """URDF joint angles (deg), in :data:`ARM_JOINTS` order -> arm joint ticks."""
+    return {j: urdf_deg_to_tick(j, float(q_deg[i]), calibration[j]) for i, j in enumerate(ARM_JOINTS)}

lerobot_robot_ax_arm/pyproject.toml

@@ -0,0 +1,19 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "lerobot_robot_ax_arm"
+version = "0.1.0"
+description = "Third-party LeRobot robot: 4-DoF Dynamixel AX-series (Protocol 1.0) arm"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "lerobot[dynamixel]",
+]
+
+[tool.setuptools.packages.find]
+include = ["lerobot_robot_ax_arm*"]
+
+[tool.setuptools.package-data]
+lerobot_robot_ax_arm = ["urdf/*.urdf"]

src/lerobot/motors/dynamixel/dynamixel.py

@@ -25,7 +25,12 @@ from typing import TYPE_CHECKING
 
 from lerobot.utils.import_utils import _dynamixel_sdk_available, require_package
 
-from ..encoding_utils import decode_twos_complement, encode_twos_complement
+from ..encoding_utils import (
+    decode_sign_magnitude,
+    decode_twos_complement,
+    encode_sign_magnitude,
+    encode_twos_complement,
+)
 from ..motors_bus import Motor, MotorCalibration, NameOrID, SerialMotorsBus, Value, get_address
 from .tables import (
     AVAILABLE_BAUDRATES,
@@ -33,6 +38,7 @@ from .tables import (
     MODEL_CONTROL_TABLE,
     MODEL_ENCODING_TABLE,
     MODEL_NUMBER_TABLE,
+    MODEL_PROTOCOL,
     MODEL_RESOLUTION,
 )
 
@@ -41,7 +47,7 @@ if TYPE_CHECKING or _dynamixel_sdk_available:
 else:
     dxl = None
 
-PROTOCOL_VERSION = 2.0
+PROTOCOL_VERSION = 2
 DEFAULT_BAUDRATE = 1_000_000
 DEFAULT_TIMEOUT_MS = 1000
 
@@ -113,23 +119,47 @@ class DynamixelMotorsBus(SerialMotorsBus):
         port: str,
         motors: dict[str, Motor],
         calibration: dict[str, MotorCalibration] | None = None,
+        protocol_version: int = PROTOCOL_VERSION,
     ):
         require_package("dynamixel-sdk", extra="dynamixel", import_name="dynamixel_sdk")
         super().__init__(port, motors, calibration)
+        self.protocol_version = protocol_version
+        self._assert_same_protocol()
         self.port_handler = dxl.PortHandler(self.port)
-        self.packet_handler = dxl.PacketHandler(PROTOCOL_VERSION)
+        self.packet_handler = dxl.PacketHandler(protocol_version)
+        logger.debug(f"Using protocol version {protocol_version}")
         self.sync_reader = dxl.GroupSyncRead(self.port_handler, self.packet_handler, 0, 0)
         self.sync_writer = dxl.GroupSyncWrite(self.port_handler, self.packet_handler, 0, 0)
         self._comm_success = dxl.COMM_SUCCESS
         self._no_error = 0x00
 
+    def _assert_same_protocol(self) -> None:
+        if any(MODEL_PROTOCOL[model] != self.protocol_version for model in self.models):
+            raise ValueError(
+                f"Some motors are incompatible with protocol_version={self.protocol_version}. "
+                f"Expected per-model protocols: "
+                f"{ {model: MODEL_PROTOCOL[model] for model in self.models} }"
+            )
+
     def _assert_protocol_is_compatible(self, instruction_name: str) -> None:
-        pass
+        if instruction_name == "sync_read" and self.protocol_version == 1:
+            raise NotImplementedError(
+                "'Sync Read' is not available with Dynamixel motors using Protocol 1. Use 'Read' sequentially instead."
+            )
+        if instruction_name == "broadcast_ping" and self.protocol_version == 1:
+            raise NotImplementedError(
+                "'Broadcast Ping' is not available with Dynamixel motors using Protocol 1. Use 'Ping' sequentially instead."
+            )
 
     def _handshake(self) -> None:
         self._assert_motors_exist()
 
     def _find_single_motor(self, motor: str, initial_baudrate: int | None = None) -> tuple[int, int]:
+        if self.protocol_version == 1:
+            return self._find_single_motor_p1(motor, initial_baudrate)
+        return self._find_single_motor_p2(motor, initial_baudrate)
+
+    def _find_single_motor_p2(self, motor: str, initial_baudrate: int | None = None) -> tuple[int, int]:
         model = self.motors[motor].model
         search_baudrates = (
             [initial_baudrate] if initial_baudrate is not None else self.model_baudrate_table[model]
@@ -151,6 +181,29 @@ class DynamixelMotorsBus(SerialMotorsBus):
 
         raise RuntimeError(f"Motor '{motor}' (model '{model}') was not found. Make sure it is connected.")
 
+    def _find_single_motor_p1(self, motor: str, initial_baudrate: int | None = None) -> tuple[int, int]:
+        # Protocol 1.0 has no Broadcast Ping, so scan IDs sequentially with individual pings.
+        model = self.motors[motor].model
+        search_baudrates = (
+            [initial_baudrate] if initial_baudrate is not None else self.model_baudrate_table[model]
+        )
+        expected_model_nb = self.model_number_table[model]
+
+        for baudrate in search_baudrates:
+            self.set_baudrate(baudrate)
+            for id_ in range(dxl.MAX_ID + 1):
+                found_model = self.ping(id_)
+                if found_model is not None:
+                    if found_model != expected_model_nb:
+                        raise RuntimeError(
+                            f"Found one motor on {baudrate=} with id={id_} but it has a "
+                            f"model number '{found_model}' different than the one expected: '{expected_model_nb}'. "
+                            f"Make sure you are connected only connected to the '{motor}' motor (model '{model}')."
+                        )
+                    return baudrate, id_
+
+        raise RuntimeError(f"Motor '{motor}' (model '{model}') was not found. Make sure it is connected.")
+
     def configure_motors(self, return_delay_time=0) -> None:
         # By default, Dynamixel motors have a 500µs delay response time (corresponding to a value of 250 on
         # the 'Return_Delay_Time' address). We ensure this is reduced to the minimum of 2µs (value of 0).
@@ -162,6 +215,20 @@ class DynamixelMotorsBus(SerialMotorsBus):
         return self.calibration == self.read_calibration()
 
     def read_calibration(self) -> dict[str, MotorCalibration]:
+        if self.protocol_version == 1:
+            # AX-series (Protocol 1.0) has no Homing_Offset/Drive_Mode registers, uses CW/CCW angle
+            # limits as its position range, and does not support Sync Read.
+            calibration = {}
+            for motor, m in self.motors.items():
+                calibration[motor] = MotorCalibration(
+                    id=m.id,
+                    drive_mode=0,
+                    homing_offset=0,
+                    range_min=int(self.read("CW_Angle_Limit", motor, normalize=False)),
+                    range_max=int(self.read("CCW_Angle_Limit", motor, normalize=False)),
+                )
+            return calibration
+
         offsets = self.sync_read("Homing_Offset", normalize=False)
         mins = self.sync_read("Min_Position_Limit", normalize=False)
         maxes = self.sync_read("Max_Position_Limit", normalize=False)
@@ -181,9 +248,13 @@ class DynamixelMotorsBus(SerialMotorsBus):
 
     def write_calibration(self, calibration_dict: dict[str, MotorCalibration], cache: bool = True) -> None:
         for motor, calibration in calibration_dict.items():
-            self.write("Homing_Offset", motor, calibration.homing_offset)
-            self.write("Min_Position_Limit", motor, calibration.range_min)
-            self.write("Max_Position_Limit", motor, calibration.range_max)
+            if self.protocol_version == 1:
+                self.write("CW_Angle_Limit", motor, calibration.range_min)
+                self.write("CCW_Angle_Limit", motor, calibration.range_max)
+            else:
+                self.write("Homing_Offset", motor, calibration.homing_offset)
+                self.write("Min_Position_Limit", motor, calibration.range_min)
+                self.write("Max_Position_Limit", motor, calibration.range_max)
 
         if cache:
             self.calibration = calibration_dict
@@ -205,8 +276,12 @@ class DynamixelMotorsBus(SerialMotorsBus):
             model = self._id_to_model(id_)
             encoding_table = self.model_encoding_table.get(model)
             if encoding_table and data_name in encoding_table:
-                n_bytes = encoding_table[data_name]
-                ids_values[id_] = encode_twos_complement(ids_values[id_], n_bytes)
+                param = encoding_table[data_name]
+                # Protocol 1.0 (AX-series) uses sign-magnitude; Protocol 2.0 (X-series) two's complement.
+                if self.protocol_version == 1:
+                    ids_values[id_] = encode_sign_magnitude(ids_values[id_], param)
+                else:
+                    ids_values[id_] = encode_twos_complement(ids_values[id_], param)
 
         return ids_values
 
@@ -215,8 +290,12 @@ class DynamixelMotorsBus(SerialMotorsBus):
             model = self._id_to_model(id_)
             encoding_table = self.model_encoding_table.get(model)
             if encoding_table and data_name in encoding_table:
-                n_bytes = encoding_table[data_name]
-                ids_values[id_] = decode_twos_complement(ids_values[id_], n_bytes)
+                param = encoding_table[data_name]
+                # Protocol 1.0 (AX-series) uses sign-magnitude; Protocol 2.0 (X-series) two's complement.
+                if self.protocol_version == 1:
+                    ids_values[id_] = decode_sign_magnitude(ids_values[id_], param)
+                else:
+                    ids_values[id_] = decode_twos_complement(ids_values[id_], param)
 
         return ids_values
 
@@ -225,6 +304,11 @@ class DynamixelMotorsBus(SerialMotorsBus):
         On Dynamixel Motors:
         Present_Position = Actual_Position + Homing_Offset
         """
+        if self.protocol_version == 1:
+            raise NotImplementedError(
+                "AX-series motors (Protocol 1.0) have no Homing_Offset register; "
+                "calibrate the position range via CW/CCW angle limits instead."
+            )
         half_turn_homings: dict[NameOrID, Value] = {}
         for motor, pos in positions.items():
             model = self._get_motor_model(motor)
@@ -248,6 +332,7 @@ class DynamixelMotorsBus(SerialMotorsBus):
         return data
 
     def broadcast_ping(self, num_retry: int = 0, raise_on_error: bool = False) -> dict[int, int] | None:
+        self._assert_protocol_is_compatible("broadcast_ping")
         for n_try in range(1 + num_retry):
             data_list, comm = self.packet_handler.broadcastPing(self.port_handler)
             if self._is_comm_success(comm):

src/lerobot/motors/dynamixel/tables.py

@@ -33,6 +33,58 @@
 # 2. We can change the value of the MyControlTableKey enums without impacting the client code
 
 
+# {data_name: (address, size_byte)}
+# https://emanual.robotis.com/docs/en/dxl/ax/{MODEL}/#control-table
+AX_SERIES_CONTROL_TABLE = {
+    # EEPROM Area
+    "Model_Number": (0, 2),
+    "Firmware_Version": (2, 1),
+    "ID": (3, 1),
+    "Baud_Rate": (4, 1),
+    "Return_Delay_Time": (5, 1),
+    "CW_Angle_Limit": (6, 2),
+    "CCW_Angle_Limit": (8, 2),
+    "Temperature_Limit": (11, 1),
+    "Min_Voltage_Limit": (12, 1),
+    "Max_Voltage_Limit": (13, 1),
+    "Max_Torque": (14, 2),
+    "Status_Return_Level": (16, 1),
+    "Alarm_LED": (17, 1),
+    "Shutdown": (18, 1),
+    # RAM Area
+    "Torque_Enable": (24, 1),
+    "LED": (25, 1),
+    "CW_Compliance_Margin": (26, 1),
+    "CCW_Compliance_Margin": (27, 1),
+    "CW_Compliance_Slope": (28, 1),
+    "CCW_Compliance_Slope": (29, 1),
+    "Goal_Position": (30, 2),
+    "Moving_Speed": (32, 2),
+    "Torque_Limit": (34, 2),
+    "Present_Position": (36, 2),
+    "Present_Speed": (38, 2),
+    "Present_Load": (40, 2),
+    "Present_Voltage": (42, 1),
+    "Present_Temperature": (43, 1),
+    "Registered": (44, 1),
+    "Moving": (46, 1),
+    "Lock": (47, 1),
+    "Punch": (48, 2),
+}
+
+# https://emanual.robotis.com/docs/en/dxl/ax/{MODEL}/#baud-rate4
+AX_SERIES_BAUDRATE_TABLE = {
+    9_600: 207,
+    19_200: 103,
+    57_600: 34,
+    115_200: 16,
+    200_000: 9,
+    250_000: 7,
+    400_000: 4,
+    500_000: 3,
+    1_000_000: 1,
+}
+
 # {data_name: (address, size_byte)}
 # https://emanual.robotis.com/docs/en/dxl/x/{MODEL}/#control-table
 X_SERIES_CONTROL_TABLE = {
@@ -114,6 +166,15 @@ X_SERIES_ENCODINGS_TABLE = {
     "Present_Velocity": X_SERIES_CONTROL_TABLE["Present_Velocity"][1],
 }
 
+# {data_name: sign_bit_index}
+# AX-series speed/load use sign-magnitude encoding with the direction stored in bit 10.
+# Position registers (0-1023) are unsigned and therefore need no sign handling.
+AX_SERIES_ENCODINGS_TABLE = {
+    "Moving_Speed": 10,
+    "Present_Speed": 10,
+    "Present_Load": 10,
+}
+
 MODEL_ENCODING_TABLE = {
     "x_series": X_SERIES_ENCODINGS_TABLE,
     "xl330-m077": X_SERIES_ENCODINGS_TABLE,
@@ -122,6 +183,8 @@ MODEL_ENCODING_TABLE = {
     "xm430-w350": X_SERIES_ENCODINGS_TABLE,
     "xm540-w270": X_SERIES_ENCODINGS_TABLE,
     "xc430-w150": X_SERIES_ENCODINGS_TABLE,
+    "ax_series": AX_SERIES_ENCODINGS_TABLE,
+    "ax-12a": AX_SERIES_ENCODINGS_TABLE,
 }
 
 # {model: model_resolution}
@@ -134,6 +197,8 @@ MODEL_RESOLUTION = {
     "xm430-w350": 4096,
     "xm540-w270": 4096,
     "xc430-w150": 4096,
+    "ax_series": 1024,
+    "ax-12a": 1024,
 }
 
 # {model: model_number}
@@ -145,10 +210,13 @@ MODEL_NUMBER_TABLE = {
     "xm430-w350": 1020,
     "xm540-w270": 1120,
     "xc430-w150": 1070,
+    "ax-12a": 12,
 }
 
 # {model: available_operating_modes}
 # https://emanual.robotis.com/docs/en/dxl/x/{MODEL}/#operating-mode11
+# Note: AX-series (Protocol 1.0) has no Operating_Mode register; joint vs wheel mode is
+# selected via the CW/CCW angle limits, so it is intentionally absent from this table.
 MODEL_OPERATING_MODES = {
     "xl330-m077": [0, 1, 3, 4, 5, 16],
     "xl330-m288": [0, 1, 3, 4, 5, 16],
@@ -166,6 +234,8 @@ MODEL_CONTROL_TABLE = {
     "xm430-w350": X_SERIES_CONTROL_TABLE,
     "xm540-w270": X_SERIES_CONTROL_TABLE,
     "xc430-w150": X_SERIES_CONTROL_TABLE,
+    "ax_series": AX_SERIES_CONTROL_TABLE,
+    "ax-12a": AX_SERIES_CONTROL_TABLE,
 }
 
 MODEL_BAUDRATE_TABLE = {
@@ -176,6 +246,22 @@ MODEL_BAUDRATE_TABLE = {
     "xm430-w350": X_SERIES_BAUDRATE_TABLE,
     "xm540-w270": X_SERIES_BAUDRATE_TABLE,
     "xc430-w150": X_SERIES_BAUDRATE_TABLE,
+    "ax_series": AX_SERIES_BAUDRATE_TABLE,
+    "ax-12a": AX_SERIES_BAUDRATE_TABLE,
+}
+
+# {model: protocol_version}
+# AX series communicate over Protocol 1.0, X series over Protocol 2.0.
+MODEL_PROTOCOL = {
+    "x_series": 2,
+    "xl330-m077": 2,
+    "xl330-m288": 2,
+    "xl430-w250": 2,
+    "xm430-w350": 2,
+    "xm540-w270": 2,
+    "xc430-w150": 2,
+    "ax_series": 1,
+    "ax-12a": 1,
 }
 
 AVAILABLE_BAUDRATES = [

tests/motors/test_dynamixel.py

@@ -107,6 +107,39 @@ def test_abc_implementation(dummy_motors):
     DynamixelMotorsBus(port="/dev/dummy-port", motors=dummy_motors)
 
 
+def test_assert_same_protocol_rejects_mismatch():
+    """AX-series motors (Protocol 1.0) cannot be used with the default Protocol 2.0."""
+    motors = {"ax": Motor(5, "ax-12a", MotorNormMode.RANGE_M100_100)}
+    with pytest.raises(ValueError, match="incompatible with protocol_version"):
+        DynamixelMotorsBus(port="/dev/dummy-port", motors=motors)
+
+
+@pytest.mark.parametrize("instruction", ["sync_read", "broadcast_ping"])
+def test_protocol1_unsupported_instructions(instruction):
+    motors = {"ax": Motor(5, "ax-12a", MotorNormMode.RANGE_M100_100)}
+    bus = DynamixelMotorsBus(port="/dev/dummy-port", motors=motors, protocol_version=1)
+    with pytest.raises(NotImplementedError):
+        bus._assert_protocol_is_compatible(instruction)
+
+
+@pytest.mark.parametrize(
+    "protocol_version, model, data_name, value",
+    [
+        (1, "ax-12a", "Present_Speed", -300),
+        (1, "ax-12a", "Moving_Speed", 300),
+        (2, "xl430-w250", "Present_Velocity", -300),
+    ],
+    ids=["ax-sign-magnitude-neg", "ax-sign-magnitude-pos", "x-twos-complement-neg"],
+)
+def test_sign_encoding_roundtrip(protocol_version, model, data_name, value):
+    motors = {"m": Motor(5, model, MotorNormMode.RANGE_M100_100)}
+    bus = DynamixelMotorsBus(port="/dev/dummy-port", motors=motors, protocol_version=protocol_version)
+    encoded = bus._encode_sign(data_name, {5: value})
+    assert encoded[5] >= 0
+    decoded = bus._decode_sign(data_name, encoded)
+    assert decoded[5] == value
+
+
 @pytest.mark.parametrize("id_", [1, 2, 3])
 def test_ping(id_, mock_motors, dummy_motors):
     expected_model_nb = MODEL_NUMBER_TABLE[dummy_motors[f"dummy_{id_}"].model]