diff --git a/data_only_viz/action_head_pub.py b/data_only_viz/action_head_pub.py index d6398e8..92f3820 100644 --- a/data_only_viz/action_head_pub.py +++ b/data_only_viz/action_head_pub.py @@ -149,6 +149,12 @@ class ActionHeadPublisher(threading.Thread): self._last_pids = current_pids if getattr(self, "_stab", None) is not None: self._stab_slotted = self._step_stab() + # Reset finger-strike state for slots that just transitioned from + # "held replay" back to a real hand, preventing phantom strikes from + # the compressed motion delta accumulated during the hold window. + for i, resumed in enumerate(self._stab.resumed_flags()): + if resumed: + self._finger_det.reset_slot(i) else: self._stab_slotted = None self._emit_hands(t_now) @@ -160,6 +166,7 @@ class ActionHeadPublisher(threading.Thread): with self.state.lock(): hands = list(getattr(self.state, "persons_hands", None) or []) chirality = list(getattr(self.state, "persons_hands_chirality", None) or []) + mirror = bool(getattr(self.state, "mirror_2d", False)) 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) @@ -169,7 +176,8 @@ class ActionHeadPublisher(threading.Thread): fist_enabled = stab.active_flags() if stab is not None else (True, True) feats = self._hand_ext.step(hands, chirality or None, swap=getattr(self, "_hand_swap_lr", False), - fist_enabled=fist_enabled) + fist_enabled=fist_enabled, + mirror=mirror) with self.state.lock(): self.state.hand_feats = feats self.bridge.send_hands(feats, t_now) @@ -217,22 +225,29 @@ class ActionHeadPublisher(threading.Thread): near_off=cfg.hand_near_off, ) - def _pick_hands(self) -> tuple[list, list, str, bool]: - """Return (hands, chirality, source_label, ip_fresh) per FINGER_SOURCE policy.""" + def _pick_hands(self) -> tuple[list, list, str, bool, bool]: + """Return (hands, chirality, source_label, ip_fresh, mirror). + + mirror is read from state.mirror_2d under the lock so it is consistent + with the chirality dropout decision made in the same tick. Callers + pass it through to route_hands so the cx fallback slot assignment stays + coherent with the renderer's panel layout when chirality is absent. + """ 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 []) + mirror = bool(getattr(self.state, "mirror_2d", False)) 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, ip_chir, "iphone", ip_fresh + return ip_hands, ip_chir, "iphone", ip_fresh, mirror if src == "mediapipe": - return mp_hands, [], "mediapipe", ip_fresh + return mp_hands, [], "mediapipe", ip_fresh, mirror if ip_fresh: - return ip_hands, ip_chir, "iphone", ip_fresh - return mp_hands, [], "mediapipe", ip_fresh + return ip_hands, ip_chir, "iphone", ip_fresh, mirror + return mp_hands, [], "mediapipe", ip_fresh, mirror def _step_stab(self) -> list: """Pick hands, route with near_min=0, step gesture stabilizer. @@ -242,11 +257,12 @@ class ActionHeadPublisher(threading.Thread): _stab_slotted is not yet set (e.g. in unit tests that call emitters without going through _tick). """ - hands, chir, _, _ = self._pick_hands() + hands, chir, _, _, mirror = self._pick_hands() raw = route_hands( hands, chir or None, swap=getattr(self, "_hand_swap_lr", False), near_min=0.0, + mirror=mirror, ) return self._stab.step(raw) @@ -254,7 +270,7 @@ class ActionHeadPublisher(threading.Thread): """Detect finger strikes (air-piano) and emit /pose/finger.""" if not getattr(self, "_finger_enabled", False): return - hands, chirality, used, ip_fresh = self._pick_hands() + hands, chirality, used, ip_fresh, _ = self._pick_hands() slotted = getattr(self, "_stab_slotted", None) if slotted is None: slotted = self._step_stab() @@ -272,7 +288,7 @@ class ActionHeadPublisher(threading.Thread): /pose/pinch. Gated by PINCH_ENABLE, independently of the air-piano.""" if not getattr(self, "_pinch_enabled", False): return - hands, chirality, used, _ip = self._pick_hands() + hands, chirality, used, _ip, _ = self._pick_hands() slotted = getattr(self, "_stab_slotted", None) if slotted is None: slotted = self._step_stab() diff --git a/data_only_viz/finger_strike.py b/data_only_viz/finger_strike.py index c2680fa..c8331b5 100644 --- a/data_only_viz/finger_strike.py +++ b/data_only_viz/finger_strike.py @@ -56,6 +56,21 @@ class FingerStrikeDetector: self._state = [[_FingerState() for _ in range(5)] for _ in range(history_slots)] + def reset_slot(self, slot: int) -> None: + """Clear per-finger prev_rel and re-arm for the given slot. + + Call this BEFORE step() when GestureSlotStabilizer.resumed_flags() + reports a held -> real transition for that slot. During a Vision hole + the stabilizer replays the last hand (frozen coords, vel=0), but when + the real hand returns within the hold window the first delta compresses + several frames of motion into one large velocity spike, causing a + phantom strike. Clearing prev_rel makes the first real frame a + no-op prime that restarts tracking from the new position. + """ + for f in range(5): + self._state[slot][f].prev_rel = None + self._state[slot][f].armed = True + def step(self, slotted: list, t_now: float) -> list[StrikeEvent]: """Process a pre-routed 2-slot hand list [left|None, right|None]. diff --git a/data_only_viz/hand_features.py b/data_only_viz/hand_features.py index 491aaa1..0144759 100644 --- a/data_only_viz/hand_features.py +++ b/data_only_viz/hand_features.py @@ -80,11 +80,13 @@ class HandFeatureExtractor: return _clamp(spd, 0.0, 1.0) def step(self, hands: list, chirality=None, swap: bool = False, - fist_enabled: tuple = (True, True)) -> dict: + fist_enabled: tuple = (True, True), mirror: bool = False) -> dict: from data_only_viz.hand_slots import route_hands # swap must mirror the gesture path (HAND_SWAP_LR) or per-voice # L/R mods contradict pinch/strike in the same performance. - left_lm, right_lm = route_hands(hands, chirality, swap=swap) + # mirror must match state.mirror_2d so the cx fallback slot assignment + # stays coherent with the renderer panel layout on chirality dropout. + left_lm, right_lm = route_hands(hands, chirality, swap=swap, mirror=mirror) if left_lm is None and right_lm is None: self._buf[0].clear() diff --git a/data_only_viz/hand_slots.py b/data_only_viz/hand_slots.py index c57239c..30b5eb5 100644 --- a/data_only_viz/hand_slots.py +++ b/data_only_viz/hand_slots.py @@ -158,6 +158,10 @@ class GestureSlotStabilizer: self._miss: list[int] = [0, 0] # consecutive miss count self._near_active: list[bool] = [False, False] self._out: list = [None, None] # last step() output (for active_flags) + # Hold-tracking: True when previous step() output was a held replay. + self._held_flag: list[bool] = [False, False] + # Resumed: True when this step() transitioned held -> real for that slot. + self._resumed: list[bool] = [False, False] def step(self, slotted: list) -> list: """Stabilize a 2-element [left|None, right|None] slot list. @@ -171,10 +175,15 @@ class GestureSlotStabilizer: out: list = [None, None] for i in range(2): hand = slotted[i] if i < len(slotted) else None + prev_held = self._held_flag[i] if hand is not None: - # Hand present this frame - size = _hand_size(hand) + # Real hand present this frame. + # Transition held -> real: mark resumed so callers can reset + # per-finger state before consuming this step's output. + self._resumed[i] = prev_held + self._held_flag[i] = False # Update near hysteresis + size = _hand_size(hand) if self._near_active[i]: if size < self._near_off: self._near_active[i] = False @@ -188,15 +197,20 @@ class GestureSlotStabilizer: out[i] = hand if self._near_active[i] else None else: # Hand absent this frame + self._resumed[i] = False self._miss[i] += 1 if self._miss[i] <= self._hold: # Hold: output last hand if it was near-active if self._near_active[i] and self._last[i] is not None: + self._held_flag[i] = True out[i] = self._last[i] - # else: not near-active, nothing to hold + else: + # Not near-active: no hold replay; slot stays None + self._held_flag[i] = False else: # Past hold limit — clear near_active so next appearance # must re-qualify from near_on. + self._held_flag[i] = False self._near_active[i] = False out[i] = None self._out = out @@ -210,3 +224,20 @@ class GestureSlotStabilizer: action_head_pub._emit_hands to gate the fist feature. """ return (self._out[0] is not None, self._out[1] is not None) + + def resumed_flags(self) -> tuple[bool, bool]: + """Return (slot0_resumed, slot1_resumed). + + True for a slot when the current step() transitioned from a "held + replay" (Vision hole, last hand replayed) back to a real hand. Callers + that track per-frame deltas (e.g. FingerStrikeDetector) should reset + their state for that slot BEFORE consuming the step output to avoid + phantom events caused by the compressed motion accumulated during the + hold window. + + Semantics: True only for the ONE step where the transition occurs; + False on all subsequent frames where the hand stays real, and False + after an expired hold (miss > hold_frames, slot went None) since the + absent-slot logic in gesture detectors already resets per-finger state. + """ + return (self._resumed[0], self._resumed[1]) 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 0e7c230..d861c76 100644 --- a/data_only_viz/tests/test_action_head_finger_emit.py +++ b/data_only_viz/tests/test_action_head_finger_emit.py @@ -2,6 +2,7 @@ from __future__ import annotations import threading +import pytest from unittest.mock import MagicMock from data_only_viz.finger_strike import FINGERTIPS, FINGER_BASES @@ -10,6 +11,7 @@ from data_only_viz.finger_strike import FINGERTIPS, FINGER_BASES class _FakeState: def __init__(self) -> None: self.persons_hands = [] + self.mirror_2d = False self._lock = threading.RLock() def lock(self): @@ -118,3 +120,43 @@ def test_pinch_gate_env_defaults_are_strict(monkeypatch): assert pub._pinch_det.ext_ratio == 1.35 assert pub._pinch_det.ext_min == 2 assert pub._pinch_det.debounce_frames == 3 + + +# --------------------------------------------------------------------------- +# Item 1a: mirror_2d coherence — _emit_hands passes mirror to extractor +# --------------------------------------------------------------------------- + +def _make_hand_row(cx: float) -> list: + """21 landmarks as [x, y, z] rows with enough geometry for features.""" + lm = [[cx, 0.5, 0.0] for _ in range(21)] + lm[0] = [cx, 0.6, 0.0] # wrist + lm[9] = [cx, 0.45, 0.0] # middle-MCP -> hand_size 0.15 > near threshold + return lm + + +def test_emit_hands_mirror_2d_flips_cx_slot_assignment(monkeypatch): + """With state.mirror_2d=True, the cx fallback in _emit_hands assigns + the RIGHT-screen hand to the L slot — coherent with the mirrored renderer.""" + from data_only_viz.hand_features import HandFeatureExtractor + pub = _pub(monkeypatch, enabled=False) + pub._hand_ext = HandFeatureExtractor() # not set by _pub/_init_finger_piano + + h_low_cx = _make_hand_row(0.2) # screen-left + h_high_cx = _make_hand_row(0.8) # screen-right + pub.state.persons_hands = [h_low_cx, h_high_cx] + # No chirality -> cx fallback path + # (persons_hands_chirality absent -> getattr returns None -> [] -> None) + + # Without mirror: screen-left (cx=0.2) -> slot L + pub.state.mirror_2d = False + pub._emit_hands(0.0) + feats_plain = pub.state.hand_feats + assert feats_plain["L"] is not None + assert feats_plain["L"]["cx"] == pytest.approx(0.2, abs=0.05) + + # With mirror: cx fallback flips -> screen-right (cx=0.8) -> slot L + pub.state.mirror_2d = True + pub._emit_hands(0.0) + feats_mirror = pub.state.hand_feats + assert feats_mirror["L"] is not None + assert feats_mirror["L"]["cx"] == pytest.approx(0.8, abs=0.05) diff --git a/data_only_viz/tests/test_finger_strike.py b/data_only_viz/tests/test_finger_strike.py index 4387c81..ae1938a 100644 --- a/data_only_viz/tests/test_finger_strike.py +++ b/data_only_viz/tests/test_finger_strike.py @@ -254,3 +254,84 @@ def test_release_immediate_after_debounced_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 + + +# --------------------------------------------------------------------------- +# FingerStrikeDetector.reset_slot — phantom-strike prevention +# --------------------------------------------------------------------------- + +def _strike_hand(tip_y: float, base_y: float = 0.4, cx: float = 0.3) -> list: + """Hand where index finger tip_y controls the relative position.""" + return _hand({1: tip_y}, base_y=base_y, cx=cx) + + +def test_reset_slot_prevents_phantom_strike_on_reappear(): + """After reset_slot, the first real-hand frame primes without firing. + + Scenario: prime detector, hold the last hand (frozen coords), + then reset_slot before the real hand returns with a large delta. + Expected: no strike on reappear (first frame after reset is a prime). + """ + det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=0) + + # Prime: index tip at 0.40 + det.step([_strike_hand(0.40), None], t_now=0.00) + + # "Held" frames: stabilizer would replay the same hand — same coords, + # vel=0, no strike, but prev_rel stays at 0.40. + det.step([_strike_hand(0.40), None], t_now=0.03) + det.step([_strike_hand(0.40), None], t_now=0.06) + + # Simulate held -> real transition: reset before consuming new position. + det.reset_slot(0) + + # Real hand reappears with a large downward delta (0.40 -> 0.70 = 0.30 >> thresh). + events = det.step([_strike_hand(0.70), None], t_now=0.09) + assert events == [], ( + "after reset_slot, first frame should prime without firing " + f"(got {events})" + ) + + +def test_without_reset_big_delta_fires_strike(): + """Control: WITHOUT reset_slot, the same scenario fires a phantom strike.""" + det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=0) + + det.step([_strike_hand(0.40), None], t_now=0.00) # prime + det.step([_strike_hand(0.40), None], t_now=0.03) # held + det.step([_strike_hand(0.40), None], t_now=0.06) # held + + # No reset — prev_rel is still 0.40 → delta 0.30 > thresh → fires + events = det.step([_strike_hand(0.70), None], t_now=0.09) + assert len(events) > 0, "without reset_slot, phantom strike must fire (control)" + + +def test_reset_slot_clears_all_five_fingers(): + """reset_slot clears every finger state for the specified slot.""" + det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=0) + # Prime all 5 fingers by stepping a neutral hand + neutral = _hand({f: 0.40 for f in range(5)}, base_y=0.40) + det.step([neutral, None], t_now=0.00) + # All prev_rel should be non-None now + for f in range(5): + assert det._state[0][f].prev_rel is not None + + det.reset_slot(0) + + for f in range(5): + assert det._state[0][f].prev_rel is None, f"finger {f} prev_rel not cleared" + assert det._state[0][f].armed is True, f"finger {f} not re-armed" + + +def test_reset_slot_does_not_affect_other_slot(): + """Resetting slot 0 leaves slot 1 state intact.""" + det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=0) + h = _hand({1: 0.40}) + det.step([h, h], t_now=0.00) # prime both slots + # slot 1 finger 1 should have a non-None prev_rel + assert det._state[1][1].prev_rel is not None + + det.reset_slot(0) + + # slot 1 unchanged + assert det._state[1][1].prev_rel is not None, "slot 1 must be unaffected" diff --git a/data_only_viz/tests/test_hand_features.py b/data_only_viz/tests/test_hand_features.py index bf8dfeb..30e69cd 100644 --- a/data_only_viz/tests/test_hand_features.py +++ b/data_only_viz/tests/test_hand_features.py @@ -178,6 +178,28 @@ def test_fist_enabled_true_computes_fist(): assert hand["fist"] > 0.9 +def test_mirror_flips_cx_fallback_slot_assignment(): + """mirror=True inverts the cx fallback so screen-right hand goes to slot L.""" + ext = HandFeatureExtractor() + h_low = _hand(0.2, 0.5, span=0.10) # cx=0.2 -> slot L normally (leftmost) + h_high = _hand(0.8, 0.5, span=0.10) # cx=0.8 -> slot R normally + # No chirality -> cx fallback + plain = ext.step([h_low, h_high], chirality=None, mirror=False) + mirrored = ext.step([h_low, h_high], chirality=None, mirror=True) + assert plain["L"]["cx"] == pytest.approx(0.2, abs=0.02), "non-mirrored: L gets leftmost" + assert mirrored["L"]["cx"] == pytest.approx(0.8, abs=0.02), "mirrored: L gets rightmost" + + +def test_mirror_default_false_unchanged(): + """mirror defaults to False; existing tests that pass no mirror kwarg are unaffected.""" + ext = HandFeatureExtractor() + h_a = _hand(0.2, 0.5, span=0.10) + h_b = _hand(0.8, 0.5, span=0.10) + no_kwarg = ext.step([h_a, h_b]) + explicit = ext.step([h_a, h_b], mirror=False) + assert no_kwarg["L"]["cx"] == pytest.approx(explicit["L"]["cx"], abs=1e-6) + + def test_fist_enabled_per_slot_independence(): """fist_enabled masks only the specified slot.""" ext = HandFeatureExtractor() diff --git a/data_only_viz/tests/test_hand_slots.py b/data_only_viz/tests/test_hand_slots.py index 29fdd6e..a6367a1 100644 --- a/data_only_viz/tests/test_hand_slots.py +++ b/data_only_viz/tests/test_hand_slots.py @@ -301,3 +301,58 @@ def test_stab_active_flags_reflect_current_output(): stab.step([None, None]) # miss=1 -> hold flags2 = stab.active_flags() assert flags2 == (True, False) # still holding + + +# --------------------------------------------------------------------------- +# GestureSlotStabilizer.resumed_flags +# --------------------------------------------------------------------------- + +def test_resumed_flags_true_exactly_one_step_on_reappear(): + """resumed_flags True on the one step that transitions held -> real.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + h2 = _near_hand(cx=0.4) + stab.step([h, None]) # activate slot 0 + stab.step([None, None]) # miss=1 -> held replay + stab.step([h2, None]) # real hand returns -> resumed + assert stab.resumed_flags()[0] is True, "should be True on first real frame after hold" + # next step, still real -> False + stab.step([h2, None]) + assert stab.resumed_flags()[0] is False, "should be False once hand stays real" + + +def test_resumed_flags_false_steady(): + """Continuously present hand -> resumed_flags never True.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + stab.step([h, None]) + stab.step([h, None]) + assert stab.resumed_flags()[0] is False + + +def test_resumed_flags_false_after_expired_hold(): + """Hand returning after hold window expired does not trigger resumed (slot went None).""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + h2 = _near_hand(cx=0.4) + stab.step([h, None]) # activate + stab.step([None, None]) # miss=1, held + stab.step([None, None]) # miss=2, held + stab.step([None, None]) # miss=3 > hold -> near_active cleared, out=None + stab.step([h2, None]) # reappear after expired hold + # _held_flag was False (hold expired), so resumed must be False + assert stab.resumed_flags()[0] is False, "expired hold: held_flag reset, no resume" + + +def test_resumed_flags_independent_per_slot(): + """Each slot tracks held/resumed independently.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h0 = _near_hand(cx=0.3) + h1 = _near_hand(cx=0.7) + h0b = _near_hand(cx=0.3) + stab.step([h0, h1]) # activate both + stab.step([None, h1]) # slot 0 held, slot 1 real + stab.step([h0b, h1]) # slot 0 resumed, slot 1 unchanged + r = stab.resumed_flags() + assert r[0] is True, "slot 0 just transitioned held -> real" + assert r[1] is False, "slot 1 was never held"