Connect
The three lanes into the robot, the credential store, and the menlo command.
Describe the robot's wires once in a ConnectionConfig, bind a Robot to it, then pick the
lane at connect(). robot.py does not know which wire it is on: verbs, waits and errors are
identical everywhere.
connect(mode) | control + state | video + audio | needs a LiveKit server | config slots |
|---|---|---|---|---|
"udp" | UDP 8850 / 8851 | — | no | udp=UdpConfig(host) |
"hybrid" | UDP 8850 / 8851 | LiveKit tracks | yes | udp=… and livekit=… |
"livekit" | LiveKit data channel | LiveKit tracks | yes | livekit=… |
The same bytes travel on every lane: one asimov.io.RobotCommand per datagram or packet out,
one asimov.io.RobotState in. Every lane lands in the edge's arbiter and passes the same
safety layer.
Zero Config
from menlo.asimov import Robot
with Robot().connect() as robot: ...Robot() resolves, in order: MENLO_MANAGER_URL + MENLO_CREDENTIAL in the environment;
then the default entry in ~/.menlo/robots.toml (MENLO_ROBOT picks a named one); else a
ConnectError naming both. A manager URL and credential is a one-lane config, so the mode is
livekit and connect() needs no argument.
The store is written by menlo login, or by connect(persist=True) (or MENLO_PERSIST=1)
after a connect that worked. MENLO_HOME moves the directory. The file is 0600.
menlo login http://<robot> --credential <credential> # validate, then save
menlo robots # what is saved, default first
menlo use menlo-0001 # the default robot
menlo logout menlo-0001 # forget oneUDP: On the LAN, No Server
from menlo.asimov import ConnectionConfig, Robot, UdpConfig
cfg = ConnectionConfig(udp=UdpConfig(host="asimov.local"))
with Robot(cfg).connect("udp") as robot: ...The edge must run with --udp-control and push state to this machine
(--udp-state-host <your ip>). Both machines need to reach each other on ports 8850 and
8851. This lane carries drive and state only: robot.camera raises UnsupportedError.
The edge pushes state to one destination, so one client at a time on this lane.
Hybrid: UDP Control, Media Over the Room
from menlo.asimov import ConnectionConfig, ManagerConfig, Robot, UdpConfig
cfg = ConnectionConfig(
udp=UdpConfig(host="asimov.local"),
livekit=ManagerConfig(url="http://asimov.local", credential=CRED),
)
with Robot(cfg).connect("hybrid") as robot:
jpeg = robot.camera.photo(timeout=5.0).to_jpeg()ManagerConfig asks the robot's manager for the room name and a fresh join token on every
connect, so no LiveKit token or secret is ever in a script. The manager URL is whatever the
robot's web UI answers on; no port is assumed.
LiveKit: Everything Over the Room
from menlo.asimov import ConnectionConfig, ManagerConfig, Robot
cfg = ConnectionConfig(livekit=ManagerConfig(url="http://asimov.local", credential=CRED))
with Robot(cfg).connect() as robot: ... # one lane, so the mode is impliedFor a robot that is not routable from your machine. Commands go on the reliable data topic
commands; state comes back on the data track state, one frame per sample, in order.
Your Own LiveKit Server
from menlo.asimov import LiveKitConfig
livekit=LiveKitConfig(url="wss://sfu.example", room="robot-0001", token=mint)token is a join token, or a callable that mints a fresh one per join. The SDK has no
api_key, api_secret or identity parameter: a participant's identity is a claim inside the
token, and the SDK reads it back as robot.info.endpoint. One token is one participant; two
sessions in a room need two tokens.
Two scripts on one credential
With ManagerConfig every session gets its own identity, sdk-<credential id>-<host>-<random>, so two scripts on one credential coexist. ManagerConfig(label="agent") fixes the suffix when you want a recognisable name; two sessions with the same label then evict each other.
Switching and Options
robot = Robot(cfg) # bound, no network yet
cfg.available_modes() # ("udp", "hybrid", "livekit")
robot.connect("hybrid", timeout=5.0) # returns robot once the first state arrived
robot.close(); robot.connect("udp") # switch lanes on the same Robot; limits and hooks kept
robot.connect("livekit", require_state=False) # media now; verbs unlock when the firmware reports
robot.connect("livekit", persist=True) # save URL + credential after successrequire_state=False opens the media lane without waiting for the firmware: camera,
microphone and speaker work at once, and robot.state and every motion verb raise
NotConnectedError until the robot reports, then unlock on their own.
Next
How is this guide?