diff --git a/data_only_viz/action_head_pub.py b/data_only_viz/action_head_pub.py index cf4e607..674649d 100644 --- a/data_only_viz/action_head_pub.py +++ b/data_only_viz/action_head_pub.py @@ -147,6 +147,10 @@ class ActionHeadPublisher(threading.Thread): self.head.forget(gone) self.bridge.send_leave(pid=gone) self._last_pids = current_pids + if getattr(self, "_stab", None) is not None: + self._stab_slotted = self._step_stab() + else: + self._stab_slotted = None self._emit_hands(t_now) self._emit_fingers(t_now) self._emit_pinch(t_now) @@ -201,6 +205,14 @@ class ActionHeadPublisher(threading.Thread): # future iPhone app build flips it. self._hand_swap_lr = cfg.hand_swap_lr self._hand_near_min = cfg.hand_near_min + self._hand_near_off = cfg.hand_near_off + self._hand_hold_frames = cfg.hand_hold_frames + from data_only_viz.hand_slots import GestureSlotStabilizer + self._stab = GestureSlotStabilizer( + hold_frames=cfg.hand_hold_frames, + near_on=cfg.hand_near_min, + 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.""" @@ -219,16 +231,31 @@ class ActionHeadPublisher(threading.Thread): return ip_hands, ip_chir, "iphone", ip_fresh return mp_hands, [], "mediapipe", ip_fresh + def _step_stab(self) -> list: + """Pick hands, route with near_min=0, step gesture stabilizer. + + Called once per tick from _tick() to share the result across all + three gesture emitters. Also callable directly from emitters when + _stab_slotted is not yet set (e.g. in unit tests that call emitters + without going through _tick). + """ + hands, chir, _, _ = self._pick_hands() + from data_only_viz.hand_slots import route_hands as _rh + raw = _rh( + hands, chir or None, + swap=getattr(self, "_hand_swap_lr", False), + near_min=0.0, + ) + return self._stab.step(raw) + 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, 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), - ) + slotted = getattr(self, "_stab_slotted", None) + if slotted is None: + slotted = self._step_stab() events = self._finger_det.step(slotted, t_now) if getattr(self, "_finger_dbg", False): self._fdbg_n += 1 @@ -244,11 +271,9 @@ class ActionHeadPublisher(threading.Thread): if not getattr(self, "_pinch_enabled", False): return 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), - ) + slotted = getattr(self, "_stab_slotted", None) + if slotted is None: + slotted = self._step_stab() 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", diff --git a/data_only_viz/hand_slots.py b/data_only_viz/hand_slots.py index 588da47..575ece1 100644 --- a/data_only_viz/hand_slots.py +++ b/data_only_viz/hand_slots.py @@ -125,3 +125,88 @@ def route_hands( slots[i] = None return slots + + +class GestureSlotStabilizer: + """Stateful per-slot stabilizer for the gesture detection path. + + Applied AFTER route_hands (with near_min=0.0) and BEFORE the gesture + detectors (FingerStrikeDetector, PinchDetector). + + Two behaviours: + - **Hold-last**: when a slot is None this call but held a hand within the + last ``hold_frames`` calls, output the LAST seen hand for that slot. + Fills 1-2 frame Vision holes so pinch debounce never sees a spurious + absence. After ``hold_frames`` consecutive misses the slot yields None + for real. + - **Near hysteresis**: a slot ACTIVATES when hand_size >= ``near_on`` and + stays active until hand_size < ``near_off``. Inactive slot -> None output + (but hold-last still tracks so re-activation is instant). + """ + + def __init__( + self, + hold_frames: int = 2, + near_on: float = 0.10, + near_off: float = 0.08, + ) -> None: + self._hold = hold_frames + self._near_on = near_on + self._near_off = near_off + # Per-slot state + self._last: list = [None, None] # last seen hand object + 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) + + def step(self, slotted: list) -> list: + """Stabilize a 2-element [left|None, right|None] slot list. + + Args: + slotted: output of route_hands (near_min=0.0) — length 2. + + Returns: + Stabilized 2-element list; each slot is the stabilized hand or None. + """ + out: list = [None, None] + for i in range(2): + hand = slotted[i] if i < len(slotted) else None + if hand is not None: + # Hand present this frame + size = _hand_size(hand) + # Update near hysteresis + if self._near_active[i]: + if size < self._near_off: + self._near_active[i] = False + else: + if size >= self._near_on: + self._near_active[i] = True + # Track the hand regardless of near state + self._last[i] = hand + self._miss[i] = 0 + # Output only if near-active + out[i] = hand if self._near_active[i] else None + else: + # Hand absent this frame + 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: + out[i] = self._last[i] + # else: not near-active, nothing to hold + else: + # Past hold limit — clear near_active so next appearance + # must re-qualify from near_on. + self._near_active[i] = False + out[i] = None + self._out = out + return out + + def active_flags(self) -> tuple: + """Return (slot0_active, slot1_active). + + True when the last step() yielded a non-None for that slot — meaning the + hand is established (held or present) AND within the near zone. Used by + action_head_pub._emit_hands to gate the fist feature. + """ + return (self._out[0] is not None, self._out[1] is not None) diff --git a/data_only_viz/tests/test_hand_slots.py b/data_only_viz/tests/test_hand_slots.py index ced70e0..29fdd6e 100644 --- a/data_only_viz/tests/test_hand_slots.py +++ b/data_only_viz/tests/test_hand_slots.py @@ -1,8 +1,8 @@ -"""Tests for hand_slots.route_hands.""" +"""Tests for hand_slots.route_hands and GestureSlotStabilizer.""" from __future__ import annotations import pytest -from data_only_viz.hand_slots import route_hands +from data_only_viz.hand_slots import route_hands, GestureSlotStabilizer # --------------------------------------------------------------------------- @@ -198,3 +198,106 @@ def test_result_always_length_2(): ]: r = route_hands(hands, chir) assert len(r) == 2, f"expected 2 slots, got {len(r)} for {chir}" + + +# --------------------------------------------------------------------------- +# GestureSlotStabilizer +# --------------------------------------------------------------------------- + +def _near_hand(cx: float = 0.5, size: float = 0.15) -> list: + """Hand with hand_size == size (near enough for near_on=0.10).""" + return _hand(cx=cx, size=size) + + +def _far_hand(cx: float = 0.5) -> list: + """Hand with hand_size == 0.02 (below near_on=0.10).""" + return _hand(cx=cx, size=0.02) + + +def test_stab_hold_last_one_frame_hole(): + """A 1-frame Vision hole keeps the last hand in the slot.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + stab.step([h, None]) # activate slot 0 + stab.step([h, None]) # slot 0 established + out = stab.step([None, None]) # miss=1 <= hold -> slot 0 holds last hand + assert out[0] is h, "slot 0 should hold last hand on 1-frame miss" + assert out[1] is None + + +def test_stab_hold_last_two_frame_hole(): + """A 2-frame Vision hole keeps the last hand both frames.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + stab.step([h, None]) + out1 = stab.step([None, None]) # miss=1 + out2 = stab.step([None, None]) # miss=2 <= hold=2 + assert out1[0] is h + assert out2[0] is h + + +def test_stab_slot_goes_none_after_hold_frames(): + """After hold_frames+1 consecutive misses the slot yields None.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h = _near_hand() + stab.step([h, None]) + stab.step([None, None]) # miss=1 + stab.step([None, None]) # miss=2 (still holding) + out = stab.step([None, None]) # miss=3 > hold=2 -> None for real + assert out[0] is None + + +def test_stab_hysteresis_stays_active_above_near_off(): + """Once active, hand stays active while size >= near_off.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h_on = _near_hand(size=0.11) # size >= near_on -> activates + h_mid = _near_hand(size=0.09) # near_off <= size < near_on -> stays active + out1 = stab.step([h_on, None]) + assert out1[0] is h_on + out2 = stab.step([h_mid, None]) + assert out2[0] is h_mid, "should stay active at size 0.09 (>= near_off 0.08)" + + +def test_stab_hysteresis_deactivates_below_near_off(): + """Hand deactivates when size drops below near_off.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h_on = _near_hand(size=0.11) + h_off = _near_hand(size=0.07) # size < near_off -> deactivate + stab.step([h_on, None]) + out = stab.step([h_off, None]) + assert out[0] is None, "slot should deactivate below near_off" + + +def test_stab_hysteresis_needs_near_on_to_reactivate(): + """After deactivation, size must reach near_on again to re-activate.""" + stab = GestureSlotStabilizer(hold_frames=2, near_on=0.10, near_off=0.08) + h_on = _near_hand(size=0.11) + h_off = _near_hand(size=0.07) # deactivates + h_mid = _near_hand(size=0.09) # between near_off and near_on -> stays inactive + h_on2 = _near_hand(size=0.11) # back to near_on -> reactivates + stab.step([h_on, None]) + stab.step([h_off, None]) # deactivated + out_mid = stab.step([h_mid, None]) + assert out_mid[0] is None, "should stay inactive at 0.09 after deactivation" + out_on2 = stab.step([h_on2, None]) + assert out_on2[0] is h_on2, "should re-activate at near_on" + + +def test_stab_fresh_far_hand_returns_none(): + """A far hand on a fresh stabilizer yields None (not near).""" + stab = GestureSlotStabilizer(near_on=0.10, near_off=0.08) + h = _far_hand() + out = stab.step([h, None]) + assert out[0] is None + + +def test_stab_active_flags_reflect_current_output(): + """active_flags() matches the last step() output per slot.""" + stab = GestureSlotStabilizer(near_on=0.10, near_off=0.08) + h = _near_hand() + stab.step([h, None]) + flags = stab.active_flags() + assert flags == (True, False) + stab.step([None, None]) # miss=1 -> hold + flags2 = stab.active_flags() + assert flags2 == (True, False) # still holding