Robot Control

import itertools
import time
from asimov_protocol.v1.edge_cloud_pb2 import CloudCommand, VelocityCommand

# Monotonic sequence counter shared by all examples on this page.
_seq = itertools.count(1)
def seq() -> int:
    return next(_seq)


cmd = CloudCommand(
    timestamp_us=int(time.time() * 1e6),
    sequence=seq(),
    velocity=VelocityCommand(vx=0.5, vy=0.0, vyaw=0.0)
)
await room.local_participant.publish_data(
    cmd.SerializeToString(), topic="commands", reliable=True
)

import time
from asimov_protocol.v1.edge_cloud_pb2 import (
    CloudCommand, TrajectoryRequest, FullTrajectory, JointSegment, EdgeTelemetry
)

# Keep a reference to the latest telemetry frame; update it from your
# telemetry subscription handler (see "Telemetry" below).
latest_telemetry: EdgeTelemetry | None = None

# ... once `latest_telemetry` has been populated by your subscription handler:
assert latest_telemetry is not None, "wait for the first telemetry frame"

# Seed positions from the latest telemetry pose so unspecified joints
# hold their current value; override only the ones you're driving.
positions = list(latest_telemetry.joint_pos)   # 25 floats
positions[12] = 0.5           # L_Shoulder_Pitch
positions[13] = -0.2          # L_Shoulder_Roll
positions[15] = 1.2           # L_Elbow

cmd = CloudCommand(
    timestamp_us=int(time.time() * 1e6),
    sequence=seq(),
    trajectory=TrajectoryRequest(
        full=FullTrajectory(segments=[
            JointSegment(positions=positions)
        ])
    )
)
await room.local_participant.publish_data(
    cmd.SerializeToString(), topic="commands", reliable=True
)

import time
from asimov_protocol.v1.edge_cloud_pb2 import CloudCommand, ModeCommand, Mode


# Stand up
cmd = CloudCommand(
    timestamp_us=int(time.time() * 1e6),
    sequence=seq(),
    mode=ModeCommand(mode=Mode.MODE_STAND)
)

# Request software DAMP (not a hardware emergency stop)
cmd = CloudCommand(
    timestamp_us=int(time.time() * 1e6),
    sequence=seq(),
    mode=ModeCommand(mode=Mode.MODE_DAMP)
)

import asyncio
from asimov_protocol.v1.edge_cloud_pb2 import EdgeTelemetry


async def read_telemetry(track):
    stream = track.subscribe()
    async for frame in stream:
        t = EdgeTelemetry.FromString(frame.payload)
        print(f"Mode: {t.fw_mode}, Joints: {list(t.joint_pos)}")


# Subscribe to the robot's telemetry DataTrack
@room.on("data_track_published")
def on_data_track(track):
    asyncio.create_task(read_telemetry(track))