"""Real-time viewer: watch the capsule drive, live in your browser. The viewer is itself a plain AICC client: it polls the bridge through the protocol sensors (world_query once, proprioception + vision a few times per second), renders a top-down map from what it observes, and pushes the frames to any number of browser tabs over its own WebSocket. Run it, then open http://127.0.0.1:8000 and start the demo in another terminal (scripts/run_demo.sh) — you will see the capsule move in real time. Usage: python -m testbed.live [--bridge ws://127.0.0.1:8765] [--http-port 8000] """ from __future__ import annotations import argparse import asyncio import base64 import io import json import math from typing import Any from aicc.client import AICCClient from aicc.transport.websocket import WebSocketClientTransport from testbed.room.mapview import draw_sensor_map HTTP_PORT = 8000 WS_PORT = 8001 POLL_SECONDS = 0.25 # ~4 fps HTML = """ AICC capsule — live view

AICC capsule — real-time view

connecting…

first person (vision)

vision

top-down (sensors)

map
""" def _png_b64(img) -> str: buf = io.BytesIO() img.save(buf, format="PNG") return base64.b64encode(buf.getvalue()).decode("ascii") class LiveViewer: """Polls the bridge via AICC sensors and broadcasts frames to browsers.""" def __init__( self, bridge_url: str, http_port: int, ws_port: int, poll_seconds: float = POLL_SECONDS, ): self.bridge_url = bridge_url self.http_port = http_port self.ws_port = ws_port self.poll = poll_seconds self.clients: set[Any] = set() self.layout: dict[str, Any] | None = None self.path: list[tuple[float, float]] = [] # -- sensor polling ----------------------------------------------------- async def _sense(self, client: AICCClient) -> None: while True: try: if self.layout is None: self.layout = (await client.call_tool("world_query", {})).output prop = (await client.call_tool("proprioception", {})).output vision = (await client.call_tool("vision", {})).output pos = prop["position"] yaw = prop["rotation"]["yaw_deg"] % 360.0 self.path.append((pos["x"], pos["z"])) if len(self.path) > 4000: self.path = self.path[-2000:] beacon = self.layout.get("beacon", {}) if self.layout else {} bx = beacon.get("x", 12.5) bz = beacon.get("z", 12.5) dist = math.hypot(bx - pos["x"], bz - pos["z"]) await self._broadcast( { "type": "view", "png_b64": vision["png_b64"], "tick": vision["tick"], } ) map_img = draw_sensor_map( self.layout, (pos["x"], pos["z"]), yaw, self.path, step=prop["tick"] ) await self._broadcast( { "type": "map", "png_b64": _png_b64(map_img), "tick": prop["tick"], "pos": [pos["x"], pos["z"]], "yaw": yaw, "dist": dist, } ) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - keep polling through transient errors print(f"[live] sense error: {type(exc).__name__}: {exc}") await asyncio.sleep(self.poll) async def _broadcast(self, msg: dict[str, Any]) -> None: if not self.clients: return data = json.dumps(msg) for ws in list(self.clients): try: await ws.send(data) except Exception: # noqa: BLE001 self.clients.discard(ws) # -- servers ------------------------------------------------------------ async def run(self) -> None: import websockets async def handler(ws) -> None: self.clients.add(ws) try: await ws.wait_closed() finally: self.clients.discard(ws) async def http_page( reader: asyncio.StreamReader, writer: asyncio.StreamWriter ) -> None: try: await reader.read(4096) body = HTML.encode("utf-8") writer.write( b"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n" + f"Content-Length: {len(body)}\r\n\r\n".encode() + body ) await writer.drain() finally: writer.close() async def bridge_loop() -> None: while True: try: transport = WebSocketClientTransport(self.bridge_url) async with AICCClient(transport) as client: await client.handshake() print(f"[live] connected to bridge {self.bridge_url}") await self._sense(client) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 - wait for the bridge print(f"[live] bridge not reachable: {type(exc).__name__}: {exc}") await asyncio.sleep(1.0) ws_server = await websockets.serve(handler, "127.0.0.1", self.ws_port) http_server = await asyncio.start_server(http_page, "127.0.0.1", self.http_port) print( f"[live] open http://127.0.0.1:{self.http_port} (frames on ws://127.0.0.1:{self.ws_port})" ) print( "[live] start the demo in another terminal: scripts/run_demo.sh --agent scripted" ) poll_task = asyncio.create_task(bridge_loop()) try: await asyncio.sleep(3600) finally: poll_task.cancel() ws_server.close() http_server.close() def main() -> int: parser = argparse.ArgumentParser( description="Real-time browser viewer for the AICC capsule testbed." ) parser.add_argument("--bridge", default="ws://127.0.0.1:8765") parser.add_argument("--http-port", type=int, default=HTTP_PORT) parser.add_argument("--ws-port", type=int, default=WS_PORT) parser.add_argument( "--poll", type=float, default=POLL_SECONDS, help="sensor poll interval in seconds", ) args = parser.parse_args() viewer = LiveViewer(args.bridge, args.http_port, args.ws_port, args.poll) try: asyncio.run(viewer.run()) except KeyboardInterrupt: pass return 0 if __name__ == "__main__": raise SystemExit(main())