armcade

#!/usr/bin/env python
# Copyright 2026 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
"""Remote teleop against the SIMULATOR — one command, no real arm or cameras needed.

This is the REMOTE tier built on top of the LOCAL sim (`tomctl.control`): it just composes
the local sim with the LiveKit control bridge. It launches and supervises two child processes:

  1. python -m tomctl.control --web --remote-port P     (the local sim + the UDP remote receiver)
  2. python -m tomctl.livekit.bridge --room R --arms ... --remote-port P
                                                       (the queue — ONE arm controllable at a time)

Camera feeds are OPTIONAL and do not block control: open the sim's page in a browser and click
"Stream cameras" (or open it with ?stream=1), and it publishes SIMULATED wrist/top/side views to the
same LiveKit room — so a remote operator can watch the arm move without any real cameras.

The operator drives from the armcade `tools/livekit_viewer.html` with an operator token (see the
printed mint command). Ctrl-C tears everything down.

    python -m tomctl.remote_sim --arms arm1 arm2 --room tom-sim   # two arms, cycled (run from robot/)
    python -m tomctl.remote_sim --urdf-path assets/tom.urdf       # override the sim URDF
"""

from __future__ import annotations

import argparse
import os
import signal
import subprocess
import sys
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
ROBOT_ROOT = HERE.parent  # robot/ — the sys.path root for `python -m tomctl.<x>` + `import lerobot`
PY = sys.executable


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--room", default="tom-sim", help="LiveKit room (operator + cameras + bridge share it)")
    p.add_argument("--url", default="ws://localhost:7880", help="LiveKit server ws URL")
    p.add_argument("--arms", nargs="+", default=["arm1"], metavar="ARM",
                   help="Arm ids (1 or 2). With 2, the queue cycles arm1 -> arm2 -> arm1.")
    p.add_argument("--remote-port", type=int, default=9090, help="Loopback UDP port: bridge -> sim receiver")
    p.add_argument("--sim-port", type=int, default=7010, help="Sim web/control-panel HTTP port (informational)")
    p.add_argument("--max-session-s", type=float, default=None, help="Max seconds per session (default: unlimited)")
    p.add_argument("--idle-timeout-s", type=float, default=30.0, help="Release a session after this idle (0 disables)")
    p.add_argument("--operator", action="append", metavar="IDENTITY",
                   help="Optional allow-list of operator identities (repeatable). Omit for open access.")
    p.add_argument("--urdf-path", default=None, help="Override the sim URDF (passed through to the sim).")
    p.add_argument("--web-host", default="127.0.0.1",
                   help="Bind for the sim web panel. Use 0.0.0.0 to open the panel + stream sim cameras "
                        "from another machine on the LAN.")
    return p.parse_args()


def main() -> None:
    args = parse_args()
    if not 1 <= len(args.arms) <= 2:
        raise SystemExit(f"--arms takes 1 or 2 ids, got {args.arms}")
    single = len(args.arms) == 1

    procs: list[tuple[str, subprocess.Popen]] = []

    def launch(name: str, cmd: list[str], env: dict | None = None) -> None:
        print(f"[remote-sim] start {name}: {' '.join(cmd)}", flush=True)
        procs.append((name, subprocess.Popen(cmd, env=env, cwd=ROBOT_ROOT)))

    # 1) local sim + remote receiver. BROWSER=true stops webbrowser.open() from hanging on a headless host.
    sim_cmd = [PY, "-m", "tomctl.control", "--web",
               "--remote-port", str(args.remote_port),
               "--livekit-url", args.url, "--livekit-room", args.room,
               "--web-host", args.web_host]
    if single:
        sim_cmd.append("--single-arm")
    if args.urdf_path:
        sim_cmd += ["--urdf-path", args.urdf_path]
    launch("sim", sim_cmd, env={**os.environ, "BROWSER": "true"})

    # 2) the queue/control bridge
    bridge_cmd = [PY, "-m", "tomctl.livekit.bridge", "--url", args.url, "--room", args.room,
                  "--arms", *args.arms, "--remote-port", str(args.remote_port),
                  "--idle-timeout-s", str(args.idle_timeout_s)]
    if args.max_session_s is not None:
        bridge_cmd += ["--max-session-s", str(args.max_session_s)]
    for op in args.operator or []:
        bridge_cmd += ["--operator", op]
    launch("bridge", bridge_cmd)

    print(f"""
==================== REMOTE SIM is up ({len(args.arms)} arm{'s' if not single else ''}) ====================
  Sim panel + CAMERA streaming : http://127.0.0.1:{args.sim_port}/   (click "Stream cameras", or add ?stream=1)
  Operator drives via          : armcade  tools/livekit_viewer.html   (url {args.url})
  Mint an operator token       : python -m tomctl.livekit.tokens --role operator \\
                                     --identity <player-id> --room {args.room} --url {args.url}
  Cameras are OPTIONAL — control works whether or not the sim page is streaming.
  Ctrl-C to stop everything.
================================================================================
""", flush=True)

    stop = {"now": False}
    signal.signal(signal.SIGINT, lambda *_: stop.update(now=True))
    signal.signal(signal.SIGTERM, lambda *_: stop.update(now=True))
    try:
        while not stop["now"]:
            for name, p in procs:
                if p.poll() is not None:
                    print(f"[remote-sim] {name} exited (code {p.returncode}); shutting down", flush=True)
                    stop["now"] = True
            time.sleep(0.3)
    finally:
        for name, p in reversed(procs):
            if p.poll() is None:
                p.terminate()
        for _name, p in reversed(procs):
            try:
                p.wait(timeout=5)
            except subprocess.TimeoutExpired:
                p.kill()
        print("[remote-sim] stopped.", flush=True)


if __name__ == "__main__":
    main()