python api

worlds and lifecycle

every environment owns a list of things:

from botblocks import Box, MujocoEnv, Plane, Robot

robot = Robot().load('public/so101')
env = MujocoEnv([Plane(), Box(size=0.03), robot]).start()

Thing accepts pos=[x, y, z], rot=[roll, pitch, yaw], and an optional name. Plane adds size and color; Box adds cube edge size and color.

@robot.ready
def setup(bot, env):
    pass                    # once at start and again after reset

@robot.loop
def tick(bot, env):
    pass                    # repeatedly, synchronized to frames

useful world methods:

env.time()                  # seconds since this run started
env.next_frame()            # wait for the next frame
env.reset(wait=True)        # reset things, subsystems, and physics
env.quit()                  # stop the script

thing.pose                  # live Pose(pos, rot)
thing.set_pos([x, y, z])    # move a thing immediately

start(wait=True, view=True, loop=True) normally blocks and runs lifecycle callbacks. wait=False returns the environment. view=False creates a headless env. loop=False is for callers such as GymEnv that drive simulation steps themselves.

environments

MujocoEnv: physics, contacts, free bodies, sensors, and simulated actuators
ViewerEnv: physics-free kinematics; servo targets snap to their poses and motors integrate as angular velocity
LocalEnv: providers read and command hardware on the local machine

MujocoEnv.drive(on) toggles realtime physics. step_sim(n=1, ctrl=None) advances a manually driven simulation.

robots

robot = Robot().load('public/so101')
robot = Robot().load('menagerie/unitree_go2')
robot = Robot().load('my-project/my-robot')

robot['shoulder']            # named subsystem
robot['goal'] = Command(n=3) # add an in-script subsystem

use the gallery or project page to inspect a robot's configured subsystem names. code should not assume that every robot has an arm, drive, camera, or imu.

actuators and sensors

servo.target(angle)          # radians
servo.read()
servo.velocity()

motor.power(value)           # normally -1..1
motor.read()
motor.velocity()

encoder.read()
encoder.velocity()

drive.power(fwd, turn=0)     # DiffDrive

imu.gravity()                # body-frame gravity direction
imu.pos()                    # world position
imu.vel()                    # body-frame linear velocity
imu.vel_world()              # world-frame linear velocity
imu.ang_vel()                # body-frame angular velocity
imu.ang()                    # roll, pitch, yaw
imu.accel()
imu.quat()

camera.snap()                # raw rgba BytesId, also appears in snaps
camera.snap_img()            # rgb PIL.Image
subsystem.pose               # world pose of its mounted body

a camera needs an attached viewer or the host renderer before it can answer a snapshot.

arm ik

an Arm is a composite over a configured servo chain:

arm.move([dx, dy, dz])
arm.move([dx, dy, dz], aim=[ax, ay, az])
arm.goto([x, y, z])
arm.goto([x, y, z], rpy=[roll, pitch, yaw])
arm.fk()                     # current 4x4 tool transform
arm.describe()               # joint angles and tool pose

move is relative and refuses components larger than 6 cm. goto uses absolute world coordinates. both return a refused: ... string and emit a warning when the requested pose is not reachable; check the result and verify with describe().

keyboard input

Keyboard exposes the currently held keys. KeyboardCounter integrates key pairs into numeric axes:

keys = KeyboardCounter(pairs=['qa', 'ws'], init=[0, -0.5], rate=0.8)

@robot.loop
def drive(bot, env):
    shoulder, elbow = keys.get_axes()
    bot['shoulder'].target(shoulder)
    bot['elbow'].target(elbow)

ViewerEnv([Plane(), robot, keys]).start()

reinforcement learning

GymEnv turns selected subsystems into a gymnasium observation and action space. subsystems with observe() become observations; those with act() become actions.

from botblocks import GymEnv, Plane, Robot
import math

furuta = Robot().load('public/furuta')

@furuta.loop
def reward(bot, env):
    angle = bot['encoder'].read()
    env.reward(-math.cos(angle))

GymEnv([Plane(), furuta], subsys=[furuta['encoder'], furuta['servo']]).train(
    n=32, shards=4, steps=1_000_000, save='/tmp/furuta'
)

training uses sharded stable-baselines3 ppo. metrics stream to the workspace. rl works on every container size; medium and gpu boxes provide more compute:

botblocks deploy train.py --size medium

play a saved policy with:

policy = Policy('/tmp/furuta', subsys=[furuta['encoder'], furuta['servo']], vecnorm='/tmp/furuta.pkl')

@furuta.loop
def act(bot, env):
    policy.step()

hardware providers

LocalEnv([robot], providers=[ServoSTS]).start()

ServoSTS auto-detects a feetech sts bus and assigns ids in subsystem order. CameraUSB(camera_id) provides a local usb camera. custom providers can implement the same subsystem interfaces.