Scheduling Waypoints

Controllers and skills send sparse, timestamped targets. The Module owns a TrajectoryInterpolator, upsamples at its frequency, and writes the dense stream each control_step. That is the same UMI pattern used before UR servoL — only the interpolator type and write target change.

Controller / skill  (~10–30 Hz)
        │  SCHEDULE_L / SCHEDULE_J  fields + wall-clock timestamp
        ▼
Module  (frequency: 90–120, interpolator_type: pose | scalar)
        │  sample interpolator @ module rate
        │  write targets
        ▼
Actuator bus  (IK joints, servo, sim joint, …)

Yaml selects the interpolator with interpolator_type. ArmModule handles SCHEDULE_L (pose) and SCHEDULE_J (joints); GripperModule handles SCHEDULE_J on the scalar interpolator. Subclasses usually only declare actuators (and optional unit conversion).

Contract

1. Set interpolator_type and max_speed on the module yaml.
2. Run the module at a higher frequency than the controller (upsample rate).
3. Emit ActionType.SCHEDULE_L (Cartesian) or SCHEDULE_J (joints / gripper) with field values and a wall-clock timestamp.
4. Stamp targets in the future. schedule_waypoint drops waypoints with time <= curr_time (UMI behaviour).

Do not run a second interpolator in the skill/controller for the same stream — that double-smooths and adds lag. Forward policy waypoints straight onto the module.

Go-to and streaming moves are different action types: MOVE_L / MOVE_J (moveL / moveJ, duration from max_speed) and SERVO_L / SERVO_J (adaptive drive horizon). See Motion actions.

Yaml

frequency: 90
interpolator_type: pose   # or scalar
max_speed: [0.25, 0.6]    # meaning depends on type (see below)

Example: arm (pose)

Cartesian MinkIK arms use interpolator_type: pose. Controllers schedule x..rz; the module samples a 6D pose and writes the Cartesian actuators; IKActuatorBus solves joints.

Module + yaml

Declare six CartesianIKActuators and a MinkIK backend. No custom schedule handler:

class ArmModule(Module):
    shoulder_pan = FeetechServo()
    # ... other joints ...
    x = CartesianIKActuator()
    y = CartesianIKActuator()
    z = CartesianIKActuator()
    rx = CartesianIKActuator()
    ry = CartesianIKActuator()
    rz = CartesianIKActuator()

    ik_backend = MinkIK(
        mjcf="assets/so106.xml",
        joints={...},
        frame="tcp",
    )

# conf/modules/right_arm.yaml
frequency: 90
interpolator_type: pose
max_speed: [0.25, 0.6]   # m/s, rad/s

See Adding a new arm for the full SO-106 contract.

Scheduling from a controller

from solarian.core.action import Action, ActionType

target_time = time.time() + 1.0 / controller_fps
data = {
    axis: {"value": float(pose[i]), "timestamp": target_time}
    for i, axis in enumerate(("x", "y", "z", "rx", "ry", "rz"))
}
return Action(
    action_type=ActionType.MODULE_ACTION,
    data={
        "right_arm": {
            "action_type": ActionType.SCHEDULE_L,
            "data": data,
        }
    },
)

Each module tick: sample PoseTrajectoryInterpolator → write x..rz → IK.

Example: gripper (scalar)

Grippers use the same action type with interpolator_type: scalar. The interpolator works in millimetres of aperture; GripperModule converts to the servo's normalized [-100, 100] at the write and seed edges, so a subclass only declares which actuator to drive.

Module + yaml

class UmiSlimModule(GripperModule):
    scalar_actuator = "gripper"   # field name on SCHEDULE_J
    gripper = FeetechServo()

# conf/modules/right_gripper.yaml
frequency: 120
interpolator_type: scalar
max_speed: 150.0           # mm/s aperture rate limit
aperture_table:            # written by `solarian calibrate-gripper-ranges`
    norm: [-100.0, -50.0, 0.0, 50.0, 100.0]
    aperture_mm: [0.0, 8.0, 25.0, 55.0, 85.0]

The mm → normalized conversion follows aperture_table when it is present, and a straight line between aperture_closed_mm and aperture_open_mm otherwise.

Scheduling from a controller

Use the same future timestamp as the arm — a stamp of time.time() is often already in the past when the module receives it, and those waypoints are dropped:

target_time = time.time() + 1.0 / controller_fps
gripper_action = {
    "action_type": ActionType.SCHEDULE_J,
    "data": {
        "gripper": {"value": aperture_mm, "timestamp": target_time},
    },
}
return Action(
    action_type=ActionType.MODULE_ACTION,
    data={
        "right_arm": arm_action,
        "right_gripper": gripper_action,
    },
)

Each module tick: sample ScalarTrajectoryInterpolator → _scalar_to_actuator → write_actuators.

Policy / skill path

Diffusion (and similar) policies emit a chunk of absolute target times. For each step, push one SCHEDULE_L onto the arm and one SCHEDULE_J onto the gripper with that wall-clock time (minus action latency). Module-side interpolators own the dense stream — skills should not keep a parallel PoseTrajectoryInterpolator / ScalarTrajectoryInterpolator for the same modules.

Pitfalls

• Past timestamps are ignored. Always schedule ahead of curr_time (teleop: one controller period; policies: absolute future times).
• Units must match max_speed. Pose speeds are m/s and rad/s; scalar speed is in the same units as the waypoint value (mm/s for an aperture).
• One interpolator owns the stream. Controller/skill = sparse setpoints; module = upsample + write.
• Override control_step carefully. Call super().control_step() if you still want core pose/scalar sampling, or replace it entirely for latch-only modules.

See also

• Modules — process loop, config, handlers
• Motion actions — SCHEDULE_L / SCHEDULE_J
• Adding a new arm — full MinkIK pose contract (SO-106)
• Inverse Kinematics — Cartesian actuators and IKActuatorBus