Stream joint positions to an arm

MoveThroughJointPositionsStreamed opens a stream onto which you push joint trajectory points while the arm is already moving: the arm executes the points it has while you keep appending. The other joint-space methods take the whole motion up front, the configurations to hit and optional ceilings on how fast to get there. The arm module picks the motion profile from there.

Streaming hands the arm a time-parameterized trajectory: be at this configuration at this time, moving at this velocity. You produce that schedule.

Reach for streaming when the trajectory is produced as the motion runs: a teleoperation feed, a visual-servoing loop, a force-feedback correction on a surface-finishing path, or a trajectory long enough that you want to keep only part of it in memory.

MoveThroughJointPositionsStreamed is safety-heartbeat monitored: if the session that last called it stops sending heartbeats, viam-server stops the arm. A client that dies mid-trajectory leaves the arm stopped.

Prerequisites

• A configured arm component and an SDK client.
• An arm module that implements streaming. Streaming is currently supported for the viam:ufactory and viam:universal-robots modules.
• A source of trajectory points that stays within the arm’s joint limits.

What a trajectory point contains

A trajectory point carries the following:

• Time, measured from the start of the motion. The first point must be zero, and every point after it must be strictly later than the one before.
• Positions, one value per joint, matching the arm’s degrees of freedom.
• Constraints, optional and set per point. A point either carries no constraints at all or carries a velocity for every joint, with accelerations an optional addition on top. The arm starts from rest, so the first point is the arm standing still: either leave constraints off that point, or give every joint a velocity of zero. Only velocities have to be zero there; an acceleration on the first point is allowed.

Each SDK uses different units for positions and constraints. For reference, see Units.

How batching works

A batch is a request of one or more trajectory points. Point times are offsets from the start of the motion, so the same trajectory sent whole and sent in three batches produces the same motion. Batching controls delivery; the point times control the arm.

Batch size controls how far ahead of the arm you commit. A point you have sent is final: points cannot be replaced or revoked. Large batches cost fewer round trips and leave more of the trajectory queued if your producer falls behind, at the price of a longer committed stretch. Small batches keep the last committed point close to where the arm is now, so a fresh sensor reading can still change the next move.

Stream a trajectory

Each SDK exposes the same stream through a different control flow. Python takes an async iterator and gives you one back. Go hands you two channels that you own.

from datetime import timedelta

from viam.components.arm import Arm

my_arm = Arm.from_robot(machine, "my-arm")

# Times are offsets from the start of the motion; positions are degrees.
first_batch = [
    Arm.TrajectoryPoint(
        time=timedelta(0),
        positions=[0, -45, 90, 0, 45, 0],
        # Velocities on the t=0 point must be zero.
        constraints=Arm.KinematicConstraints(velocities=[0, 0, 0, 0, 0, 0]),
    ),
    Arm.TrajectoryPoint(
        time=timedelta(milliseconds=500),
        positions=[0, -22.5, 90, 0, 22.5, 0],
    ),
    Arm.TrajectoryPoint(
        time=timedelta(seconds=1),
        positions=[0, 0, 90, 0, 0, 0],
    ),
]


async def batches():
    # Yield a list per batch. Returning ends the trajectory.
    yield first_batch


async for update in my_arm.move_through_joint_positions_streamed(batches()):
    # Updates arrive as the arm executes. Stopping this loop early closes
    # the stream.
    pass

import (
    "math"
    "time"

    "go.viam.com/rdk/components/arm"
    "go.viam.com/rdk/referenceframe"
)

batches := make(chan []arm.TrajectoryPoint)
responses := make(chan arm.Response)

// The arm is free to acknowledge nothing at all, so this goroutine drains the
// channel rather than tracking progress.
go func() {
    for range responses {
    }
}()

go func() {
    defer close(batches)

    // Give up if the call returns early, so this goroutine never blocks on a
    // channel nobody is reading.
    send := func(b []arm.TrajectoryPoint) bool {
        select {
        case batches <- b:
            return true
        case <-ctx.Done():
            return false
        }
    }

    // Ten waypoints, 100ms apart, sent five at a time.
    batch := make([]arm.TrajectoryPoint, 0, 5)
    for i := 0; i < 10; i++ {
        batch = append(batch, arm.TrajectoryPoint{
            Time: time.Duration(i*100) * time.Millisecond,
            // Revolute joint values are radians, matching referenceframe.Input.
            Positions: []referenceframe.Input{
                0, -math.Pi/4 + float64(i)*math.Pi/40, math.Pi / 2, 0, math.Pi / 4, 0,
            },
        })
        if len(batch) == 5 {
            if !send(batch) {
                return
            }
            batch = make([]arm.TrajectoryPoint, 0, 5)
        }
    }
    if len(batch) > 0 {
        send(batch)
    }
}()

// Blocks until the arm finishes the trajectory, the stream fails, or another
// operation cancels it.
err := myArm.MoveThroughJointPositionsStreamed(ctx, batches, responses, nil)
close(responses)
if err != nil {
    logger.Fatal(err)
}

Troubleshooting

What’s next

• Move by joint positions: the unary methods, for trajectories you have in hand before the arm moves.
• Move an arm to a pose: Cartesian motion with obstacle avoidance through motion.Move.
• Arm kinematics: the kinematic file that declares joint limits.