#!/usr/bin/env python3 """ hal_client.py — minimal Python client for zelpi's native HAL protocol v1 (lib/hal/protocol.mjs). Lets a Python inference loop (a VLA/policy) connect directly to ANY robot-agent — the bundled sim driver, a real robot's native SDK driver, or the ROS bridge (server/createRosDriver.mjs) — over the exact same WebSocket JSON wire format zelpi's own JS controller (cli/hal.mjs) uses. This is deliberately NOT robot-specific: point --url at whichever robot-agent is running (sim for testing, or a real one for actual hardware) and the same inference loop works against all of them. Usage (as a library): from hal_client import HalClient client = HalClient("ws://127.0.0.1:9091", robot_id="rb-vla-0") client.connect(timeout=5) print(client.capabilities) # {"drive": "...", "dof": ..., "skills": [...]} obs = client.telemetry # {"x":.., "y":.., "theta":.., "joints": [...]} client.send_joints([0.1, 0.2, ...], gripper=0.5) # "setJoints" skill client.send_cmd(v=0.3, w=0.0) # unicycle velocity, for diff/omni/holonomic bases client.estop("manual"); client.release() client.close() """ from __future__ import annotations import json import threading import time from typing import Optional PROTOCOL_VERSION = 1 class HalClient: def __init__(self, url: str, robot_id: str = "vla-0", ping_interval_s: float = 0.4): self.url = url self.robot_id = robot_id self.ping_interval_s = ping_interval_s self._ws = None self._lock = threading.Lock() self._telemetry = {} self._capabilities = {} self._skill_results = {} # reqId -> result dict self._seq = 0 self._req_seq = 0 self._connected_evt = threading.Event() self._stop = False self._recv_thread = None self._ping_thread = None @property def telemetry(self) -> dict: with self._lock: return dict(self._telemetry) @property def capabilities(self) -> dict: with self._lock: return dict(self._capabilities) def connect(self, timeout: float = 5.0): import websocket # websocket-client package self._ws = websocket.create_connection(self.url, timeout=timeout) self._recv_thread = threading.Thread(target=self._recv_loop, daemon=True) self._recv_thread.start() if not self._connected_evt.wait(timeout): raise TimeoutError(f"no 'hello' from robot-agent at {self.url} within {timeout}s") # Accept the link — mirrors cli/hal.mjs's controller sending `welcome`. self._send({"v": PROTOCOL_VERSION, "op": "welcome", "accept": True, "controller": "hal_client.py"}) self._ping_thread = threading.Thread(target=self._ping_loop, daemon=True) self._ping_thread.start() def seed_pose(self, x: float = 0.0, y: float = 0.0, theta: float = 0.0): self._send({"v": PROTOCOL_VERSION, "op": "pose", "id": self.robot_id, "x": x, "y": y, "theta": theta}) def send_cmd(self, v: float, w: float): """Unicycle velocity command — for diff/omni/holonomic mobile bases.""" self._seq += 1 self._send({"v": PROTOCOL_VERSION, "op": "cmd", "id": self.robot_id, "V": v, "w": w, "seq": self._seq}) def send_joints(self, joints, gripper: Optional[float] = None, wait: bool = False, timeout: float = 2.0): """Continuous joint-space action (the 'setJoints' skill convention — see docs/HARDWARE.md) — for arm-class robots driven by a VLA/policy.""" args = {"joints": [float(j) for j in joints]} if gripper is not None: args["gripper"] = float(gripper) return self._send_skill("setJoints", args, wait=wait, timeout=timeout) def send_skill(self, skill: str, args: Optional[dict] = None, wait: bool = False, timeout: float = 5.0): return self._send_skill(skill, args or {}, wait=wait, timeout=timeout) def estop(self, reason: str = "manual"): self._send({"v": PROTOCOL_VERSION, "op": "estop", "reason": reason}) def release(self): self._send({"v": PROTOCOL_VERSION, "op": "release"}) def close(self): self._stop = True try: if self._ws: self._ws.close() except Exception: pass # ── internal ───────────────────────────────────────────────────────────── def _send_skill(self, skill, args, wait, timeout): self._req_seq += 1 req_id = f"hal-client-{self._req_seq}" self._send({"v": PROTOCOL_VERSION, "op": "skill", "id": self.robot_id, "skill": skill, "args": args, "reqId": req_id}) if not wait: return None deadline = time.monotonic() + timeout while time.monotonic() < deadline: with self._lock: if req_id in self._skill_results: return self._skill_results.pop(req_id) time.sleep(0.02) raise TimeoutError(f"skill '{skill}' timed out waiting for skillResult") def _send(self, obj): if not self._ws: raise RuntimeError("not connected — call connect() first") self._ws.send(json.dumps(obj)) def _ping_loop(self): # Keeps the robot-agent's 500ms watchdog happy during slow inference # steps (a real VLA forward pass can take longer than that) — setJoints # calls already reset the watchdog too, so this is a safety net, not # the only liveness signal. while not self._stop: try: self._send({"v": PROTOCOL_VERSION, "op": "ping", "t": time.time() * 1000}) except Exception: break time.sleep(self.ping_interval_s) def _recv_loop(self): while not self._stop: try: raw = self._ws.recv() except Exception: break if not raw: continue try: m = json.loads(raw) except Exception: continue op = m.get("op") if op == "hello": with self._lock: self._capabilities = m.get("caps", {}) self._connected_evt.set() elif op == "telemetry": with self._lock: self._telemetry = {k: v for k, v in m.items() if k not in ("v", "op")} elif op == "skillResult": req_id = m.get("reqId") if req_id: with self._lock: self._skill_results[req_id] = m elif op == "fault": print(f"[hal_client] fault: {m.get('code')} — {m.get('message')}") # pong/welcome: no action needed if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="Smoke-test the HAL client against a running robot-agent") ap.add_argument("--url", default="ws://127.0.0.1:9091") ap.add_argument("--robot-id", default="vla-test-0") args = ap.parse_args() client = HalClient(args.url, robot_id=args.robot_id) print(f"connecting to {args.url}...") client.connect(timeout=5) print("capabilities:", client.capabilities) client.seed_pose(0, 0, 0) time.sleep(0.2) print("telemetry after seed:", client.telemetry) result = client.send_joints([0.1, -0.2, 0.3, 0.0, 0.0, 0.0, 0.0], gripper=0.8, wait=True) print("setJoints result:", result) time.sleep(0.3) print("telemetry after setJoints:", client.telemetry) client.close() print("done")