steam-frame/DeeJanuz--frametop
1
0 Fork 0
mirror of https://github.com/DeeJanuz/frametop.git synced 2026-10-06 08:00:09 +02:00
Code Issues Packages Projects Releases 4 Wiki Activity
Files
79eb251d6c3bbaabe163db15bf0fdeb5f869f792
DeeJanuz--frametop/hands/rec/tests/test_session.py
T
DeeJanuzandClaude Opus 5.5 79eb251d6c Hands: the headset button as Next, clearer push steps 
The headset's right-side click button (KEY_SELECT on gpio-keys, read without
a grab) now works the session: Next while a step waits, pause during a hold,
resume while paused. With no mouse connected the hints lead with it.

The push sections say plainly to push straight out from the headset and pull
back, with a side-view picture of the head, the headset and the arrow, and
the bar's ends read "At your chest" and "Arm out". The bar labels are sent
as one field, so labels with spaces no longer split.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 17:19:16 -06:00

429 lines
18 KiB
Python
Raw Blame History

#!/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")
# 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))
# /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 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()
Reference in new issue View Git Blame Copy Permalink