diff --git a/data_only_viz/action_head_pub.py b/data_only_viz/action_head_pub.py index 9ed4cad..3d0bf15 100644 --- a/data_only_viz/action_head_pub.py +++ b/data_only_viz/action_head_pub.py @@ -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, diff --git a/data_only_viz/finger_strike.py b/data_only_viz/finger_strike.py index 15116d8..f59f9bd 100644 --- a/data_only_viz/finger_strike.py +++ b/data_only_viz/finger_strike.py @@ -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 diff --git a/data_only_viz/pose_bridge.py b/data_only_viz/pose_bridge.py index f9242e7..c0c46b8 100644 --- a/data_only_viz/pose_bridge.py +++ b/data_only_viz/pose_bridge.py @@ -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)]) diff --git a/data_only_viz/tests/test_finger_strike.py b/data_only_viz/tests/test_finger_strike.py index 686063b..1278aae 100644 --- a/data_only_viz/tests/test_finger_strike.py +++ b/data_only_viz/tests/test_finger_strike.py @@ -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 == []