Overview

LIVEKIT_ENABLE_DATA_TRACKS=true livekit-server --dev --bind 0.0.0.0

livekit-cli create-token \
    --api-key devkey --api-secret secret \
    --join --room robot --identity my-client \
    --valid-for 24h

import asyncio
import time
from livekit import rtc
from edge.generated.edge_cloud_pb2 import (
    CloudCommand, VelocityCommand, ModeCommand, Mode, EdgeTelemetry
)

LIVEKIT_URL = "ws://localhost:7880"
LIVEKIT_TOKEN = "your-token"
seq = 0


def next_seq():
    global seq
    seq += 1
    return seq


async def main():
    room = rtc.Room()
    await room.connect(LIVEKIT_URL, LIVEKIT_TOKEN)
    print(f"Connected to room: {room.name}")

    # Subscribe to robot's video track
    @room.on("track_subscribed")
    def on_track(track, publication, participant):
        if isinstance(track, rtc.RemoteVideoTrack):
            pass  # handle video frames

    # Subscribe to robot's telemetry DataTrack
    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)[:5]}...")

    @room.on("data_track_published")
    def on_data_track(track):
        asyncio.create_task(read_telemetry(track))

    # Stand up first (robot boots in DAMP).
    # Note: command Mode and telemetry FirmwareMode use different numbering -
    # always use the named constants instead of raw ints.
    stand = CloudCommand(
        timestamp_us=int(time.time() * 1e6),
        sequence=next_seq(),
        mode=ModeCommand(mode=Mode.MODE_STAND)
    )
    await room.local_participant.publish_data(
        stand.SerializeToString(), topic="commands", reliable=True
    )
    await asyncio.sleep(3)  # wait for standing pose to settle

    # Walk forward for 5 seconds at 10 Hz.
    # Velocity commands persist - once set, the robot keeps walking at the
    # last commanded velocity until you change it or send DAMP. Streaming at
    # 10 Hz here is just to demonstrate updates, not a safety requirement.
    for _ in range(50):
        cmd = CloudCommand(
            timestamp_us=int(time.time() * 1e6),
            sequence=next_seq(),
            velocity=VelocityCommand(vx=0.5, vy=0, vyaw=0)
        )
        await room.local_participant.publish_data(
            cmd.SerializeToString(), topic="commands", reliable=True
        )
        await asyncio.sleep(0.1)

    # Stop - send DAMP
    stop = CloudCommand(
        timestamp_us=int(time.time() * 1e6),
        sequence=next_seq(),
        mode=ModeCommand(mode=Mode.MODE_DAMP)
    )
    await room.local_participant.publish_data(
        stop.SerializeToString(), topic="commands", reliable=True
    )


if __name__ == "__main__":
    asyncio.run(main())