Menlo
Python SDK

Reference

The public surface of menlo-sdk: Robot, configs, Sent, State, media, transports and errors.

Everything here is importable from menlo.asimov. menlo.__version__ is the distribution version. Units are metres per second, radians per second, radians, degrees Celsius.

Jump To

Do thisUse
say where the robot isConnectionConfig · Robot
walk, strafe, turnset_velocity · stop
wake it, or make it limpstand · damp
pose a jointgoto · trajectory
wait for somethingwait_for · wait_until
read what it reportsState · Robot.state
check what it can dohas / require · RobotInfo
see what actually went outSent · Outcomes
pictures and soundMedia
record a runrecord
save a robot for next timeStore
handle failureErrors

ConnectionConfig

ConnectionConfig(udp: UdpConfig | None = None, livekit: ManagerConfig | LiveKitConfig | None = None)
Method
available_modes() -> tuple[ConnectMode, ...]which of "udp", "hybrid", "livekit" this config can reach
only_mode() -> ConnectModethe one lane; ValueError with several
ConnectionConfig.from_environment()from MENLO_MANAGER_URL + MENLO_CREDENTIAL
Slot
UdpConfig(host, command_port=8850, state_bind=("0.0.0.0", 8851), state_source=None)the LAN lane. state_source pins the one address state may come from
ManagerConfig(url, credential, label=None, timeout=5.0)the robot's manager mints the room and a join token per connect. url needs neither scheme nor port. .host
LiveKitConfig(url, room, token)details you already hold; token is a string or a callable minting one per join

Robot

Robot(source: ConnectionConfig | Transport | None = None, *, limits=None, link_timeout=2.0)

None resolves the robot from the environment, else the store.

connect

connect(mode=None, *, timeout=5.0, media_timeout=3.0, connect_timeout=10.0,
        allow_version_skew=False, require_state=True, persist=None) -> Robot

Attach on "udp", "hybrid" or "livekit"; mode may be left out when the config has one lane. Returns once the first State has arrived and its protocol version matches. Raises ConnectError before touching the network when the config lacks a slot the mode needs. require_state=False returns as soon as the lane is open. persist=True saves the manager URL and credential to the store after success. close() then connect() again to switch lanes.

open(*, timeout=5.0, allow_version_skew=False, require_state=True) does the same over a Transport passed to the constructor.

Verbs — Each Returns a Sent Immediately

set_velocity

set_velocity(vx=0.0, vy=0.0, vyaw=0.0, *, duration=None, wait=False) -> Sent

Walk. Clamped to limits; held at 10 Hz until superseded, until stop(), or until duration seconds pass (then zero is sent). Returns at once; wait=True blocks until the hold has ended and its zero has gone out, or another verb superseded it. ValueError on a non-finite value or a non-positive duration. See Move the Robot.

stop

stop() -> Sent

Zero velocity. Stays in MOVE, still balancing: where a walk should end.

stand

stand() -> Sent

Stiffen into the standing pose and hold it. One-shot. Required on the way out of DAMP: a velocity sent to a DAMPed robot is dropped by the edge. No balance loop — see Safety.

damp

damp() -> Sent

Motors compliant now. One-shot. Overwritten if something is streaming setpoints — see Safety. A software state, not an emergency stop.

goto

goto(positions, *, duration=2.0, hz=50.0, kp=None, kd=None,
     wait=True, tolerance=0.05, timeout=None) -> Sent

Interpolate (minimum jerk) from the current reported pose to positions over duration, clocking trajectory() setpoints from a thread, then hold the target until another verb. Any verb from any thread stops the goto thread before its next setpoint leaves. Refuses to plan from a pose older than 0.5 s. wait=True blocks until every joint is within tolerance rad. See Control Joints Directly.

trajectory

trajectory(positions, *, kp=None, kd=None) -> Sent

One setpoint, radians, firmware order — clock these yourself or use goto. ValueError unless len(positions) == info.dof. A kp/kd of 0 isn't "no gain": the firmware substitutes its damping constants, i.e. limp. The edge DAMPs a trajectory about two seconds after the last setpoint.

Waits — Read the Robot's Own Report

wait_for

wait_for(mode, *, timeout=10.0, stale_after=None) -> State

Blocks until state.mode is mode and returns that State. Raises WaitTimeoutError, StateStaleError or RobotFaultedError. A quiet link raises StateStaleError, not a plain timeout — "never got there" and "stopped talking" are different failures.

wait_until

wait_until(predicate, *, timeout=10.0, stale_after=None, poll=0.05) -> State

Blocks until predicate(state) is true and returns that first State. Same errors as wait_for. stale_after defaults to link_timeout.

Properties and Callbacks

NameTypeNotes
configConnectionConfigwhat this Robot was bound to
infoRobotInfotransport, endpoint, dof, joint names, protocol version, limits, capabilities
has(cap) / require(*caps)bool / raises UnsupportedErrorthe transport's live set: drive, state, battery, camera, microphone, speaker
camera / microphone / speakerCamera / Microphone / Speakersee Media; UnsupportedError on a lane that does not carry them
record(path)Recordingcontext manager: every state sample and every command as JSON lines
stateStatethe latest sample; NotConnectedError before the first
connectedboolopen and not LinkLostError
outcomes()Iterator[Refused]drain received refusals, oldest first
on_stateCallable[[State], None] | Noneevery accepted sample, on the transport thread
on_alertCallable[[Alert, "raised" | "cleared"], None] | Nonean alert appearing or disappearing
on_mode_changeCallable[[Mode, Mode], None] | Noneprevious, current
on_refusedCallable[[Refused], None] | None
on_link_lostCallable[[LinkLostError], None] | None
close()zero velocity if held, drop the link; idempotent; never raises

Robot is a context manager; __exit__ calls close().

Sent

MemberTypeMeaning
namestrthe verb: set_velocity, stop, stand, damp, trajectory
sequenceint-1 if the command was superseded before it was sent
commandVelocity | ModeCommand | Trajectorywhat was actually sent, after clamping
clampedboolcommand differs from what you asked
sent_atfloattime.monotonic()
outcomeApplied | Refused | Nonenon-blocking; None = pending
wait_outcome(timeout=None)Applied | Refused | UnknownNone uses the transport default
require(timeout=None, *, unknown_ok=True)as aboveraises CommandRefusedError; with unknown_ok=False also OutcomeUnknownError

Outcomes

TypeFields
Appliedsequence
Refusedsequence, reason: Refusal, detail: str (diagnostic text; never branch on it)
Unknownsequence, waited_s

Refusal (IntEnum): UNSPECIFIED, FW_DAMPED, FAULT_DAMPED, GATE, UNSUPPORTED_SOURCE, BAD_LENGTH, EMPTY_TRAJECTORY, NON_FINITE, PARSE, UNKNOWN_COMMAND, UNKNOWN_MODE, SHUTTING_DOWN, NO_CAMERA, NOT_ACTIVE, UNRECOGNIZED. Property retryable.

State

@dataclass(frozen=True)
class State:
    mode: Mode                                   # DAMP | STAND | MOVE | UNKNOWN
    joints: tuple[Joint, ...]                    # firmware order
    gravity: tuple[float, float, float] | None   # body frame; z ≈ -1 upright
    gyro: tuple[float, float, float] | None      # rad/s
    quat: tuple[float, float, float, float] | None   # w, x, y, z
    error_flags: int
    alerts: tuple[Alert, ...]
    sequence: int
    fw_timestamp_us: int
    protocol_version: int
    battery: Battery | None                      # None when the robot reports no BMS
    received_at: float
    edge_timestamp_us: int                       # the edge's clock; 0 on the UDP lane
    # properties
    age_s: float
    upright: bool | None
    euler: tuple[float, float, float] | None     # roll, pitch, yaw (rad) from quat
    yaw: float | None                            # rad, counter-clockwise, (-π, π]
    faulted: bool                                # error_flags or any critical alert
    joint_pos: tuple[float, ...]
    def joint(self, name: str) -> Joint          # KeyError on an unknown name

Joint(name, pos, vel, current, temp) — vel, current and temp are None when the robot did not report them (never a guessed zero). Alert(id, severity, value, threshold, source_id, first_set_us); alert.critical is severity == 0. Battery(voltage_v, current_a, soc_percent, max_cell_temp_c, protection: BatteryProtection) with charging and protecting.

RobotInfo(transport, endpoint, dof, joint_names, protocol_version, limits, capabilities) with joint_index(name) and has(capability).

Limits(vx=0.6, vy=0.6, vyaw=1.5).

Media

robot.camera.photo(timeout=5.0) -> Frame        # ONE fresh frame; WaitTimeoutError when quiet
robot.camera.latest() -> Frame | None           # newest frame; None before the first
robot.camera.frames(timeout=5.0)                # iterator, always the newest frame
robot.camera.subscribe(cb)                      # cb(Frame) on the transport thread
robot.camera.capture_clip(seconds, audio=True) -> Clip
robot.microphone.chunks(timeout=5.0)            # every AudioChunk in order (bounded; .dropped counts loss)
robot.speaker.play(chunk); robot.speaker.play_pcm(pcm_s16le, sample_rate_hz=16000, channels=1)

Frame(width, height, encoding, data, stride_bytes, key_frame, frame_id, timestamp_ns, sequence) with shape, age_s, to_numpy(), to_jpeg(quality=85). AudioChunk(sample_rate_hz, channels, samples_per_channel, encoding, data, stream_id, timestamp_ns, sequence) with duration_s, to_numpy(). Clip(frames, audio, started_at) with duration_s, fps, save_wav(path), save_frames(dir), save_mp4(path, fps=None), frames_as_numpy(). numpy, Pillow and OpenCV are needed only at the call that uses them, and named in the ImportError.

Recording

with robot.record("run.jsonl") as rec:
    ...
rec.samples, rec.commands_written
from menlo.asimov import recording
for line in recording.load("run.jsonl"): ...   # dicts with kind "state" | "sent"

Store and CLI

~/.menlo/robots.toml (MENLO_HOME moves it), 0600 in a 0700 directory.

RobotStore()get(name=None), put(...), remove(name), use(name), names
StoredRobotname, manager_url, credential, default; .manager(label=), .connection(label=)
menlo login <url> [--credential] [--name]validates against the manager, then saves
menlo robots / menlo use <name> / menlo logout <name>list, set default, forget
menlo --version

Environment: MENLO_MANAGER_URL, MENLO_CREDENTIAL, MENLO_ROBOT, MENLO_PERSIST, MENLO_HOME.

Transports

UdpTransport(host, *, command_port, state_bind, state_source), LiveKitTransport(url, room, *, token, media_timeout, connect_timeout), HybridTransport(host, *, livekit_url, room, token, command_port, state_bind, ...). All implement Transport; a ConnectionConfig builds the right one for the mode. LiveKitTransport.identity and .room read back what the token claimed.

Errors

MenloError
├── ConnectError
│   └── ProtocolMismatchError(expected, observed)
├── NotConnectedError
├── LinkLostError
├── WaitTimeoutError(last)            # also TimeoutError
│   └── StateStaleError
├── RobotFaultedError(state)
├── CommandRefusedError(refused)
├── OutcomeUnknownError(unknown)      # also TimeoutError
└── UnsupportedError(capability, transport)

Caller mistakes are builtins: ValueError (non-finite velocity, limit or duration, wrong trajectory length, a lone kp or kd, a mode the config cannot reach), KeyError (unknown joint name), RuntimeError (open() on an open robot).

Threading

Robot is synchronous and thread-safe. Callbacks — on_state, on_alert, on_mode_change, on_refused, on_link_lost — fire on a background thread. Keep them short, and do not call back into the robot from inside one.

Per-Robot Tables

menlo.asimov.robots.ASIMOV_1_BIPED_JOINTS is the 25-motor order of the asimov_1_biped firmware profile, used to name joints until the robot reports them itself. menlo.asimov.robots.PROTOCOL_VERSION is the asimov.io version the SDK was built against; connect() compares it with what the robot echoes.

How is this guide?

On this page