Without the SDK

class NoRedirect(urllib.request.HTTPRedirectHandler):
    """Refuse every redirect: urllib would send the credential on to the new address."""

    def redirect_request(self, *args: Any, **kwargs: Any) -> None:
        return None


def fetch_grant(manager: str = MANAGER_URL, credential: str = CREDENTIAL) -> dict[str, Any]:
    """The LiveKit URL, the robot's room and a join token, from Asimov Manager."""
    if not credential:
        sys.exit("Set MENLO_CREDENTIAL to an SDK credential from Asimov Manager.")
    request = urllib.request.Request(
        manager.rstrip("/") + "/api/livekit/token",
        data=b"{}",
        method="POST",
        headers={"Authorization": f"Bearer {credential}", "Content-Type": "application/json"},
    )
    with urllib.request.build_opener(NoRedirect).open(request, timeout=5) as response:
        grant: dict[str, Any] = json.load(response)  # url, room, token, identity, role
    # The URL is the one the robot uses itself; localhost there is Asimov Manager's host.
    url = urlsplit(grant["url"])
    if url.hostname in ("localhost", "127.0.0.1"):
        host = urlsplit(manager).hostname or ""
        netloc = host if url.port is None else f"{host}:{url.port}"
        grant["url"] = url._replace(netloc=netloc).geturl()
    return grant

async def print_states(track: rtc.RemoteDataTrack) -> None:
    async for frame in track.subscribe():
        state = asimov_state_pb2.RobotState.FromString(bytes(frame.payload))
        mode = asimov_common_pb2.ControlMode.Name(state.current_mode)
        print(f"{mode}  protocol {state.protocol_version}  sequence {state.sequence}")


async def main() -> None:
    grant = fetch_grant()
    room = rtc.Room()
    readers: set[asyncio.Task[None]] = set()

    def on_data_track(track: rtc.RemoteDataTrack) -> None:
        if track.info.name == "state":
            task = asyncio.ensure_future(print_states(track))
            readers.add(task)  # keep a reference so the task is not collected
            task.add_done_callback(readers.discard)

    room.on("data_track_published", on_data_track)
    await room.connect(grant["url"], grant["token"])
    try:
        await asyncio.sleep(SECONDS)
    finally:
        await room.disconnect()

reason = why_not_stand(*latest) if latest else "no state received"
if reason is not None:
    print("not sending STAND:", reason)
    return 1
payload = command(asimov_common_pb2.CONTROL_MODE_STAND)
await room.local_participant.publish_data(payload, reliable=True, topic="commands")

async def save_one_frame(track: rtc.Track, done: asyncio.Future[str]) -> None:
    stream = rtc.VideoStream(track)
    async for event in stream:
        rgb = event.frame.convert(rtc.VideoBufferType.RGB24)
        header = f"P6 {rgb.width} {rgb.height} 255\n".encode()
        Path(PATH).write_bytes(header + bytes(rgb.data))
        done.set_result(f"saved {PATH}, {rgb.width}x{rgb.height}")
        break
    await stream.aclose()


async def main() -> None:
    grant = fetch_grant()
    room = rtc.Room()
    done: asyncio.Future[str] = asyncio.get_running_loop().create_future()
    tasks: set[asyncio.Task[None]] = set()

    def on_track(track: rtc.Track, *_: object) -> None:
        if track.kind == rtc.TrackKind.KIND_VIDEO and not tasks:
            tasks.add(asyncio.ensure_future(save_one_frame(track, done)))

    room.on("track_subscribed", on_track)
    await room.connect(grant["url"], grant["token"])
    try:
        print(await asyncio.wait_for(done, TIMEOUT_S))
    finally:
        await room.disconnect()

async def record(track: rtc.Track, done: asyncio.Future[str]) -> None:
    stream = rtc.AudioStream(track)
    chunks: list[bytes] = []
    rate = channels = 0
    async for event in stream:
        frame = event.frame
        rate, channels = frame.sample_rate, frame.num_channels
        chunks.append(bytes(frame.data))
        if sum(len(c) for c in chunks) >= SECONDS * rate * channels * 2:
            break
    await stream.aclose()
    with wave.open(PATH, "wb") as wav:
        wav.setnchannels(channels)
        wav.setsampwidth(2)  # 16-bit samples
        wav.setframerate(rate)
        wav.writeframes(b"".join(chunks))
    done.set_result(f"saved {PATH}, {SECONDS:.0f} s at {rate} Hz")


async def main() -> None:
    grant = fetch_grant()
    room = rtc.Room()
    done: asyncio.Future[str] = asyncio.get_running_loop().create_future()
    tasks: set[asyncio.Task[None]] = set()

    def on_track(track: rtc.Track, *_: object) -> None:
        if track.kind == rtc.TrackKind.KIND_AUDIO and not tasks:
            tasks.add(asyncio.ensure_future(record(track, done)))

    room.on("track_subscribed", on_track)
    await room.connect(grant["url"], grant["token"])
    try:
        print(await asyncio.wait_for(done, SECONDS + TIMEOUT_S))
    finally:
        await room.disconnect()