diff --git a/hands/rec/script.json b/hands/rec/script.json new file mode 100644 index 0000000..03d0fd4 --- /dev/null +++ b/hands/rec/script.json @@ -0,0 +1,178 @@ +{ + "version": 1, + "intro_s": 4, + "between_s": 3, + "welcome": {"title": "Hand recorder", "seconds": 7, + "text": "Follow the steps on this panel. Each one has a bar that shows its time left.|Pause: Space in the Hand recorder window."}, + "done": {"title": "All done", "seconds": 6, + "text": "Thank you! You can take off the headset.|Review your recordings in the Hand recorder window."}, + "stopped": {"title": "Stopped", "seconds": 4, + "text": "What was recorded so far is saved. Review it in the Hand recorder window."}, + "objects": {"pencil": "pencil or pen", "phone": "phone", "cup": "cup", "keyboard": "keyboard", + "mouse": "mouse", "gamepad": "gamepad", "small": "small thing"}, + "sections": [ + {"id": "hand-size", "title": "Hand size", "requires": [], + "intro": "First, your hand size. Hold each pose still until the bar runs out.", + "defaults": {"hands": "both", "distance": "mid", "position": "centre"}, + "prompts": [ + {"text": "Both hands flat, palms toward you, about 40 cm away. Fingers together.", "seconds": 8, "pose": "flat"}, + {"text": "Turn them around: backs of your hands toward you.", "seconds": 6, "pose": "flat-back"}, + {"text": "Palms toward you again, fingers spread wide.", "seconds": 6, "pose": "spread"} + ]}, + + {"id": "static-poses", "title": "Hand poses", "requires": [], + "intro": "Now some poses. Close means a hand's length away. Far means your arm stretched out.", + "prompts": [ + {"text": "Both hands: open and relaxed. Halfway out, in front of you.", "seconds": 5, "hands": "both", "pose": "open", "distance": "mid", "position": "centre"}, + {"text": "Both hands: make fists. Close to you, off to your left.", "seconds": 5, "hands": "both", "pose": "fist", "distance": "near", "position": "left"}, + {"text": "Both hands: point with your index fingers. Arms out, off to your right.", "seconds": 5, "hands": "both", "pose": "point", "distance": "far", "position": "right"}, + {"text": "Both hands: pinch, thumb and index tips touching. Halfway out, up high.", "seconds": 5, "hands": "both", "pose": "pinch", "distance": "mid", "position": "up"}, + {"text": "Both hands: make an OK sign. Close to you, down low.", "seconds": 5, "hands": "both", "pose": "ok", "distance": "near", "position": "down"}, + {"text": "Both hands: thumbs up. Arms out, in front of you.", "seconds": 5, "hands": "both", "pose": "thumbs-up", "distance": "far", "position": "centre"}, + {"text": "Both hands: spread your fingers wide. Halfway out, off to your left.", "seconds": 5, "hands": "both", "pose": "spread", "distance": "mid", "position": "left"}, + {"text": "Both hands: curl your fingers like claws. Close to you, off to your right.", "seconds": 5, "hands": "both", "pose": "claw", "distance": "near", "position": "right"}, + + {"text": "Now your left hand only. Rest your right hand in your lap.", "seconds": 4, "hands": "left"}, + {"text": "Left hand: open and relaxed. Close to you, off to your right.", "seconds": 5, "hands": "left", "pose": "open", "distance": "near", "position": "right"}, + {"text": "Left hand: a fist. Arm out, up high.", "seconds": 5, "hands": "left", "pose": "fist", "distance": "far", "position": "up"}, + {"text": "Left hand: point with your index finger. Halfway out, down low.", "seconds": 5, "hands": "left", "pose": "point", "distance": "mid", "position": "down"}, + {"text": "Left hand: pinch. Close to you, in front of you.", "seconds": 5, "hands": "left", "pose": "pinch", "distance": "near", "position": "centre"}, + {"text": "Left hand: an OK sign. Arm out, off to your left.", "seconds": 5, "hands": "left", "pose": "ok", "distance": "far", "position": "left"}, + {"text": "Left hand: thumbs up. Halfway out, off to your right.", "seconds": 5, "hands": "left", "pose": "thumbs-up", "distance": "mid", "position": "right"}, + {"text": "Left hand: spread your fingers wide. Close to you, up high.", "seconds": 5, "hands": "left", "pose": "spread", "distance": "near", "position": "up"}, + {"text": "Left hand: curl your fingers like a claw. Arm out, down low.", "seconds": 5, "hands": "left", "pose": "claw", "distance": "far", "position": "down"}, + {"text": "Left hand, halfway out: show one finger.", "seconds": 4, "hands": "left", "pose": "count-1", "distance": "mid", "position": "centre"}, + {"text": "Two fingers.", "seconds": 4, "hands": "left", "pose": "count-2", "distance": "mid", "position": "centre"}, + {"text": "Three fingers.", "seconds": 4, "hands": "left", "pose": "count-3", "distance": "mid", "position": "centre"}, + {"text": "Four fingers.", "seconds": 4, "hands": "left", "pose": "count-4", "distance": "mid", "position": "centre"}, + {"text": "Five fingers.", "seconds": 4, "hands": "left", "pose": "count-5", "distance": "mid", "position": "centre"}, + + {"text": "Now your right hand only. Rest your left hand in your lap.", "seconds": 4, "hands": "right"}, + {"text": "Right hand: open and relaxed. Arm out, down low.", "seconds": 5, "hands": "right", "pose": "open", "distance": "far", "position": "down"}, + {"text": "Right hand: a fist. Halfway out, in front of you.", "seconds": 5, "hands": "right", "pose": "fist", "distance": "mid", "position": "centre"}, + {"text": "Right hand: point with your index finger. Close to you, off to your left.", "seconds": 5, "hands": "right", "pose": "point", "distance": "near", "position": "left"}, + {"text": "Right hand: pinch. Arm out, off to your right.", "seconds": 5, "hands": "right", "pose": "pinch", "distance": "far", "position": "right"}, + {"text": "Right hand: an OK sign. Halfway out, up high.", "seconds": 5, "hands": "right", "pose": "ok", "distance": "mid", "position": "up"}, + {"text": "Right hand: thumbs up. Close to you, down low.", "seconds": 5, "hands": "right", "pose": "thumbs-up", "distance": "near", "position": "down"}, + {"text": "Right hand: spread your fingers wide. Arm out, in front of you.", "seconds": 5, "hands": "right", "pose": "spread", "distance": "far", "position": "centre"}, + {"text": "Right hand: curl your fingers like a claw. Halfway out, off to your left.", "seconds": 5, "hands": "right", "pose": "claw", "distance": "mid", "position": "left"}, + {"text": "Right hand, halfway out: show one finger.", "seconds": 4, "hands": "right", "pose": "count-1", "distance": "mid", "position": "centre"}, + {"text": "Two fingers.", "seconds": 4, "hands": "right", "pose": "count-2", "distance": "mid", "position": "centre"}, + {"text": "Three fingers.", "seconds": 4, "hands": "right", "pose": "count-3", "distance": "mid", "position": "centre"}, + {"text": "Four fingers.", "seconds": 4, "hands": "right", "pose": "count-4", "distance": "mid", "position": "centre"}, + {"text": "Five fingers.", "seconds": 4, "hands": "right", "pose": "count-5", "distance": "mid", "position": "centre"} + ]}, + + {"id": "wrist-rotations", "title": "Wrist turns", "requires": [], + "intro": "Slow wrist turns. Keep your fingers relaxed.", + "defaults": {"pose": "open", "distance": "mid"}, + "prompts": [ + {"text": "Both hands open, palms toward you. Slowly turn them palms away, then back.", "seconds": 8, "hands": "both", "pose": "turn-in-out"}, + {"text": "Left hand, fingers up. Slowly turn your wrist: palm in, palm out.", "seconds": 7, "hands": "left", "pose": "turn-in-out"}, + {"text": "Right hand, fingers up. Slowly turn your wrist: palm in, palm out.", "seconds": 7, "hands": "right", "pose": "turn-in-out"}, + {"text": "Both hands flat, palms down. Slowly turn them palms up, then down again.", "seconds": 8, "hands": "both", "pose": "turn-up-down"}, + {"text": "Bend both wrists slowly up and down, then side to side.", "seconds": 8, "hands": "both", "pose": "bend"} + ]}, + + {"id": "gestures", "title": "Gestures", "requires": [], + "intro": "Now some movements. Go at a calm, normal pace.", + "defaults": {"hands": "both"}, + "prompts": [ + {"text": "Tap your thumb and index finger together a few times, both hands.", "seconds": 6, "pose": "pinch-tap"}, + {"text": "Pinch, move your hand slowly to the side, then let go. Do it again.", "seconds": 8, "pose": "pinch-drag"}, + {"text": "Close your hands as if grabbing a bar, then open them. A few times.", "seconds": 6, "pose": "grab"}, + {"text": "Cross your hands in front of you, then uncross them. Slowly.", "seconds": 8, "pose": "cross"}, + {"text": "Hold one hand over the other, palms toward you, and slide them around.", "seconds": 6, "pose": "overlap"}, + {"text": "Bring your hands close to your face and move them around it. Don't touch the headset.", "seconds": 6, "pose": "near-face", "distance": "near"}, + {"text": "Reach out to where a screen would be, about arm's length, and point at things on it.", "seconds": 6, "pose": "screen-point", "distance": "far"} + ]}, + + {"id": "desk-work", "title": "Desk work", "requires": [], + "intro": "Now your desk. Look down at your hands while you work.", + "defaults": {"hands": "both", "position": "down"}, + "prompts": [ + {"when": "objects:keyboard", "text": "Type on your keyboard the way you normally do. Any text is fine.", "seconds": 10, "pose": "typing", "object": "keyboard"}, + {"when": "!objects:keyboard", "text": "Rest your hands on the desk and pretend to type.", "seconds": 10, "pose": "typing"}, + {"when": "objects:keyboard", "text": "Lift your hands off the keyboard and put them back, a few times.", "seconds": 6, "pose": "lift", "object": "keyboard"}, + {"when": "objects:mouse", "text": "Use your mouse: move it around and click a few times.", "seconds": 10, "hands": "right", "pose": "mouse", "object": "mouse"}, + {"when": "!objects:mouse", "text": "Put one hand flat on the desk and slide it around, as if using a mouse.", "seconds": 8, "hands": "right", "pose": "mouse"}, + {"when": "objects:keyboard", "text": "Switch between typing and the mouse a few times.", "seconds": 8, "pose": "switch", "object": "keyboard"} + ]}, + + {"id": "objects", "title": "Things you hold", "requires": ["objects"], "for_each": "object", + "skip_objects": ["keyboard", "mouse"], + "intro": "Now the things you set out. Use each one the way you normally would.", + "defaults": {"hands": "both"}, + "prompts": [ + {"text": "Pick up the {object}, use it for a moment, then put it down.", "seconds": 12, "pose": "hold", "object": "{object}"} + ]}, + + {"id": "touch", "title": "Touch the dot", "kind": "targets", "requires": [], + "intro": "A small dot will appear. Touch it with your index fingertip and hold still until it turns green.", + "defaults": {"hands": "any", "pose": "point"}, + "hold_s": 1.0, "timeout_s": 8, + "text": "Touch the dot with your index fingertip and hold still.", + "targets": [[0.00, -0.15, -0.40], [-0.20, -0.10, -0.40], [0.20, -0.10, -0.40], [0.00, -0.05, -0.50], + [-0.15, -0.25, -0.35], [0.15, -0.25, -0.35], [0.00, -0.20, -0.25], [-0.25, -0.05, -0.50], + [0.25, -0.05, -0.50], [0.05, -0.10, -0.55]]}, + + {"id": "controller-push", "title": "Depth with controllers", "kind": "bar", "requires": ["controllers"], + "before": {"text": "Put on both controllers and tighten the straps.|Take your time: Space pauses.", "seconds": 25}, + "intro": "Follow the marker on the bar with both hands: out as it moves right, back as it moves left.", + "defaults": {"hands": "both", "pose": "open", "controller": true, "controllers": ["left", "right"]}, + "near_m": 0.2, "far_m": 0.6, "period_s": 6, "lead_s": 3, "reps": 5, + "heights": [ + {"id": "chest", "text": "Chest height: hands open, palms out like a stop sign. Push out and back with the marker."}, + {"id": "desk", "text": "Now at desk height, palms out. Push out and back with the marker."}, + {"id": "eye", "text": "Now at eye level, palms out. Push out and back with the marker."}, + {"id": "left", "text": "Now off to your left side. Push out and back with the marker.", "reps": 2}, + {"id": "right", "text": "Now off to your right side. Push out and back with the marker.", "reps": 2} + ], + "prompts": [ + {"text": "Slowly turn your wrists: palms out, palms toward you, and back.", "seconds": 8, "pose": "turn-in-out", "distance": "mid"}, + {"text": "Open and close your hands a few times.", "seconds": 6, "pose": "open-close", "distance": "mid"} + ]}, + + {"id": "bridge", "title": "Controller and fingertip", "requires": ["controllers"], + "intro": "Now one controller on and one hand bare. The thumbstick is the point to touch.", + "defaults": {"controller": true}, + "prompts": [ + {"text": "Keep the left controller on. Take the right one off and put it on the desk.", "seconds": 8, "hands": "both", "controllers": ["left"]}, + {"text": "Left hand close to you. Touch the left thumbstick with your right index fingertip.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "near", "controllers": ["left"]}, + {"text": "Keep touching it while you move your left hand halfway out.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "mid", "controllers": ["left"]}, + {"text": "Keep touching it with your left arm stretched out.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "far", "controllers": ["left"]}, + {"text": "Swap: put the right controller on, then take the left one off.", "seconds": 12, "hands": "both"}, + {"text": "Right hand close to you. Touch the right thumbstick with your left index fingertip.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "near", "controllers": ["right"]}, + {"text": "Keep touching it while you move your right hand halfway out.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "mid", "controllers": ["right"]}, + {"text": "Keep touching it with your right arm stretched out.", "seconds": 6, "hands": "both", "pose": "touch-stick", "distance": "far", "controllers": ["right"]}, + {"text": "Take off the right controller and put it on the desk in front of you.", "seconds": 8, "hands": "both"}, + {"text": "Touch its thumbstick with your index fingertip, from the front.", "seconds": 5, "hands": "any", "pose": "touch-stick", "position": "down"}, + {"text": "Now touch the thumbstick reaching from the left.", "seconds": 5, "hands": "any", "pose": "touch-stick", "position": "left"}, + {"text": "Now from the right.", "seconds": 5, "hands": "any", "pose": "touch-stick", "position": "right"}, + {"text": "Now from straight above, with your other hand.", "seconds": 5, "hands": "any", "pose": "touch-stick", "position": "up"} + ]}, + + {"id": "bare-push", "title": "Depth, bare hands", "kind": "bar", "requires": [], + "before": {"when": "controllers", "text": "Take the controllers off and put them out of view.", "seconds": 15}, + "intro": "Follow the marker on the bar with both hands: out as it moves right, back as it moves left.", + "defaults": {"hands": "both", "pose": "open"}, + "near_m": 0.2, "far_m": 0.6, "period_s": 6, "lead_s": 3, "reps": 4, + "heights": [ + {"id": "chest", "text": "Chest height: hands open, palms out like a stop sign. Push out and back with the marker."}, + {"id": "desk", "text": "Now at desk height, palms out. Push out and back with the marker."}, + {"id": "eye", "text": "Now at eye level, palms out. Push out and back with the marker."}, + {"id": "left", "text": "Now off to your left side. Push out and back with the marker.", "reps": 2}, + {"id": "right", "text": "Now off to your right side. Push out and back with the marker.", "reps": 2} + ], + "prompts": [ + {"text": "Slowly turn your wrists: palms out, palms toward you, and back.", "seconds": 8, "pose": "turn-in-out", "distance": "mid"}, + {"text": "Open and close your hands a few times.", "seconds": 6, "pose": "open-close", "distance": "mid"} + ]}, + + {"id": "no-hands", "title": "No hands", "requires": [], + "intro": "Last one: put your hands down, out of view.", + "prompts": [ + {"text": "Keep your hands out of view and look slowly around the room.", "seconds": 10, "hands": "none"} + ]} + ] +} diff --git a/hands/rec/session.py b/hands/rec/session.py new file mode 100755 index 0000000..5e8663c --- /dev/null +++ b/hands/rec/session.py @@ -0,0 +1,1506 @@ +#!/usr/bin/env python3 +"""The hand recorder's session runner (DESIGN.md: "Processes during a session", "Files", "The +script", "Feedback while recording", "Controls"). + +It reads script.json, starts what the session needs (ft-camd and a tracking ft-hands, only if +they aren't running), and runs the sections in order. Each section is one take: a recording +ft-hands child process (--record-only, 10 sets a second) plus prompts.jsonl, take.json and the +panel's poses.jsonl. The headset panel (ft-handpanel) shows the prompts; the live hands file +gives feedback ("I can't see your left hand"). + +Plain Python, standard library only: ft_handrec.py imports it, and it runs from the command line +for testing (in the dev container): + + python3 hands/rec/session.py --dry-run --speed 20 # no processes: prints the panel commands + python3 hands/rec/session.py --ring /tmp/ring --base /tmp/hr # ft-ringplay's frames, no headset needed + +All _ns times are CLOCK_MONOTONIC nanoseconds. +""" +import argparse +import glob +import json +import math +import mmap +import os +import re +import shutil +import signal +import socket +import struct +import subprocess +import sys +import threading +import time +import traceback + +HERE = os.path.dirname(os.path.abspath(__file__)) +REPO = os.path.dirname(os.path.dirname(HERE)) +HANDS = os.path.join(REPO, "hands") +FT_HANDS = os.path.join(HANDS, "build", "ft-hands") +FT_CAMD = os.path.join(HANDS, "build", "ft-camd") +PANEL_BIN = os.path.join(HERE, "build", "ft-handpanel") +SCRIPT_PATH = os.path.join(HERE, "script.json") +BASE_DIR = os.path.expanduser("~/.local/share/frametop/hands/contrib") +PANEL_SOCKET = "ft_handpanel" +CAMD_UNIT = "frametop-handrec-camd.service" +HANDS_UNIT = "frametop-handrec-hands.service" + +RECORD_HZ = 10 +TICK_S = 0.05 # the session loop's step (real time) +FRESH_S = 0.3 # the hands file counts as live if published this recently +LOST_S = 1.5 # an asked-for hand lost this long gets a note +CONTROLLER_LOST_S = 1.0 # a controller off 200 (Running_OK) this long gets a note +TOUCH_M = 0.03 # touch the dot: the index tip within this of the dot +MIN_FREE = 1.5e9 # stop the session before the disk fills +RESUME_HINT = "Paused. Resume: Space in the Hand recorder window" + + +def mono_ns(): + return time.clock_gettime_ns(time.CLOCK_MONOTONIC) + + +def run_dir(): + return "/run/user/%d/frametop-hands" % os.getuid() + + +def in_container(): + return os.path.exists("/run/.containerenv") or os.path.exists("/.dockerenv") + + +def host_command(*cmd): + """argv to run a command on the SteamOS host from the dev container (as in + display-settings: distrobox-host-exec needs the user's real session bus).""" + exe = shutil.which("distrobox-host-exec") + if not in_container() or not exe: + return list(cmd) + return ["env", "DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/%d/bus" % os.getuid(), exe] + list(cmd) + + +def host_path(path): + """A host file, from the container (/run/host) or the host itself.""" + for p in ("/run/host" + path, path): + if os.path.exists(p): + return p + return None + + +def write_json(path, data): + tmp = path + ".tmp" + with open(tmp, "w") as f: + json.dump(data, f, indent=1) + f.write("\n") + os.replace(tmp, path) + + +def clean_text(s): + """Text for a panel command: one line; "|" is the panel's line break, so callers use it.""" + return " ".join(str(s).replace("\r", " ").replace("\n", " ").split()) + + +# ------------------------------------------------------------------------------------------ +# The camera ring (camd/fhring.h; tools/ring.py's layout, struct only) + +RING_HDR = struct.Struct("<8sIIIIQqQ16x") # 64 bytes +RING_CAM = struct.Struct("<32s32siIIIIIQQQQQIf24x") # 160 bytes +RING_SLOT = struct.Struct("": {"mean": float, "dark_mean": float}}, the + newest frames' mean luma and ft-camd's near-black frame mean (the room's IR light), averaged + over about a second. None if no camera ring is running.""" + path = ring_path or os.path.join(run_dir(), "cam-ring") + try: + ring = Ring(path) + except (OSError, ValueError): + return None + try: + if not ring.alive(): + return None + sums = {} + for k in range(samples): + for cam in ring.mono(): + m = ring.frame_mean(cam) + s = sums.setdefault(cam["name"], [0.0, 0, 0.0, 0]) + if m is not None: + s[0] += m + s[1] += 1 + s[2] += ring.dark_mean(cam) + s[3] += 1 + if k + 1 < samples: + time.sleep(interval) + out = {name: {"mean": round(s[0] / s[1], 2) if s[1] else 0.0, + "dark_mean": round(s[2] / s[3], 2) if s[3] else 0.0} for name, s in sums.items()} + return out or None + finally: + ring.close() + + +def _light_levels(ring): + vals = [v for v in (ring or {}).values() if isinstance(v, dict)] + if not vals: + return None + return (sum(float(v.get("mean", 0)) for v in vals) / len(vals), + sum(float(v.get("dark_mean", 0)) for v in vals) / len(vals)) + + +def _close(a, b, share=0.15, floor=1.0): + """Within 15% of the larger; the floor keeps near-black levels (a dim room's dark mean is a + few levels) from looking different over sensor noise.""" + return abs(a - b) <= max(share * max(abs(a), abs(b)), floor) + + +def similar_lighting(base_dir, lighting): + """An earlier session whose cameras saw about the same light (the mean of every mono + camera's mean and dark_mean, each within 15%): (session_id, chosen), the latest such, or + None. lighting is ring_lighting()'s dict, or session.json's {"chosen", "ring"}.""" + if not lighting: + return None + ring = lighting.get("ring") if isinstance(lighting.get("ring"), dict) else lighting + now = _light_levels(ring) + if not now: + return None + for path in sorted(glob.glob(os.path.join(base_dir, "sessions", "*", "session.json")), reverse=True): + try: + with open(path) as f: + light = json.load(f).get("lighting") or {} + except (OSError, ValueError): + continue + then = _light_levels(light.get("ring")) + if then and _close(now[0], then[0]) and _close(now[1], then[1]): + return os.path.basename(os.path.dirname(path)), light.get("chosen", "") + return None + + +# ------------------------------------------------------------------------------------------ +# The factory calibration, without what identifies the unit + +ID_TOKENS = {"serial", "sn", "uuid", "guid", "mac", "id", "ids", "identifier"} + + +def _key_identifies(key): + k = str(key) + if re.search(r"serial|uuid", k, re.I): + return True + tokens = re.findall(r"[A-Z]+(?![a-z])|[A-Z]?[a-z]+|\d+", k) + return any(t.lower() in ID_TOKENS for t in tokens) + + +def _value_identifies(value): + """A string holding something like a serial number: a run of 8 or more letters and digits + with at least 4 digits and a letter (sensor names like og01a1b are shorter).""" + if not isinstance(value, str): + return False + for run in re.findall(r"[A-Za-z0-9]+", value): + if len(run) >= 8 and sum(c.isdigit() for c in run) >= 4 and any(c.isalpha() for c in run): + return True + return False + + +def strip_calibration(xrservice_json_dict): + """/persist/xrservice.json without what identifies the unit: keys naming a serial, sn, + uuid, mac or id, and string values that look like serial numbers. Returns (cleaned dict, + removed key paths such as "cameras[0].serial").""" + removed = [] + + def walk(o, path): + if isinstance(o, dict): + out = {} + for k, v in o.items(): + p = "%s.%s" % (path, k) if path else str(k) + if _key_identifies(k) or _value_identifies(v): + removed.append(p) + continue + out[k] = walk(v, p) + return out + if isinstance(o, list): + out = [] + for i, v in enumerate(o): + p = "%s[%d]" % (path, i) + if _value_identifies(v): + removed.append(p) + out.append(None) # keep the other items' positions + continue + out.append(walk(v, p)) + return out + return o + + return walk(xrservice_json_dict, ""), removed + + +# ------------------------------------------------------------------------------------------ +# The live hands file (include/fh_hands.h) + +HANDS_HDR = struct.Struct("<8sIIQQQII16x") # 64 bytes +HAND = struct.Struct(" FRESH_S: + return None + out = {"left": None, "right": None} + for k in range(min(nhands, 2)): + f = HAND.unpack_from(data, HANDS_HDR.size + k * HAND.size) + side = "right" if f[1] & FH_HAND_RIGHT else "left" + if out[side]: + continue + pts = [f[4 + 3 * i: 7 + 3 * i] for i in range(21)] + palm = [sum(pts[i][j] for i in PALM) / len(PALM) for j in range(3)] + out[side] = {"palm": palm, "palm_m": math.sqrt(sum(v * v for v in palm)), + "tip": list(pts[INDEX_TIP])} + return out + return None + finally: + os.close(fd) + + +# ------------------------------------------------------------------------------------------ +# The panel (ft-handpanel, @ft_handpanel) + +class Panel: + """Commands to ft-handpanel over its datagram socket. Commands that need an answer wait + for it; the rest don't, and their replies are read (and errors logged) later. With + dry=True nothing is sent: commands are printed and replies made up.""" + + def __init__(self, name=PANEL_SOCKET, dry=False, log=None, out=None): + self.name, self.dry, self.log, self.out = name, dry, log or (lambda s: None), out or print + self.sock = None + self.sent = {} + self.pending = 0 # commands sent and not answered yet + if not dry: + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC) + self.sock.bind("\0ft_handrec.%d.%d" % (os.getpid(), id(self) & 0xffff)) + self.sock.setblocking(False) + + def close(self): + if self.sock: + self.sock.close() + self.sock = None + + def _drain(self): + while self.sock: + try: + r = self.sock.recv(4096).decode(errors="replace") + except (BlockingIOError, OSError): + return + self.pending = max(0, self.pending - 1) + if r.startswith("error"): + self.log("panel: %s" % r) + + def cmd(self, text, reply=False, timeout=0.5): + """Send a command; with reply=True, its answer ("ok ..." or "error ..."), or None. + The panel answers every command in order, so the answer is the one after those of + the commands still unanswered.""" + if self.dry: + self.out("panel: %s" % text) + if not reply: + return None + if text.startswith("devices"): + return "ok hmd 200 left 200 right 200" + return "ok shown" if text == "ping" else "ok" + if not self.sock: + return None + self._drain() + try: + self.sock.sendto(text.encode(), "\0" + self.name) + except OSError: + return None + self.pending += 1 + if not reply: + return None + end = time.monotonic() + timeout + while time.monotonic() < end: + try: + self.sock.settimeout(max(0.01, end - time.monotonic())) + r = self.sock.recv(4096).decode(errors="replace") + except OSError: + break + finally: + self.sock.setblocking(False) + self.pending = max(0, self.pending - 1) + if self.pending == 0: + return r + if r.startswith("error"): + self.log("panel: %s" % r) + self.pending = 0 # a lost answer: start counting afresh + return None + + def set(self, key, text): + """Send a command only if it changes what that part of the panel shows.""" + if self.sent.get(key) == text: + return + self.sent[key] = text + self.cmd(text) + + +# ------------------------------------------------------------------------------------------ +# Processes + +def find_processes(name): + """[(pid, argv)] of running processes called name (the container shares the host's PIDs).""" + out = [] + for pid in os.listdir("/proc"): + if not pid.isdigit(): + continue + try: + with open("/proc/%s/comm" % pid) as f: + if f.read().strip() != name: + continue + with open("/proc/%s/cmdline" % pid, "rb") as f: + argv = [a.decode(errors="replace") for a in f.read().split(b"\0") if a] + except OSError: + continue + out.append((int(pid), argv)) + return out + + +def tracker_running(): + """A tracking ft-hands (not a --record-only recorder).""" + return any("--record-only" not in argv for _, argv in find_processes("ft-hands")) + + +class Recorder: + """One part of a take's recording: ft-hands --record-only as a child process. Part 1 + writes TAKE/sets.bin; part N (after a pause) records into TAKE/.part-N and is moved to + TAKE/sets-N.bin when it ends (ft-hands never overwrites, and always writes DIR/sets.bin).""" + + def __init__(self, take_dir, part, seconds, ring, log_file): + self.take_dir, self.part = take_dir, part + self.dir = take_dir if part == 1 else os.path.join(take_dir, ".part-%d" % part) + argv = [FT_HANDS, "--record-only", "--record", self.dir, "--record-for", "%.0f" % max(seconds, 5), + "--record-hz", str(RECORD_HZ), "--status", "0"] + if ring: + argv += ["--ring", ring] + if not in_container(): + argv = [os.path.expanduser("~/.local/bin/distrobox"), "enter", "dev", "--"] + argv + self.proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=log_file, stderr=log_file) + self.started_ns = mono_ns() + + def exited(self): + return self.proc.poll() is not None + + def stop(self): + """End it (SIGTERM: ft-hands writes out its queue) and put the part in place.""" + if self.proc.poll() is None: + self.proc.send_signal(signal.SIGTERM) + try: + self.proc.wait(15) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait() + if self.part > 1: + src = os.path.join(self.dir, "sets.bin") + if os.path.exists(src): + os.replace(src, os.path.join(self.take_dir, "sets-%d.bin" % self.part)) + try: + os.rmdir(self.dir) + except OSError: + pass + return self.proc.returncode + + +# ------------------------------------------------------------------------------------------ +# The script + +HAND_VALUES = ("left", "right", "both", "none", "any", "") +PROMPT_KEYS = ("text", "seconds", "hands", "pose", "distance", "position", "object", "controller") + + +def _when_ok(cond, ctx): + """A prompt's or block's "when": "controllers", "objects:keyboard", or either negated with "!".""" + if not cond: + return True + neg = cond.startswith("!") + c = cond.lstrip("!") + value = c[len("objects:"):] in ctx["objects"] if c.startswith("objects:") else bool(ctx.get(c)) + return value != neg + + +def _prompt(section, raw, ctx, subst=None): + p = dict(section.get("defaults") or {}) + p.update(raw) + p.pop("when", None) + for k, v in (subst or {}).items(): + for key in ("text", "object"): + if isinstance(p.get(key), str): + p[key] = p[key].replace("{%s}" % k, v) + if subst and "object_key" in subst: + p["object"] = subst["object_key"] + for key in PROMPT_KEYS: + p.setdefault(key, False if key == "controller" else 0 if key == "seconds" else "") + p["controllers"] = list(p.get("controllers") or []) + if p["hands"] not in HAND_VALUES: + raise ValueError("%s: hands %r" % (section["id"], p["hands"])) + return p + + +def _set_ids(section, prompts): + seen = {} + for p in prompts: + if not p.get("id"): + parts = [section["id"]] + [re.sub(r"[^a-z0-9]+", "-", str(p[k]).lower()).strip("-") + for k in ("pose", "hands", "distance", "position", "object") if p[k]] + p["id"] = "/".join(parts) if len(parts) > 1 else "%s/%d" % (section["id"], len(seen) + 1) + n = seen.get(p["id"], 0) + 1 + seen[p["id"]] = n + if n > 1: + p["id"] = "%s#%d" % (p["id"], n) + + +def load_script(path): + with open(path) as f: + script = json.load(f) + if script.get("version") != 1 or not isinstance(script.get("sections"), list): + raise ValueError("%s: not a version 1 script" % path) + return script + + +def build_plan(script, checklist): + """The sections this session runs, prompts expanded for the checklist, and the ones + skipped: (plan, skipped) with skipped = [{"section", "reason"}].""" + objects = [o for o in (checklist.get("objects") or []) if o] + own = [o for o in (checklist.get("own_objects") or []) if str(o).strip()] + ctx = {"objects": set(objects) | set(own), "controllers": checklist.get("controllers") == "straps"} + names = script.get("objects") or {} + plan, skipped = [], [] + for raw in script["sections"]: + sid = raw["id"] + missing = [r for r in raw.get("requires") or [] if not ctx.get(r)] + if missing: + skipped.append({"section": sid, "reason": "needs " + ", ".join(missing)}) + continue + s = dict(raw) + s["kind"] = raw.get("kind", "prompts") + s["intro_s"] = float(raw.get("intro_s", script.get("intro_s", 4))) + prompts = [] + if raw.get("for_each") == "object": + skip = set(raw.get("skip_objects") or []) + for o in [o for o in objects if o not in skip] + own: + label = clean_text(names.get(o, o)).replace("|", "/") + for p in raw.get("prompts") or []: + if _when_ok(p.get("when"), ctx): + prompts.append(_prompt(s, p, ctx, {"object": label, "object_key": clean_text(o)})) + else: + prompts = [_prompt(s, p, ctx) for p in raw.get("prompts") or [] if _when_ok(p.get("when"), ctx)] + _set_ids(s, prompts) + s["prompts"] = prompts + if s["kind"] == "bar": + heights = [] + for h in raw.get("heights") or []: + h = {"id": h} if isinstance(h, str) else dict(h) + h.setdefault("text", (raw.get("height_text") or {}).get(h["id"], raw.get("text", ""))) + h.setdefault("reps", raw.get("reps", 5)) + heights.append(h) + s["heights"] = heights + before = raw.get("before") + s["before"] = before if before and _when_ok(before.get("when"), ctx) else None + if s["kind"] == "prompts" and not prompts: + skipped.append({"section": sid, "reason": "nothing to do"}) + continue + if s["kind"] == "targets" and not raw.get("targets"): + skipped.append({"section": sid, "reason": "no targets"}) + continue + plan.append(s) + return plan, skipped + + +def section_seconds(s, worst=False): + """A section's length in script seconds (targets: about 4 s each, or the timeout if worst).""" + t = s["intro_s"] + sum(p["seconds"] for p in s["prompts"]) + if s["kind"] == "targets": + t += len(s["targets"]) * (s.get("timeout_s", 8) if worst else min(4.0, s.get("timeout_s", 8))) + if s["kind"] == "bar": + t += sum(s.get("lead_s", 3) + h["reps"] * s.get("period_s", 6) for h in s["heights"]) + return t + + +def plan_seconds(script, plan, worst=False): + t = (script.get("welcome") or {}).get("seconds", 0) + (script.get("done") or {}).get("seconds", 0) + for i, s in enumerate(plan): + t += section_seconds(s, worst) + t += s["before"]["seconds"] if s.get("before") else (script.get("between_s", 3) if i else 0) + return t + + +# ------------------------------------------------------------------------------------------ +# The session + +class _Skip(Exception): + pass + + +class _Stop(Exception): + pass + + +class _Fail(Exception): + pass + + +def _git_describe(): + try: + r = subprocess.run(["git", "-C", REPO, "describe", "--always", "--dirty", "--tags"], + capture_output=True, text=True, timeout=5) + return r.stdout.strip() or "unknown" + except (OSError, subprocess.TimeoutExpired): + return "unknown" + + +def _os_version(): + path = "/run/host/etc/os-release" if in_container() else "/etc/os-release" + try: + with open(path) as f: + for line in f: + if line.startswith("VERSION_ID="): + return line.split("=", 1)[1].strip().strip('"') + except OSError: + pass + return "" + + +def _steamvr_version(): + for db in ("/usr/lib/holo/pacmandb", "/var/lib/pacman"): # SteamOS keeps it in the image + found = sorted(glob.glob(("/run/host" if in_container() else "") + db + "/local/deckard-steamvr-rel-*")) + if found: + return os.path.basename(found[-1])[len("deckard-steamvr-rel-"):] + p = host_path("/opt/steamvr/bin/version.txt") + if p: + try: + with open(p) as f: + return "build " + f.read().strip() + except OSError: + pass + return "" + + +class Session: + """One recording session. start() runs it in its own thread; pause(), resume(), skip() + and stop() steer it from any thread. on_status(dict) is called from the session thread + whenever something changes (see STATUS_KEYS).""" + + def __init__(self, base_dir, profile, checklist, lighting_choice, script_path, *, ring=None, + start_processes=True, dry_run=False, speed=1.0, on_status=None, hands_dir=None, panel_bin=None): + self.base_dir = os.path.abspath(os.path.expanduser(base_dir or BASE_DIR)) + self.profile = dict(profile or {}) + self.checklist = dict(checklist or {}) + self.lighting_choice = lighting_choice or "" + self.script_path = script_path or SCRIPT_PATH + self.ring = ring + self.start_processes = start_processes + self.dry_run = dry_run + self.speed = max(float(speed or 1.0), 0.01) + self.on_status = on_status + self.hands_dir = hands_dir or run_dir() + self.panel_bin = panel_bin or PANEL_BIN + self.print = print # where dry-run panel commands go (the CLI's stdout) + self.script = load_script(self.script_path) + self.plan, self.skipped = build_plan(self.script, self.checklist) + self.session_dir = "" + self._thread = None + self._lock = threading.Lock() + self._want = {"pause": False, "skip": False, "stop": False} + self._wake = threading.Event() + self._status = {"state": "starting", "section": "", "title": "", "section_index": 0, + "section_count": len(self.plan), "prompt": "", "seconds_left": 0.0, "note": "", + "hands": {"left": None, "right": None}, "take": None, "error": ""} + self._last_emit = 0.0 + self._log_file = None + self._panel = None + self._panel_proc = None + self._units = [] # transient units this session started + self._session_json = None + self._take = None # the take in progress: dict + self._recorder = None + self._prompt = None # the prompt shown: dict (hands, controllers) + self._live = None # the hands file's last read + self._hands_file = HandsFile(os.path.join(self.hands_dir, "hands")) + self._paused = False + self._fb = {} # feedback timers + + # --- controls (any thread) + def start(self): + if self._thread: + raise RuntimeError("the session has started already") + self._thread = threading.Thread(target=self._run, name="handrec-session", daemon=True) + self._thread.start() + + def _set(self, key, value): + with self._lock: + self._want[key] = value + self._wake.set() + + def pause(self): + self._set("pause", True) + + def resume(self): + self._set("pause", False) + + def skip(self): + self._set("skip", True) + + def stop(self, wait=20.0): + """Stop. The take in progress is kept as far as it got. Blocks until the session has + written its files (up to wait seconds), unless called from the session thread.""" + self._set("stop", True) + if self._thread and threading.current_thread() is not self._thread and wait: + self._thread.join(wait) + + def join(self, timeout=None): + if self._thread: + self._thread.join(timeout) + + @property + def state(self): + return self._status["state"] + + # --- status and logging + def _emit(self, force=True, **changes): + self._status.update(changes) + now = time.monotonic() + if not force and now - self._last_emit < 0.25: + return + self._last_emit = now + if self.on_status: + try: + self.on_status(dict(self._status, hands=dict(self._status["hands"]))) + except Exception: + self._log("on_status: " + traceback.format_exc()) + + def _log(self, text): + if self._log_file: + self._log_file.write("%s %s\n" % (time.strftime("%H:%M:%S"), text)) + self._log_file.flush() + + def _event(self, event, **fields): + if not self._take: + return + line = {"t": mono_ns(), "event": event} + line.update(fields) + self._take["prompts"].write(json.dumps(line) + "\n") + self._take["prompts"].flush() + + # --- the run + def _run(self): + try: + self._make_dir() + self._emit(state="starting") + self._setup() + self._screen("starting", self.script.get("welcome") or {}) + for i, s in enumerate(self.plan): + self._section(i, s) + self._finish("done") + except _Stop: + self._finish("stopped") + except _Fail as e: + self._finish("error", str(e)) + except Exception as e: + self._log(traceback.format_exc()) + self._finish("error", "%s: %s" % (type(e).__name__, e)) + + def _make_dir(self): + sessions = os.path.join(self.base_dir, "sessions") + os.makedirs(sessions, exist_ok=True) + sid = time.strftime("%Y%m%d-%H%M%S") + path, n = os.path.join(sessions, sid), 1 + while True: + try: + os.mkdir(path) + break + except FileExistsError: + n += 1 + path = os.path.join(sessions, "%s-%d" % (sid, n)) + os.mkdir(os.path.join(path, "takes")) + self.session_dir = path + self._log_file = open(os.path.join(path, "session.log"), "a", buffering=1) + self._log("session %s%s, script %s" % (os.path.basename(path), " (dry run)" if self.dry_run else "", + self.script_path)) + for s in self.skipped: + self._log("skipping %s: %s" % (s["section"], s["reason"])) + + def _setup(self): + ring_path = self.ring or os.path.join(self.hands_dir, "cam-ring") + if not self.dry_run: + if not os.access(FT_HANDS, os.X_OK): + raise _Fail("ft-hands isn't built: hands/build.sh") + self._ensure_ring(ring_path) + if not tracker_running(): + if self.start_processes: + self._start_tracker() + else: + self._log("no tracking ft-hands found: feedback only if one publishes") + lighting = ring_lighting(ring_path) + cams = [] + try: + ring = Ring(ring_path) + cams = [{"name": c["name"], "width": c["width"], "height": c["height"]} + for c in ring.cams if not c["flags"] & FH_CAM_DARK and not c["name"].endswith("_dk")] + ring.close() + except (OSError, ValueError): + pass + removed = self._write_calibration() + self._session_json = { + "schema": 1, "tool": "ft-handrec " + _git_describe(), "started": time.strftime("%Y-%m-%dT%H:%M:%S%z"), + "contributor": self.profile.get("contributor", ""), + "lighting": {"chosen": self.lighting_choice, "ring": lighting or {}}, + "checklist": self.checklist, + "device": {"steamos": _os_version(), "steamvr": _steamvr_version(), "cameras": cams}, + "calibration_removed": removed, + "script": {"version": self.script.get("version"), "sections": [s["id"] for s in self.plan], + "skipped": self.skipped}, + "takes": [], "status": "recording"} + if self.dry_run: + self._session_json["dry_run"] = True + if self.speed != 1: + self._session_json["speed"] = self.speed + self._save_session() + self._panel = Panel(dry=self.dry_run, log=self._log, out=self.print) + if not self.dry_run: + self._start_panel() + self._panel.cmd("show") + self._panel.cmd("paused off") + for key, c in (("note", "note "), ("countdown", "countdown off"), ("hands", "hands off off"), + ("bar", "bar off"), ("target", "target off")): + self._panel.set(key, c) + + def _write_calibration(self): + src = host_path("/persist/xrservice.json") + if not src: + self._log("no /persist/xrservice.json: no calibration.json") + return [] + with open(src) as f: + clean, removed = strip_calibration(json.load(f)) + write_json(os.path.join(self.session_dir, "calibration.json"), clean) + return removed + + def _save_session(self): + if self._session_json is not None: + write_json(os.path.join(self.session_dir, "session.json"), self._session_json) + + def _ensure_ring(self, path): + def alive(): + try: + r = Ring(path) + except (OSError, ValueError): + return False + try: + return r.alive() + finally: + r.close() + + if alive(): + return + if self.ring: + raise _Fail("No frames in %s (start ft-ringplay first)" % path) + if not self.start_processes: + raise _Fail("ft-camd isn't running (and --no-start)") + if not os.access(FT_CAMD, os.X_OK): + raise _Fail("ft-camd isn't built: hands/build.sh") + caps = subprocess.run(["getcap", FT_CAMD], capture_output=True, text=True) if shutil.which("getcap") else None + if caps is not None and "cap_sys_ptrace" not in caps.stdout: + raise _Fail("ft-camd needs its capabilities: hands/run.sh caps (asks for sudo)") + self._start_unit(CAMD_UNIT, "the camera broker", [FT_CAMD, "--status", "60"]) + for _ in range(150): + if alive(): + return + if self._want["stop"]: + raise _Stop() + time.sleep(0.1) + raise _Fail("ft-camd didn't start (is SteamVR running?): journalctl --user -u " + CAMD_UNIT) + + def _start_tracker(self): + up = os.path.join(REPO, "scripts", "container-up.sh") + if os.access(up, os.X_OK): + subprocess.run(host_command(up), capture_output=True, timeout=120) + argv = [os.path.expanduser("~/.local/bin/distrobox"), "enter", "dev", "--", FT_HANDS, + "--no-gestures", "--status", "0"] + if self.ring: + argv += ["--ring", self.ring] + self._start_unit(HANDS_UNIT, "hand tracking for feedback", argv) + + def _start_unit(self, unit, what, argv): + def active(): + return subprocess.run(host_command("systemctl", "--user", "-q", "is-active", unit), + capture_output=True, timeout=30).returncode == 0 + + if active(): + self._log("%s is running already" % unit) + return + cmd = host_command("systemd-run", "--user", "--quiet", "--collect", "--unit=" + unit, + "--description=Frametop hand recorder: " + what, + "-p", "PartOf=steamvr.service", "-p", "After=steamvr.service", + "-p", "Restart=on-failure", "-p", "RestartSec=3", "-p", "TimeoutStopSec=5", *argv) + for attempt in range(3): # distrobox-host-exec has failed once, silently, and worked again + r = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + if r.returncode == 0 or active(): + self._units.append(unit) + self._log("started %s" % unit) + return + self._log("starting %s failed (exit %d): %s" % (unit, r.returncode, (r.stderr or r.stdout).strip())) + time.sleep(0.5) + raise _Fail("couldn't start %s (exit %d): %s" % (unit, r.returncode, (r.stderr or r.stdout).strip())) + + def _stop_units(self): + for unit in reversed(self._units): + subprocess.run(host_command("systemctl", "--user", "stop", unit), capture_output=True, timeout=30) + self._log("stopped %s" % unit) + self._units = [] + + def _start_panel(self): + if self._panel.cmd("ping", reply=True): + self._log("using the ft-handpanel that's running") + return + if not os.access(self.panel_bin, os.X_OK): + raise _Fail("ft-handpanel isn't built: hands/rec/build.sh") + self._panel_proc = subprocess.Popen([self.panel_bin, "--watch-stdin"], stdin=subprocess.PIPE, + stdout=self._log_file, stderr=self._log_file) + end = time.monotonic() + 15 + while time.monotonic() < end: + if self._panel.cmd("ping", reply=True, timeout=0.2): + return + if self._panel_proc.poll() is not None: + raise _Fail("ft-handpanel exited (is SteamVR running?): see session.log") + time.sleep(0.1) + raise _Fail("ft-handpanel doesn't answer") + + def _stop_panel(self): + if self._panel: + self._panel.cmd("hide") + if self._panel_proc: + try: + self._panel_proc.stdin.close() + self._panel_proc.wait(3) + except (OSError, subprocess.TimeoutExpired): + self._panel_proc.terminate() + try: + self._panel_proc.wait(3) + except subprocess.TimeoutExpired: + self._panel_proc.kill() + self._panel_proc = None + if self._panel: + self._panel.close() + + def _finish(self, state, error=""): + if self._take: + self._end_take({"done": "complete", "stopped": "stopped"}.get(state, "stopped")) + if self._session_json is not None: + self._session_json["status"] = state + self._session_json["ended"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") + if error: + self._session_json["error"] = error + self._save_session() + self._log("%s%s" % (state, ": " + error if error else "")) + self._stop_units() + if self._panel: + screen = {"done": self.script.get("done") or {}, "stopped": self.script.get("stopped") or {}, + "error": {"title": "Something went wrong", "text": "See the Hand recorder window.", + "seconds": 4}}[state] + try: + self._panel.cmd("paused off") + self._screen(state, screen, controls=False) + except (_Stop, _Skip, _Fail): + pass + self._stop_panel() + if self._log_file: + self._log_file.close() + self._log_file = None + screen = {"done": self.script.get("done"), "stopped": self.script.get("stopped")}.get(state) or {} + self._emit(state=state, error=error, seconds_left=0.0, note="", take=None, section="", + prompt=error or screen.get("text", "").replace("|", "\n"), hands={"left": None, "right": None}) + + # --- the timing loop + def _controls(self): + """Handle a pause (blocking until resumed), skip and stop. True if a pause happened.""" + with self._lock: + want = dict(self._want) + self._want["skip"] = False + if want["stop"]: + raise _Stop() + if want["skip"]: + raise _Skip() + if not want["pause"]: + return False + self._pause() + outcome = None + while outcome is None: + self._wake.wait(0.2) + self._wake.clear() + with self._lock: + want = dict(self._want) + self._want["skip"] = False + if want["skip"]: + self._want["pause"] = False + if want["stop"]: + outcome = _Stop + elif want["skip"]: + outcome = _Skip + elif not want["pause"]: + outcome = True + self._unpause(record=outcome is True) + if outcome is not True: + raise outcome() + return True + + def _pause(self): + self._paused = True + self._state_before = self._status["state"] + if self._take: + self._stop_recording() + self._event("pause") + self._panel.cmd("paused on") + self._panel.set("note", "note " + RESUME_HINT) + self._emit(state="paused", note=RESUME_HINT) + self._log("paused") + + def _unpause(self, record=True): + self._paused = False + self._panel.cmd("paused off") + self._panel.set("note", "note ") + self._fb.clear() + if self._take and record: + self._event("resume") + self._start_recording() + self._emit(state=self._state_before, note="") + self._log("resumed") + + def _wait(self, seconds, tick=None, countdown=True): + """Let `seconds` (script time) pass, handling the controls and feedback. tick(dt) runs + every step with the script time passed; it ends the wait early by returning True. + Returns True if tick ended it, else False.""" + left = float(seconds) + last = time.monotonic() + last_cd = -1.0 + while True: + if self._controls(): + last = time.monotonic() + now = time.monotonic() + dt = (now - last) * self.speed + last = now + left -= dt + self._feedback() + if tick and tick(dt): + return True + if countdown and seconds > 0 and (now - last_cd >= 0.25 or left <= 0): + last_cd = now + self._panel.set("countdown", "countdown %.2f" % max(0.0, min(1.0, left / seconds))) + self._emit(force=False, seconds_left=round(max(left, 0.0), 1)) + if left <= 0: + return False + self._wake.wait(TICK_S) + self._wake.clear() + + # --- feedback + def _feedback(self): + now = time.monotonic() + fb = self._fb + if self._panel_proc and self._panel_proc.poll() is not None: + raise _Fail("The headset panel closed (did SteamVR quit?)") + if self._recorder and self._recorder.exited(): + raise _Fail("The recording stopped by itself: see session.log") + if not self.dry_run and now - fb.get("disk", 0) > 5: + fb["disk"] = now + if shutil.disk_usage(self.session_dir).free < MIN_FREE: + raise _Fail("The disk is nearly full: the session stopped") + live = None if self.dry_run else self._hands_file.read() + self._live = live + p = self._prompt or {} + asked = p.get("hands", "") + seen = {s: (bool(live[s]) if live else None) for s in ("left", "right")} + # the chips: the asked-for hands, seen or lost, while the tracker publishes + chips = [] + for side in ("left", "right"): + show = live is not None and (asked in (side, "both", "any")) + chips.append(("seen" if seen[side] else "lost") if show else "off") + self._panel.set("hands", "hands %s %s" % tuple(chips)) + # a note when an asked-for hand stays lost, or a hand shows when none is wanted + notes = [] + if live is not None and asked: + if asked == "none": + missing = [] if not (seen["left"] or seen["right"]) else ["shown"] + elif asked == "any": + missing = [] if (seen["left"] or seen["right"]) else ["any"] + else: + missing = [s for s in ("left", "right") if asked in (s, "both") and not seen[s]] + key = ",".join(missing) + if key != fb.get("lost_key"): + fb["lost_key"], fb["lost_since"] = key, now + if missing and now - fb["lost_since"] > LOST_S: + if missing == ["shown"]: + notes.append("I can see a hand: keep them out of view") + elif missing == ["any"] or len(missing) == 2: + notes.append("I can't see your hands: bring them into view") + else: + notes.append("I can't see your %s hand: bring it into view" % missing[0]) + else: + fb.pop("lost_key", None) + notes = self._controller_feedback(now, p) + notes + note = notes[0] if notes else "" + if not self._paused: + self._panel.set("note", "note " + note) + hands = {"left": seen["left"], "right": seen["right"]} + if note != self._status["note"] or hands != self._status["hands"]: + self._emit(note=note, hands=hands) + if self._take and live is not None and now - fb.get("logged", 0) >= 0.5: + fb["logged"] = now + self._event("feedback", left=seen["left"], right=seen["right"], + palm_m=[round(live[s]["palm_m"], 4) if live[s] else None for s in ("left", "right")]) + + def _controller_feedback(self, now, p): + """Sections with controllers: `devices` once a second; a result other than 200 for + more than a second gets a note, and each change goes into prompts.jsonl.""" + sides = p.get("controllers") or [] + fb = self._fb + if not sides: + fb.pop("ctl", None) + return [] + ctl = fb.setdefault("ctl", {"polled": 0, "r": {}, "bad_since": {}, "lost": []}) + if now - ctl["polled"] >= 1.0: + ctl["polled"] = now + reply = self._panel.cmd("devices", reply=True, timeout=0.3) or "" + tok = reply.split() + vals = dict(zip(tok[1::2], tok[2::2])) + if tok[:1] == ["ok"] and "hmd" in vals: + ctl["r"] = {s: vals.get(s, "-") for s in ("left", "right")} + for s in sides: + r = ctl["r"].get(s) + if r is None or r == "200": + ctl["bad_since"].pop(s, None) + else: + ctl["bad_since"].setdefault(s, now) + lost = sorted(s for s, t in ctl["bad_since"].items() if s in sides and now - t > CONTROLLER_LOST_S) + if lost != ctl["lost"]: + ctl["lost"] = lost + self._event("feedback", controller={s: (None if ctl["r"].get(s) in (None, "-") else int(ctl["r"][s])) + for s in ("left", "right")}, controller_lost=lost) + notes = [] + for s in lost: + if ctl["r"].get(s) == "-": + notes.append("I can't find the %s controller: is it on?" % s) + else: + notes.append("The %s controller lost tracking: turn your palm slightly toward you" % s) + return notes + + # --- sections and takes + def _show(self, title=None, step=None, text=None): + if title is not None: + self._panel.set("title", "title " + clean_text(title)) + if step is not None: + self._panel.set("step", "step " + clean_text(step)) + if text is not None: + self._panel.set("text", "text " + clean_text(text)) + + def _screen(self, state, screen, controls=True): + """A screen of its own (welcome, done): title, text, a few seconds. controls=False: the + session's end, which reports its state once all is done.""" + if not screen: + return + self._prompt = None + self._show(screen.get("title", ""), "", screen.get("text", "")) + self._panel.set("countdown", "countdown off") + if controls: + self._emit(state=state, title=screen.get("title", ""), prompt=screen.get("text", "").replace("|", "\n"), + section="", seconds_left=float(screen.get("seconds", 0))) + try: + self._wait(screen.get("seconds", 0), countdown=False) + except _Skip: + pass + else: + end = time.monotonic() + screen.get("seconds", 0) / self.speed + while time.monotonic() < end and not self._want["stop"]: + time.sleep(TICK_S) + + def _section(self, i, s): + step = "Section %d of %d" % (i + 1, len(self.plan)) + status = {"section": s["id"], "title": s["title"], "section_index": i + 1, "section_count": len(self.plan)} + try: + before = s.get("before") + if before or i > 0: + self._prompt = None + text = before["text"] if before else "Next: %s" % s["title"] + secs = before.get("seconds", 10) if before else self.script.get("between_s", 3) + self._show("Get ready" if before else s["title"], step, text) + self._emit(state="between", prompt=text.replace("|", "\n"), take=None, **status) + self._wait(secs) + self._start_take(i, s) + self._status.update(state="intro", take=self._take["id"], **status) # sent with the intro + self._run_prompt(s, {"id": s["id"] + "/intro", "text": s.get("intro", ""), "seconds": s["intro_s"], + "hands": "", "pose": "", "distance": "", "position": "", "object": "", + "controller": False, "controllers": []}, step, intro=True) + self._status["state"] = "running" # sent with the first prompt + if s["kind"] == "targets": + self._targets(s, step) + elif s["kind"] == "bar": + self._bar(s, step) + for p in s["prompts"]: + self._run_prompt(s, p, step) + self._end_take("complete") + except _Skip: + self._log("skipped %s" % s["id"]) + if self._take: + self._end_take("skipped") + except _Stop: + raise + finally: + self._panel.set("bar", "bar off") + self._panel.set("target", "target off") + + def _start_take(self, i, s): + n = len(self._session_json["takes"]) + 1 + take_id = "%02d-%s" % (n, s["id"]) + d = os.path.join(self.session_dir, "takes", take_id) + os.makedirs(d) + self._take = {"id": take_id, "dir": d, "section": s, "part": 0, + "prompts": open(os.path.join(d, "prompts.jsonl"), "a", buffering=1), + "json": {"section": s["id"], "title": s["title"], "started_ns": mono_ns(), "ended_ns": None, + "status": "stopped", "deleted": [], "notes": ""}} + write_json(os.path.join(d, "take.json"), self._take["json"]) + self._session_json["takes"].append(take_id) + self._save_session() + self._event("take", section=s["id"], take=take_id) + self._start_recording() + self._log("take %s" % take_id) + + def _start_recording(self): + t = self._take + t["part"] += 1 + if not self.dry_run: + # a safety net only: the session ends the recording itself + remaining = section_seconds(t["section"], worst=True) / self.speed + self._recorder = Recorder(t["dir"], t["part"], remaining * 1.5 + 60, self.ring, self._log_file) + self._panel.cmd("poses start " + os.path.join(t["dir"], "poses.jsonl")) + + def _stop_recording(self): + self._panel.cmd("poses stop", reply=not self.dry_run, timeout=1.0) + if self._recorder: + rec, self._recorder = self._recorder, None + code = rec.stop() + self._log("recording part %d ended (%s)" % (rec.part, code)) + + def _end_take(self, status): + t = self._take + try: + if not self._paused: + self._stop_recording() + finally: + self._event("end", status=status) + self._take = None + t["prompts"].close() + t["json"]["ended_ns"] = mono_ns() + t["json"]["status"] = status + write_json(os.path.join(t["dir"], "take.json"), t["json"]) + self._log("take %s %s" % (t["id"], status)) + + def _prompt_event(self, p): + self._event("prompt", id=p["id"], text=p["text"], hands=p["hands"], pose=p["pose"], distance=p["distance"], + position=p["position"], object=p["object"], controller=bool(p["controller"]), + **({"controllers": p["controllers"]} if p["controllers"] else {})) + + def _begin_prompt(self, s, p, step): + self._prompt = p + self._fb.pop("lost_key", None) + self._show(s["title"], step, p["text"]) + self._prompt_event(p) + self._emit(prompt=p["text"].replace("|", "\n"), seconds_left=float(p["seconds"])) + self._log(" %s" % p["id"]) + + def _run_prompt(self, s, p, step, intro=False): + if intro and not p["text"]: + return + self._begin_prompt(s, p, step) + self._wait(p["seconds"]) + + def _targets(self, s, step): + hold_s, timeout_s = float(s.get("hold_s", 1.0)), float(s.get("timeout_s", 8)) + defaults = s.get("defaults") or {} + for k, pt in enumerate(s["targets"]): + p = {"id": "%s/%d" % (s["id"], k + 1), "text": s.get("text", ""), "seconds": timeout_s, + "hands": defaults.get("hands", "any"), "pose": defaults.get("pose", "point"), "distance": "", + "position": "", "object": "", "controller": False, "controllers": []} + self._begin_prompt(s, p, step) + xyz = "%.3f %.3f %.3f" % tuple(pt) + reply = self._panel.cmd("target %s show" % xyz, reply=True) or "" + self._panel.sent["target"] = "target %s show" % xyz + tok = reply.split() + room = [float(v) for v in tok[1:4]] if tok[:1] == ["ok"] and len(tok) >= 4 else None + target = {"id": p["id"], "head": list(pt), "room": room} + self._event("target", state="show", **target) + st = {"hold": 0.0, "sent": 0.0, "holding": False} + + def tick(dt, pt=pt, room=room, target=target, st=st, xyz=xyz): + d = self._tip_distance(pt, room) + if d is not None and d <= TOUCH_M: + if not st["holding"]: + st["holding"] = True + self._event("target", state="hold", **target) + st["hold"] += dt + elif st["holding"]: + st["holding"], st["hold"] = False, 0.0 + self._panel.set("target", "target %s show" % xyz) + if st["holding"]: + frac = min(1.0, st["hold"] / hold_s) + if time.monotonic() - st["sent"] >= 0.1 or frac >= 1: + st["sent"] = time.monotonic() + self._panel.set("target", "target %s hold %.2f" % (xyz, frac)) + return st["hold"] >= hold_s + + done = self._wait(timeout_s, tick) + state = "done" if done else "timeout" + if done: + self._panel.set("target", "target %s done" % xyz) + self._event("target", state=state, **target) + self._log(" %s %s" % (p["id"], state)) + if done: + self._wait(0.6, countdown=False) + self._panel.set("target", "target off") + + def _tip_distance(self, pt, room): + """The nearest seen index tip's distance from the target: in the room (with the head's + pose now) when the panel placed the target there, else in the head frame.""" + live = self._live + if not live: + return None + tips = [live[s]["tip"] for s in ("left", "right") if live[s]] + if not tips: + return None + ref = pt + if room is not None: + reply = (self._panel.cmd("head", reply=True, timeout=0.1) or "").split() + if reply[:1] == ["ok"] and len(reply) == 13: + m = [float(v) for v in reply[1:]] + tips = [[m[4 * r] * t[0] + m[4 * r + 1] * t[1] + m[4 * r + 2] * t[2] + m[4 * r + 3] for r in range(3)] + for t in tips] + ref = room + return min(math.dist(t, ref) for t in tips) + + def _palm_share(self, near, far): + live = self._live + if not live: + return -1.0 + ds = [live[s]["palm_m"] for s in ("left", "right") if live[s]] + if not ds: + return -1.0 + return max(0.0, min(1.0, (sum(ds) / len(ds) - near) / (far - near))) + + def _bar(self, s, step): + near, far = float(s.get("near_m", 0.2)), float(s.get("far_m", 0.6)) + period, lead = float(s.get("period_s", 6)), float(s.get("lead_s", 3)) + labels = "%s %s" % (s.get("near_label", "Near"), s.get("far_label", "Far")) + defaults = s.get("defaults") or {} + for h in s["heights"]: + p = {"id": "%s/%s" % (s["id"], h["id"]), "text": h["text"], "hands": defaults.get("hands", "both"), + "pose": defaults.get("pose", "open"), "distance": "", "position": h["id"], "object": "", + "controller": bool(defaults.get("controller", False)), + "controllers": list(defaults.get("controllers") or [])} + total = lead + h["reps"] * period + p["seconds"] = total + self._begin_prompt(s, p, step) + st = {"t": 0.0, "sent": 0.0} + + def tick(dt, st=st): + st["t"] += dt + sweep = st["t"] - lead + target = 0.0 if sweep <= 0 else 1 - abs(1 - 2 * ((sweep % period) / period)) + if time.monotonic() - st["sent"] >= 0.1: + st["sent"] = time.monotonic() + cur = self._palm_share(near, far) + self._panel.set("bar", "bar %.3f %.3f %s" % (target, cur, labels)) + if sweep > 0: + self._event("bar", target=round(target, 3), current=None if cur < 0 else round(cur, 3)) + return False + + self._wait(total, tick) + self._panel.set("bar", "bar off") + + +# ------------------------------------------------------------------------------------------ +# The command line + +def main(): + ap = argparse.ArgumentParser(description="Run a hand recording session (the hand recorder's session runner).") + ap.add_argument("--dry-run", action="store_true", help="run no processes; print the panel commands") + ap.add_argument("--speed", type=float, default=1.0, help="run the script this many times faster") + ap.add_argument("--ring", help="read frames from this ring (ft-ringplay's) instead of ft-camd's") + ap.add_argument("--no-start", action="store_true", help="start no ft-camd or tracking ft-hands") + ap.add_argument("--base", help="where sessions go (default %s; a temporary folder with --dry-run)" % BASE_DIR) + ap.add_argument("--objects", default="", help="ticked objects, comma-separated (unknown names are your own)") + ap.add_argument("--controllers", action="store_true", help="controllers with the straps") + ap.add_argument("--lighting", default="room", choices=("dim", "room", "daylight")) + ap.add_argument("--script", default=SCRIPT_PATH) + ap.add_argument("--panel", help="the panel program (default hands/rec/build/ft-handpanel)") + ap.add_argument("--hands-dir", help="where the hands file is (default /run/user/UID/frametop-hands)") + ap.add_argument("--plan", action="store_true", help="print the sections and their length, and exit") + a = ap.parse_args() + + known = ("pencil", "phone", "cup", "keyboard", "mouse", "gamepad", "small") + names = [o.strip() for o in a.objects.split(",") if o.strip()] + checklist = {"objects": [o for o in names if o in known], "own_objects": [o for o in names if o not in known], + "controllers": "straps" if a.controllers else "none", "sleeves": "", "rings": False, + "watch": False, "notes": ""} + base = a.base + if not base and a.dry_run: + import tempfile + base = tempfile.mkdtemp(prefix="handrec-dry-") + base = base or BASE_DIR + profile = {} + try: + with open(os.path.join(base, "profile.json")) as f: + profile = json.load(f) + except (OSError, ValueError): + pass + + t0 = time.monotonic() + last = {} + + def on_status(st): + key = (st["state"], st["section"], st["prompt"], st["note"], st["take"], + st["hands"]["left"], st["hands"]["right"], st["error"]) + if key == last.get("key"): + return + last["key"] = key + hands = "".join("%s%s" % (s[0].upper(), {True: "+", False: "-", None: "?"}[st["hands"][s]]) + for s in ("left", "right")) + line = "[%6.1f] %-8s %d/%d %-16s %s %s" % (time.monotonic() - t0, st["state"], st["section_index"], + st["section_count"], st["section"] or "-", hands, st["prompt"]) + if st["note"]: + line += " (%s)" % st["note"] + if st["error"]: + line += " ERROR: %s" % st["error"] + print(line, flush=True) + + s = Session(base, profile, checklist, a.lighting, a.script, ring=a.ring, start_processes=not a.no_start, + dry_run=a.dry_run, speed=a.speed, on_status=on_status, hands_dir=a.hands_dir, panel_bin=a.panel) + est, worst = plan_seconds(s.script, s.plan), plan_seconds(s.script, s.plan, worst=True) + print("%d sections, about %.1f min (at most %.1f)%s" % (len(s.plan), est / 60, worst / 60, + ", %gx speed" % a.speed if a.speed != 1 else "")) + for sk in s.skipped: + print(" skipping %s: %s" % (sk["section"], sk["reason"])) + if a.plan: + for sec in s.plan: + print(" %-18s %-9s %3d prompts %5.0f s" % (sec["id"], sec["kind"], len(sec["prompts"]), section_seconds(sec))) + return 0 + if not a.dry_run: + light = ring_lighting(a.ring) + match = similar_lighting(base, {"chosen": a.lighting, "ring": light}) if light else None + print("lighting: %s" % (json.dumps(light) if light else "no camera ring")) + if match: + print(" about the same light as session %s (%s)" % match) + + def on_signal(*_): + print("stopping", flush=True) + threading.Thread(target=s.stop, daemon=True).start() + + signal.signal(signal.SIGINT, on_signal) + signal.signal(signal.SIGTERM, on_signal) + if sys.stdin.isatty(): + def keys(): + print("keys: p pause, r resume, s skip, q stop (then Enter)", flush=True) + for line in sys.stdin: + c = line.strip()[:1] + if c == "p": + s.pause() + elif c == "r": + s.resume() + elif c == "s": + s.skip() + elif c == "q": + s.stop(wait=0) + threading.Thread(target=keys, daemon=True).start() + s.start() + while s._thread.is_alive(): + s.join(0.5) + print("session: %s" % s.session_dir) + return 0 if s.state in ("done", "stopped") else 1 + + +if __name__ == "__main__": + sys.exit(main())