diff --git a/data_only_viz/action_head_pub.py b/data_only_viz/action_head_pub.py index 40a1cf0..e0b24db 100644 --- a/data_only_viz/action_head_pub.py +++ b/data_only_viz/action_head_pub.py @@ -154,11 +154,13 @@ class ActionHeadPublisher(threading.Thread): """Lock state, extract hand features, emit /pose/hands once per tick.""" with self.state.lock(): hands = list(getattr(self.state, "persons_hands", None) or []) + chirality = list(getattr(self.state, "persons_hands_chirality", None) or []) if not hands and getattr(self.state, "hands_present", False): lkp = getattr(self.state, "left_hand_kp", None) rkp = getattr(self.state, "right_hand_kp", None) hands = [h for h in (lkp, rkp) if h is not None and len(h) > 0] - feats = self._hand_ext.step(hands) + chirality = [] + feats = self._hand_ext.step(hands, chirality or None) with self.state.lock(): self.state.hand_feats = feats self.bridge.send_hands(feats, t_now) @@ -198,28 +200,42 @@ class ActionHeadPublisher(threading.Thread): # "iphone" / "mediapipe". self._finger_source = os.environ.get("FINGER_SOURCE", "auto").lower() self._fdbg_n = 0 + # Safety knob: swap the two routed slots if the iPhone app's camera + # config inverts Vision chirality (fix live without redeploy). + self._hand_swap_lr = os.environ.get("HAND_SWAP_LR", "0") not in ( + "0", "", "false", "False", + ) + self._hand_near_min = float(os.environ.get("HAND_NEAR_MIN", "0.10")) - def _pick_hands(self) -> tuple[list, str, bool]: - """Return (hands, source_label, ip_fresh) per FINGER_SOURCE policy.""" + def _pick_hands(self) -> tuple[list, list, str, bool]: + """Return (hands, chirality, source_label, ip_fresh) per FINGER_SOURCE policy.""" with self.state.lock(): mp_hands = list(getattr(self.state, "persons_hands", None) or []) ip_hands = list(getattr(self.state, "persons_hands_iphone", None) or []) ip_t = getattr(self.state, "persons_hands_iphone_t", 0.0) + ip_chir = list(getattr(self.state, "persons_hands_chirality", None) or []) ip_fresh = bool(ip_hands) and (time.perf_counter() - ip_t) < 0.5 src = getattr(self, "_finger_source", "auto") if src == "iphone": - return ip_hands, "iphone", ip_fresh + return ip_hands, ip_chir, "iphone", ip_fresh if src == "mediapipe": - return mp_hands, "mediapipe", ip_fresh - return (ip_hands, "iphone", ip_fresh) if ip_fresh else ( - mp_hands, "mediapipe", ip_fresh) + return mp_hands, [], "mediapipe", ip_fresh + if ip_fresh: + return ip_hands, ip_chir, "iphone", ip_fresh + return mp_hands, [], "mediapipe", ip_fresh def _emit_fingers(self, t_now: float) -> None: """Detect finger strikes (air-piano) and emit /pose/finger.""" if not getattr(self, "_finger_enabled", False): return - hands, used, ip_fresh = self._pick_hands() - events = self._finger_det.step(hands, t_now) + from data_only_viz.hand_slots import route_hands + hands, chirality, used, ip_fresh = self._pick_hands() + slotted = route_hands( + hands, chirality or None, + swap=getattr(self, "_hand_swap_lr", False), + near_min=getattr(self, "_hand_near_min", 0.10), + ) + events = self._finger_det.step(slotted, t_now) if getattr(self, "_finger_dbg", False): self._fdbg_n += 1 if events or self._fdbg_n % 30 == 0: @@ -233,8 +249,14 @@ class ActionHeadPublisher(threading.Thread): /pose/pinch. Gated by PINCH_ENABLE, independently of the air-piano.""" if not getattr(self, "_pinch_enabled", False): return - hands, used, _ip = self._pick_hands() - for ev in self._pinch_det.step(hands, t_now): + from data_only_viz.hand_slots import route_hands + hands, chirality, used, _ip = self._pick_hands() + slotted = route_hands( + hands, chirality or None, + swap=getattr(self, "_hand_swap_lr", False), + near_min=getattr(self, "_hand_near_min", 0.10), + ) + for ev in self._pinch_det.step(slotted, t_now): if getattr(self, "_finger_dbg", False): LOG.info("pinch dbg: src=%s hand=%d finger=%d", used, ev.hand, ev.finger) diff --git a/data_only_viz/finger_strike.py b/data_only_viz/finger_strike.py index af260ea..c2680fa 100644 --- a/data_only_viz/finger_strike.py +++ b/data_only_viz/finger_strike.py @@ -1,9 +1,12 @@ -"""Air-piano finger strike detection from raw MediaPipe hand joints. +"""Air-piano finger strike detection from pre-routed hand slots. -Mirrors hand_features.py conventions (L/R slotting by cx, finite-guarded -coordinate extraction). A "strike" is a fast downward motion of a fingertip -RELATIVE to its base knuckle, so translating the whole hand does not fire all -fingers. Output feeds the OSC /pose/finger route consumed by SuperCollider. +A "strike" is a fast downward motion of a fingertip RELATIVE to its base +knuckle, so translating the whole hand does not fire all fingers. Output feeds +the OSC /pose/finger route consumed by SuperCollider. + +Consumers call route_hands() (hand_slots) once per tick and pass the resulting +2-element [left|None, right|None] list to FingerStrikeDetector.step() and +PinchDetector.step(). slot 0 = left hand (matrix "MG"), slot 1 = right. """ from __future__ import annotations @@ -24,26 +27,9 @@ MIDDLE_MCP: int = 9 PINCH_TIPS: tuple[int, ...] = (8, 12, 16, 20) -def order_hands_by_cx(hands: list, max_slots: int) -> list: - """Return up to max_slots hand-landmark lists ordered leftmost-first - (slot 0 = L), matching HandFeatureExtractor / FingerStrikeDetector.""" - cand = [] - for lm in hands: - try: - if lm is None or len(lm) < 21: - continue - except TypeError: - continue - xs = [_finite(_coord(p, "x", 0), 0.5) for p in lm[:21]] - cx = _clamp(sum(xs) / len(xs), 0.0, 1.0) - cand.append((cx, lm)) - cand.sort(key=lambda c: c[0]) - return [lm for _cx, lm in cand[:max_slots]] - - @dataclass class StrikeEvent: - hand: int # 0 = left slot (leftmost cx), 1 = right slot + hand: int # 0 = left slot, 1 = right slot finger: int # 0..4 = thumb, index, middle, ring, pinky strike_speed: float z: float @@ -70,27 +56,19 @@ class FingerStrikeDetector: self._state = [[_FingerState() for _ in range(5)] for _ in range(history_slots)] - def _slot_hands(self, hands: list) -> list: - """Validate and order hands leftmost-first (slot 0 = L), like - HandFeatureExtractor. Returns up to 2 (cx, landmarks) ordered.""" - cand = [] - for lm in hands: - try: - if lm is None or len(lm) < 21: - continue - except TypeError: - continue - xs = [_finite(_coord(p, "x", 0), 0.5) for p in lm[:21]] - cx = _clamp(sum(xs) / len(xs), 0.0, 1.0) - cand.append((cx, lm)) - cand.sort(key=lambda c: c[0]) - return cand[:len(self._state)] + def step(self, slotted: list, t_now: float) -> list[StrikeEvent]: + """Process a pre-routed 2-slot hand list [left|None, right|None]. - def step(self, hands: list, t_now: float) -> list[StrikeEvent]: - ordered = self._slot_hands(hands) - present = set(range(len(ordered))) + Absent slots (None) reset their finger state so re-entry does not + produce spurious velocity spikes. + """ events: list[StrikeEvent] = [] - for slot, (_cx, lm) in enumerate(ordered): + for slot, lm in enumerate(slotted): + if lm is None: + for f in range(5): + self._state[slot][f].prev_rel = None + self._state[slot][f].armed = True + continue for f in range(5): tip = lm[FINGERTIPS[f]] base = lm[FINGER_BASES[f]] @@ -116,18 +94,12 @@ class FingerStrikeDetector: tipx=_finite(_coord(tip, "x", 0), 0.5), tipy=tip_y, )) - # reset slots not present this tick so re-entry does not spike - for slot in range(len(self._state)): - if slot not in present: - for f in range(5): - self._state[slot][f].prev_rel = None - self._state[slot][f].armed = True return events @dataclass class PinchEvent: - hand: int # 0 = left slot (leftmost cx), 1 = right slot + hand: int # 0 = left slot, 1 = right slot finger: int # 1..4 = index, middle, ring, pinky (thumb is trigger) state: int = 1 # 1 = engage edge, 0 = release edge @@ -169,17 +141,27 @@ class PinchDetector: self.ext_ratio = ext_ratio self.ext_min = int(ext_min) # Engage fires only after debounce_frames consecutive qualifying - # frames; release stays immediate. 1 = no debounce. (Enforced in - # step(); wired fully by the debounce task.) + # frames; release stays immediate. 1 = no debounce. self.debounce_frames = max(1, int(debounce_frames)) self._state = [[_PinchState() for _ in range(4)] for _ in range(history_slots)] - def step(self, hands: list, t_now: float) -> list[PinchEvent]: - ordered = order_hands_by_cx(hands, len(self._state)) - present = set(range(len(ordered))) + def step(self, slotted: list, t_now: float) -> list[PinchEvent]: + """Process a pre-routed 2-slot hand list [left|None, right|None]. + + Absent slots synthesise release edges for any engaged fingers and + reset debounce state. + """ events: list[PinchEvent] = [] - for slot, lm in enumerate(ordered): + for slot, lm in enumerate(slotted): + if lm is None: + for i in range(4): + st = self._state[slot][i] + st.qual = 0 + if st.engaged: + st.engaged = False + events.append(PinchEvent(hand=slot, finger=i + 1, state=0)) + continue tx = _finite(_coord(lm[THUMB_TIP], "x", 0), 0.5) ty = _finite(_coord(lm[THUMB_TIP], "y", 1), 0.5) wx = _finite(_coord(lm[WRIST], "x", 0), 0.5) @@ -228,12 +210,4 @@ class PinchDetector: events.append(PinchEvent(hand=slot, finger=i + 1, state=1)) else: st.qual = 0 - for slot in range(len(self._state)): - if slot not in present: - for i in range(4): - st = self._state[slot][i] - st.qual = 0 - if st.engaged: - st.engaged = False - events.append(PinchEvent(hand=slot, finger=i + 1, state=0)) return events diff --git a/data_only_viz/hand_features.py b/data_only_viz/hand_features.py index 4602aca..9882d95 100644 --- a/data_only_viz/hand_features.py +++ b/data_only_viz/hand_features.py @@ -78,31 +78,37 @@ class HandFeatureExtractor: buf.append((cx, cy)) return _clamp(spd, 0.0, 1.0) - def step(self, hands: list) -> dict: - feats = [] - for lm in hands: - try: - valid = lm is not None and len(lm) >= 21 - except TypeError: - continue - if valid: - feats.append(self._features(lm)) - if not feats: + def step(self, hands: list, chirality=None) -> dict: + from data_only_viz.hand_slots import route_hands + left_lm, right_lm = route_hands(hands, chirality) + + if left_lm is None and right_lm is None: self._buf[0].clear() self._buf[1].clear() return {"L": None, "R": None, "dist": 0.0} - feats.sort(key=lambda f: f["cx"]) # leftmost first - left = feats[0] - right = feats[-1] if len(feats) > 1 else None - left["speed"] = self._speed(0, left["cx"], left["cy"]) - out_l = left + + out_l = None out_r = None - dist = 0.0 - if right is not None: - right["speed"] = self._speed(1, right["cx"], right["cy"]) - out_r = right - dist = _clamp(math.hypot(right["cx"] - left["cx"], - right["cy"] - left["cy"]), 0.0, 1.0) + + if left_lm is not None: + lf = self._features(left_lm) + lf["speed"] = self._speed(0, lf["cx"], lf["cy"]) + out_l = lf + else: + self._buf[0].clear() + + if right_lm is not None: + rf = self._features(right_lm) + rf["speed"] = self._speed(1, rf["cx"], rf["cy"]) + out_r = rf else: self._buf[1].clear() + + dist = 0.0 + if out_l is not None and out_r is not None: + dist = _clamp( + math.hypot(out_r["cx"] - out_l["cx"], out_r["cy"] - out_l["cy"]), + 0.0, 1.0, + ) + return {"L": out_l, "R": out_r, "dist": dist} diff --git a/data_only_viz/renderer.py b/data_only_viz/renderer.py index 6b48816..2726bdb 100644 --- a/data_only_viz/renderer.py +++ b/data_only_viz/renderer.py @@ -53,6 +53,7 @@ from .hand_display import ( panel_frame, panel_segments, ) +from .hand_slots import route_hands from .mesh_topology import ( BODY_TRIANGLES, FACE_TRIANGLES, @@ -164,6 +165,11 @@ class MetalRenderer(NSObject): self._hand_gate = HandPersistenceGate( min_frames=int(os.environ.get("HAND_PERSIST_FRAMES", "3")) ) + # Safety knob: swap the two routed hand slots if the iPhone app's + # camera config inverts Vision chirality (fix live without redeploy). + self._hand_swap_lr = os.environ.get("HAND_SWAP_LR", "0") not in ( + "0", "", "false", "False", + ) self._init_skel_cpu_buffer() self._init_mesh_cpu_buffer() self._build_pipelines() @@ -485,39 +491,29 @@ class MetalRenderer(NSObject): # even when use_arkit is True). Hands from s.persons_hands only. if self._mp_bones is not None: _body_b, _face_b, lhand_bones_p, _ = self._mp_bones - # Filter hands by persistence gate before routing to panels. - hands = [h for h, ok in zip(s.persons_hands, _hand_draw_flags) if ok] + # Filter hands by persistence gate + plausibility before routing + # to panels (a ghost hand must not steal a slot from a real one). + keep = [ + ok and hand_plausible(h, conf_min=self._hand_conf_min) + for h, ok in zip(s.persons_hands, _hand_draw_flags) + ] + gated_hands = [h for h, k in zip(s.persons_hands, keep) if k] chir_src = getattr(s, "persons_hands_chirality", None) or [] + # Keep chirality aligned with the gated hands (same filter mask). if chir_src and len(chir_src) == len(s.persons_hands): - chir = [c for c, ok in zip(chir_src, _hand_draw_flags) if ok] + gated_chir = [c for c, k in zip(chir_src, keep) if k] else: - chir = [] + gated_chir = None asp = float(s.width) / float(s.height) if s.width and s.height else 1.0 - # Route hands to left/right panels via chirality when available, - # otherwise by screen position (mirror-aware). - left_kp = None - right_kp = None - if chir and len(chir) == len(hands): - # chirality: 0 = left hand, 1 = right hand - for h_kp, c in zip(hands, chir): - if c == 0 and left_kp is None: - left_kp = h_kp - elif c == 1 and right_kp is None: - right_kp = h_kp - else: - # Fallback: order plausible hands by screen cx, mirror-aware. - # Screen left has low cx normally; high cx when video is mirrored. - plausible = [ - (h_kp, sum(p.x for p in h_kp) / len(h_kp)) - for h_kp in hands - if hand_plausible(h_kp, conf_min=self._hand_conf_min) - ] - plausible.sort(key=lambda t: -t[1] if mirror else t[1]) - if plausible: - left_kp = plausible[0][0] - if len(plausible) >= 2: - right_kp = plausible[1][0] + # Route to L/R panels. mirror= passed for cx fallback path; + # no near_min — panels show far hands as user proximity feedback. + slotted = route_hands( + gated_hands, gated_chir, + mirror=mirror, + swap=self._hand_swap_lr, + ) + left_kp, right_kp = slotted # Draw frame (pid 7) then wireframe (pid 5 left / 6 right) for side_name, h_kp, pid_s in ( diff --git a/data_only_viz/tests/test_action_head_finger_emit.py b/data_only_viz/tests/test_action_head_finger_emit.py index a55df98..0e7c230 100644 --- a/data_only_viz/tests/test_action_head_finger_emit.py +++ b/data_only_viz/tests/test_action_head_finger_emit.py @@ -18,6 +18,8 @@ class _FakeState: def _hand(index_tip_y: float, base_y: float = 0.4): lm = [[0.3, base_y, 0.0] for _ in range(21)] + lm[0] = [0.3, base_y + 0.15, 0.0] # wrist 0.15 below center -> hand_size 0.15 + lm[9] = [0.3, base_y, 0.0] # middle-MCP at center -> size 0.15 > near_min lm[FINGERTIPS[1]] = [0.3, index_tip_y, 0.0] lm[FINGER_BASES[1]] = [0.3, base_y, 0.0] return lm diff --git a/data_only_viz/tests/test_finger_strike.py b/data_only_viz/tests/test_finger_strike.py index f70dec4..4387c81 100644 --- a/data_only_viz/tests/test_finger_strike.py +++ b/data_only_viz/tests/test_finger_strike.py @@ -27,11 +27,11 @@ def _hand(tip_y_by_finger: dict[int, float], base_y: float = 0.4, def test_downward_spike_fires_exactly_one_strike(): det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=120.0) # frame 0: neutral (primes prev), index tip level with base - det.step([_hand({1: 0.40})], t_now=0.00) + det.step([_hand({1: 0.40}), None], t_now=0.00) # frame 1: index tip drops 0.06 below -> downward velocity 0.06 > thresh - e1 = det.step([_hand({1: 0.46})], t_now=0.04) + e1 = det.step([_hand({1: 0.46}), None], t_now=0.04) # frame 2: tip stays down -> velocity ~0, must NOT refire - e2 = det.step([_hand({1: 0.46})], t_now=0.08) + e2 = det.step([_hand({1: 0.46}), None], t_now=0.08) strikes = e1 + e2 assert len(strikes) == 1 assert strikes[0].finger == 1 @@ -44,16 +44,16 @@ def test_whole_hand_translation_does_not_fire(): out = [] # tip and base move down together each frame -> relative y constant for i, by in enumerate((0.40, 0.50, 0.60, 0.70)): - out += det.step([_hand({1: by}, base_y=by)], t_now=i * 0.04) + out += det.step([_hand({1: by}, base_y=by), None], t_now=i * 0.04) assert out == [] def test_refractory_blocks_second_strike(): det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=120.0) - det.step([_hand({1: 0.40})], t_now=0.00) # prime - a = det.step([_hand({1: 0.46})], t_now=0.02) # strike 1 - det.step([_hand({1: 0.40})], t_now=0.04) # lift -> rearm - b = det.step([_hand({1: 0.46})], t_now=0.06) # within 120 ms -> blocked + det.step([_hand({1: 0.40}), None], t_now=0.00) # prime + a = det.step([_hand({1: 0.46}), None], t_now=0.02) # strike 1 + det.step([_hand({1: 0.40}), None], t_now=0.04) # lift -> rearm + b = det.step([_hand({1: 0.46}), None], t_now=0.06) # within 120 ms -> blocked assert len(a) == 1 assert b == [] @@ -61,10 +61,10 @@ def test_refractory_blocks_second_strike(): def test_strike_speed_scales_with_velocity(): soft = FingerStrikeDetector(vel_thresh=0.02, speed_scale=0.10) hard = FingerStrikeDetector(vel_thresh=0.02, speed_scale=0.10) - soft.step([_hand({1: 0.40})], t_now=0.0) - hard.step([_hand({1: 0.40})], t_now=0.0) - s = soft.step([_hand({1: 0.44})], t_now=0.04) # delta 0.04 - h = hard.step([_hand({1: 0.50})], t_now=0.04) # delta 0.10 + soft.step([_hand({1: 0.40}), None], t_now=0.0) + hard.step([_hand({1: 0.40}), None], t_now=0.0) + s = soft.step([_hand({1: 0.44}), None], t_now=0.04) # delta 0.04 + h = hard.step([_hand({1: 0.50}), None], t_now=0.04) # delta 0.10 assert h[0].strike_speed > s[0].strike_speed @@ -89,8 +89,8 @@ _PINCH = ((0.5, 0.5), (0.52, 0.5)) # dist 0.02 -> ratio 0.067 def test_pinch_fires_on_thumb_index_contact(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65) - assert det.step([_pinch_hand(*_OPEN)], 0.0) == [] - ev = det.step([_pinch_hand(*_PINCH)], 0.1) + assert det.step([_pinch_hand(*_OPEN), None], 0.0) == [] + ev = det.step([_pinch_hand(*_PINCH), None], 0.1) assert len(ev) == 1 assert ev[0].finger == 1 # index = finger 1 assert ev[0].hand == 0 @@ -98,29 +98,29 @@ def test_pinch_fires_on_thumb_index_contact(): def test_pinch_does_not_refire_while_held(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand(*_OPEN)], 0.0) - a = det.step([_pinch_hand(*_PINCH)], 0.1) # fire - b = det.step([_pinch_hand(*_PINCH)], 0.2) # still held -> no refire + det.step([_pinch_hand(*_OPEN), None], 0.0) + a = det.step([_pinch_hand(*_PINCH), None], 0.1) # fire + b = det.step([_pinch_hand(*_PINCH), None], 0.2) # still held -> no refire assert len(a) == 1 assert b == [] def test_pinch_rearms_after_release(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand(*_OPEN)], 0.0) - a = det.step([_pinch_hand(*_PINCH)], 0.1) # fire - det.step([_pinch_hand(*_OPEN)], 0.2) # release (ratio > off) - c = det.step([_pinch_hand(*_PINCH)], 0.3) # pinch again -> fire + det.step([_pinch_hand(*_OPEN), None], 0.0) + a = det.step([_pinch_hand(*_PINCH), None], 0.1) # fire + det.step([_pinch_hand(*_OPEN), None], 0.2) # release (ratio > off) + c = det.step([_pinch_hand(*_PINCH), None], 0.3) # pinch again -> fire assert len(a) == 1 assert len(c) == 1 def test_pinch_refractory_blocks(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=200) - det.step([_pinch_hand(*_OPEN)], 0.0) - a = det.step([_pinch_hand(*_PINCH)], 0.05) # fire t=0.05 - det.step([_pinch_hand(*_OPEN)], 0.08) # release - b = det.step([_pinch_hand(*_PINCH)], 0.10) # within 200 ms -> blocked + det.step([_pinch_hand(*_OPEN), None], 0.0) + a = det.step([_pinch_hand(*_PINCH), None], 0.05) # fire t=0.05 + det.step([_pinch_hand(*_OPEN), None], 0.08) # release + b = det.step([_pinch_hand(*_PINCH), None], 0.10) # within 200 ms -> blocked assert len(a) == 1 assert b == [] @@ -128,9 +128,9 @@ def test_pinch_refractory_blocks(): def test_pinch_engage_release_emits_both_edges(): """Engage emits state=1; subsequent open emits state=0 (same finger).""" det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand(*_OPEN)], 0.0) - engage = det.step([_pinch_hand(*_PINCH)], 0.1) # engage edge - release = det.step([_pinch_hand(*_OPEN)], 0.2) # release edge + det.step([_pinch_hand(*_OPEN), None], 0.0) + engage = det.step([_pinch_hand(*_PINCH), None], 0.1) # engage edge + release = det.step([_pinch_hand(*_OPEN), None], 0.2) # release edge assert len(engage) == 1 and engage[0].state == 1 assert len(release) == 1 and release[0].state == 0 assert release[0].hand == 0 and release[0].finger == 1 @@ -139,9 +139,9 @@ def test_pinch_engage_release_emits_both_edges(): def test_pinch_hand_disappear_emits_release(): """When an engaged hand disappears, a release edge is synthesised.""" det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand(*_OPEN)], 0.0) - engage = det.step([_pinch_hand(*_PINCH)], 0.1) # engage - release = det.step([], 0.2) # hand gone + det.step([_pinch_hand(*_OPEN), None], 0.0) + engage = det.step([_pinch_hand(*_PINCH), None], 0.1) # engage + release = det.step([None, None], 0.2) # hand gone assert len(engage) == 1 and engage[0].state == 1 assert len(release) == 1 and release[0].state == 0 @@ -163,9 +163,9 @@ def _pinch_hand_multi(thumb_xy, tips): def test_pinch_closest_finger_wins(): # index ratio ~0.067, middle ratio ~0.33 -> margin 0.27 >= 0.20 -> index only. det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand_multi((0.5, 0.5), {})], 0.0) # open + det.step([_pinch_hand_multi((0.5, 0.5), {}), None], 0.0) # open ev = det.step([_pinch_hand_multi((0.5, 0.5), - {0: (0.52, 0.5), 1: (0.60, 0.5)})], 0.1) + {0: (0.52, 0.5), 1: (0.60, 0.5)}), None], 0.1) assert len(ev) == 1 assert ev[0].finger == 1 and ev[0].state == 1 # index only @@ -174,9 +174,9 @@ def test_pinch_ambiguous_adjacent_fires_nothing(): # index ratio ~0.067, middle ratio ~0.10 -> margin 0.033 < 0.20 -> no winner. # (both are below ratio_on, so the OLD per-finger logic would fire both.) det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0) - det.step([_pinch_hand_multi((0.5, 0.5), {})], 0.0) # open + det.step([_pinch_hand_multi((0.5, 0.5), {}), None], 0.0) # open ev = det.step([_pinch_hand_multi((0.5, 0.5), - {0: (0.52, 0.5), 1: (0.53, 0.5)})], 0.1) + {0: (0.52, 0.5), 1: (0.53, 0.5)}), None], 0.1) assert ev == [] @@ -201,7 +201,7 @@ def test_relaxed_hand_rejected_by_extension_gate(): ext_min=2) out = [] for i in range(10): - out += det.step([_relaxed_hand()], i * 0.033) + out += det.step([_relaxed_hand(), None], i * 0.033) assert out == [] @@ -210,8 +210,8 @@ def test_open_hand_pinch_passes_extension_gate(): # (ext 1.67 / 1.49 / 1.37, all >= 1.35) -> 3 extended >= ext_min 2. det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0, ext_min=2) - det.step([_pinch_hand(*_OPEN)], 0.0) - ev = det.step([_pinch_hand(*_PINCH)], 0.1) + det.step([_pinch_hand(*_OPEN), None], 0.0) + ev = det.step([_pinch_hand(*_PINCH), None], 0.1) assert len(ev) == 1 and ev[0].state == 1 and ev[0].finger == 1 @@ -219,17 +219,17 @@ def test_extension_gate_defeat_reproduces_old_behavior(): # ext_min=0 disables the gate: the relaxed hand fires like today. det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0, ext_min=0) - ev = det.step([_relaxed_hand()], 0.1) + ev = det.step([_relaxed_hand(), None], 0.1) assert len(ev) == 1 and ev[0].state == 1 def test_debounce_delays_engage_to_nth_frame(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0, debounce_frames=3) - det.step([_pinch_hand(*_OPEN)], 0.00) - a = det.step([_pinch_hand(*_PINCH)], 0.10) # qualifying 1 - b = det.step([_pinch_hand(*_PINCH)], 0.13) # qualifying 2 - c = det.step([_pinch_hand(*_PINCH)], 0.16) # qualifying 3 -> engage + det.step([_pinch_hand(*_OPEN), None], 0.00) + a = det.step([_pinch_hand(*_PINCH), None], 0.10) # qualifying 1 + b = det.step([_pinch_hand(*_PINCH), None], 0.13) # qualifying 2 + c = det.step([_pinch_hand(*_PINCH), None], 0.16) # qualifying 3 -> engage assert a == [] and b == [] assert len(c) == 1 and c[0].state == 1 and c[0].finger == 1 @@ -237,20 +237,20 @@ def test_debounce_delays_engage_to_nth_frame(): def test_debounce_jitter_resets_counter(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0, debounce_frames=3) - det.step([_pinch_hand(*_PINCH)], 0.00) # qualifying 1 - det.step([_pinch_hand(*_PINCH)], 0.03) # qualifying 2 - det.step([_pinch_hand(*_OPEN)], 0.06) # jitter -> counter resets - a = det.step([_pinch_hand(*_PINCH)], 0.09) # qualifying 1 again - b = det.step([_pinch_hand(*_PINCH)], 0.12) # qualifying 2 + det.step([_pinch_hand(*_PINCH), None], 0.00) # qualifying 1 + det.step([_pinch_hand(*_PINCH), None], 0.03) # qualifying 2 + det.step([_pinch_hand(*_OPEN), None], 0.06) # jitter -> counter resets + a = det.step([_pinch_hand(*_PINCH), None], 0.09) # qualifying 1 again + b = det.step([_pinch_hand(*_PINCH), None], 0.12) # qualifying 2 assert a == [] and b == [] def test_release_immediate_after_debounced_engage(): det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0, debounce_frames=3) - det.step([_pinch_hand(*_PINCH)], 0.00) - det.step([_pinch_hand(*_PINCH)], 0.03) - eng = det.step([_pinch_hand(*_PINCH)], 0.06) # 3rd frame -> engage - rel = det.step([_pinch_hand(*_OPEN)], 0.09) # very next frame + det.step([_pinch_hand(*_PINCH), None], 0.00) + det.step([_pinch_hand(*_PINCH), None], 0.03) + eng = det.step([_pinch_hand(*_PINCH), None], 0.06) # 3rd frame -> engage + rel = det.step([_pinch_hand(*_OPEN), None], 0.09) # very next frame assert len(eng) == 1 and eng[0].state == 1 assert len(rel) == 1 and rel[0].state == 0