mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-06 11:00:14 +02:00
DeeJanuzandClaude Opus 5.5
8321a99121
sets.bin's capture_ns is CLOCK_MONOTONIC_RAW; poses.jsonl and prompts.jsonl are CLOCK_MONOTONIC. On 2026-10-03 the two were 0.80 s apart during a session and 1.11 s apart five hours later, so images can't be paired with poses by capture_ns. take.json now samples RAW minus MONOTONIC as each recording part starts and stops (as ft-hands' raw_minus_mono_ns), so readers can put each exposure on the poses' clock. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
635 lines
29 KiB
Python
635 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for session.py's step mode (dry runs, no processes): the ready, countdown and hold
|
|
timeline in prompts.jsonl, R (redo), the timed flow (--auto), the pose pictures' display rule,
|
|
the length estimates, and a take recorded in many parts through review, export and validate.
|
|
|
|
python3 hands/rec/tests/test_session.py
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import struct
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.dirname(HERE))
|
|
import session # noqa: E402
|
|
import takes # noqa: E402
|
|
import validate # noqa: E402
|
|
from test_validate import fhset, make_session, SESSION # noqa: E402
|
|
|
|
SCRIPT = {
|
|
"version": 1, "intro_s": 1, "between_s": 1,
|
|
"welcome": {"title": "Test", "seconds": 1, "text": "A test."},
|
|
"done": {"title": "Done", "seconds": 0, "text": "Done."},
|
|
"stopped": {"title": "Stopped", "seconds": 0, "text": "Stopped."},
|
|
"sections": [
|
|
{"id": "poses", "title": "Poses", "intro": "Some poses.", "go": "Hold",
|
|
"prompts": [{"text": "Fist.", "seconds": 4, "hands": "left", "pose": "fist", "distance": "near",
|
|
"position": "left"},
|
|
{"text": "Open.", "seconds": 4, "hands": "both", "pose": "open"}]}]}
|
|
|
|
|
|
class SessionBase(unittest.TestCase):
|
|
"""Dry-run sessions of SCRIPT, steered from the test."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="handrec-session-test-")
|
|
self.script = os.path.join(self.tmp, "script.json")
|
|
with open(self.script, "w") as f:
|
|
json.dump(SCRIPT, f)
|
|
self.panel = []
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def session(self, **kw):
|
|
kw.setdefault("speed", 10)
|
|
s = session.Session(os.path.join(self.tmp, "base"), {}, {}, "room", self.script, dry_run=True,
|
|
poses_dir=os.path.join(self.tmp, "no-poses"), **kw)
|
|
s.print = self.panel.append
|
|
return s
|
|
|
|
def wait_for(self, s, pred, what, timeout=10):
|
|
end = time.monotonic() + timeout
|
|
while time.monotonic() < end:
|
|
st = dict(s._status)
|
|
if pred(st):
|
|
return st
|
|
time.sleep(0.005)
|
|
self.fail("timed out waiting for %s: %s" % (what, s._status))
|
|
|
|
def waiting(self, s, state, prompt=None):
|
|
return self.wait_for(s, lambda st: st["state"] == state and st["waiting"]
|
|
and (prompt is None or st["prompt"] == prompt), "%s %s" % (state, prompt or ""))
|
|
|
|
def events(self, s):
|
|
with open(os.path.join(s.session_dir, "takes", "01-poses", "prompts.jsonl")) as f:
|
|
return [json.loads(line) for line in f]
|
|
|
|
|
|
|
|
class SessionTest(SessionBase):
|
|
def test_step_flow(self):
|
|
s = self.session(next_after=0.0)
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
ev = self.events(s)
|
|
self.assertEqual([e["event"] for e in ev],
|
|
["take", "ready", "prompt", "wait", "ready", "prompt", "wait", "end"])
|
|
self.assertEqual(ev[1]["id"], ev[2]["id"])
|
|
self.assertEqual(ev[1]["seconds"], session.COUNTDOWN_S)
|
|
self.assertTrue(ev[1]["t"] < ev[2]["t"] < ev[3]["t"] < ev[4]["t"])
|
|
# the countdown is recorded (3 s at 10x speed), and the hold after it
|
|
self.assertGreater(ev[2]["t"] - ev[1]["t"], 0.25e9)
|
|
with open(os.path.join(s.session_dir, "session.json")) as f:
|
|
self.assertEqual(json.load(f)["mode"], "step")
|
|
# a clock sample as each part starts (2 steps: 2 parts), for capture_ns (RAW) -> MONOTONIC
|
|
with open(os.path.join(s.session_dir, "takes", "01-poses", "take.json")) as f:
|
|
clock = json.load(f)["clock"]
|
|
self.assertEqual(len(clock), 2)
|
|
self.assertTrue(all(len(c) == 2 and all(isinstance(v, int) for v in c) for c in clock))
|
|
self.assertLess(clock[0][0], clock[1][0])
|
|
# the panel: Ready?, then the countdown, then the section's word for the hold, the diagram
|
|
self.assertIn("panel: action " + session.READY_TEXT, self.panel)
|
|
self.assertLess(self.panel.index("panel: big 3"), self.panel.index("panel: big 1"))
|
|
self.assertIn("panel: big Hold", self.panel)
|
|
self.assertIn("panel: where left near", self.panel)
|
|
self.assertIn("panel: rec on", self.panel)
|
|
|
|
def test_redo(self):
|
|
s = self.session()
|
|
s.start()
|
|
self.waiting(s, "starting")
|
|
s.next_step()
|
|
self.waiting(s, "intro")
|
|
s.redo() # nothing to do again yet: dropped
|
|
s.next_step()
|
|
st = self.waiting(s, "ready", "Fist.")
|
|
self.assertFalse(st["can_redo"])
|
|
s.next_step()
|
|
self.wait_for(s, lambda st: st["state"] == "running", "the hold")
|
|
s.redo() # during the hold: it starts again
|
|
self.waiting(s, "ready", "Fist.")
|
|
s.next_step()
|
|
st = self.waiting(s, "ready", "Open.")
|
|
self.assertTrue(st["can_redo"])
|
|
s.redo() # at the next step's ready screen: the one before goes again
|
|
self.waiting(s, "ready", "Fist.")
|
|
s.next_step()
|
|
self.waiting(s, "ready", "Open.")
|
|
s.next_step()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
ev = self.events(s)
|
|
names = [e["event"] for e in ev]
|
|
self.assertEqual(names, ["take", "ready", "prompt", "redo", "wait", "ready", "prompt", "wait", "redo",
|
|
"ready", "prompt", "wait", "ready", "prompt", "wait", "end"])
|
|
first, second = ev[3], ev[8]
|
|
self.assertEqual(first["id"], "poses/fist/left/near/left")
|
|
self.assertEqual(first["from"], ev[1]["t"]) # from its countdown
|
|
self.assertTrue(ev[2]["t"] < first["to"] <= ev[4]["t"])
|
|
self.assertEqual((second["from"], second["id"]), (ev[5]["t"], first["id"]))
|
|
self.assertTrue(ev[6]["t"] < second["to"] <= ev[7]["t"])
|
|
# what labels keep: prompts outside every redo range
|
|
kept = [e["id"] for e in ev if e["event"] == "prompt"
|
|
and not any(r["from"] <= e["t"] <= r["to"] for r in ev if r["event"] == "redo")]
|
|
self.assertEqual(kept, ["poses/fist/left/near/left", "poses/open/both"])
|
|
|
|
def test_pause_while_waiting(self):
|
|
s = self.session()
|
|
s.start()
|
|
self.waiting(s, "starting")
|
|
s.next_step()
|
|
self.waiting(s, "intro")
|
|
s.next_step()
|
|
self.waiting(s, "ready", "Fist.")
|
|
s.pause()
|
|
self.wait_for(s, lambda st: st["state"] == "paused", "paused")
|
|
s.next_step() # ignored while paused
|
|
s.resume()
|
|
st = self.waiting(s, "ready", "Fist.")
|
|
s.stop(wait=10)
|
|
self.assertEqual(s.state, "stopped")
|
|
# nothing recorded: no take
|
|
self.assertEqual(os.listdir(os.path.join(s.session_dir, "takes")), [])
|
|
|
|
def test_auto(self):
|
|
s = self.session(auto=True, speed=20)
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
ev = self.events(s)
|
|
self.assertEqual([e["event"] for e in ev], ["take", "prompt", "prompt", "prompt", "end"])
|
|
self.assertEqual(ev[1]["id"], "poses/intro")
|
|
|
|
def test_auto_redo_after_pause(self):
|
|
s = self.session(auto=True)
|
|
s.start()
|
|
self.wait_for(s, lambda st: st["state"] == "running" and st["prompt"] == "Fist.", "the first prompt")
|
|
s.pause()
|
|
self.wait_for(s, lambda st: st["state"] == "paused", "paused")
|
|
s.redo() # ends the pause; the prompt starts again, recording again
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
names = [e["event"] for e in self.events(s)]
|
|
self.assertEqual(names, ["take", "prompt", "prompt", "pause", "redo", "resume", "prompt", "prompt", "end"])
|
|
|
|
def test_pose_view(self):
|
|
d = os.path.join(self.tmp, "poses")
|
|
os.makedirs(d)
|
|
for name in ("fist.png", "cross.png"):
|
|
open(os.path.join(d, name), "wb").close()
|
|
with open(os.path.join(d, "poses.json"), "w") as f:
|
|
json.dump({"fist": {"file": "fist.png", "two_hands": False, "caption": "A fist"},
|
|
"cross": {"file": "../cross.png", "two_hands": True, "caption": ""},
|
|
"ok": {"file": "ok.png", "two_hands": False}}, f)
|
|
poses = session.load_poses(d)
|
|
fist = os.path.join(d, "fist.png")
|
|
self.assertEqual(session.pose_view(poses, {"pose": "fist", "hands": "right"}), (fist, "", "A fist"))
|
|
self.assertEqual(session.pose_view(poses, {"pose": "fist", "hands": "left"})[1], "mirror")
|
|
self.assertEqual(session.pose_view(poses, {"pose": "fist", "hands": "both"})[1], "both")
|
|
self.assertEqual(session.pose_view(poses, {"pose": "fist", "hands": "any"})[1], "")
|
|
# two hands drawn already; the file stays in the folder
|
|
self.assertEqual(session.pose_view(poses, {"pose": "cross", "hands": "both"})[:2],
|
|
(os.path.join(d, "cross.png"), ""))
|
|
self.assertEqual(session.pose_view(poses, {"pose": "ok", "hands": "both"}), ("", "", "")) # no file
|
|
self.assertEqual(session.pose_view(poses, {"pose": "claw", "hands": "both"}), ("", "", ""))
|
|
self.assertEqual(session.load_poses(os.path.join(self.tmp, "none")), {})
|
|
|
|
def test_plan(self):
|
|
plan, _ = session.build_plan(SCRIPT, {})
|
|
self.assertEqual(session.plan_steps(plan), 2)
|
|
self.assertEqual(session.plan_seconds(SCRIPT, plan, auto=False), 2 * session.COUNTDOWN_S + 8)
|
|
self.assertEqual(session.plan_seconds(SCRIPT, plan, auto=True), 1 + 1 + 8)
|
|
self.assertIn("2 steps", session.plan_summary(SCRIPT, plan))
|
|
|
|
|
|
SWEEP_SCRIPT = {
|
|
"version": 1, "intro_s": 1, "between_s": 1,
|
|
"welcome": {"title": "Test", "seconds": 1, "text": "A test."},
|
|
"done": {"title": "Done", "seconds": 0, "text": "Done."},
|
|
"stopped": {"title": "Stopped", "seconds": 0, "text": "Stopped."},
|
|
"cue_names": {"thumbs-up": "Thumbs up"},
|
|
"sections": [
|
|
{"id": "sweeps", "title": "Poses", "kind": "sweep", "intro": "Keep moving.", "quick": True,
|
|
"cue_s": 2, "step_s": 6,
|
|
"groups": [["open", "fist"], ["ok", "thumbs-up", "claw"]],
|
|
"sweeps": [{"hands": "both", "group": "next", "text": "Both hands."},
|
|
{"hands": "left", "group": "any", "quick": False, "text": "Left hand."}]}]}
|
|
|
|
|
|
class SweepTest(SessionBase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
with open(self.script, "w") as f:
|
|
json.dump(SWEEP_SCRIPT, f)
|
|
|
|
def events(self, s):
|
|
with open(os.path.join(s.session_dir, "takes", "01-sweeps", "prompts.jsonl")) as f:
|
|
return [json.loads(line) for line in f]
|
|
|
|
def test_sweep_flow(self):
|
|
s = self.session(next_after=0.0, speed=10, seed=1)
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
ev = self.events(s)
|
|
names = [e["event"] for e in ev]
|
|
self.assertEqual(names, ["take", "ready", "prompt", "prompt", "prompt", "wait",
|
|
"ready", "prompt", "prompt", "prompt", "wait", "end"])
|
|
plan = {p["id"]: p["cues"] for p in s.plan[0]["prompts"]}
|
|
for step, first in (("sweeps/both-1", 2), ("sweeps/left-1", 7)):
|
|
cues = [e for e in ev[first:first + 3]]
|
|
self.assertEqual([e["pose"] for e in cues], plan[step]) # one prompt per cue, in order
|
|
self.assertTrue(all(e["cue"] and e["step"] == step for e in cues))
|
|
self.assertTrue(all(e["distance"] == e["position"] == "" for e in cues))
|
|
self.assertEqual(ev[first - 1], dict(ev[first - 1], event="ready", id=step))
|
|
gaps = [(b["t"] - a["t"]) / 1e9 for a, b in zip(cues, cues[1:])]
|
|
self.assertTrue(all(0.15 < g < 0.4 for g in gaps), gaps) # cue_s 2 at 10x speed
|
|
# a cue repeated within a step (6 s of 2 s cues from a group of 2) gets its own id
|
|
both = [e["id"] for e in ev[2:5]]
|
|
self.assertEqual(len(set(both)), 3)
|
|
# the strip: every picture of the step, the cue lit
|
|
strips = [c for c in self.panel if c.startswith("panel: strip ")]
|
|
self.assertIn("panel: strip off", strips)
|
|
self.assertTrue(any(c.startswith("panel: strip 1 ") for c in strips))
|
|
with open(os.path.join(s.session_dir, "session.json")) as f:
|
|
meta = json.load(f)
|
|
self.assertEqual(meta["shuffle"]["seed"], 1)
|
|
self.assertEqual({x["id"]: x["cues"] for x in meta["shuffle"]["sweeps"]["sweeps"]}, plan)
|
|
self.assertFalse(meta["quick"])
|
|
|
|
def test_seed_from_session_id(self):
|
|
s = self.session(next_after=0.0, speed=20)
|
|
s.start()
|
|
s.join(20)
|
|
with open(os.path.join(s.session_dir, "session.json")) as f:
|
|
meta = json.load(f)
|
|
sid = os.path.basename(s.session_dir)
|
|
self.assertEqual(meta["shuffle"]["seed"], session.session_seed(sid))
|
|
again, _ = session.build_plan(SWEEP_SCRIPT, {}, seed=meta["shuffle"]["seed"]) # reproducible
|
|
self.assertEqual(session.plan_record(again), meta["shuffle"]["sweeps"])
|
|
|
|
def test_quick_session(self):
|
|
s = self.session(next_after=0.0, speed=20, quick=True)
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
self.assertEqual([e["event"] for e in self.events(s)].count("ready"), 1) # the left sweep left out
|
|
with open(os.path.join(s.session_dir, "session.json")) as f:
|
|
self.assertTrue(json.load(f)["quick"])
|
|
|
|
def test_auto(self):
|
|
s = self.session(auto=True, speed=20)
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
names = [e["event"] for e in self.events(s)]
|
|
self.assertEqual(names, ["take", "prompt"] + ["prompt"] * 6 + ["end"]) # intro, 3 cues a step
|
|
|
|
|
|
class ShuffleTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.script = session.load_script(session.SCRIPT_PATH)
|
|
|
|
def cues(self, seed, quick=False, checklist=None):
|
|
plan, _ = session.build_plan(self.script, checklist or {}, seed=seed, quick=quick)
|
|
return session.plan_record(plan)
|
|
|
|
def test_deterministic(self):
|
|
self.assertEqual(self.cues(42), self.cues(42))
|
|
self.assertNotEqual(self.cues(42)["pose-sweeps"], self.cues(43)["pose-sweeps"])
|
|
self.assertEqual(session.session_seed("20261002-202734"), session.session_seed("20261002-202734"))
|
|
self.assertNotEqual(session.session_seed("20261002-202734"), session.session_seed("20261002-202735"))
|
|
|
|
def test_groups(self):
|
|
groups = [sorted(g) for g in self.script["sections"][1]["groups"]]
|
|
for seed in range(20):
|
|
steps = self.cues(seed)["pose-sweeps"]
|
|
self.assertEqual([x["hands"] for x in steps], ["both", "both", "both", "left", "right"])
|
|
used = [sorted(set(x["cues"])) for x in steps]
|
|
self.assertEqual(sorted(used[:3]), sorted(groups)) # each group once with both hands
|
|
self.assertTrue(all(u in groups for u in used[3:]))
|
|
self.assertNotEqual(used[3], used[4]) # the one-hand sweeps differ
|
|
self.assertTrue(all(len(x["cues"]) == 5 for x in steps)) # 20 s of 4 s cues
|
|
# unshuffled: the script's order
|
|
steps = self.cues(None)["pose-sweeps"]
|
|
self.assertEqual(steps[0]["cues"], ["open", "fist", "point", "pinch", "open"])
|
|
# gestures and hand size keep their order
|
|
for seed in (1, 2, 3):
|
|
self.assertEqual([x["cues"] for x in self.cues(seed)["gestures"]],
|
|
[["pinch-tap", "pinch-drag"], ["grab", "cross", "overlap"], ["near-face", "screen-point"]])
|
|
self.assertEqual(self.cues(seed)["hand-size"][0]["cues"], ["flat", "flat-back", "spread"])
|
|
|
|
def test_plans(self):
|
|
full, skipped = session.build_plan(self.script, {}, seed=5)
|
|
self.assertEqual([s["id"] for s in full],
|
|
["hand-size", "pose-sweeps", "gestures", "desk-work", "touch", "bare-push", "no-hands"])
|
|
self.assertEqual(session.plan_steps(full), 15)
|
|
self.assertLess(session.plan_seconds(self.script, full, auto=False), 6 * 60)
|
|
everything, _ = session.build_plan(self.script, {"objects": ["pencil", "keyboard", "mouse"],
|
|
"controllers": "straps"}, seed=5)
|
|
self.assertEqual(len(everything), 10)
|
|
quick, skipped = session.build_plan(self.script, {"objects": ["pencil"], "controllers": "straps"},
|
|
seed=5, quick=True)
|
|
self.assertEqual([s["id"] for s in quick], ["hand-size", "pose-sweeps", "touch", "no-hands"])
|
|
self.assertEqual([p["hands"] for p in quick[1]["prompts"]], ["both"] * 3)
|
|
self.assertEqual(len(quick[2]["targets"]), 6)
|
|
self.assertLess(session.plan_seconds(self.script, quick, auto=False), 150)
|
|
self.assertIn({"section": "objects", "reason": "not in a quick round"}, skipped)
|
|
# desk work: the mouse-and-keyboard step only with both ticked
|
|
desk = [s for s in everything if s["id"] == "desk-work"][0]["prompts"]
|
|
self.assertEqual([p.get("cues") for p in desk], [None, ["mouse", "switch"]])
|
|
|
|
|
|
class StripPanelTest(unittest.TestCase):
|
|
"""The panel's strip command on ft-handpanel --no-vr (skipped if it isn't built)."""
|
|
|
|
def test_strip(self):
|
|
import subprocess
|
|
binary = session.PANEL_BIN
|
|
if not os.access(binary, os.X_OK):
|
|
self.skipTest("ft-handpanel isn't built")
|
|
name = "hrtest-strip-%d" % os.getpid()
|
|
proc = subprocess.Popen([binary, "--no-vr", "--socket", name], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
text=True)
|
|
try:
|
|
panel = session.Panel(name=name)
|
|
for _ in range(50):
|
|
if panel.cmd("ping", reply=True, timeout=0.1):
|
|
break
|
|
if proc.poll() is not None:
|
|
self.skipTest("ft-handpanel doesn't run here: " + proc.stderr.read().strip()[-200:])
|
|
time.sleep(0.1)
|
|
else:
|
|
self.fail("ft-handpanel --no-vr doesn't answer")
|
|
poses = session.POSES_DIR
|
|
ok = panel.cmd("strip 1 %s/fist.png|mirror|Fist;-|-|No picture;%s/ok.png|-|OK" % (poses, poses), reply=True)
|
|
bad = panel.cmd("strip 1", reply=True)
|
|
missing = panel.cmd("strip 0 /nonexistent.png|-|Gone", reply=True)
|
|
panel.cmd("show", reply=True)
|
|
panel.cmd("strip off", reply=True)
|
|
panel.close()
|
|
finally:
|
|
proc.terminate()
|
|
out = proc.communicate(timeout=5)[0]
|
|
self.assertEqual(ok, "ok")
|
|
self.assertTrue(bad.startswith("error usage: strip"))
|
|
self.assertTrue(missing.startswith("error can't read"))
|
|
self.assertIn("strip: cue 0: Gone(/nonexistent.png)", out)
|
|
self.assertIn("strip: off", out)
|
|
|
|
|
|
# /proc/bus/input/devices as the Frame has it (2026-10-02), trimmed, plus a USB mouse.
|
|
DEVICES = """I: Bus=0019 Vendor=0001 Product=0001 Version=0100
|
|
N: Name="gpio-keys"
|
|
P: Phys=gpio-keys/input0
|
|
S: Sysfs=/devices/platform/gpio-keys/input/input3
|
|
U: Uniq=
|
|
H: Handlers=kbd kbd event3 qcom_pon_combo_timer_listener
|
|
B: PROP=0
|
|
B: EV=3
|
|
B: KEY=100000000000 0 0 0 0 200000000 0 0 0 0 0
|
|
|
|
I: Bus=0006 Vendor=4d44 Product=0001 Version=0001
|
|
N: Name="frametop virtual mouse"
|
|
P: Phys=
|
|
S: Sysfs=/devices/virtual/input/input4
|
|
U: Uniq=
|
|
H: Handlers=event4 qcom_pon_combo_timer_listener
|
|
B: PROP=0
|
|
B: EV=7
|
|
B: KEY=ffffff 0 0 0 0
|
|
B: REL=ffff
|
|
|
|
I: Bus=0003 Vendor=0001 Product=0001 Version=0001
|
|
N: Name="frame-voice keyboard"
|
|
P: Phys=py-evdev-uinput
|
|
S: Sysfs=/devices/virtual/input/input7
|
|
U: Uniq=
|
|
H: Handlers=kbd sysrq kbd event7 qcom_pon_combo_timer_listener
|
|
B: PROP=0
|
|
B: EV=200003
|
|
B: KEY=1ffffffffffffff ffffffffffffffff ffffffffffffffff fffffffffffffffe
|
|
|
|
I: Bus=0005 Vendor=214e Product=0035 Version=0001
|
|
N: Name="Z3 Keyboard"
|
|
P: Phys=90:82:c3:5f:4a:a7
|
|
S: Sysfs=/devices/virtual/misc/uhid/0005:214E:0035.0005/input/input27
|
|
U: Uniq=da:ef:2f:d2:be:e2
|
|
H: Handlers=kbd sysrq kbd event9 qcom_pon_combo_timer_listener
|
|
B: PROP=0
|
|
B: EV=10001f
|
|
B: KEY=300000000000 33eff 0 0 483ffff17aff32d bfd4444600000000 1 130ff38b17c007 ffff7bfad9415fff feb2ffdfffefffff fffffffffffffffe
|
|
B: REL=1040
|
|
B: MSC=10
|
|
"""
|
|
BT_MOUSE = """
|
|
I: Bus=0005 Vendor=214e Product=0035 Version=0001
|
|
N: Name="Z3 Mouse"
|
|
P: Phys=90:82:c3:5f:4a:a7
|
|
S: Sysfs=/devices/virtual/misc/uhid/0005:214E:0035.0005/input/input26
|
|
U: Uniq=da:ef:2f:d2:be:e2
|
|
H: Handlers=event8 qcom_pon_combo_timer_listener
|
|
B: PROP=0
|
|
B: EV=17
|
|
B: KEY=ffff0000 0 0 0 0
|
|
B: REL=1943
|
|
B: MSC=10
|
|
"""
|
|
USB_MOUSE = """
|
|
I: Bus=0003 Vendor=046d Product=c077 Version=0111
|
|
N: Name="Logitech USB Optical Mouse"
|
|
P: Phys=usb-xhci-hcd.1.auto-1/input0
|
|
S: Sysfs=/devices/platform/soc@0/a600000.usb/xhci-hcd.1.auto/usb1/1-1/1-1:1.0/0003:046D:C077.0001/input/input40
|
|
U: Uniq=
|
|
H: Handlers=mouse0 event13
|
|
B: PROP=0
|
|
B: EV=17
|
|
B: KEY=70000 0 0 0 0
|
|
B: REL=903
|
|
B: MSC=10
|
|
"""
|
|
|
|
|
|
def event(etype, code, value):
|
|
return session.INPUT_EVENT.pack(1, 2, etype, code, value)
|
|
|
|
|
|
PRESS = event(session.EV_KEY, session.KEY_SELECT, 1) + event(0, 0, 0)
|
|
RELEASE = event(session.EV_KEY, session.KEY_SELECT, 0) + event(0, 0, 0)
|
|
|
|
|
|
class InputTest(unittest.TestCase):
|
|
def test_devices(self):
|
|
devs = session.parse_input_devices(DEVICES)
|
|
self.assertEqual([d["name"] for d in devs], ["gpio-keys", "frametop virtual mouse", "frame-voice keyboard",
|
|
"Z3 Keyboard"])
|
|
self.assertEqual(session.find_button(devs), "/dev/input/event3")
|
|
self.assertFalse(any(session.real_mouse(d) for d in devs)) # the virtual mouse, keyboards with wheels
|
|
for extra, name in ((BT_MOUSE, "Z3 Mouse"), (USB_MOUSE, "Logitech USB Optical Mouse")):
|
|
mice = [d["name"] for d in session.parse_input_devices(DEVICES + extra) if session.real_mouse(d)]
|
|
self.assertEqual(mice, [name])
|
|
# another gpio-keys without KEY_SELECT isn't the button
|
|
self.assertIsNone(session.find_button(session.parse_input_devices(DEVICES.replace("200000000", "0"))))
|
|
|
|
def test_presses(self):
|
|
data = (PRESS + event(session.EV_KEY, session.KEY_SELECT, 2) * 3 # autorepeat: not presses
|
|
+ RELEASE + event(session.EV_KEY, 115, 1) + PRESS) # another key
|
|
self.assertEqual(session.button_presses(data), (2, b""))
|
|
n, rest = session.button_presses(PRESS + PRESS[:10])
|
|
self.assertEqual((n, rest), (1, PRESS[:10])) # half an event waits for the rest
|
|
self.assertEqual(session.button_presses(rest + PRESS[10:])[0], 1)
|
|
|
|
def test_reader_fifo(self):
|
|
tmp = tempfile.mkdtemp(prefix="handrec-button-test-")
|
|
try:
|
|
fifo = os.path.join(tmp, "button")
|
|
os.mkfifo(fifo)
|
|
presses = []
|
|
reader = session.ButtonReader(fifo, lambda: presses.append(time.monotonic()), debounce_s=0.3).start()
|
|
with open(fifo, "wb", buffering=0) as w:
|
|
w.write(PRESS + RELEASE)
|
|
w.write(PRESS[:7]) # a press split across writes, a bounce
|
|
time.sleep(0.05)
|
|
w.write(PRESS[7:] + RELEASE)
|
|
time.sleep(0.4)
|
|
w.write(PRESS + RELEASE)
|
|
time.sleep(0.2)
|
|
with open(fifo, "wb", buffering=0) as w: # the writer comes back: read again
|
|
time.sleep(0.4)
|
|
w.write(PRESS)
|
|
time.sleep(0.3)
|
|
reader.stop()
|
|
self.assertEqual(len(presses), 3)
|
|
self.assertTrue(reader.ok)
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def test_missing_device(self):
|
|
logs = []
|
|
reader = session.ButtonReader("/nonexistent/event99", lambda: None, log=logs.append).start()
|
|
time.sleep(0.1)
|
|
reader.stop()
|
|
self.assertFalse(reader.ok)
|
|
self.assertIn("can't open", logs[0])
|
|
|
|
|
|
class ButtonSessionTest(SessionBase):
|
|
"""A dry run steered with a simulated headset button (a FIFO), with no mouse, then one."""
|
|
|
|
def test_button(self):
|
|
fifo = os.path.join(self.tmp, "button")
|
|
os.mkfifo(fifo)
|
|
procfile = os.path.join(self.tmp, "devices")
|
|
with open(procfile, "w") as f:
|
|
f.write(DEVICES)
|
|
writer = os.open(fifo, os.O_RDWR) # kept open, so the reader never sees the end
|
|
press = lambda: os.write(writer, PRESS + RELEASE)
|
|
try:
|
|
with mock.patch.object(session, "BUTTON_DEBOUNCE_S", 0.0):
|
|
s = self.session(speed=4, button_device=fifo)
|
|
s.input_devices = procfile
|
|
s.start()
|
|
st = self.waiting(s, "starting")
|
|
self.assertEqual((st["button"], st["mouse"], st["ready_text"]), (True, False, session.READY_BUTTON))
|
|
self.assertIn("panel: action " + session.READY_BUTTON, self.panel)
|
|
self.assertIn("panel: keys " + session.KEYS_STEP_BUTTON, self.panel)
|
|
press()
|
|
self.waiting(s, "intro")
|
|
press()
|
|
self.waiting(s, "ready", "Fist.")
|
|
press() # Next
|
|
self.wait_for(s, lambda st: st["state"] == "running", "the hold")
|
|
press() # pause
|
|
self.wait_for(s, lambda st: st["state"] == "paused", "paused")
|
|
with open(procfile, "a") as f: # a mouse arrives
|
|
f.write(BT_MOUSE)
|
|
press() # resume
|
|
st = self.waiting(s, "ready", "Open.")
|
|
self.assertEqual((st["mouse"], st["ready_text"]), (True, session.READY_BUTTON_MOUSE))
|
|
press()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
names = [e["event"] for e in self.events(s)]
|
|
self.assertEqual(names, ["take", "ready", "prompt", "pause", "resume", "wait", "ready", "prompt", "wait",
|
|
"end"])
|
|
finally:
|
|
os.close(writer)
|
|
|
|
def test_no_button(self):
|
|
s = self.session(next_after=0.0, button=False, button_device="/nonexistent")
|
|
s.start()
|
|
s.join(20)
|
|
self.assertEqual(s.state, "done")
|
|
self.assertIn("panel: action " + session.READY_TEXT, self.panel)
|
|
self.assertIn("panel: keys " + session.KEYS_STEP, self.panel)
|
|
|
|
|
|
class LightingTest(unittest.TestCase):
|
|
"""The measured lighting label: the mono cameras' ambient infrared."""
|
|
|
|
def ring(self, slam, upper):
|
|
return {"slam_left": {"mean": 70.0, "dark_mean": slam}, "slam_right": {"mean": 80.0, "dark_mean": slam},
|
|
"upper_left": {"mean": 50.0, "dark_mean": upper}, "upper_right": {"mean": 20.0, "dark_mean": upper}}
|
|
|
|
def test_indoor_and_daylight(self):
|
|
self.assertEqual(session.classify_lighting(self.ring(3.0, 0.5)), "indoor") # one lamp, 2026-10-02
|
|
self.assertEqual(session.classify_lighting(self.ring(3.7, 0.7)), "indoor") # a lamp-lit room
|
|
self.assertEqual(session.classify_lighting(self.ring(20.0, 8.0)), "daylight")
|
|
self.assertEqual(session.classify_lighting(self.ring(0.0, 0.0)), "") # no dark frames yet
|
|
self.assertEqual(session.classify_lighting(None), "")
|
|
self.assertEqual(session.ambient_ir(self.ring(3.0, 0.5)), 1.75)
|
|
|
|
def test_record(self):
|
|
rec = session.lighting_record("auto", self.ring(3.0, 0.5))
|
|
self.assertEqual((rec["chosen"], rec["source"], rec["measured"]), ("indoor", "measured", "indoor"))
|
|
rec = session.lighting_record("dim", self.ring(3.0, 0.5))
|
|
self.assertEqual((rec["chosen"], rec["source"], rec["measured"]), ("dim", "picked", "indoor"))
|
|
rec = session.lighting_record("auto", None)
|
|
self.assertEqual((rec["chosen"], rec["source"], rec["ring"]), ("", "measured", {}))
|
|
|
|
|
|
class ManyPartsTest(unittest.TestCase):
|
|
"""A take recorded in 40 parts (step mode stops the recording between steps): review reads
|
|
them in order, export makes one stream, validate passes."""
|
|
|
|
def setUp(self):
|
|
if not takes.find_zstd():
|
|
self.skipTest("no zstd")
|
|
self.tmp = tempfile.mkdtemp(prefix="handrec-parts-test-")
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def test_parts(self):
|
|
make_session(self.tmp)
|
|
tdir = os.path.join(self.tmp, "sessions", SESSION, "takes", "01-hand-size")
|
|
t0 = 5 * 10 ** 12
|
|
for part in range(1, 41):
|
|
name = "sets.bin" if part == 1 else "sets-%d.bin" % part
|
|
with open(os.path.join(tdir, name), "wb") as f:
|
|
for i in range(3): # 3 sets a part, 10 s apart between parts
|
|
f.write(fhset(t0 + part * 10 ** 10 + i * 10 ** 8, part))
|
|
store = takes.Store(self.tmp)
|
|
index = takes.take_index(tdir)
|
|
self.assertEqual(len(index.parts), 40)
|
|
self.assertEqual(len(index), 120)
|
|
times = [index.time_ns(i) for i in range(len(index))]
|
|
self.assertEqual(times, sorted(times)) # sets-10.bin after sets-9.bin
|
|
self.assertAlmostEqual(index.duration_s(), 40 * 0.2) # the gaps between parts don't count
|
|
path = store.export(SESSION, low_priority=False)
|
|
r = validate.validate(path)
|
|
self.assertEqual(r.errors, [])
|
|
self.assertEqual(r.summary["sets"], 120 + 30)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|