feat: thumb-finger pinch detection
PinchDetector (hysteresis + refractory, normalized by hand size) emits /pose/pinch. Publisher picks iPhone Vision hands when fresh (FINGER_SOURCE). 4 pinch tests.
This commit is contained in:
clement
@@ -147,6 +147,7 @@ class ActionHeadPublisher(threading.Thread):
|
||||
self._last_pids = current_pids
|
||||
self._emit_hands(t_now)
|
||||
self._emit_fingers(t_now)
|
||||
self._emit_pinch(t_now)
|
||||
|
||||
def _emit_hands(self, t_now: float) -> None:
|
||||
"""Lock state, extract hand features, emit /pose/hands once per tick."""
|
||||
@@ -162,9 +163,11 @@ class ActionHeadPublisher(threading.Thread):
|
||||
self.bridge.send_hands(feats, t_now)
|
||||
|
||||
def _init_finger_piano(self) -> None:
|
||||
"""Read FINGER_PIANO env config and build the strike detector."""
|
||||
"""Read FINGER_* env config and build the strike + pinch detectors."""
|
||||
import os
|
||||
from data_only_viz.finger_strike import FingerStrikeDetector
|
||||
from data_only_viz.finger_strike import (
|
||||
FingerStrikeDetector, PinchDetector,
|
||||
)
|
||||
self._finger_enabled = os.environ.get("FINGER_PIANO", "0") not in (
|
||||
"0", "", "false", "False",
|
||||
)
|
||||
@@ -173,6 +176,11 @@ class ActionHeadPublisher(threading.Thread):
|
||||
self._finger_det = FingerStrikeDetector(
|
||||
vel_thresh=vel, refractory_ms=refr,
|
||||
)
|
||||
self._pinch_det = PinchDetector(
|
||||
ratio_on=float(os.environ.get("PINCH_RATIO_ON", "0.45")),
|
||||
ratio_off=float(os.environ.get("PINCH_RATIO_OFF", "0.65")),
|
||||
refractory_ms=float(os.environ.get("PINCH_REFRACTORY_MS", "250")),
|
||||
)
|
||||
self._finger_dbg = os.environ.get("FINGER_DEBUG", "0") not in (
|
||||
"0", "", "false", "False",
|
||||
)
|
||||
@@ -181,14 +189,8 @@ class ActionHeadPublisher(threading.Thread):
|
||||
self._finger_source = os.environ.get("FINGER_SOURCE", "auto").lower()
|
||||
self._fdbg_n = 0
|
||||
|
||||
def _emit_fingers(self, t_now: float) -> None:
|
||||
"""Detect finger strikes and emit /pose/finger.
|
||||
|
||||
Source: iPhone Vision hands (stable, rotation-invariant) when fresh,
|
||||
else MediaPipe hands from the (possibly rotated) video frame.
|
||||
"""
|
||||
if not getattr(self, "_finger_enabled", False):
|
||||
return
|
||||
def _pick_hands(self) -> tuple[list, str, bool]:
|
||||
"""Return (hands, 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 [])
|
||||
@@ -196,11 +198,17 @@ class ActionHeadPublisher(threading.Thread):
|
||||
ip_fresh = bool(ip_hands) and (time.perf_counter() - ip_t) < 0.5
|
||||
src = getattr(self, "_finger_source", "auto")
|
||||
if src == "iphone":
|
||||
hands, used = ip_hands, "iphone"
|
||||
elif src == "mediapipe":
|
||||
hands, used = mp_hands, "mediapipe"
|
||||
else:
|
||||
hands, used = (ip_hands, "iphone") if ip_fresh else (mp_hands, "mediapipe")
|
||||
return ip_hands, "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)
|
||||
|
||||
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)
|
||||
if getattr(self, "_finger_dbg", False):
|
||||
self._fdbg_n += 1
|
||||
@@ -210,6 +218,17 @@ class ActionHeadPublisher(threading.Thread):
|
||||
for ev in events:
|
||||
self.bridge.send_finger(ev)
|
||||
|
||||
def _emit_pinch(self, t_now: float) -> None:
|
||||
"""Detect thumb-to-finger pinches (clip toggles) and emit /pose/pinch."""
|
||||
if not getattr(self, "_finger_enabled", False):
|
||||
return
|
||||
hands, used, _ip = self._pick_hands()
|
||||
for ev in self._pinch_det.step(hands, t_now):
|
||||
if getattr(self, "_finger_dbg", False):
|
||||
LOG.info("pinch dbg: src=%s hand=%d finger=%d",
|
||||
used, ev.hand, ev.finger)
|
||||
self.bridge.send_pinch(ev)
|
||||
|
||||
def _read_sources(self) -> tuple[
|
||||
list[tuple[int, np.ndarray, np.ndarray, float, np.ndarray]] | None,
|
||||
float, str, bool,
|
||||
|
||||
@@ -7,6 +7,7 @@ fingers. Output feeds the OSC /pose/finger route consumed by SuperCollider.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from dataclasses import dataclass
|
||||
|
||||
from data_only_viz.hand_features import _clamp, _coord, _finite
|
||||
@@ -16,6 +17,29 @@ from data_only_viz.hand_features import _clamp, _coord, _finite
|
||||
FINGERTIPS: tuple[int, ...] = (4, 8, 12, 16, 20)
|
||||
FINGER_BASES: tuple[int, ...] = (2, 5, 9, 13, 17)
|
||||
|
||||
# Pinch: thumb tip vs the 4 opposable finger tips (index..pinky).
|
||||
THUMB_TIP: int = 4
|
||||
WRIST: int = 0
|
||||
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:
|
||||
@@ -99,3 +123,66 @@ class FingerStrikeDetector:
|
||||
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
|
||||
finger: int # 1..4 = index, middle, ring, pinky (thumb is trigger)
|
||||
|
||||
|
||||
class _PinchState:
|
||||
__slots__ = ("engaged", "last_t")
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.engaged: bool = False
|
||||
self.last_t: float = -1e9
|
||||
|
||||
|
||||
class PinchDetector:
|
||||
"""Edge-triggered thumb-to-finger pinch with hysteresis.
|
||||
|
||||
Fires one PinchEvent when thumb tip contacts a finger tip (distance,
|
||||
normalized by hand size, drops below ratio_on). Re-arms only after the
|
||||
distance rises back above ratio_off, so one pinch = one event.
|
||||
"""
|
||||
|
||||
def __init__(self, ratio_on: float = 0.45, ratio_off: float = 0.65,
|
||||
refractory_ms: float = 250.0, history_slots: int = 2) -> None:
|
||||
self.ratio_on = ratio_on
|
||||
self.ratio_off = ratio_off
|
||||
self.refractory_s = refractory_ms / 1000.0
|
||||
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)))
|
||||
events: list[PinchEvent] = []
|
||||
for slot, lm in enumerate(ordered):
|
||||
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)
|
||||
wy = _finite(_coord(lm[WRIST], "y", 1), 0.5)
|
||||
mx = _finite(_coord(lm[MIDDLE_MCP], "x", 0), 0.5)
|
||||
my = _finite(_coord(lm[MIDDLE_MCP], "y", 1), 0.5)
|
||||
size = math.hypot(mx - wx, my - wy)
|
||||
size = size if size > 1e-4 else 1e-4
|
||||
for i, tip_idx in enumerate(PINCH_TIPS):
|
||||
fx = _finite(_coord(lm[tip_idx], "x", 0), 0.5)
|
||||
fy = _finite(_coord(lm[tip_idx], "y", 1), 0.5)
|
||||
ratio = math.hypot(fx - tx, fy - ty) / size
|
||||
st = self._state[slot][i]
|
||||
if st.engaged:
|
||||
if ratio > self.ratio_off:
|
||||
st.engaged = False
|
||||
elif ratio < self.ratio_on and (
|
||||
t_now - st.last_t) >= self.refractory_s:
|
||||
st.engaged = True
|
||||
st.last_t = t_now
|
||||
events.append(PinchEvent(hand=slot, finger=i + 1))
|
||||
for slot in range(len(self._state)):
|
||||
if slot not in present:
|
||||
for i in range(4):
|
||||
self._state[slot][i].engaged = False
|
||||
return events
|
||||
|
||||
@@ -387,6 +387,15 @@ class PoseSoundBridge:
|
||||
pass
|
||||
self._vj("/pose/finger", args)
|
||||
|
||||
def send_pinch(self, ev) -> None:
|
||||
"""Emit one thumb-to-finger pinch event (clip toggle trigger)."""
|
||||
args = [0, int(ev.hand), int(ev.finger)]
|
||||
try:
|
||||
self._client.send_message("/pose/pinch", args)
|
||||
except OSError:
|
||||
pass
|
||||
self._vj("/pose/pinch", args)
|
||||
|
||||
def send_enter(self, pid: int) -> None:
|
||||
"""Send lifecycle event when person enters frame."""
|
||||
self._client.send_message("/pose/enter", [int(pid)])
|
||||
|
||||
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
|
||||
from data_only_viz.finger_strike import (
|
||||
FingerStrikeDetector,
|
||||
PinchDetector,
|
||||
StrikeEvent,
|
||||
FINGERTIPS,
|
||||
FINGER_BASES,
|
||||
@@ -65,3 +66,60 @@ def test_strike_speed_scales_with_velocity():
|
||||
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
|
||||
assert h[0].strike_speed > s[0].strike_speed
|
||||
|
||||
|
||||
def _pinch_hand(thumb_xy, index_xy):
|
||||
"""21-kp hand with fixed wrist & middle-MCP (hand size = 0.3). Middle/
|
||||
ring/pinky tips are parked far from the thumb so only the thumb-index
|
||||
pair can pinch."""
|
||||
lm = [[0.5, 0.5, 0.0] for _ in range(21)]
|
||||
lm[0] = [0.5, 0.8, 0.0] # WRIST
|
||||
lm[9] = [0.5, 0.5, 0.0] # MIDDLE_MCP -> size 0.3
|
||||
lm[4] = [thumb_xy[0], thumb_xy[1], 0.0] # THUMB_TIP
|
||||
lm[8] = [index_xy[0], index_xy[1], 0.0] # INDEX_TIP
|
||||
lm[12] = [0.9, 0.5, 0.0] # MIDDLE_TIP (far)
|
||||
lm[16] = [0.9, 0.6, 0.0] # RING_TIP (far)
|
||||
lm[20] = [0.9, 0.7, 0.0] # LITTLE_TIP (far)
|
||||
return lm
|
||||
|
||||
|
||||
_OPEN = ((0.2, 0.5), (0.8, 0.5)) # thumb-index dist 0.6 -> ratio 2.0
|
||||
_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 len(ev) == 1
|
||||
assert ev[0].finger == 1 # index = finger 1
|
||||
assert ev[0].hand == 0
|
||||
|
||||
|
||||
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
|
||||
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
|
||||
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
|
||||
assert len(a) == 1
|
||||
assert b == []
|
||||
|
||||
Reference in New Issue
Block a user