Run a policy

uv add dreamscale

export DREAMSCALE_API_KEY="dreamscale_sk_..."

import dreamscale

observation = dreamscale.so101.observe(
    side_frame=side_rgb,    # H×W×3 uint8 RGB
    wrist_frame=wrist_rgb,  # H×W×3 uint8 RGB
    joint_positions=joints, # 6 values, LeRobot degrees
)

with dreamscale.connect("molmoact2-so101") as policy:
    result = policy.predict(observation, instruction="pick up the red block")

print(len(result.actions), "actions of", len(result.actions[0]), "values")
print(f"{result.timing.obs_to_action_ms:.0f} ms")

import time

with dreamscale.connect("molmoact2-so101") as policy:
    period = 1.0 / policy.action_hz
    while running:
        start = time.perf_counter()
        action = policy.next_action(read_observation(), instruction="pick up the red block")
        send_to_robot(action)
        time.sleep(max(0.0, period - (time.perf_counter() - start)))

def observe(context):
    return read_observation()


def act(action, context):
    send_to_robot(action)
    return False  # return True to stop early


with dreamscale.connect("molmoact2-so101") as policy:
    result = policy.run(
        instruction="pick up the red block",
        observe=observe,
        act=act,
        max_actions=300,
    )

print(result.actions, "actions,", result.underruns, "underruns")

dreamscale.connect(
    "pi05-yam",
    keep_warm=0,            # release the GPU on close (default: keep it 300 s, billed)
    speed="fast",           # faster GPU where the model offers it
    startup_timeout=600,    # give up if a cold start takes longer
)