mirror of
https://github.com/DeeJanuz/frametop.git
synced 2026-10-08 17:00:13 +02:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aef18d3601 | ||
|
|
983a4654b3 | ||
|
|
85b8717bb8 | ||
|
|
dfd1bbca0f |
No files matched your search
+5
-3
@@ -10,6 +10,8 @@ CFLAGS ?= -O2 -g -Wall -Wextra -Wno-unused-parameter
|
||||
CXXFLAGS ?= -O2 -g -Wall -Wextra -Wno-unused-parameter -Wno-psabi
|
||||
CXXFLAGS += -std=c++17 -fopenmp -I$(NCNN)/include/ncnn
|
||||
LDLIBS = $(NCNN)/lib/libncnn.a -ljsoncpp -fopenmp -lpthread
|
||||
# The standalone Hand Recorder's release build passes LDFLAGS=-static (it runs on the host).
|
||||
LDFLAGS ?=
|
||||
|
||||
CAMD = camd/camd.c camd/tp.c camd/xrcams.c
|
||||
TRACK = track/calib.cpp track/nets.cpp track/tracker.cpp track/io.cpp track/record.cpp track/pinch.cpp track/sides.cpp
|
||||
@@ -24,15 +26,15 @@ build/ft-camd: $(CAMD) camd/tp.h camd/xrcams.h camd/fhring.h
|
||||
|
||||
build/ft-hands: track/main.cpp $(TRACK) $(HDR) $(NCNN)/lib/libncnn.a
|
||||
@mkdir -p build
|
||||
$(CXX) $(CXXFLAGS) -o $@ track/main.cpp $(TRACK) $(LDLIBS)
|
||||
$(CXX) $(CXXFLAGS) $(LDFLAGS) -o $@ track/main.cpp $(TRACK) $(LDLIBS)
|
||||
|
||||
build/ft-handreplay: track/replay.cpp $(TRACK) $(HDR) $(NCNN)/lib/libncnn.a
|
||||
@mkdir -p build
|
||||
$(CXX) $(CXXFLAGS) -o $@ track/replay.cpp $(TRACK) $(LDLIBS)
|
||||
$(CXX) $(CXXFLAGS) $(LDFLAGS) -o $@ track/replay.cpp $(TRACK) $(LDLIBS)
|
||||
|
||||
build/sides-test: tests/sides_test.cpp $(TRACK) $(HDR) $(NCNN)/lib/libncnn.a
|
||||
@mkdir -p build
|
||||
$(CXX) $(CXXFLAGS) -o $@ tests/sides_test.cpp $(TRACK) $(LDLIBS)
|
||||
$(CXX) $(CXXFLAGS) $(LDFLAGS) -o $@ tests/sides_test.cpp $(TRACK) $(LDLIBS)
|
||||
|
||||
check: build/sides-test
|
||||
build/sides-test
|
||||
|
||||
+18
-3
@@ -23,6 +23,17 @@ The plan this belongs to is `~/Desktop/Projects/frame-hands/notes/hands-plan.md`
|
||||
|
||||
Every part runs in the dev container, as ft-hands and Input Settings do. The host Python isn't used: it lacks PySide6 and zstd. `setup/dev-container.sh` gains `zstd`.
|
||||
|
||||
### The standalone Hand Recorder
|
||||
|
||||
[frametop-hand-recorder](https://github.com/DeeJanuz/frametop-hand-recorder) is the recorder for people without Frametop. It pins this repo as a submodule and ships the parts above with its own window, a Qt Quick Controls `main.qml` sized for SteamVR's dashboard. Its release builds the binaries for the SteamOS host: ft-camd and ft-hands statically (`make LDFLAGS=-static`), and ft-handpanel against SteamVR's `libopenvr_api`. Everything runs on the host, the Python from a venv (`ft_handrec.py --qml ITS_QML --style Basic`).
|
||||
|
||||
Its tarball keeps this tree's layout (`hands/build/`, `hands/rec/`, `hands/models/`), so the paths here don't change. A `standalone.json` at the top marks it (`takes.standalone()`):
|
||||
- `session.py` starts ft-hands directly, not through distrobox.
|
||||
- session.json's `tool` and the manifest's are `ft-handrec <describe> (frametop-hand-recorder <version>)`.
|
||||
- The repair hints say to run its install command again.
|
||||
|
||||
The recordings go to the same `~/.local/share/frametop/hands/contrib`, so either recorder shows both's sessions.
|
||||
|
||||
## Processes during a session
|
||||
|
||||
- **ft-camd** publishes the camera ring. If it isn't running, the session starts it as the transient user unit `frametop-handrec-camd.service`, the way `hands/ft-cutouts` starts `frametop-cutouts-camd.service` (needs `hands/build/ft-camd` with capabilities: `hands/run.sh caps`). An ft-camd already running from ft-cutouts or ft-handsctl is used as it is.
|
||||
@@ -296,11 +307,15 @@ On 2026-10-02 a whole session showed "I can't see your hands": after the headset
|
||||
### Controls
|
||||
|
||||
- The window has Start, a big Next (while a step waits; its hint is the panel's), Pause/Resume, Redo step, Skip section and Stop. While it has focus: Space is Next, P pauses or resumes, R redoes, S skips the section, Esc stops. The panel's bottom line and the window list them. The window also shows the step's picture, diagram, countdown and "Hold".
|
||||
- **The headset's button.** The Frame has a click button on its right side, for use without controllers: `KEY_SELECT` (353) on the evdev device `gpio-keys`. While a step waits (the welcome, a section's intro, a step's ready screen) a press is Next; during a countdown or hold (and auto mode's timed screens) it pauses; while paused it resumes. The session finds the device in `/proc/bus/input/devices` by name and its KEY bitmap (`event3` on the maintainer's Frame) and reads `input_event` structs with plain `struct` (no python-evdev), from a thread. Only key-downs count (value 1: releases and autorepeat, value 2, don't), and presses closer than 0.3 s count once.
|
||||
- **The headset's button.** The Frame has a click button on its right side, for use without controllers: `KEY_SELECT` (353) on the evdev device `gpio-keys`. It has three gestures (`ButtonGestures`), so a session can run with the window out of reach (the standalone Hand Recorder's window is in SteamVR's dashboard, closed while recording):
|
||||
- **A press.** While a step waits (the welcome, a section's intro, a step's ready screen, the no-hands question) it's Next; during a countdown or hold (and auto mode's timed screens) it pauses; while paused it resumes. It counts once 0.45 s have passed without a second press, so Next and pause come that much after the press.
|
||||
- **Two presses**, the second down within 0.45 s of the first one's release: redo, as R. It counts at the second press. Where R does nothing (nothing recorded yet, a section's intro), neither does this.
|
||||
- **A hold** of 1.5 s: stop, as Esc. It counts after 1.5 s, while the button is still down.
|
||||
- The session finds the device in `/proc/bus/input/devices` by name and its KEY bitmap (`event3` on the maintainer's Frame) and reads `input_event` structs with plain `struct` (no python-evdev), from a thread. Downs (value 1) and ups (value 0) count; autorepeat (value 2) doesn't. The gaps between events come from the kernel's timestamps, so a double press read in one go still counts as two. A release and a press closer than 30 ms are one press (gpio-keys debounces too).
|
||||
- It's read, never grabbed (`EVIOCGRAB`). Frametop's input relay (`input/input-relay.py`), ft-powerd, SteamVR and gamescope read the same device, and the relay remaps its volume keys (the experimental branch's `docs/hazards.md`). A grab would take the volume keys from the relay.
|
||||
- `steamos` is in the `input` group, so it needs no sudo. The dev container sees the host's `/dev/input` and `/proc/bus/input/devices`, and the group carries over (checked 2026-10-02), so it works from the window there and from `session.py` on the host. If the device is missing or can't be opened, the session logs it, keeps trying every 5 s, and the hints don't mention the button.
|
||||
- **Not known yet (needs the headset):** what SteamVR and gamescope do with the same press. They read it too, so it may also click whatever is under the head or gaze pointer in VR, or open something. Check on the first session; if it does, the fix is on their side or a different button, not a grab.
|
||||
- **The Next hint follows what's there.** No mouse connected: "Ready? Press the button on the right side of the headset" (the window's Next can't be clicked, and Space needs the window focused). A mouse: "Ready? Press Space, click Next, or press the headset button". No button: "Ready? Press Space or click Next". The panel's bottom line leads with "Headset button: next, pause". A mouse is a device in `/proc/bus/input/devices` with EV_REL, REL_X and REL_Y that isn't made in software: uinput devices (Frametop's virtual mouse, frame-voice's keyboard) sit under `/devices/virtual/input` or on the virtual bus (6), and are left out; Bluetooth mice come through uhid, under `/devices/virtual/misc/uhid`, and count. It's looked at again at each step, so a mouse plugged in mid-session counts from the next step.
|
||||
- **Not known yet (needs the headset):** what SteamVR and gamescope do with the same press. They read it too, so it may also click whatever is under the head or gaze pointer in VR, or open something, and a hold or a double press may mean something else to them. Check on the first session; if it does, the fix is on their side or a different button, not a grab.
|
||||
- **The Next hint follows what's there.** No mouse connected: "Ready? Press the button on the right side of the headset" (the window's Next can't be clicked, and Space needs the window focused). A mouse: "Ready? Press Space, click Next, or press the headset button". No button: "Ready? Press Space or click Next". The panel's bottom line leads with "Headset button: press for next or pause · press twice to redo · hold to stop", then the window's keys by letter. With the button, the pause note and the no-hands question say how to resume or stop with it too. A mouse is a device in `/proc/bus/input/devices` with EV_REL, REL_X and REL_Y that isn't made in software: uinput devices (Frametop's virtual mouse, frame-voice's keyboard) sit under `/devices/virtual/input` or on the virtual bus (6), and are left out; Bluetooth mice come through uhid, under `/devices/virtual/misc/uhid`, and count. It's looked at again at each step, so a mouse plugged in mid-session counts from the next step.
|
||||
- **R (redo).** During a step's countdown or hold: that step starts again from its ready screen. At a step's ready screen: the step before it (in this section) goes again. Either way a `redo` event marks the range of the try being redone, so its labels are skipped; the sets stay, to delete in review if wanted.
|
||||
- A pause stops the take's recording and starts it again on resume as the next part of the same take (`sets-2.bin`, and so on). takes.py reads all parts in order. A pause while a step waits only shows "Paused"; a Next pressed while paused doesn't count.
|
||||
|
||||
|
||||
+21
-5
@@ -24,7 +24,9 @@ Everything lives under ~/.local/share/frametop/hands/contrib (--base). Nothing i
|
||||
unless the person presses Upload; while CONSENT.md or UPLOAD.md is a draft, Upload stays off
|
||||
unless FT_HANDREC_ALLOW_UPLOAD=1 (the maintainer's rehearsal against a test repo, picked with
|
||||
FT_HANDREC_DATASET). --hub-dry-run does everything but the network calls.
|
||||
Launch with hands/rec/ft-handrec (host wrapper).
|
||||
Launch with hands/rec/ft-handrec (host wrapper). The standalone Hand Recorder
|
||||
(github.com/DeeJanuz/frametop-hand-recorder) runs this backend on the host, from a venv, with
|
||||
its own window for SteamVR's dashboard: --qml its main.qml, --style Basic.
|
||||
"""
|
||||
import argparse
|
||||
import datetime
|
||||
@@ -43,6 +45,7 @@ from PySide6.QtGui import QColor, QDesktopServices, QFont, QGuiApplication, QIco
|
||||
from PySide6.QtQml import QQmlApplicationEngine
|
||||
from PySide6.QtQuick import QQuickImageProvider
|
||||
from PySide6.QtQuickControls2 import QQuickStyle
|
||||
import shiboken6
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
@@ -292,6 +295,11 @@ class Backend(QObject):
|
||||
def consentText(self):
|
||||
return read_text(CONSENT_PATH) or "CONSENT.md is missing."
|
||||
|
||||
@Property(str, constant=True)
|
||||
def toolVersion(self):
|
||||
"""What session.json records as the tool (takes.tool_version), for the window to show."""
|
||||
return takes.tool_version()
|
||||
|
||||
@Property(str, constant=True)
|
||||
def consentVersion(self):
|
||||
return consent_version()
|
||||
@@ -389,7 +397,8 @@ class Backend(QObject):
|
||||
measured = ""
|
||||
try:
|
||||
ring = mod.ring_lighting()
|
||||
if ring is None and not self.sessionActive:
|
||||
# A dry run starts nothing, the camera broker included.
|
||||
if ring is None and not self.sessionActive and not self._session_options.get("dry_run"):
|
||||
if mod.start_camd():
|
||||
self._camd_started = True
|
||||
time.sleep(1.0) # the first near-black frames
|
||||
@@ -1173,6 +1182,9 @@ def main():
|
||||
help="test: Upload checks the export and says what it would send, with no network calls")
|
||||
ap.add_argument("--ignore-cameras", action="store_true",
|
||||
help="start sessions even if the camera check (hands/camcheck.py) finds the upper cameras off")
|
||||
ap.add_argument("--qml", default=os.path.join(HERE, "main.qml"),
|
||||
help="the window (default: this folder's main.qml, which needs Kirigami)")
|
||||
ap.add_argument("--style", default="org.kde.desktop", help="the Qt Quick Controls style")
|
||||
a, qt_args = ap.parse_known_args()
|
||||
app = QGuiApplication([sys.argv[0]] + qt_args)
|
||||
app.setApplicationName("ft-handrec")
|
||||
@@ -1180,7 +1192,7 @@ def main():
|
||||
app.setDesktopFileName("ft-handrec")
|
||||
if not QIcon.themeName():
|
||||
QIcon.setThemeName("breeze")
|
||||
QQuickStyle.setStyle("org.kde.desktop")
|
||||
QQuickStyle.setStyle(a.style)
|
||||
store = takes.Store(a.base)
|
||||
engine = QQmlApplicationEngine()
|
||||
engine.addImageProvider("frames", FrameProvider(store))
|
||||
@@ -1195,7 +1207,7 @@ def main():
|
||||
app.aboutToQuit.connect(backend.shutdown)
|
||||
engine.rootContext().setContextProperty("backend", backend)
|
||||
engine.rootContext().setContextProperty("startPage", a.page)
|
||||
engine.load(QUrl.fromLocalFile(os.path.join(HERE, "main.qml")))
|
||||
engine.load(QUrl.fromLocalFile(os.path.abspath(a.qml)))
|
||||
if not engine.rootObjects():
|
||||
sys.exit(1)
|
||||
|
||||
@@ -1208,7 +1220,11 @@ def main():
|
||||
signal.signal(signal.SIGINT, on_signal)
|
||||
tick = QTimer(interval=500, timeout=lambda: None)
|
||||
tick.start()
|
||||
sys.exit(app.exec())
|
||||
code = app.exec()
|
||||
# The window goes before the backend: otherwise its bindings run again against a deleted
|
||||
# backend as Python tears down ("Cannot read property ... of null", one per binding).
|
||||
shiboken6.delete(engine)
|
||||
sys.exit(code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+4
-2
@@ -486,7 +486,8 @@ Kirigami.ApplicationWindow {
|
||||
+ "Each step waits until you're ready: press the button on the right side of the headset, "
|
||||
+ "or Space or Next in this window. A 3-2-1 "
|
||||
+ "countdown follows, then hold the pose until the bar runs out. Nothing is recorded while "
|
||||
+ "a step waits. The headset button also pauses and resumes a recording. In this window P "
|
||||
+ "a step waits. The headset button also pauses and resumes a recording; press it twice "
|
||||
+ "to record a step again, or hold it to stop. In this window P "
|
||||
+ "pauses, R records the last step again, S skips a section and Esc stops.\n\n"
|
||||
+ "Nothing leaves the headset. Afterwards you watch the takes in Review, delete anything "
|
||||
+ "you don't want to share, and only then export."
|
||||
@@ -919,7 +920,8 @@ Kirigami.ApplicationWindow {
|
||||
opacity: 0.7
|
||||
text: "The instructions appear in the headset. "
|
||||
+ (sessionView.st.button ? "The button on the right side of the headset: "
|
||||
+ (sessionView.stepMode ? "next, " : "") + "pause or resume. " : "")
|
||||
+ (sessionView.stepMode ? "next, " : "") + "pause or resume; twice: record "
|
||||
+ "the last step again; hold: stop. " : "")
|
||||
+ "While this window has focus: "
|
||||
+ (sessionView.stepMode ? "Space: next · " : "")
|
||||
+ "P: pause or resume · R: record the last step again · S: skip section · Esc: stop. "
|
||||
|
||||
+150
-57
@@ -51,6 +51,7 @@ sys.path.insert(0, HANDS)
|
||||
import camcheck # noqa: E402 (hands/camcheck.py: are all four mono cameras running?)
|
||||
sys.path.insert(0, HERE)
|
||||
import sides # noqa: E402 (hands/rec/sides.py: which side camera is which)
|
||||
import takes # noqa: E402 (hands/rec/takes.py: the version, and the standalone recorder's build info)
|
||||
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")
|
||||
@@ -60,6 +61,9 @@ 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"
|
||||
# The standalone Hand Recorder (takes.standalone()): its binaries run on the host, so ft-hands
|
||||
# starts directly rather than in the dev container, and repairs mean its install command.
|
||||
STANDALONE = takes.standalone()
|
||||
|
||||
RECORD_HZ = 10
|
||||
SIDES_READ_S = 0.5 # how often the tracking ft-hands' side camera decision is read
|
||||
@@ -72,15 +76,19 @@ MIN_FREE = 1.5e9 # stop the session before the disk fills
|
||||
COUNTDOWN_S = 3 # step mode: the 3-2-1 before each step, recorded
|
||||
FIRST_SET_S = 3.0 # step mode: how long the hold may wait for its recording's first set
|
||||
RESUME_HINT = "Paused. Resume: P in the Hand recorder window"
|
||||
RESUME_HINT_BUTTON = "Paused. Resume: press the headset button, or P in the Hand recorder window"
|
||||
READY_TEXT = "Ready? Press Space or click Next"
|
||||
# With the headset's button: it leads when no mouse is connected (the window's Next can't be clicked).
|
||||
READY_BUTTON = "Ready? Press the button on the right side of the headset"
|
||||
READY_BUTTON_MOUSE = "Ready? Press Space, click Next, or press the headset button"
|
||||
KEYS_STEP = "Hand recorder window: Space next \u00b7 P pause \u00b7 R redo \u00b7 S skip section \u00b7 Esc stop"
|
||||
KEYS_AUTO = "Hand recorder window: P pause \u00b7 R redo \u00b7 S skip section \u00b7 Esc stop"
|
||||
KEYS_STEP_BUTTON = ("Headset button: next, pause \u00b7 Window: Space next \u00b7 P pause \u00b7 R redo "
|
||||
"\u00b7 S skip \u00b7 Esc stop")
|
||||
KEYS_AUTO_BUTTON = "Headset button: pause \u00b7 Window: P pause \u00b7 R redo \u00b7 S skip section \u00b7 Esc stop"
|
||||
# The headset button: a press, two (redo) and a hold (stop): ButtonGestures. The window's keys follow
|
||||
# by letter: the window lists what they do.
|
||||
KEYS_STEP_BUTTON = ("Headset button: press for next or pause \u00b7 press twice to redo \u00b7 hold to stop "
|
||||
"\u00b7 Window keys: Space P R S Esc")
|
||||
KEYS_AUTO_BUTTON = ("Headset button: press to pause \u00b7 press twice to redo \u00b7 hold to stop "
|
||||
"\u00b7 Window keys: P R S Esc")
|
||||
# The early no-hands stop (DESIGN.md, "Camera check"): the first step of this section has both
|
||||
# hands up. If the live tracker publishes through its hold and never sees a hand, the session
|
||||
# stops that step and asks: try again, or stop. Only when the tracker published in at least
|
||||
@@ -93,6 +101,8 @@ NO_HANDS_TITLE = "I can't see your hands"
|
||||
NO_HANDS_TEXT = "The hand tracker didn't see either of your hands during that whole step."
|
||||
NO_HANDS_RETRY = ("Try again: hold both hands up in front of you, about 40 cm away. To stop instead: "
|
||||
"Esc or Stop in the Hand recorder window.")
|
||||
NO_HANDS_RETRY_BUTTON = ("Try again: hold both hands up in front of you, about 40 cm away, and press the headset "
|
||||
"button. To stop instead: hold the button, or Esc or Stop in the Hand recorder window.")
|
||||
|
||||
|
||||
def mono_ns():
|
||||
@@ -131,6 +141,13 @@ def host_command(*cmd):
|
||||
exe] + list(cmd)
|
||||
|
||||
|
||||
def fix_hint(frametop):
|
||||
"""How to repair a missing or broken part: Frametop's command, or the standalone install's."""
|
||||
if STANDALONE:
|
||||
return STANDALONE.get("reinstall") or "run the Hand Recorder's install command again"
|
||||
return frametop
|
||||
|
||||
|
||||
def host_path(path):
|
||||
"""A host file, from the container (/run/host) or the host itself."""
|
||||
for p in ("/run/host" + path, path):
|
||||
@@ -369,10 +386,10 @@ def start_camd(path=None, log=lambda line: None, stop=lambda: False, timeout=15.
|
||||
if ring_alive(path):
|
||||
return False
|
||||
if not os.access(FT_CAMD, os.X_OK):
|
||||
raise RuntimeError("ft-camd isn't built: hands/build.sh")
|
||||
raise RuntimeError("ft-camd isn't built: " + fix_hint("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 RuntimeError("ft-camd needs its capabilities: hands/run.sh caps (asks for sudo)")
|
||||
raise RuntimeError("ft-camd needs its capabilities: " + fix_hint("hands/run.sh caps (asks for sudo)"))
|
||||
started = start_unit(CAMD_UNIT, "the camera broker", [FT_CAMD, "--status", "60"], log)
|
||||
end = time.monotonic() + timeout
|
||||
while time.monotonic() < end:
|
||||
@@ -614,7 +631,7 @@ class Recorder:
|
||||
"--sides", "auto" if swap is None else "1" if swap else "0"]
|
||||
if ring:
|
||||
argv += ["--ring", ring]
|
||||
if not in_container():
|
||||
if not in_container() and not STANDALONE:
|
||||
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()
|
||||
@@ -673,7 +690,9 @@ EV_KEY, EV_REL = 0x01, 0x02
|
||||
REL_X, REL_Y = 0x00, 0x01
|
||||
KEY_SELECT = 353
|
||||
BUTTON_DEVICE = "gpio-keys"
|
||||
BUTTON_DEBOUNCE_S = 0.3 # presses closer than this count once
|
||||
BUTTON_DOUBLE_S = 0.45 # a second press this soon after the first one's release: a double press
|
||||
BUTTON_HOLD_S = 1.5 # held down this long: a hold
|
||||
BUTTON_BOUNCE_S = 0.03 # a release and press closer than this are one press (gpio-keys debounces too)
|
||||
|
||||
|
||||
def parse_input_devices(text):
|
||||
@@ -753,29 +772,84 @@ def mouse_connected(path=INPUT_DEVICES):
|
||||
return any(real_mouse(d) for d in read_input_devices(path))
|
||||
|
||||
|
||||
def button_presses(data):
|
||||
"""KEY_SELECT key-downs in a run of input_event structs (value 1; releases and autorepeat
|
||||
are left out). Returns (how many, the bytes left over after the last whole event)."""
|
||||
n, usable = 0, len(data) - len(data) % INPUT_EVENT.size
|
||||
def button_events(data):
|
||||
"""KEY_SELECT downs and ups in a run of input_event structs: [(True for a down or False for
|
||||
an up, the event's time in seconds), ...] (autorepeat, value 2, is left out). Returns (them,
|
||||
the bytes left over after the last whole event)."""
|
||||
out, usable = [], len(data) - len(data) % INPUT_EVENT.size
|
||||
for off in range(0, usable, INPUT_EVENT.size):
|
||||
_, _, etype, code, value = INPUT_EVENT.unpack_from(data, off)
|
||||
if etype == EV_KEY and code == KEY_SELECT and value == 1:
|
||||
n += 1
|
||||
return n, data[usable:]
|
||||
sec, usec, etype, code, value = INPUT_EVENT.unpack_from(data, off)
|
||||
if etype == EV_KEY and code == KEY_SELECT and value in (0, 1):
|
||||
out.append((value == 1, sec + usec / 1e6))
|
||||
return out, data[usable:]
|
||||
|
||||
|
||||
class ButtonGestures:
|
||||
"""The headset button's downs and ups, with their times, as gestures:
|
||||
"press": one short press, reported once the double-press time has passed without a second;
|
||||
"double": two presses, the second down within double_s of the first one's release, reported
|
||||
at the second down;
|
||||
"hold": held down for hold_s, reported while still down.
|
||||
A release and a press closer than bounce_s are one press. Each call returns the gestures that
|
||||
are due, in order; tick() is called now and then to report the ones that are due by time."""
|
||||
|
||||
def __init__(self, double_s=BUTTON_DOUBLE_S, hold_s=BUTTON_HOLD_S, bounce_s=BUTTON_BOUNCE_S):
|
||||
self.double_s, self.hold_s, self.bounce_s = double_s, hold_s, bounce_s
|
||||
self.down_at = None # when the button went down, while it's down
|
||||
self.done = False # the press that's down was already reported (a double, a hold)
|
||||
self.released = None # when a short press was let go, until it's reported or doubled
|
||||
self._up = None # the last release, for the bounce: (time, down_at, done, released before it)
|
||||
|
||||
def down(self, t):
|
||||
if self.down_at is not None: # a missed release
|
||||
return []
|
||||
if self._up and t - self._up[0] < self.bounce_s: # a bounce: still the same press
|
||||
_, self.down_at, self.done, self.released = self._up
|
||||
self._up = None
|
||||
return []
|
||||
out = self.tick(t)
|
||||
self.down_at, self.done = t, False
|
||||
if self.released is not None: # within the double-press time (else tick reported it)
|
||||
self.released, self.done = None, True
|
||||
out.append("double")
|
||||
return out
|
||||
|
||||
def up(self, t):
|
||||
if self.down_at is None:
|
||||
return []
|
||||
out = self.tick(t)
|
||||
self._up = (t, self.down_at, self.done, self.released)
|
||||
if not self.done:
|
||||
self.released = t
|
||||
self.down_at, self.done = None, False
|
||||
return out
|
||||
|
||||
def tick(self, t):
|
||||
if self.down_at is not None and not self.done and t - self.down_at >= self.hold_s:
|
||||
self.done = True
|
||||
return ["hold"]
|
||||
if self.released is not None and t - self.released > self.double_s:
|
||||
self.released = None
|
||||
return ["press"]
|
||||
return []
|
||||
|
||||
def waiting(self):
|
||||
"""Something is due by time alone: a hold, or a press that may still become a double."""
|
||||
return (self.down_at is not None and not self.done) or self.released is not None
|
||||
|
||||
|
||||
class ButtonReader:
|
||||
"""Reads the headset button on a thread and calls on_press() per press, debounced. path:
|
||||
an event device or, for testing, a FIFO carrying input_event structs. It's opened read-only
|
||||
and shared; if it can't be opened (no device, no permission, /dev/input not reachable in a
|
||||
container) it says so in the log and tries again now and then."""
|
||||
"""Reads the headset button on a thread and calls on_gesture("press" | "double" | "hold")
|
||||
(ButtonGestures). path: an event device or, for testing, a FIFO carrying input_event
|
||||
structs. It's opened read-only and shared; if it can't be opened (no device, no permission,
|
||||
/dev/input not reachable in a container) it says so in the log and tries again now and then."""
|
||||
|
||||
def __init__(self, path, on_press, log=None, debounce_s=BUTTON_DEBOUNCE_S):
|
||||
self.path, self.on_press, self.log = path, on_press, log or (lambda s: None)
|
||||
self.debounce_s = debounce_s
|
||||
def __init__(self, path, on_gesture, log=None, double_s=BUTTON_DOUBLE_S, hold_s=BUTTON_HOLD_S,
|
||||
bounce_s=BUTTON_BOUNCE_S):
|
||||
self.path, self.on_gesture, self.log = path, on_gesture, log or (lambda s: None)
|
||||
self.gestures = ButtonGestures(double_s, hold_s, bounce_s)
|
||||
self.ok = False # opened at least once
|
||||
self._stop = threading.Event()
|
||||
self._last = -1e9
|
||||
self._thread = threading.Thread(target=self._run, name="handrec-button", daemon=True)
|
||||
|
||||
def start(self):
|
||||
@@ -786,11 +860,9 @@ class ButtonReader:
|
||||
self._stop.set()
|
||||
self._thread.join(2)
|
||||
|
||||
def _press(self, n):
|
||||
now = time.monotonic()
|
||||
if n and now - self._last >= self.debounce_s:
|
||||
self._last = now
|
||||
self.on_press()
|
||||
def _report(self, gestures):
|
||||
for g in gestures:
|
||||
self.on_gesture(g)
|
||||
|
||||
def _run(self):
|
||||
failed = False
|
||||
@@ -809,8 +881,9 @@ class ButtonReader:
|
||||
rest = b""
|
||||
try:
|
||||
while not self._stop.is_set():
|
||||
r, _, _ = select.select([fd], [], [], 0.2)
|
||||
r, _, _ = select.select([fd], [], [], 0.02 if self.gestures.waiting() else 0.2)
|
||||
if not r:
|
||||
self._report(self.gestures.tick(time.monotonic()))
|
||||
continue
|
||||
try:
|
||||
data = os.read(fd, INPUT_EVENT.size * 64)
|
||||
@@ -819,8 +892,13 @@ class ButtonReader:
|
||||
if not data: # a FIFO's writer left: open it again
|
||||
self._stop.wait(0.2)
|
||||
break
|
||||
n, rest = button_presses(rest + data)
|
||||
self._press(n)
|
||||
events, rest = button_events(rest + data)
|
||||
# The kernel's times (the realtime clock) keep the gaps between events that
|
||||
# came in one read; the last one is taken as now.
|
||||
now = time.monotonic()
|
||||
for is_down, t in events:
|
||||
at = now - min(max(events[-1][1] - t, 0.0), 2.0)
|
||||
self._report(self.gestures.down(at) if is_down else self.gestures.up(at))
|
||||
except OSError as e: # the device went away
|
||||
self.log("headset button: %s: %s" % (self.path, e.strerror))
|
||||
self._stop.wait(2)
|
||||
@@ -1117,15 +1195,6 @@ class _Redo(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:
|
||||
@@ -1193,8 +1262,8 @@ def camera_text(result):
|
||||
return camcheck.USER_TEXT
|
||||
if camcheck.is_ring_short(result):
|
||||
return ("The headset's tracking cameras are all running, but the recorder can't read some of them (%s). "
|
||||
"Close the Hand Recorder, run ~/frametop/hands/rec/install.sh again, and open it again. If that "
|
||||
"doesn't help, ask in the Frametop Discord." % result.get("reason", ""))
|
||||
"Close the Hand Recorder, %s, and open it again. If that doesn't help, ask in the Frametop "
|
||||
"Discord." % (result.get("reason", ""), fix_hint("run ~/frametop/hands/rec/install.sh again")))
|
||||
return ("Not all of the headset's tracking cameras are running (%s). Restart SteamVR, or restart the "
|
||||
"headset if that doesn't fix it." % result.get("reason", ""))
|
||||
|
||||
@@ -1297,8 +1366,25 @@ class Session:
|
||||
self._want[key] = value
|
||||
self._wake.set()
|
||||
|
||||
def button_gesture(self, gesture):
|
||||
"""The headset's button (ButtonGestures). A press: Next while a step waits, pause during
|
||||
a countdown or hold (and auto mode's timed screens), resume while paused. A double press:
|
||||
redo, as R. A hold: stop, as Esc."""
|
||||
if gesture == "press":
|
||||
self.button_press()
|
||||
elif gesture == "double":
|
||||
if self._status["state"] in ("done", "stopped", "error") or not self._status.get("can_redo"):
|
||||
return
|
||||
self.redo()
|
||||
self._log("headset button (redo)")
|
||||
elif gesture == "hold":
|
||||
if self._status["state"] in ("done", "stopped", "error"):
|
||||
return
|
||||
self._log("headset button (stop)")
|
||||
self.stop(wait=0)
|
||||
|
||||
def button_press(self):
|
||||
"""The headset's button: Next while a step waits, pause during a countdown or hold
|
||||
"""The headset button's press: Next while a step waits, pause during a countdown or hold
|
||||
(and auto mode's timed screens), resume while paused."""
|
||||
with self._lock:
|
||||
paused = self._want["pause"]
|
||||
@@ -1444,7 +1530,7 @@ class Session:
|
||||
self._ring_path = ring_path
|
||||
if not self.dry_run:
|
||||
if not os.access(FT_HANDS, os.X_OK):
|
||||
raise _Fail("ft-hands isn't built: hands/build.sh")
|
||||
raise _Fail("ft-hands isn't built: " + fix_hint("hands/build.sh"))
|
||||
self._ensure_ring(ring_path)
|
||||
if not tracker_running():
|
||||
if self.start_processes:
|
||||
@@ -1462,7 +1548,7 @@ class Session:
|
||||
pass
|
||||
removed = self._write_calibration() + self._write_device()
|
||||
self._session_json = {
|
||||
"schema": 1, "tool": "ft-handrec " + _git_describe(), "started": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"schema": 1, "tool": takes.tool_version(), "started": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
|
||||
"contributor": self.profile.get("contributor", ""),
|
||||
"lighting": lighting_record(self.lighting_choice, lighting),
|
||||
"checklist": self.checklist,
|
||||
@@ -1507,7 +1593,8 @@ class Session:
|
||||
if not path:
|
||||
self._log("headset button: no %s device with KEY_SELECT in %s" % (BUTTON_DEVICE, INPUT_DEVICES))
|
||||
return
|
||||
self._button = ButtonReader(path, self.button_press, log=self._log, debounce_s=BUTTON_DEBOUNCE_S).start()
|
||||
self._button = ButtonReader(path, self.button_gesture, log=self._log, double_s=BUTTON_DOUBLE_S,
|
||||
hold_s=BUTTON_HOLD_S, bounce_s=BUTTON_BOUNCE_S).start()
|
||||
end = time.monotonic() + 0.5 # opened in a moment, or it isn't reachable
|
||||
while not self._button.ok and time.monotonic() < end:
|
||||
time.sleep(0.01)
|
||||
@@ -1599,11 +1686,12 @@ class Session:
|
||||
raise _Stop()
|
||||
|
||||
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"]
|
||||
argv = [FT_HANDS, "--no-gestures", "--status", "0"]
|
||||
if not STANDALONE: # built in the dev container: it runs there
|
||||
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", "--"] + argv
|
||||
if self.ring:
|
||||
argv += ["--ring", self.ring]
|
||||
self._start_unit(HANDS_UNIT, "hand tracking for feedback", argv)
|
||||
@@ -1626,7 +1714,7 @@ class Session:
|
||||
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")
|
||||
raise _Fail("ft-handpanel isn't built: " + fix_hint("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
|
||||
@@ -1742,9 +1830,10 @@ class Session:
|
||||
if self._recording:
|
||||
self._stop_recording()
|
||||
self._event("pause")
|
||||
hint = RESUME_HINT_BUTTON if self._button_ok() else RESUME_HINT
|
||||
self._panel.cmd("paused on")
|
||||
self._panel.set("note", "note " + RESUME_HINT)
|
||||
self._emit(state="paused", note=RESUME_HINT)
|
||||
self._panel.set("note", "note " + hint)
|
||||
self._emit(state="paused", note=hint)
|
||||
self._log("paused")
|
||||
|
||||
def _unpause(self, record=True):
|
||||
@@ -1930,10 +2019,13 @@ class Session:
|
||||
for e in items)))
|
||||
self._status.update(strip=items, cue=cue)
|
||||
|
||||
def _button_ok(self):
|
||||
return bool(self._button and self._button.ok)
|
||||
|
||||
def _hints(self, action=True):
|
||||
"""The Next hint (with action) and the key line for what's there now: the headset button
|
||||
leads when no mouse is connected. Looked at again for each step, so a mouse plugged in counts."""
|
||||
button = bool(self._button and self._button.ok)
|
||||
button = self._button_ok()
|
||||
mouse = mouse_connected(self.input_devices)
|
||||
text = (READY_BUTTON_MOUSE if mouse else READY_BUTTON) if button else READY_TEXT
|
||||
keys = (KEYS_AUTO_BUTTON if self.auto else KEYS_STEP_BUTTON) if button else (KEYS_AUTO if self.auto else KEYS_STEP)
|
||||
@@ -2239,7 +2331,7 @@ class Session:
|
||||
for line in (cam or {}).get("evidence", []):
|
||||
self._log(" " + line)
|
||||
text = "%s|%s|%s" % (NO_HANDS_TEXT, cam_text or "The camera check found nothing wrong (%s)." % summary,
|
||||
NO_HANDS_RETRY)
|
||||
NO_HANDS_RETRY_BUTTON if self._button_ok() else NO_HANDS_RETRY)
|
||||
self._stop_note = "Stopped: no hands were seen in the first step. Camera check: %s." % summary
|
||||
if cam_text:
|
||||
self._stop_note += " " + cam_text
|
||||
@@ -2492,7 +2584,8 @@ def main():
|
||||
help="test: press Next by itself after S seconds of waiting (real time)")
|
||||
ap.add_argument("--poses", help="the pose pictures' folder, with poses.json (default hands/rec/poses)")
|
||||
ap.add_argument("--no-headset-button", action="store_true",
|
||||
help="don't read the headset's button (gpio-keys KEY_SELECT: Next, pause, resume)")
|
||||
help="don't read the headset's button (gpio-keys KEY_SELECT: a press is Next, pause or resume; "
|
||||
"two are redo; a hold is stop)")
|
||||
ap.add_argument("--button-device", metavar="PATH",
|
||||
help="test: read the button from this event device or FIFO of input_event structs (also in a dry run)")
|
||||
ap.add_argument("--quick", action="store_true",
|
||||
|
||||
+19
-1
@@ -83,8 +83,26 @@ def find_zstd():
|
||||
return next((p for p in ZSTD_PATHS if os.access(p, os.X_OK)), None)
|
||||
|
||||
|
||||
def standalone(path=None):
|
||||
"""The standalone Hand Recorder's build info, or None in a Frametop checkout. Its release
|
||||
(github.com/DeeJanuz/frametop-hand-recorder) ships this repo's tree with standalone.json at
|
||||
the top: {"name", "version", "frametop": this repo's git describe, "reinstall": how to repair
|
||||
an install}. Its binaries are built for the SteamOS host, so nothing runs in the dev container."""
|
||||
try:
|
||||
with open(path or os.path.join(REPO, "standalone.json")) as f:
|
||||
info = json.load(f)
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return info if isinstance(info, dict) else None
|
||||
|
||||
|
||||
def tool_version():
|
||||
"""ft-handrec plus the checkout's git describe, for session.json and the manifest."""
|
||||
"""ft-handrec plus the checkout's git describe, for session.json and the manifest. The
|
||||
standalone Hand Recorder has no checkout: the describe it was built from, and its version."""
|
||||
info = standalone()
|
||||
if info:
|
||||
return "ft-handrec %s (%s %s)" % (info.get("frametop") or "unknown", info.get("name") or "standalone",
|
||||
info.get("version") or "unknown")
|
||||
try:
|
||||
out = subprocess.run(["git", "-C", REPO, "describe", "--always", "--dirty", "--tags"],
|
||||
capture_output=True, text=True, timeout=5)
|
||||
|
||||
@@ -2,7 +2,8 @@
|
||||
"""main.qml against ft_handrec.Backend: every backend.name(...) the window calls is a slot, and
|
||||
every backend.name it reads is a property or a slot. A method that lost its @Slot shows up in
|
||||
QML only as "is not a function" when its button is pressed (2026-10-03: Export did nothing).
|
||||
Needs PySide6 (the dev container); skipped without it.
|
||||
Needs PySide6 (the dev container); skipped without it. FT_HANDREC_QML=PATH checks another
|
||||
window against the same backend (the standalone Hand Recorder's, in its CI).
|
||||
|
||||
python3 hands/rec/tests/test_qml_backend.py
|
||||
"""
|
||||
@@ -14,6 +15,7 @@ import unittest
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
REC = os.path.dirname(HERE)
|
||||
sys.path.insert(0, REC)
|
||||
QML = os.environ.get("FT_HANDREC_QML") or os.path.join(REC, "main.qml")
|
||||
|
||||
|
||||
class QmlBackendTest(unittest.TestCase):
|
||||
@@ -25,7 +27,7 @@ class QmlBackendTest(unittest.TestCase):
|
||||
meta = self.meta = ft_handrec.Backend.staticMetaObject
|
||||
self.slots = {bytes(meta.method(i).name()).decode() for i in range(meta.methodCount())}
|
||||
self.props = {meta.property(i).name() for i in range(meta.propertyCount())}
|
||||
with open(os.path.join(REC, "main.qml")) as f:
|
||||
with open(QML) as f:
|
||||
self.qml = f.read()
|
||||
|
||||
def test_calls_are_slots(self):
|
||||
|
||||
+154
-17
@@ -458,8 +458,15 @@ B: MSC=10
|
||||
"""
|
||||
|
||||
|
||||
def event(etype, code, value):
|
||||
return session.INPUT_EVENT.pack(1, 2, etype, code, value)
|
||||
def event(etype, code, value, t=1.000002):
|
||||
return session.INPUT_EVENT.pack(int(t), round(t % 1 * 1e6), etype, code, value)
|
||||
|
||||
|
||||
def taps(*times):
|
||||
"""Presses, each 50 ms down, at these times (the kernel's clock), as one write."""
|
||||
return b"".join(event(session.EV_KEY, session.KEY_SELECT, 1, t) + event(0, 0, 0, t)
|
||||
+ event(session.EV_KEY, session.KEY_SELECT, 0, t + 0.05) + event(0, 0, 0, t + 0.05)
|
||||
for t in times)
|
||||
|
||||
|
||||
PRESS = event(session.EV_KEY, session.KEY_SELECT, 1) + event(0, 0, 0)
|
||||
@@ -479,35 +486,74 @@ class InputTest(unittest.TestCase):
|
||||
# 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
|
||||
def test_events(self):
|
||||
data = (PRESS + event(session.EV_KEY, session.KEY_SELECT, 2) * 3 # autorepeat: left out
|
||||
+ 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)
|
||||
self.assertEqual(session.button_events(data), ([(True, 1.000002), (False, 1.000002), (True, 1.000002)], b""))
|
||||
out, rest = session.button_events(PRESS + PRESS[:10])
|
||||
self.assertEqual((out, rest), ([(True, 1.000002)], PRESS[:10])) # half an event waits for the rest
|
||||
self.assertEqual(session.button_events(rest + PRESS[10:])[0], [(True, 1.000002)])
|
||||
self.assertEqual([t for _, t in session.button_events(taps(5.25))[0]], [5.25, 5.3])
|
||||
|
||||
def test_gestures(self):
|
||||
def run(steps, **kw):
|
||||
"""steps: (time, "down" | "up" | "tick"). The gestures, with the time each came."""
|
||||
g, out = session.ButtonGestures(double_s=0.4, hold_s=1.5, bounce_s=0.05, **kw), []
|
||||
for t, what in steps:
|
||||
out += [(t, x) for x in getattr(g, what)(t)]
|
||||
return out
|
||||
# a press comes once the double-press time has passed
|
||||
self.assertEqual(run([(0, "down"), (0.1, "up"), (0.3, "tick"), (0.6, "tick")]), [(0.6, "press")])
|
||||
# two: a double, at the second down, and nothing at its release
|
||||
self.assertEqual(run([(0, "down"), (0.1, "up"), (0.4, "down"), (0.5, "up"), (2, "tick")]),
|
||||
[(0.4, "double")])
|
||||
# the second too late: two presses (the first reported when the second goes down)
|
||||
self.assertEqual(run([(0, "down"), (0.1, "up"), (0.6, "down"), (0.7, "up"), (1.2, "tick")]),
|
||||
[(0.6, "press"), (1.2, "press")])
|
||||
# a hold: while still down, and nothing at the release
|
||||
self.assertEqual(run([(0, "down"), (1.0, "tick"), (1.5, "tick"), (3, "up"), (4, "tick")]),
|
||||
[(1.5, "hold")])
|
||||
# a hold noticed only at the release (a slow read) still counts
|
||||
self.assertEqual(run([(0, "down"), (2, "up"), (3, "tick")]), [(2, "hold")])
|
||||
# a bounce (up and down 20 ms apart) is one press
|
||||
self.assertEqual(run([(0, "down"), (0.1, "up"), (0.12, "down"), (0.2, "up"), (1, "tick")]),
|
||||
[(1, "press")])
|
||||
# a press, then a hold soon after: a double (the hold counts from a fresh down)
|
||||
self.assertEqual(run([(0, "down"), (0.1, "up"), (0.3, "down"), (2, "tick"), (2.1, "up")]),
|
||||
[(0.3, "double")])
|
||||
# a release with no down, and two downs: left alone
|
||||
self.assertEqual(run([(0, "up"), (1, "down"), (1.1, "down"), (1.2, "up"), (2, "tick")]),
|
||||
[(2, "press")])
|
||||
|
||||
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()
|
||||
got = []
|
||||
reader = session.ButtonReader(fifo, got.append, double_s=0.15, hold_s=0.5).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 + RELEASE) # a press
|
||||
time.sleep(0.3)
|
||||
w.write(PRESS[:7]) # two, the first split across writes
|
||||
time.sleep(0.02)
|
||||
w.write(PRESS[7:] + RELEASE)
|
||||
time.sleep(0.4)
|
||||
time.sleep(0.07)
|
||||
w.write(PRESS + RELEASE)
|
||||
time.sleep(0.3)
|
||||
w.write(taps(10, 10.1)) # two in one read: the kernel's times tell them apart
|
||||
time.sleep(0.3)
|
||||
w.write(PRESS) # held
|
||||
time.sleep(0.7)
|
||||
self.assertEqual(got, ["press", "double", "double", "hold"]) # the hold came before the release
|
||||
w.write(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)
|
||||
w.write(PRESS + RELEASE)
|
||||
time.sleep(0.3)
|
||||
reader.stop()
|
||||
self.assertEqual(len(presses), 3)
|
||||
self.assertEqual(got, ["press", "double", "double", "hold", "press"])
|
||||
self.assertTrue(reader.ok)
|
||||
finally:
|
||||
shutil.rmtree(tmp, ignore_errors=True)
|
||||
@@ -533,7 +579,7 @@ class ButtonSessionTest(SessionBase):
|
||||
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):
|
||||
with mock.patch.object(session, "BUTTON_DOUBLE_S", 0.1):
|
||||
s = self.session(speed=4, button_device=fifo)
|
||||
s.input_devices = procfile
|
||||
s.start()
|
||||
@@ -563,6 +609,41 @@ class ButtonSessionTest(SessionBase):
|
||||
finally:
|
||||
os.close(writer)
|
||||
|
||||
def test_redo_and_stop(self):
|
||||
"""Two presses redo the step, a hold stops the session; the hints say so."""
|
||||
fifo = os.path.join(self.tmp, "button")
|
||||
os.mkfifo(fifo)
|
||||
writer = os.open(fifo, os.O_RDWR)
|
||||
press = lambda: os.write(writer, PRESS + RELEASE)
|
||||
try:
|
||||
with mock.patch.multiple(session, BUTTON_DOUBLE_S=0.15, BUTTON_HOLD_S=0.4):
|
||||
s = self.session(speed=2, button_device=fifo)
|
||||
s.start()
|
||||
self.waiting(s, "starting")
|
||||
press()
|
||||
self.waiting(s, "intro")
|
||||
press()
|
||||
self.waiting(s, "ready", "Fist.")
|
||||
press()
|
||||
self.wait_for(s, lambda st: st["state"] == "running", "the hold")
|
||||
os.write(writer, taps(1, 1.1)) # twice: redo
|
||||
self.waiting(s, "ready", "Fist.")
|
||||
time.sleep(0.1) # (sooner would be a bounce of the last)
|
||||
press()
|
||||
self.wait_for(s, lambda st: st["state"] == "running", "the hold again")
|
||||
press() # pause
|
||||
st = self.wait_for(s, lambda st: st["state"] == "paused", "paused")
|
||||
self.assertEqual(st["note"], session.RESUME_HINT_BUTTON)
|
||||
os.write(writer, PRESS) # held: stop
|
||||
s.join(20)
|
||||
os.write(writer, RELEASE)
|
||||
self.assertEqual(s.state, "stopped")
|
||||
names = [e["event"] for e in self.events(s)]
|
||||
self.assertEqual(names, ["take", "ready", "prompt", "redo", "wait", "ready", "prompt", "pause", "end"])
|
||||
self.assertIn("panel: keys " + session.KEYS_STEP_BUTTON, self.panel)
|
||||
finally:
|
||||
os.close(writer)
|
||||
|
||||
def test_no_button(self):
|
||||
s = self.session(next_after=0.0, button=False, button_device="/nonexistent")
|
||||
s.start()
|
||||
@@ -630,5 +711,61 @@ class ManyPartsTest(unittest.TestCase):
|
||||
self.assertEqual(r.summary["sets"], 120 + 30)
|
||||
|
||||
|
||||
class StandaloneTest(unittest.TestCase):
|
||||
"""The standalone Hand Recorder's tree (standalone.json at the top): ft-hands runs on the
|
||||
host, the version comes from the build info, and repairs point at its install command."""
|
||||
INFO = {"name": "frametop-hand-recorder", "version": "0.1.0", "frametop": "4ba49af",
|
||||
"reinstall": "run the install command again"}
|
||||
|
||||
def test_build_info(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
path = os.path.join(d, "standalone.json")
|
||||
self.assertIsNone(takes.standalone(path))
|
||||
with open(path, "w") as f:
|
||||
json.dump(self.INFO, f)
|
||||
self.assertEqual(takes.standalone(path), self.INFO)
|
||||
with open(path, "w") as f:
|
||||
f.write("[1]")
|
||||
self.assertIsNone(takes.standalone(path))
|
||||
with mock.patch.object(takes, "standalone", return_value=self.INFO):
|
||||
self.assertEqual(takes.tool_version(), "ft-handrec 4ba49af (frametop-hand-recorder 0.1.0)")
|
||||
|
||||
def recorder_argv(self, standalone):
|
||||
with tempfile.TemporaryDirectory() as d, mock.patch.object(session, "STANDALONE", standalone), \
|
||||
mock.patch.object(session, "in_container", return_value=False), \
|
||||
mock.patch.object(session.subprocess, "Popen") as popen:
|
||||
session.Recorder(d, 1, 10, None, None)
|
||||
return popen.call_args[0][0]
|
||||
|
||||
def tracker_argv(self, standalone):
|
||||
fake = mock.Mock(ring=None)
|
||||
with mock.patch.object(session, "STANDALONE", standalone), \
|
||||
mock.patch.object(session.subprocess, "run") as run:
|
||||
session.Session._start_tracker(fake)
|
||||
return fake._start_unit.call_args[0][2], run
|
||||
|
||||
def test_ft_hands_on_the_host(self):
|
||||
self.assertEqual(self.recorder_argv(self.INFO)[0], session.FT_HANDS)
|
||||
argv, run = self.tracker_argv(self.INFO)
|
||||
self.assertEqual(argv[0], session.FT_HANDS)
|
||||
self.assertIn("--no-gestures", argv)
|
||||
run.assert_not_called() # no container to bring up
|
||||
|
||||
def test_ft_hands_in_the_dev_container(self):
|
||||
self.assertTrue(self.recorder_argv(None)[0].endswith("distrobox"))
|
||||
argv, _ = self.tracker_argv(None)
|
||||
self.assertEqual(argv[:4], [os.path.expanduser("~/.local/bin/distrobox"), "enter", "dev", "--"])
|
||||
self.assertEqual(argv[4], session.FT_HANDS)
|
||||
|
||||
def test_fix_hint(self):
|
||||
with mock.patch.object(session, "STANDALONE", self.INFO):
|
||||
self.assertEqual(session.fix_hint("hands/build.sh"), "run the install command again")
|
||||
text = session.camera_text({"status": "degraded", "reason": "2 of 4", "ring_missing": ["upper_left"]})
|
||||
with mock.patch.object(session, "STANDALONE", None):
|
||||
self.assertEqual(session.fix_hint("hands/build.sh"), "hands/build.sh")
|
||||
self.assertIn("run the install command again", text)
|
||||
self.assertNotIn("~/frametop", text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user