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 this | Use |
|---|---|
| say where the robot is | ConnectionConfig · Robot |
| walk, strafe, turn | set_velocity · stop |
| wake it, or make it limp | stand · damp |
| pose a joint | goto · trajectory |
| wait for something | wait_for · wait_until |
| read what it reports | State · Robot.state |
| check what it can do | has / require · RobotInfo |
| see what actually went out | Sent · Outcomes |
| pictures and sound | Media |
| record a run | record |
| save a robot for next time | Store |
| handle failure | Errors |
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() -> ConnectMode | the 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) -> RobotAttach 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) -> SentWalk. 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() -> SentZero velocity. Stays in MOVE, still balancing: where a walk should end.
stand
stand() -> SentStiffen 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() -> SentMotors 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) -> SentInterpolate (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) -> SentOne 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) -> StateBlocks 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) -> StateBlocks until predicate(state) is true and returns that first State. Same errors as
wait_for. stale_after defaults to link_timeout.
Properties and Callbacks
| Name | Type | Notes |
|---|---|---|
config | ConnectionConfig | what this Robot was bound to |
info | RobotInfo | transport, endpoint, dof, joint names, protocol version, limits, capabilities |
has(cap) / require(*caps) | bool / raises UnsupportedError | the transport's live set: drive, state, battery, camera, microphone, speaker |
camera / microphone / speaker | Camera / Microphone / Speaker | see Media; UnsupportedError on a lane that does not carry them |
record(path) | Recording | context manager: every state sample and every command as JSON lines |
state | State | the latest sample; NotConnectedError before the first |
connected | bool | open and not LinkLostError |
outcomes() | Iterator[Refused] | drain received refusals, oldest first |
on_state | Callable[[State], None] | None | every accepted sample, on the transport thread |
on_alert | Callable[[Alert, "raised" | "cleared"], None] | None | an alert appearing or disappearing |
on_mode_change | Callable[[Mode, Mode], None] | None | previous, current |
on_refused | Callable[[Refused], None] | None | |
on_link_lost | Callable[[LinkLostError], None] | None | |
close() | zero velocity if held, drop the link; idempotent; never raises |
Robot is a context manager; __exit__ calls close().
Sent
| Member | Type | Meaning |
|---|---|---|
name | str | the verb: set_velocity, stop, stand, damp, trajectory |
sequence | int | -1 if the command was superseded before it was sent |
command | Velocity | ModeCommand | Trajectory | what was actually sent, after clamping |
clamped | bool | command differs from what you asked |
sent_at | float | time.monotonic() |
outcome | Applied | Refused | None | non-blocking; None = pending |
wait_outcome(timeout=None) | Applied | Refused | Unknown | None uses the transport default |
require(timeout=None, *, unknown_ok=True) | as above | raises CommandRefusedError; with unknown_ok=False also OutcomeUnknownError |
Outcomes
| Type | Fields |
|---|---|
Applied | sequence |
Refused | sequence, reason: Refusal, detail: str (diagnostic text; never branch on it) |
Unknown | sequence, 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 nameJoint(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 |
StoredRobot | name, 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?