fix(pose): mirror-coherent gesture fallback

Read state.mirror_2d in _pick_hands and thread it through
_step_stab (route_hands mirror=) and _emit_hands
(HandFeatureExtractor.step mirror=) so the cx fallback slot
assignment matches the renderer panel layout on chirality dropout.

Add mirror kwarg to HandFeatureExtractor.step plumbed to route_hands.
Tests: extractor-level mirror flip + publisher _emit_hands mirror_2d.
This commit is contained in:
L'électron rare
2026-07-02 14:13:36 +02:00
parent caa41fd7f1
commit 05d0fc1df1
8 changed files with 279 additions and 15 deletions
+26 -10
View File
@@ -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()
+15
View File
@@ -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].
+4 -2
View File
@@ -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()
+34 -3
View File
@@ -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])
@@ -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)
+81
View File
@@ -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"
+22
View File
@@ -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()
+55
View File
@@ -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"