#!/usr/bin/env python3 """ Inspire hand web dashboard: sliders + arm/hand combos + live status. - hand poses -> C++ hand_bridge (127.0.0.1:7799) -> rt/inspire/cmd - arm actions -> build/arm_action (Unitree G1ArmActionClient) - status/restart/iface controls Needs inspire_g1 + hand_bridge (./manage.sh dashboard starts both). """ import os, socket, json, argparse, subprocess, glob, re, time, pty, signal, threading, queue, shlex from flask import Flask, request, jsonify, Response REPO = os.environ.get("HAND_REPO") or os.path.dirname(os.path.dirname(os.path.abspath(__file__))) BUILD = os.path.join(REPO, "build") BRIDGE_HOST, BRIDGE_PORT = "127.0.0.1", 7799 # Saved shapes/combos live in HAND_DATA, which run.sh bind-mounts from the host. They used # to sit inside the image at /opt/hand, so every `docker rm -f` in run.sh silently destroyed # the whole library. DATA_HOME = os.environ.get("HAND_DATA") or REPO os.makedirs(DATA_HOME, exist_ok=True) POSES_FILE = os.path.join(DATA_HOME, "hand_poses.txt") COMBOS_FILE = os.path.join(DATA_HOME, "hand_combos.json") IFACE_FILE = os.path.join(DATA_HOME, ".dashboard_iface") PORT = 8088 # set from --port; the recorder pulls shapes back off this port # Record/replay is driven by g1_record_replay.py as a CHILD PROCESS rather than # reimplemented here. That script is ~1000 lines of arm gravity feed-forward, touch # thresholding and closure remapping, and a second copy inside the dashboard would drift # from the CLI the first time either one changed. RECORDER_DIR overrides the search. REC_SCRIPT = "g1_record_replay.py" # recorder/ inside this project comes first: the recorder ships WITH the dashboard so the # two can never fall out of step, and a checkout is self-contained. The outside paths stay # as fallbacks for a host that still keeps its own working copy. REC_DIR = os.path.abspath(os.environ.get("RECORDER_DIR") or next( (p for p in (os.path.join(REPO, "recorder"), os.path.join(REPO, "..", "Manual_Recorder"), os.path.expanduser("~/Manual_Recorder"), os.path.expanduser("~/Robotics_workspace/yslootahtech/Project/G1/G1_Lootah/" "Manual_Recorder")) if os.path.isfile(os.path.join(p, REC_SCRIPT))), "")) # The recorder needs unitree_sdk2py, which on a workstation lives in the g1_env conda env — # point RECORDER_PY at that interpreter if the default python3 cannot import it. REC_PY = os.environ.get("RECORDER_PY") or "python3" TAKES_DIR = os.path.join(REC_DIR, "DataG1") if REC_DIR else "" JOINTS = ["pinky", "ring", "middle", "index", "thumb_bend", "thumb_rot"] ARM_ACTIONS = {"shake hand": 27, "right hand up": 23, "hands up": 15, "high five": 18, "hug": 19, "high wave": 26, "face wave": 25, "clap": 17, "heart": 20, "reject": 22, "release arm": 99} app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 256 * 1024 * 1024 # take uploads; a 60Hz minute is ~4MB def get_iface(): try: v = open(IFACE_FILE).read().strip() return v or "eth0" except Exception: return os.environ.get("IFACE", "eth0") def sh(args, timeout=20): try: r = subprocess.run(args, capture_output=True, timeout=timeout) return r.returncode, (r.stdout + r.stderr).decode(errors="ignore") except Exception as e: return 1, str(e) def running(name): return sh(["pgrep", "-x", name], timeout=5)[0] == 0 def iface_status(name): _, out = sh(["ip", "-brief", "addr", "show", name], timeout=4) m = re.search(r"(\d+\.\d+\.\d+\.\d+)", out) return {"up": "UP" in out, "ip": m.group(1) if m else None} def detected_ports(): """Which tty did inspire_g1 open for each hand, per its startup log. Handles BOTH log formats. The old one put them on one line: Inspire hands detected: right=/dev/ttyUSB1 left=/dev/ttyUSB0 the current one splits them and adds the USB path, which the old single-line regex could not match — so the dashboard showed a red "no ports" on a perfectly healthy service: Inspire hands identified by usb path order: RIGHT = /dev/ttyUSB0 (usb 1-2.2.3, rs485 id 1) LEFT = /dev/ttyUSB1 (usb 1-2.2.1.1, rs485 id 1) """ try: txt = open("/tmp/inspire_g1.log").read() except Exception: return [] m = re.search(r"right=(\S+)\s+left=(\S+)", txt) # old single-line form if m: return [m.group(1), m.group(2)] r = re.findall(r"^\s*RIGHT\s*=\s*(\S+)", txt, re.M) # current two-line form l = re.findall(r"^\s*LEFT\s*=\s*(\S+)", txt, re.M) if r and l: return [r[-1], l[-1]] return [] def send_pose(vals): if len(vals) != 12: return False, "need 12 values" line = " ".join("%.3f" % float(v) for v in vals) + "\n" try: s = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT), timeout=2) s.sendall(line.encode()); r = s.recv(16); s.close() return (b"ok" in r), r.decode(errors="ignore").strip() except Exception as e: return False, "bridge down: %s" % e def read_state(want_force=False): """Ask the bridge for state. 'S' returns 24 values (12 angles + 12 forces); 'R' returns the legacy 12 angles. Falls back automatically so an un-rebuilt bridge still works.""" def ask(cmd, n): try: s = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT), timeout=2) s.sendall(cmd); r = s.recv(512); s.close() vals = [float(x) for x in r.decode().split()] return vals if len(vals) == n else None except Exception: return None if want_force: v = ask(b"S\n", 24) if v: return v[:12], v[12:] v = ask(b"R\n", 12) return (v, None) if v else (None, None) v = ask(b"S\n", 24) if v: return v[:12] return ask(b"R\n", 12) def run_arm(aid, timeout=15): rc, out = sh([os.path.join(BUILD, "arm_action"), str(int(aid)), get_iface()], timeout=timeout) return rc == 0, out[-200:] def load_poses(): m = {} if os.path.exists(POSES_FILE): for line in open(POSES_FILE): p = line.split() if len(p) == 13 and not line.startswith("#"): try: m[p[0]] = [float(x) for x in p[1:]] except ValueError: pass return m def backup_file(path): """Keep the previous contents next to a library file before overwriting it. Both libraries are single files rewritten IN FULL on every save, rename and delete, so deleting the last entry silently empties the file. Five calibrated finger shapes were lost exactly that way, with no copy anywhere on the robot to restore from. A .bak costs a few KB and makes that one click recoverable. """ try: if os.path.exists(path) and os.path.getsize(path) > 0: with open(path, "rb") as a, open(path + ".bak", "wb") as b: b.write(a.read()) except OSError: pass def write_poses(m): backup_file(POSES_FILE) with open(POSES_FILE, "w") as f: f.write("# name \n" % " ".join(JOINTS)) for k in sorted(m): f.write(k + " " + " ".join("%.3f" % x for x in m[k]) + "\n") def save_pose(name, vals): m = load_poses(); m[name] = [float(v) for v in vals]; write_poses(m) def load_combos(): try: return json.load(open(COMBOS_FILE)) except Exception: return {} def write_combos(m): backup_file(COMBOS_FILE) json.dump(m, open(COMBOS_FILE, "w"), indent=2) @app.route("/") def index(): return Response(HTML, mimetype="text/html") @app.route("/api/status") def api_status(): ports, live = detected_ports(), sorted(glob.glob("/dev/ttyUSB*")) mismatch = bool(ports) and any(p not in live for p in ports) return jsonify(service=running("inspire_g1"), bridge=running("hand_bridge"), ports=ports, live=live, mismatch=mismatch, iface=get_iface(), eth0=iface_status("eth0"), wlan0=iface_status("wlan0")) IN_CONTAINER = os.path.exists("/.dockerenv") @app.route("/api/restart", methods=["POST"]) def api_restart(): """Restart the RS-485 service. In the container the ONLY correct move is to kill inspire_g1 and let docker/start.sh's supervisor bring a fresh one up (~3s, with ClearError and the force limits re-applied), because that supervisor owns the process. manage.sh must not be used here. It is written for a host shell where nothing else supervises: it stops the service and starts its own copy, so the supervisor's restart and manage.sh's start race — two writers on one RS-485 bus, which is exactly how the hands end up with latched actuator faults. Measured in the container: the call also never returned, leaving inspire_g1 parented to a blocked manage.sh. """ if IN_CONTAINER: sh(["pkill", "-x", "inspire_g1"], timeout=5) for _ in range(30): # the supervisor polls every 3s time.sleep(0.5) if running("inspire_g1"): break return jsonify(ok=running("inspire_g1"), via="supervisor") env = dict(os.environ); env["IFACE"] = get_iface() subprocess.run([os.path.join(REPO, "manage.sh"), "restart"], capture_output=True, timeout=25, env=env) return jsonify(ok=running("inspire_g1"), via="manage.sh") @app.route("/api/iface", methods=["POST"]) def api_iface(): name = request.get_json(force=True).get("iface", "eth0") if name not in ("eth0", "wlan0"): return jsonify(ok=False, msg="eth0 or wlan0") open(IFACE_FILE, "w").write(name) sh(["pkill", "-x", "hand_bridge"], timeout=5); time.sleep(1) subprocess.Popen(["setsid", os.path.join(BUILD, "hand_bridge"), name, str(BRIDGE_PORT)], stdout=open("/tmp/hand_bridge.log", "w"), stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL) time.sleep(1) return jsonify(ok=True, iface=name) @app.route("/api/apply", methods=["POST"]) def api_apply(): ok, msg = send_pose(request.get_json(force=True).get("q", [])) return jsonify(ok=ok, msg=msg) @app.route("/api/release", methods=["POST"]) def api_release(): # NOT a release, despite the name. q<0 sends 0xFFFF, which the RH56 reads as "no change": # the joint keeps its previous target and stays driven. There is no motor-release command, # and the worm drive is non-backdrivable, so a finger can never be posed by hand. The UI # button that promised this has been removed; the route stays so old callers don't 404. ok, msg = send_pose([-1.0] * 12) return jsonify(ok=ok, msg=msg, note="no-op: RH56 cannot release a finger") ERR_BITS = [(0x01, "LOCKED-ROTOR"), (0x02, "OVER-TEMP"), (0x04, "OVER-CURRENT"), (0x08, "ABNORMAL"), (0x10, "COMMS")] STATUS_TXT = {0: "unclenching", 1: "grasping", 2: "reached pos", 3: "reached force", 5: "current prot", 6: "locked rotor", 7: "fault"} def read_diag(): """Full per-finger diagnostics from the bridge 'D' command: 8 blocks of 12 + publish rate.""" try: s = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT), timeout=2) s.sendall(b"D\n") buf = b"" while len(buf) < 4096 and b"\n" not in buf: c = s.recv(4096) if not c: break buf += c s.close() v = [float(x) for x in buf.decode().split()] if len(v) < 97: return None b = [v[i * 12:(i + 1) * 12] for i in range(8)] return dict(q=b[0], f=b[1], cur=b[2], fset=b[3], temp=b[4], err=[int(x) for x in b[5]], sta=[int(x) for x in b[6]], lost=[int(x) for x in b[7]], rate=v[96]) except Exception: return None @app.route("/api/diag") def api_diag(): """Everything the tracker needs in one call: angle, force, current, force-limit, temp, ERROR bits, STATUS, lost counters and the measured publish rate.""" d = read_diag() if not d: q, f = read_state(want_force=True) return jsonify(ok=q is not None, q=q, f=f, full=False) d["err_txt"] = [" ".join(n for b, n in ERR_BITS if int(e) & b) or "" for e in d["err"]] d["sta_txt"] = [STATUS_TXT.get(int(x), str(int(x))) for x in d["sta"]] d["ok"] = True d["full"] = True return jsonify(d) @app.route("/api/follow", methods=["POST"]) def api_follow(): """Force-follow (admittance): press a finger and it moves under your hand. The RH56 cannot be back-driven, so a press alone moves it 0.000. This reads the press and drives the motor that way, which makes the ANGLE genuinely change — so what gets recorded is a real measured angle instead of a grams-to-degrees guess. It also removes the calibration problem: presses vary 5x between takes (41-110g one run, 48-527g the next), and no fixed scale survives that. Watching the finger move and stopping when it looks right is a closed loop with you in it. """ d = request.get_json(force=True) on = 1 if d.get("on") else 0 return jsonify(**set_follow(on, d.get("gain", 0.0006), d.get("deadband", 60), d.get("maxrate", 0.35), d.get("force", 400))) FOLLOW_ON = False # tracked so a take can disarm it — both publish rt/inspire/cmd def set_follow(on, gain=0.0006, deadband=60, maxrate=0.35, force=400): global FOLLOW_ON line = "F %d %s %s %s %s\n" % (1 if on else 0, gain, deadband, maxrate, force) try: s_ = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT), timeout=2) s_.sendall(line.encode()); r = s_.recv(64); s_.close() FOLLOW_ON = bool(on) return {"ok": True, "on": bool(on), "msg": r.decode(errors="ignore").strip()} except Exception as e: return {"ok": False, "msg": "bridge down: %s" % e} @app.route("/api/track") def api_track(): """Angle AND force in one call, for the live joint tracker. Both are needed together: a finger you press shows force but NOT angle, because the RH56 drive is non-backdrivable. Seeing the two columns side by side is what makes that obvious instead of looking like a broken sensor. """ q, f = read_state(want_force=True) return jsonify(ok=q is not None, q=q, f=f, has_force=f is not None) @app.route("/api/capture") def api_capture(): # read the current (hand-set) joint positions v = read_state() # Per-hand liveness. hand_bridge caches the last DDS state with no timestamp, so if # inspire_g1 dies or one RS-485 link drops, it keeps serving the last values forever and # the dashboard's "reads live" lamp stays green over a completely dead hand. A hand whose # six joints are ALL exactly 0.000 is not a real pose — it is an unread bus. live = [True, True] if v: for h, lo in ((0, 0), (1, 6)): live[h] = any(abs(x) > 1e-9 for x in v[lo:lo + 6]) return jsonify(ok=v is not None, q=v, live_right=live[0], live_left=live[1]) @app.route("/api/selftest", methods=["POST"]) def api_selftest(): # close -> read -> open -> read; report movement + read health send_pose([0.0] * 12); time.sleep(2.0) closed = read_state() or [0.0] * 12 send_pose([1.0] * 12); time.sleep(2.0) opened = read_state() or [0.0] * 12 moved = any(abs(closed[i] - opened[i]) > 0.05 for i in range(12)) reads_ok = any(abs(v) > 0.001 for v in closed + opened) return jsonify(closed=closed, opened=opened, moved=moved, reads_ok=reads_ok) @app.route("/api/armlist") def api_armlist(): return jsonify(ARM_ACTIONS) @app.route("/api/arm", methods=["POST"]) def api_arm(): ok, out = run_arm(request.get_json(force=True).get("id")) return jsonify(ok=ok, msg=out) @app.route("/api/combo/play", methods=["POST"]) def api_combo_play(): d = request.get_json(force=True) q, arm, release = d.get("q"), d.get("arm"), d.get("release", False) if q: send_pose(q) if arm not in (None, "", -1): ok, msg = run_arm(arm) if q: send_pose(q) if release: run_arm(99) return jsonify(ok=ok, msg=msg) return jsonify(ok=True, msg="pose applied") @app.route("/api/poses") def api_poses(): return jsonify(load_poses()) @app.route("/api/save", methods=["POST"]) def api_save(): d = request.get_json(force=True) name, q = d.get("name", "").strip(), d.get("q", []) if not name or len(q) != 12: return jsonify(ok=False, msg="name + 12 values") save_pose(name, q); return jsonify(ok=True) @app.route("/api/combos", methods=["GET", "POST"]) def api_combos(): if request.method == "GET": return jsonify(load_combos()) d = request.get_json(force=True); name = d.get("name", "").strip() if not name: return jsonify(ok=False, msg="name required") m = load_combos() m[name] = {"q": d.get("q", []), "arm": d.get("arm"), "release": bool(d.get("release", True))} write_combos(m) return jsonify(ok=True) @app.route("/api/library/restore", methods=["POST"]) def api_library_restore(): """Put back the shape (or combo) library as it was before the last write.""" which = request.get_json(force=True).get("what", "poses") path = POSES_FILE if which == "poses" else COMBOS_FILE if not os.path.exists(path + ".bak"): return jsonify(ok=False, msg="no backup yet (one is written on every save/delete)") # READ the backup before writing one, or the "make restore undoable" step overwrites # the very copy being restored — which is a restore that silently returns nothing. with open(path + ".bak", "rb") as a: saved = a.read() backup_file(path) # current becomes the new .bak: restore is undoable with open(path, "wb") as b: b.write(saved) n = len(load_poses() if which == "poses" else load_combos()) return jsonify(ok=True, count=n) @app.route("/api/pose/delete", methods=["POST"]) def api_pose_delete(): name = request.get_json(force=True).get("name", "") m = load_poses() if name in m: del m[name]; write_poses(m); return jsonify(ok=True) return jsonify(ok=False, msg="not found") @app.route("/api/pose/rename", methods=["POST"]) def api_pose_rename(): d = request.get_json(force=True) old, new = d.get("old", ""), d.get("new", "").strip() m = load_poses() if old in m and new and new not in m: m[new] = m.pop(old); write_poses(m); return jsonify(ok=True) return jsonify(ok=False, msg="bad/duplicate name") @app.route("/api/combo/delete", methods=["POST"]) def api_combo_delete(): name = request.get_json(force=True).get("name", "") m = load_combos() if name in m: del m[name]; write_combos(m); return jsonify(ok=True) return jsonify(ok=False, msg="not found") @app.route("/api/combo/rename", methods=["POST"]) def api_combo_rename(): d = request.get_json(force=True) old, new = d.get("old", ""), d.get("new", "").strip() m = load_combos() if old in m and new and new not in m: m[new] = m.pop(old); write_combos(m); return jsonify(ok=True) return jsonify(ok=False, msg="bad/duplicate name") # ---- record / replay ------------------------------------------------------------------ ANSI = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") # The recorder asks these two on stdin at the end of a take, and both are answered by the # panel rather than by a human at a terminal. PROMPT_PREVIEW = "Preview this recording" PROMPT_SAVE = "Save recording?" # Goes on a child process argv (never through a shell), so the only job here is to keep the # name inside DataG1/. Parentheses are allowed because the recorder ITSELF names collisions # that way — unique_path() writes grip1(1).jsonl — and without them those takes could not be # listed, renamed, replayed or deleted from the panel. SAFE_NAME = re.compile(r"^[A-Za-z0-9._()-]{1,64}$") TAKE = None TAKE_LOCK = threading.Lock() class Take: """One recorder run, driven over a pty. A pty and not a pipe: the recorder puts stdin in cbreak mode (tty.setcbreak) to read the in-take keys, which raises on a pipe and would kill the take before the arms ever went limp. The pty is also what lets this panel send those same keys — 1-9 shapes, o/c/[/]/;/' — which is the only way to get a finger pose into a recording, since the RH56 cannot be posed by hand. """ MAXLINES = 300 KEY_GAP = 0.15 # see _writer def __init__(self, argv, action, seconds): self.argv, self.action, self.seconds = argv, action, seconds # The console opens with the command itself, so the log doubles as a receipt of # exactly what ran — the same line you could paste into a terminal. self.lines, self.cur, self.pending = ["$ " + cmd_str(argv)], "", "" self._cr = False # a CR held over, waiting to see if an LF follows self._tail = "" # rolling raw text, for the markers below self.replaying = False # inside replay() — pause only applies there self.paused = False self.phase = "running" self.rc = None self.t0 = time.time() self.lock = threading.Lock() self.mfd, sfd = pty.openpty() env = dict(os.environ, PYTHONUNBUFFERED="1", TERM="dumb") # start_new_session: stop() signals the whole group, so the SIGINT reaches the # recorder even once it has spawned DDS threads of its own. self.proc = subprocess.Popen(argv, cwd=REC_DIR, stdin=sfd, stdout=sfd, stderr=sfd, start_new_session=True, close_fds=True, env=env) os.close(sfd) self.keyq = queue.Queue(maxsize=64) threading.Thread(target=self._pump, daemon=True).start() threading.Thread(target=self._writer, daemon=True).start() def _pump(self): while True: try: data = os.read(self.mfd, 4096) except OSError: # master closes when the child exits break if not data: break self._feed(data.decode("utf-8", "replace")) self.rc = self.proc.wait() with self.lock: if self.cur: self._push() self.phase = "done" self.keyq.put(None) try: os.close(self.mfd) except OSError: pass def _writer(self): """Send keys ONE AT A TIME, ~KEY_GAP apart. The recorder reads keys with a select() followed by sys.stdin.read(1), and Python's stdin is buffered: a burst all lands in that buffer at once, select() then reports nothing ready, and every key after the first is stranded there until the next one arrives. Measured against the real read loop: three keys 10ms apart delivered two, the same three 200ms apart delivered all three. Clicking two shape buttons quickly must not silently drop one, so the pacing lives here rather than in the browser. """ while True: item = self.keyq.get() if item is None: return if not self.alive(): continue try: os.write(self.mfd, item) except OSError: return time.sleep(self.KEY_GAP) def _push(self): self.lines.append(self.cur) self.cur = "" if len(self.lines) > self.MAXLINES: del self.lines[:len(self.lines) - self.MAXLINES] def _feed(self, txt): with self.lock: self.pending += txt # A pty turns every \n into \r\n on the way out, so a CR is only a line-redraw # (the recorder's \r progress counters) when no LF follows it. Treating them all # as redraws wiped the content of every finished line. for ch in ANSI.sub("", txt): if self._cr: self._cr = False if ch == "\n": self._push() continue self.cur = "" # bare CR: the line redraws itself in place if ch == "\r": self._cr = True elif ch == "\n": self._push() else: self.cur += ch # Both prompts arrive WITHOUT a trailing newline, so they never reach self.lines # — they have to be spotted in the raw stream. pending is cleared on every # answer so the same prompt is not re-detected after it has been dealt with. if PROMPT_SAVE in self.pending: self.phase = "save?" elif PROMPT_PREVIEW in self.pending: self.phase = "preview?" # Pause/resume state is READ BACK from the recorder rather than assumed from the # click, so the button cannot drift out of step with the process — and it also # tracks a pause typed at the terminal. A rolling tail is used because a marker # can land across two reads. self._tail = (self._tail + txt)[-400:] if "REPLAY" in self._tail: self.replaying, self.paused = True, False self._tail = "" elif "replay done" in self._tail or "replay stopped" in self._tail: self.replaying, self.paused = False, False self._tail = "" elif "PAUSED" in self._tail: self.paused = True self._tail = "" elif "resumed" in self._tail: self.paused = False self._tail = "" def alive(self): return self.proc.poll() is None def snapshot(self, n=20): with self.lock: tail = [l for l in self.lines[-n:]] if self.cur: tail.append(self.cur) return {"phase": self.phase, "lines": tail[-n:], "rc": self.rc, "action": self.action, "seconds": self.seconds, "replaying": self.replaying, "paused": self.paused, "elapsed": round(time.time() - self.t0, 1)} def key(self, k): """Relay one in-take key. Never a newline: ask() at the end of a take counts a bare Enter as YES, so a stray one would preview (arm moves) or save a discarded take.""" k = k.replace("\r", "").replace("\n", "")[:1] if not k or not self.alive(): return False try: self.keyq.put_nowait(k.encode()) except queue.Full: return False return True def answer(self, yes): if self.phase not in ("preview?", "save?") or not self.alive(): return False with self.lock: self.pending = "" self.phase = "running" # Through the same queue as the keys, so an answer can never overtake a key that is # still waiting its turn. self.keyq.put(b"y\n" if yes else b"n\n") return True def stop(self, hard=False): """SIGINT, not SIGKILL. The recorder catches KeyboardInterrupt, walks the arm home and releases it; killing it outright abandons the arm limp, mid-air.""" if not self.alive(): return sig = signal.SIGKILL if hard else signal.SIGINT try: os.killpg(os.getpgid(self.proc.pid), sig) except Exception: self.proc.send_signal(sig) def rec_ready(): return bool(REC_DIR) and os.path.isfile(os.path.join(REC_DIR, REC_SCRIPT)) def sigint_ok(): """Can Stop actually interrupt a take? A process started in the background by a non-interactive shell inherits SIGINT as SIG_IGN, and unlike a handler, SIG_IGN SURVIVES EXEC — so the recorder would inherit it and Stop would do nothing at all. manage.sh starts this dashboard exactly that way (setsid ... &), so it is the normal case, not the odd one. fix_sigint() below repairs it; this reports whether the repair held, and the panel says so if it did not. """ return signal.getsignal(signal.SIGINT) != signal.SIG_IGN def fix_sigint(): """Swap an inherited SIG_IGN for a no-op HANDLER (main thread only). A handler is reset to SIG_DFL in the child on exec, which is what makes Stop work, while the dashboard itself stays as un-killable by SIGINT as it was before. """ if not sigint_ok(): signal.signal(signal.SIGINT, lambda *_a: None) # DataG1/ also holds the arm home pose, which every take returns through on its way out # (--home). It is not a recording: listing it invites a replay, and deleting it would break # the return-to-home of every take there is. NOT_A_TAKE = ("arm_home",) def take_path(name): """Resolve a take name inside DataG1/, or None. Containment-checked: the name reaches here from a browser and ends up on an argv and an unlink().""" if not TAKES_DIR or not SAFE_NAME.match(name or "") or name in NOT_A_TAKE: return None p = os.path.abspath(os.path.join(TAKES_DIR, name + ".jsonl")) return p if p.startswith(os.path.abspath(TAKES_DIR) + os.sep) else None def num(d, key, default, lo, hi): try: v = float(d.get(key, default)) except (TypeError, ValueError): return default return lo if v < lo else (hi if v > hi else v) def rec_argv(d, action): """Build the recorder command line. ONE builder, used by both /api/rec/start and /api/rec/preview, so the command the panel shows is the command that runs — there is no second copy of this logic to drift. The panel is a front-end for the CLI, not a reimplementation of it: every option here is a flag you could type yourself, and a take made either way lands in the same DataG1/. """ name = str(d.get("name") or "take").strip() argv = [REC_PY, REC_SCRIPT, get_iface(), action] if action == "record": argv += ["--output", name, "--seconds", "%g" % num(d, "seconds", 20, 1, 600), "--thresh", "%g" % num(d, "thresh", 12, 1, 500), "--span", "%g" % num(d, "span", 120, 5, 2000), "--fingers", (d.get("fingers") if d.get("fingers") in ("touch", "position", "external", "follow") else "touch"), # Shapes come from THIS dashboard, so keys 1-9 fire the same poses the # sliders saved. The CLI default points at another host entirely. "--shapes", "http://127.0.0.1:%d" % PORT] else: argv += ["--input", name, "--speed", "%g" % num(d, "speed", 1.0, 0.1, 3.0)] if d.get("waist") == "lock": argv += ["--waist", "lock"] # Arm and hand are independent on BOTH actions: replay honours them too (the recorder # builds its DDS client from use_arm/use_hand before either branch runs), so a take with # both tracks can be replayed hand-only onto an arm you would rather not move. if d.get("no_arm"): argv += ["--no-arm"] if d.get("no_hand"): argv += ["--no-hand"] return argv def cmd_str(argv): """The same command as you would type it, cwd included — the panel prints this.""" return "cd %s && %s" % (shlex.quote(REC_DIR or "?"), " ".join(shlex.quote(a) for a in argv)) @app.route("/api/rec/status") def api_rec_status(): t = TAKE d = {"ok": True, "ready": rec_ready(), "dir": REC_DIR, "sigint": sigint_ok(), "busy": bool(t and t.alive()), "follow": FOLLOW_ON} if t: d.update(t.snapshot()) return jsonify(d) @app.route("/api/rec/start", methods=["POST"]) def api_rec_start(): """Start a take (or a replay) as a child process. Refuses to run two at once — both drive the arm over the same DDS topics, and the second one would fight the first for every joint. """ global TAKE d = request.get_json(force=True) action = "replay" if d.get("action") == "replay" else "record" name = str(d.get("name") or "take").strip() if not rec_ready(): return jsonify(ok=False, msg="%s not found — set RECORDER_DIR" % REC_SCRIPT) if not SAFE_NAME.match(name): return jsonify(ok=False, msg="name: letters, digits, dot, dash, underscore only") no_arm, no_hand = bool(d.get("no_arm")), bool(d.get("no_hand")) if no_arm and no_hand: return jsonify(ok=False, msg="--no-arm and --no-hand together records nothing") if action == "replay": # An arm-less replay still drives the hand, which is why the confirm is only # required when the arm is actually in play. if not no_arm and not d.get("confirm"): return jsonify(ok=False, msg="replay moves the arm — confirm required") if not take_path(name) or not os.path.exists(take_path(name)): return jsonify(ok=False, msg="no take named %s" % name) with TAKE_LOCK: if TAKE and TAKE.alive(): return jsonify(ok=False, msg="a %s is already running" % TAKE.action) # Force-follow publishes rt/inspire/cmd at 30Hz from the bridge, and so does the # recorder. Left armed, the two would overwrite each other every frame and the # fingers would judder — so the take disarms it and says it did. disarmed = False if FOLLOW_ON and not no_hand: disarmed = set_follow(0).get("ok", False) argv = rec_argv(d, action) secs = num(d, "seconds", 20, 1, 600) if action == "record" else None try: TAKE = Take(argv, action, secs) except Exception as e: return jsonify(ok=False, msg="spawn failed: %s" % e) return jsonify(ok=True, action=action, argv=argv, cmd=cmd_str(argv), disarmed_follow=disarmed) @app.route("/api/rec/preview", methods=["POST"]) def api_rec_preview(): """The exact command Record/Replay would run, with nothing started. This is the panel's honesty check: what it prints is what it spawns, because both come out of rec_argv(). """ d = request.get_json(force=True) action = "replay" if d.get("action") == "replay" else "record" argv = rec_argv(d, action) return jsonify(ok=True, argv=argv, cmd=cmd_str(argv), ready=rec_ready()) @app.route("/api/rec/key", methods=["POST"]) def api_rec_key(): t = TAKE if not (t and t.alive()): return jsonify(ok=False, msg="nothing running") return jsonify(ok=t.key(str(request.get_json(force=True).get("k", "")))) @app.route("/api/rec/answer", methods=["POST"]) def api_rec_answer(): """Answer the recorder's end-of-take prompts explicitly. Both are answered with a literal y or n and never a bare newline, because ask() treats an empty line as YES — and the first prompt is 'preview', which moves the arm. """ t = TAKE if not (t and t.alive()): return jsonify(ok=False, msg="nothing running") return jsonify(ok=t.answer(bool(request.get_json(force=True).get("yes"))), phase=t.phase) @app.route("/api/rec/stop", methods=["POST"]) def api_rec_stop(): t = TAKE if not (t and t.alive()): return jsonify(ok=False, msg="nothing running") t.stop(hard=bool(request.get_json(force=True).get("hard"))) return jsonify(ok=True) @app.route("/api/takes") def api_takes(): out = [] for p in sorted(glob.glob(os.path.join(TAKES_DIR, "*.jsonl")) if TAKES_DIR else []): if os.path.basename(p)[:-6] in NOT_A_TAKE: continue info = {"name": os.path.basename(p)[:-6], "kb": round(os.path.getsize(p) / 1024.0), "mtime": int(os.path.getmtime(p)), "frames": None} try: with open(p, "rb") as f: meta = json.loads(f.readline()).get("meta", {}) # Count the rest in chunks — a 60Hz take is thousands of lines and there is # no reason to parse any of them just to list the file. n = sum(c.count(b"\n") for c in iter(lambda: f.read(1 << 20), b"")) hz = float(meta.get("hz") or 0) info.update(frames=n, hz=round(hz, 1), arm=bool(meta.get("arm")), hand=bool(meta.get("hand")), secs=round(n / hz, 1) if hz > 0 else None) except Exception: pass out.append(info) # now: so the page can age a take against the ROBOT's clock. Comparing mtime to the # browser's clock would mislabel takes whenever the two hosts disagree. return jsonify(ok=True, ready=rec_ready(), dir=TAKES_DIR, takes=out, now=int(time.time()), total_kb=round(sum(t["kb"] for t in out))) @app.route("/api/take/delete", methods=["POST"]) def api_take_delete(): d = request.get_json(force=True) if d.get("all"): # Wipes the library. The token is required so that a stray {"all": true} from a # script or a mis-click cannot erase recordings that may be impossible to redo — # and the list is re-read here rather than taken from the browser, so whatever the # page happened to be showing does not decide what gets deleted. if d.get("confirm") != "DELETE ALL": return jsonify(ok=False, msg='delete-all needs confirm:"DELETE ALL"') names = [os.path.basename(p)[:-6] for p in glob.glob(os.path.join(TAKES_DIR, "*.jsonl"))] if TAKES_DIR else [] names = [n for n in names if n not in NOT_A_TAKE] if not names: return jsonify(ok=False, msg="no takes to delete") else: names = d.get("names") or [d.get("name", "")] gone, bad = [], [] for n in names[:500]: p = take_path(str(n)) if p and os.path.exists(p): os.remove(p); gone.append(n) else: bad.append(n) return jsonify(ok=bool(gone), deleted=gone, msg=("no such take: %s" % ", ".join(bad)) if bad else "") @app.route("/api/take/duplicate", methods=["POST"]) def api_take_duplicate(): """Copy a take under a free name — the safe way to try a destructive edit (a trim, a rename scheme) without risking the only copy of a pose you cannot easily re-record.""" d = request.get_json(force=True) src = take_path(str(d.get("name", ""))) if not src or not os.path.exists(src): return jsonify(ok=False, msg="no such take") stem = os.path.basename(src)[:-6] for i in range(1, 100): # same naming the recorder uses: name(1) dst = take_path("%s(%d)" % (stem, i)) if dst and not os.path.exists(dst): break else: return jsonify(ok=False, msg="too many copies") with open(src, "rb") as a, open(dst, "wb") as b: b.write(a.read()) return jsonify(ok=True, name=os.path.basename(dst)[:-6]) @app.route("/api/take/download") def api_take_download(): """Send a take to the browser. Hand-rolled rather than send_file(): the robot's container ships Flask 1.1, where the keyword is attachment_filename, while newer Flask calls it download_name. Takes are a couple of MB, so reading one into memory costs nothing and the endpoint works on both. """ name = request.args.get("name", "") p = take_path(name) if not p or not os.path.exists(p): return jsonify(ok=False, msg="no such take"), 404 with open(p, "rb") as f: data = f.read() return Response(data, mimetype="application/x-ndjson", headers={"Content-Disposition": 'attachment; filename="%s.jsonl"' % name}) @app.route("/api/take/upload", methods=["POST"]) def api_take_upload(): """Bring a take recorded elsewhere (e.g. from the workstation) into DataG1/. Validated, not trusted: the name is re-derived from the upload, and the first line must parse as the recorder's own meta header — otherwise a stray file would sit in the list looking like a take until someone pressed replay on it. """ f = request.files.get("file") if not f: return jsonify(ok=False, msg="no file") stem = os.path.basename(f.filename or "") stem = stem[:-6] if stem.endswith(".jsonl") else stem if not SAFE_NAME.match(stem) or stem in NOT_A_TAKE: return jsonify(ok=False, msg="name must be letters/digits/._- and not %s" % NOT_A_TAKE[0]) data = f.read() try: head = json.loads(data.split(b"\n", 1)[0]) if "meta" not in head: raise ValueError("no meta header") except Exception as e: return jsonify(ok=False, msg="not a recording (%s)" % e) dst = take_path(stem) if not dst: return jsonify(ok=False, msg="bad name") if os.path.exists(dst): if not request.form.get("overwrite"): return jsonify(ok=False, msg="%s already exists" % stem) with open(dst, "wb") as out: out.write(data) return jsonify(ok=True, name=stem, bytes=len(data)) @app.route("/api/take/rename", methods=["POST"]) def api_take_rename(): d = request.get_json(force=True) old, new = take_path(str(d.get("old", ""))), take_path(str(d.get("new", ""))) if not old or not os.path.exists(old): return jsonify(ok=False, msg="no such take") if not new or os.path.exists(new): return jsonify(ok=False, msg="bad or duplicate name") os.rename(old, new) return jsonify(ok=True) HTML = r""" Inspire Hand
Inspire Hand fingers + arm
service bridge ports eth0 wifi

Fingers — arm does not move

ready
right
left

Force follow — press a finger and it moves under your hand

off
Arm it, then push a fingertip pad — that finger closes while you push and holds when you stop. Raise deadband if fingers drift from incidental contact; raise sensitivity if you have to push too hard.
Only publishes while armed, so it never fights the sliders or a recording. Safety: a finger closes while you keep pushing — capped at ~3s for full travel, grip force 400g. Push with a knuckle first.

Joint tracker — watch which joints actually CHANGE

off
right
left
Trace = angle over time (fixed 0..1 scale, so flat really means flat).
Row 1 = angle · Row 2 = Δbase force, current, force limit, temp, ERROR/STATUS.
the exact signal the recorder thresholds on (amber above 12g = a touch it would capture).
Pressing a finger moves FORCE but not ANGLE — the drive is non-backdrivable, so contact cannot bend a joint. Only a command changes the angle.

Diagnostics — live read-back & self-test

reads
right (actual)
left (actual)

Read back — copy the hand's current pose into the sliders

Arm + hand combo ⚠ arm moves

Record / replay ⚠ arm moves

idle
in-take keys R rest · L rest 
idle — nothing running.
takes — DataG1/

Library — manage saved shapes & combos (modify = load → edit → Save over same name)

shapes
combos
""" if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8088) a = ap.parse_args() PORT = a.port # the recorder is told to pull its shapes off this port fix_sigint() # before any take is spawned — see sigint_ok() print("Hand dashboard http://0.0.0.0:%d iface=%s" % (a.port, get_iface())) print(" recorder: %s" % (os.path.join(REC_DIR, REC_SCRIPT) if rec_ready() else "NOT FOUND (set RECORDER_DIR)")) app.run(host="0.0.0.0", port=a.port, threaded=True)