feat(pose): gesture slot stabilizer

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",

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)

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