Skip to content

Submit a Job

A good first job is a smoke test: connect to a real arm, read its joint positions, nudge it a few degrees, put it back, and return what it saw. It exercises the whole path — image build, upload, scheduling, robot control — in about a minute, and it is safe to run against any SO-101 cell.

The three files below are complete. Drop them in an empty directory and run.

1. The runtime program

This is the code that runs inside the container, on the cell. The @main decorator marks your entry point; the platform calls it with a Context describing the robot it gave you.

so101_smoke.py
"""Move an SO-101 a few degrees and put it back."""

from __future__ import annotations

import time
from typing import Any

from armnet_runtime import Context, main
from lerobot.robots.so_follower import SO101Follower, SO101FollowerConfig

NUDGE_DEGREES = 8.0
STEPS = 30
HZ = 20


def joint_positions(observation: dict[str, Any]) -> dict[str, float]:
    """Keep the numeric joint entries, drop camera frames and anything else."""

    positions: dict[str, float] = {}
    for key, value in observation.items():
        if not key.endswith("pos"):
            continue
        # Depending on the LeRobot version these arrive as plain floats or as
        # numpy scalars, which have .item().
        value = value.item() if hasattr(value, "item") else value
        if isinstance(value, (int, float)):
            positions[key] = float(value)
    return positions


def glide(robot: SO101Follower, start: dict[str, float], target: dict[str, float]) -> None:
    """Interpolate from one pose to another so the arm never jumps."""

    for step in range(1, STEPS + 1):
        alpha = step / STEPS
        robot.send_action(
            {key: start[key] + (target[key] - start[key]) * alpha for key in target}
        )
        time.sleep(1 / HZ)


@main
def run(ctx: Context) -> dict:
    # The cell tells you where the robot is and which calibration to use. In a
    # managed cell robot_port is a connector endpoint, not a local serial device.
    robot_id, calibration_dir = ctx.cell.prepare_calibration_dir()
    robot = SO101Follower(
        SO101FollowerConfig(
            port=ctx.cell.robot_port,
            id=robot_id,
            calibration_dir=calibration_dir,
            # Caps how far a single command may move a joint. The cell supplies
            # its own limit; keeping it is what makes an unfamiliar arm safe.
            max_relative_target=ctx.cell.safety_limit,
        )
    )

    ctx.report_progress("connecting")
    robot.connect(calibrate=False)
    try:
        if not robot.is_calibrated:
            raise RuntimeError("this SO-101 is not calibrated")

        start = joint_positions(robot.get_observation())
        if not start:
            raise RuntimeError("no joint positions in the observation")

        ctx.report_progress(f"nudging every joint {NUDGE_DEGREES} degrees")
        target = {key: value + NUDGE_DEGREES for key, value in start.items()}
        glide(robot, start, target)
        time.sleep(1)

        ctx.report_progress("returning to the starting pose")
        glide(robot, target, start)

        return {"joints": sorted(start), "start_pose": start}
    finally:
        if robot.is_connected:
            robot.disconnect()

2. The image

Your program ships as a Docker image. Three details in the install are worth copying rather than reinventing:

  • pynput and evdev are Linux input packages LeRobot pulls in for local teleoperation. A container never needs them and they often fail to build, so the overrides file makes them impossible to resolve.
  • torch is pinned to a CPU build. Left alone, the resolver happily pulls a multi-gigabyte CUDA wheel that this job has no use for.
  • LeRobot goes in its own step, last, because it pins many transitive dependencies and will fight anything installed alongside it.
Dockerfile
FROM python:3.12-slim

ENV PYTHONUNBUFFERED=1

WORKDIR /app

RUN pip install --upgrade pip uv && \
    pip install armnet-runtime && \
    printf 'pynput>=0.0.0; sys_platform == "never"\n\
evdev>=0.0.0; sys_platform == "never"\n\
torch==2.10.0+cpu\n' > /tmp/overrides.txt && \
    uv pip install --system \
        --override /tmp/overrides.txt \
        --extra-index-url https://download.pytorch.org/whl/cpu \
        --index-strategy unsafe-best-match \
        'lerobot[feetech]==0.6.0'

COPY so101_smoke.py /app/so101_smoke.py

CMD ["armnet-runtime", "/app/so101_smoke.py"]

The container only needs armnet-runtime, never armnet-client. LeRobot 0.5.x works here too if your code needs it — swap the version and nothing else changes.

3. Build, push, and run it

This part runs on your machine.

submit.py
from armnet_client import Image, execute

image = Image.build(
    dockerfile="Dockerfile",
    context_dir=".",
    name="my-so101-smoke",
).push()

result = execute(
    image=image,
    embodiment="lerobot/so-101",
    task="push_green_button",
    timeout_seconds=600,
)

result.raise_for_status()
print(result.return_value)

python submit.py

You should see the image build and upload, then container logs streaming back as the job runs, then something like:

{'joints': ['elbow_flex.pos', 'gripper.pos', 'shoulder_lift.pos',
            'shoulder_pan.pos', 'wrist_flex.pos', 'wrist_roll.pos'],
 'start_pose': {'shoulder_pan.pos': 0.4, ...}}

Image.build() hashes its inputs, so a rebuild with unchanged files reuses the Docker cache and the push is a no-op. Image.push() uploads using short-lived credentials issued by the orchestrator — you never handle registry passwords.

Cell Selection

embodiment and task together select the managed robot cell that runs your job. The orchestrator routes the job to a cell advertising that pair, and rejects an unknown embodiment or task with HTTP 400.

embodiment="lerobot/so-101"      # or lerobot/bimanual_yam, lerobot/arx5
task="push_green_button"

The smoke test above ignores the task — it just moves the arm — but the task still decides which cell you land on. push_green_button is a BusyBox task, so it routes to a BusyBox cell that is available around the clock, which is what makes it a good choice for a first run. Manual tasks are staged and judged by an operator within a 7-day window instead. See Environments and Tasks for the full list.

Adding Cameras

The smoke test asks for no cameras, which keeps the observation to joint positions. To receive frames, hand the cell's camera configs to LeRobot:

SO101FollowerConfig(..., cameras=ctx.camera_configs)

Each camera then appears in robot.get_observation() as an image array alongside the joint values.

Each camera read is its own round trip to the cell. For a fast control loop, ArmnetSO101Follower gives the same observation in one round trip; see Armnet Robots.

Logs and Cancellation

By default, execute() streams container stdout/stderr back to your terminal. If the log connection drops and detach=False, the platform requests that the cell gracefully stops the job.

Use:

execute(..., detach=True)

for jobs that should keep running after your client disconnects, then check on them later with armnet jobs list and armnet jobs get <job-id>. Follow a running job's live logs, including one you submitted from another machine, with armnet jobs logs <job-id>. That only works for your own jobs, and it shows lines from when you connect onward; earlier lines are not stored yet.

Next Steps