refactor(pose): detectors consume routed slots
Convert all four consumers to accept pre-routed [L|None, R|None]: - finger_strike: remove order_hands_by_cx + _slot_hands; step() iterates enumerate(slotted), None slot resets state / synthesizes release edges in-place. - hand_features: step(hands, chirality=None) delegates to route_hands; speed buffer indexed by slot, not cx order. - action_head_pub: _pick_hands returns chirality; _emit_fingers/ _emit_pinch call route_hands with HAND_SWAP_LR/HAND_NEAR_MIN; _emit_hands plumbs chirality to HandFeatureExtractor. - renderer: panels block replaced with route_hands call; persistence gate + hand_plausible mask applied before routing so a ghost hand cannot steal a slot; HAND_SWAP_LR read at init. Test updates: [hand, None] slot lists in test_finger_strike; _hand() fixture in test_action_head_finger_emit gains realistic size so near_min=0.10 gate passes.
This commit is contained in:
L'électron rare
@@ -154,11 +154,13 @@ class ActionHeadPublisher(threading.Thread):
|
||||
"""Lock state, extract hand features, emit /pose/hands once per tick."""
|
||||
with self.state.lock():
|
||||
hands = list(getattr(self.state, "persons_hands", None) or [])
|
||||
chirality = list(getattr(self.state, "persons_hands_chirality", None) or [])
|
||||
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)
|
||||
hands = [h for h in (lkp, rkp) if h is not None and len(h) > 0]
|
||||
feats = self._hand_ext.step(hands)
|
||||
chirality = []
|
||||
feats = self._hand_ext.step(hands, chirality or None)
|
||||
with self.state.lock():
|
||||
self.state.hand_feats = feats
|
||||
self.bridge.send_hands(feats, t_now)
|
||||
@@ -198,28 +200,42 @@ class ActionHeadPublisher(threading.Thread):
|
||||
# "iphone" / "mediapipe".
|
||||
self._finger_source = os.environ.get("FINGER_SOURCE", "auto").lower()
|
||||
self._fdbg_n = 0
|
||||
# Safety knob: swap the two routed slots if the iPhone app's camera
|
||||
# config inverts Vision chirality (fix live without redeploy).
|
||||
self._hand_swap_lr = os.environ.get("HAND_SWAP_LR", "0") not in (
|
||||
"0", "", "false", "False",
|
||||
)
|
||||
self._hand_near_min = float(os.environ.get("HAND_NEAR_MIN", "0.10"))
|
||||
|
||||
def _pick_hands(self) -> tuple[list, str, bool]:
|
||||
"""Return (hands, source_label, ip_fresh) per FINGER_SOURCE policy."""
|
||||
def _pick_hands(self) -> tuple[list, list, str, bool]:
|
||||
"""Return (hands, chirality, 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 [])
|
||||
ip_t = getattr(self.state, "persons_hands_iphone_t", 0.0)
|
||||
ip_chir = list(getattr(self.state, "persons_hands_chirality", None) or [])
|
||||
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, "iphone", ip_fresh
|
||||
return ip_hands, ip_chir, "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)
|
||||
return mp_hands, [], "mediapipe", ip_fresh
|
||||
if ip_fresh:
|
||||
return ip_hands, ip_chir, "iphone", ip_fresh
|
||||
return 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)
|
||||
from data_only_viz.hand_slots import route_hands
|
||||
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),
|
||||
)
|
||||
events = self._finger_det.step(slotted, t_now)
|
||||
if getattr(self, "_finger_dbg", False):
|
||||
self._fdbg_n += 1
|
||||
if events or self._fdbg_n % 30 == 0:
|
||||
@@ -233,8 +249,14 @@ class ActionHeadPublisher(threading.Thread):
|
||||
/pose/pinch. Gated by PINCH_ENABLE, independently of the air-piano."""
|
||||
if not getattr(self, "_pinch_enabled", False):
|
||||
return
|
||||
hands, used, _ip = self._pick_hands()
|
||||
for ev in self._pinch_det.step(hands, t_now):
|
||||
from data_only_viz.hand_slots import route_hands
|
||||
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),
|
||||
)
|
||||
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",
|
||||
used, ev.hand, ev.finger)
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
"""Air-piano finger strike detection from raw MediaPipe hand joints.
|
||||
"""Air-piano finger strike detection from pre-routed hand slots.
|
||||
|
||||
Mirrors hand_features.py conventions (L/R slotting by cx, finite-guarded
|
||||
coordinate extraction). A "strike" is a fast downward motion of a fingertip
|
||||
RELATIVE to its base knuckle, so translating the whole hand does not fire all
|
||||
fingers. Output feeds the OSC /pose/finger route consumed by SuperCollider.
|
||||
A "strike" is a fast downward motion of a fingertip RELATIVE to its base
|
||||
knuckle, so translating the whole hand does not fire all fingers. Output feeds
|
||||
the OSC /pose/finger route consumed by SuperCollider.
|
||||
|
||||
Consumers call route_hands() (hand_slots) once per tick and pass the resulting
|
||||
2-element [left|None, right|None] list to FingerStrikeDetector.step() and
|
||||
PinchDetector.step(). slot 0 = left hand (matrix "MG"), slot 1 = right.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -24,26 +27,9 @@ 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:
|
||||
hand: int # 0 = left slot (leftmost cx), 1 = right slot
|
||||
hand: int # 0 = left slot, 1 = right slot
|
||||
finger: int # 0..4 = thumb, index, middle, ring, pinky
|
||||
strike_speed: float
|
||||
z: float
|
||||
@@ -70,27 +56,19 @@ class FingerStrikeDetector:
|
||||
self._state = [[_FingerState() for _ in range(5)]
|
||||
for _ in range(history_slots)]
|
||||
|
||||
def _slot_hands(self, hands: list) -> list:
|
||||
"""Validate and order hands leftmost-first (slot 0 = L), like
|
||||
HandFeatureExtractor. Returns up to 2 (cx, landmarks) ordered."""
|
||||
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 cand[:len(self._state)]
|
||||
def step(self, slotted: list, t_now: float) -> list[StrikeEvent]:
|
||||
"""Process a pre-routed 2-slot hand list [left|None, right|None].
|
||||
|
||||
def step(self, hands: list, t_now: float) -> list[StrikeEvent]:
|
||||
ordered = self._slot_hands(hands)
|
||||
present = set(range(len(ordered)))
|
||||
Absent slots (None) reset their finger state so re-entry does not
|
||||
produce spurious velocity spikes.
|
||||
"""
|
||||
events: list[StrikeEvent] = []
|
||||
for slot, (_cx, lm) in enumerate(ordered):
|
||||
for slot, lm in enumerate(slotted):
|
||||
if lm is None:
|
||||
for f in range(5):
|
||||
self._state[slot][f].prev_rel = None
|
||||
self._state[slot][f].armed = True
|
||||
continue
|
||||
for f in range(5):
|
||||
tip = lm[FINGERTIPS[f]]
|
||||
base = lm[FINGER_BASES[f]]
|
||||
@@ -116,18 +94,12 @@ class FingerStrikeDetector:
|
||||
tipx=_finite(_coord(tip, "x", 0), 0.5),
|
||||
tipy=tip_y,
|
||||
))
|
||||
# reset slots not present this tick so re-entry does not spike
|
||||
for slot in range(len(self._state)):
|
||||
if slot not in present:
|
||||
for f in range(5):
|
||||
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
|
||||
hand: int # 0 = left slot, 1 = right slot
|
||||
finger: int # 1..4 = index, middle, ring, pinky (thumb is trigger)
|
||||
state: int = 1 # 1 = engage edge, 0 = release edge
|
||||
|
||||
@@ -169,17 +141,27 @@ class PinchDetector:
|
||||
self.ext_ratio = ext_ratio
|
||||
self.ext_min = int(ext_min)
|
||||
# Engage fires only after debounce_frames consecutive qualifying
|
||||
# frames; release stays immediate. 1 = no debounce. (Enforced in
|
||||
# step(); wired fully by the debounce task.)
|
||||
# frames; release stays immediate. 1 = no debounce.
|
||||
self.debounce_frames = max(1, int(debounce_frames))
|
||||
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)))
|
||||
def step(self, slotted: list, t_now: float) -> list[PinchEvent]:
|
||||
"""Process a pre-routed 2-slot hand list [left|None, right|None].
|
||||
|
||||
Absent slots synthesise release edges for any engaged fingers and
|
||||
reset debounce state.
|
||||
"""
|
||||
events: list[PinchEvent] = []
|
||||
for slot, lm in enumerate(ordered):
|
||||
for slot, lm in enumerate(slotted):
|
||||
if lm is None:
|
||||
for i in range(4):
|
||||
st = self._state[slot][i]
|
||||
st.qual = 0
|
||||
if st.engaged:
|
||||
st.engaged = False
|
||||
events.append(PinchEvent(hand=slot, finger=i + 1, state=0))
|
||||
continue
|
||||
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)
|
||||
@@ -228,12 +210,4 @@ class PinchDetector:
|
||||
events.append(PinchEvent(hand=slot, finger=i + 1, state=1))
|
||||
else:
|
||||
st.qual = 0
|
||||
for slot in range(len(self._state)):
|
||||
if slot not in present:
|
||||
for i in range(4):
|
||||
st = self._state[slot][i]
|
||||
st.qual = 0
|
||||
if st.engaged:
|
||||
st.engaged = False
|
||||
events.append(PinchEvent(hand=slot, finger=i + 1, state=0))
|
||||
return events
|
||||
|
||||
@@ -78,31 +78,37 @@ class HandFeatureExtractor:
|
||||
buf.append((cx, cy))
|
||||
return _clamp(spd, 0.0, 1.0)
|
||||
|
||||
def step(self, hands: list) -> dict:
|
||||
feats = []
|
||||
for lm in hands:
|
||||
try:
|
||||
valid = lm is not None and len(lm) >= 21
|
||||
except TypeError:
|
||||
continue
|
||||
if valid:
|
||||
feats.append(self._features(lm))
|
||||
if not feats:
|
||||
def step(self, hands: list, chirality=None) -> dict:
|
||||
from data_only_viz.hand_slots import route_hands
|
||||
left_lm, right_lm = route_hands(hands, chirality)
|
||||
|
||||
if left_lm is None and right_lm is None:
|
||||
self._buf[0].clear()
|
||||
self._buf[1].clear()
|
||||
return {"L": None, "R": None, "dist": 0.0}
|
||||
feats.sort(key=lambda f: f["cx"]) # leftmost first
|
||||
left = feats[0]
|
||||
right = feats[-1] if len(feats) > 1 else None
|
||||
left["speed"] = self._speed(0, left["cx"], left["cy"])
|
||||
out_l = left
|
||||
|
||||
out_l = None
|
||||
out_r = None
|
||||
dist = 0.0
|
||||
if right is not None:
|
||||
right["speed"] = self._speed(1, right["cx"], right["cy"])
|
||||
out_r = right
|
||||
dist = _clamp(math.hypot(right["cx"] - left["cx"],
|
||||
right["cy"] - left["cy"]), 0.0, 1.0)
|
||||
|
||||
if left_lm is not None:
|
||||
lf = self._features(left_lm)
|
||||
lf["speed"] = self._speed(0, lf["cx"], lf["cy"])
|
||||
out_l = lf
|
||||
else:
|
||||
self._buf[0].clear()
|
||||
|
||||
if right_lm is not None:
|
||||
rf = self._features(right_lm)
|
||||
rf["speed"] = self._speed(1, rf["cx"], rf["cy"])
|
||||
out_r = rf
|
||||
else:
|
||||
self._buf[1].clear()
|
||||
|
||||
dist = 0.0
|
||||
if out_l is not None and out_r is not None:
|
||||
dist = _clamp(
|
||||
math.hypot(out_r["cx"] - out_l["cx"], out_r["cy"] - out_l["cy"]),
|
||||
0.0, 1.0,
|
||||
)
|
||||
|
||||
return {"L": out_l, "R": out_r, "dist": dist}
|
||||
|
||||
+24
-28
@@ -53,6 +53,7 @@ from .hand_display import (
|
||||
panel_frame,
|
||||
panel_segments,
|
||||
)
|
||||
from .hand_slots import route_hands
|
||||
from .mesh_topology import (
|
||||
BODY_TRIANGLES,
|
||||
FACE_TRIANGLES,
|
||||
@@ -164,6 +165,11 @@ class MetalRenderer(NSObject):
|
||||
self._hand_gate = HandPersistenceGate(
|
||||
min_frames=int(os.environ.get("HAND_PERSIST_FRAMES", "3"))
|
||||
)
|
||||
# Safety knob: swap the two routed hand slots if the iPhone app's
|
||||
# camera config inverts Vision chirality (fix live without redeploy).
|
||||
self._hand_swap_lr = os.environ.get("HAND_SWAP_LR", "0") not in (
|
||||
"0", "", "false", "False",
|
||||
)
|
||||
self._init_skel_cpu_buffer()
|
||||
self._init_mesh_cpu_buffer()
|
||||
self._build_pipelines()
|
||||
@@ -485,39 +491,29 @@ class MetalRenderer(NSObject):
|
||||
# even when use_arkit is True). Hands from s.persons_hands only.
|
||||
if self._mp_bones is not None:
|
||||
_body_b, _face_b, lhand_bones_p, _ = self._mp_bones
|
||||
# Filter hands by persistence gate before routing to panels.
|
||||
hands = [h for h, ok in zip(s.persons_hands, _hand_draw_flags) if ok]
|
||||
# Filter hands by persistence gate + plausibility before routing
|
||||
# to panels (a ghost hand must not steal a slot from a real one).
|
||||
keep = [
|
||||
ok and hand_plausible(h, conf_min=self._hand_conf_min)
|
||||
for h, ok in zip(s.persons_hands, _hand_draw_flags)
|
||||
]
|
||||
gated_hands = [h for h, k in zip(s.persons_hands, keep) if k]
|
||||
chir_src = getattr(s, "persons_hands_chirality", None) or []
|
||||
# Keep chirality aligned with the gated hands (same filter mask).
|
||||
if chir_src and len(chir_src) == len(s.persons_hands):
|
||||
chir = [c for c, ok in zip(chir_src, _hand_draw_flags) if ok]
|
||||
gated_chir = [c for c, k in zip(chir_src, keep) if k]
|
||||
else:
|
||||
chir = []
|
||||
gated_chir = None
|
||||
asp = float(s.width) / float(s.height) if s.width and s.height else 1.0
|
||||
|
||||
# Route hands to left/right panels via chirality when available,
|
||||
# otherwise by screen position (mirror-aware).
|
||||
left_kp = None
|
||||
right_kp = None
|
||||
if chir and len(chir) == len(hands):
|
||||
# chirality: 0 = left hand, 1 = right hand
|
||||
for h_kp, c in zip(hands, chir):
|
||||
if c == 0 and left_kp is None:
|
||||
left_kp = h_kp
|
||||
elif c == 1 and right_kp is None:
|
||||
right_kp = h_kp
|
||||
else:
|
||||
# Fallback: order plausible hands by screen cx, mirror-aware.
|
||||
# Screen left has low cx normally; high cx when video is mirrored.
|
||||
plausible = [
|
||||
(h_kp, sum(p.x for p in h_kp) / len(h_kp))
|
||||
for h_kp in hands
|
||||
if hand_plausible(h_kp, conf_min=self._hand_conf_min)
|
||||
]
|
||||
plausible.sort(key=lambda t: -t[1] if mirror else t[1])
|
||||
if plausible:
|
||||
left_kp = plausible[0][0]
|
||||
if len(plausible) >= 2:
|
||||
right_kp = plausible[1][0]
|
||||
# Route to L/R panels. mirror= passed for cx fallback path;
|
||||
# no near_min — panels show far hands as user proximity feedback.
|
||||
slotted = route_hands(
|
||||
gated_hands, gated_chir,
|
||||
mirror=mirror,
|
||||
swap=self._hand_swap_lr,
|
||||
)
|
||||
left_kp, right_kp = slotted
|
||||
|
||||
# Draw frame (pid 7) then wireframe (pid 5 left / 6 right)
|
||||
for side_name, h_kp, pid_s in (
|
||||
|
||||
@@ -18,6 +18,8 @@ class _FakeState:
|
||||
|
||||
def _hand(index_tip_y: float, base_y: float = 0.4):
|
||||
lm = [[0.3, base_y, 0.0] for _ in range(21)]
|
||||
lm[0] = [0.3, base_y + 0.15, 0.0] # wrist 0.15 below center -> hand_size 0.15
|
||||
lm[9] = [0.3, base_y, 0.0] # middle-MCP at center -> size 0.15 > near_min
|
||||
lm[FINGERTIPS[1]] = [0.3, index_tip_y, 0.0]
|
||||
lm[FINGER_BASES[1]] = [0.3, base_y, 0.0]
|
||||
return lm
|
||||
|
||||
@@ -27,11 +27,11 @@ def _hand(tip_y_by_finger: dict[int, float], base_y: float = 0.4,
|
||||
def test_downward_spike_fires_exactly_one_strike():
|
||||
det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=120.0)
|
||||
# frame 0: neutral (primes prev), index tip level with base
|
||||
det.step([_hand({1: 0.40})], t_now=0.00)
|
||||
det.step([_hand({1: 0.40}), None], t_now=0.00)
|
||||
# frame 1: index tip drops 0.06 below -> downward velocity 0.06 > thresh
|
||||
e1 = det.step([_hand({1: 0.46})], t_now=0.04)
|
||||
e1 = det.step([_hand({1: 0.46}), None], t_now=0.04)
|
||||
# frame 2: tip stays down -> velocity ~0, must NOT refire
|
||||
e2 = det.step([_hand({1: 0.46})], t_now=0.08)
|
||||
e2 = det.step([_hand({1: 0.46}), None], t_now=0.08)
|
||||
strikes = e1 + e2
|
||||
assert len(strikes) == 1
|
||||
assert strikes[0].finger == 1
|
||||
@@ -44,16 +44,16 @@ def test_whole_hand_translation_does_not_fire():
|
||||
out = []
|
||||
# tip and base move down together each frame -> relative y constant
|
||||
for i, by in enumerate((0.40, 0.50, 0.60, 0.70)):
|
||||
out += det.step([_hand({1: by}, base_y=by)], t_now=i * 0.04)
|
||||
out += det.step([_hand({1: by}, base_y=by), None], t_now=i * 0.04)
|
||||
assert out == []
|
||||
|
||||
|
||||
def test_refractory_blocks_second_strike():
|
||||
det = FingerStrikeDetector(vel_thresh=0.02, refractory_ms=120.0)
|
||||
det.step([_hand({1: 0.40})], t_now=0.00) # prime
|
||||
a = det.step([_hand({1: 0.46})], t_now=0.02) # strike 1
|
||||
det.step([_hand({1: 0.40})], t_now=0.04) # lift -> rearm
|
||||
b = det.step([_hand({1: 0.46})], t_now=0.06) # within 120 ms -> blocked
|
||||
det.step([_hand({1: 0.40}), None], t_now=0.00) # prime
|
||||
a = det.step([_hand({1: 0.46}), None], t_now=0.02) # strike 1
|
||||
det.step([_hand({1: 0.40}), None], t_now=0.04) # lift -> rearm
|
||||
b = det.step([_hand({1: 0.46}), None], t_now=0.06) # within 120 ms -> blocked
|
||||
assert len(a) == 1
|
||||
assert b == []
|
||||
|
||||
@@ -61,10 +61,10 @@ def test_refractory_blocks_second_strike():
|
||||
def test_strike_speed_scales_with_velocity():
|
||||
soft = FingerStrikeDetector(vel_thresh=0.02, speed_scale=0.10)
|
||||
hard = FingerStrikeDetector(vel_thresh=0.02, speed_scale=0.10)
|
||||
soft.step([_hand({1: 0.40})], t_now=0.0)
|
||||
hard.step([_hand({1: 0.40})], t_now=0.0)
|
||||
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
|
||||
soft.step([_hand({1: 0.40}), None], t_now=0.0)
|
||||
hard.step([_hand({1: 0.40}), None], t_now=0.0)
|
||||
s = soft.step([_hand({1: 0.44}), None], t_now=0.04) # delta 0.04
|
||||
h = hard.step([_hand({1: 0.50}), None], t_now=0.04) # delta 0.10
|
||||
assert h[0].strike_speed > s[0].strike_speed
|
||||
|
||||
|
||||
@@ -89,8 +89,8 @@ _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 det.step([_pinch_hand(*_OPEN), None], 0.0) == []
|
||||
ev = det.step([_pinch_hand(*_PINCH), None], 0.1)
|
||||
assert len(ev) == 1
|
||||
assert ev[0].finger == 1 # index = finger 1
|
||||
assert ev[0].hand == 0
|
||||
@@ -98,29 +98,29 @@ def test_pinch_fires_on_thumb_index_contact():
|
||||
|
||||
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
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
a = det.step([_pinch_hand(*_PINCH), None], 0.1) # fire
|
||||
b = det.step([_pinch_hand(*_PINCH), None], 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
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
a = det.step([_pinch_hand(*_PINCH), None], 0.1) # fire
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.2) # release (ratio > off)
|
||||
c = det.step([_pinch_hand(*_PINCH), None], 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
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
a = det.step([_pinch_hand(*_PINCH), None], 0.05) # fire t=0.05
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.08) # release
|
||||
b = det.step([_pinch_hand(*_PINCH), None], 0.10) # within 200 ms -> blocked
|
||||
assert len(a) == 1
|
||||
assert b == []
|
||||
|
||||
@@ -128,9 +128,9 @@ def test_pinch_refractory_blocks():
|
||||
def test_pinch_engage_release_emits_both_edges():
|
||||
"""Engage emits state=1; subsequent open emits state=0 (same finger)."""
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0)
|
||||
det.step([_pinch_hand(*_OPEN)], 0.0)
|
||||
engage = det.step([_pinch_hand(*_PINCH)], 0.1) # engage edge
|
||||
release = det.step([_pinch_hand(*_OPEN)], 0.2) # release edge
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
engage = det.step([_pinch_hand(*_PINCH), None], 0.1) # engage edge
|
||||
release = det.step([_pinch_hand(*_OPEN), None], 0.2) # release edge
|
||||
assert len(engage) == 1 and engage[0].state == 1
|
||||
assert len(release) == 1 and release[0].state == 0
|
||||
assert release[0].hand == 0 and release[0].finger == 1
|
||||
@@ -139,9 +139,9 @@ def test_pinch_engage_release_emits_both_edges():
|
||||
def test_pinch_hand_disappear_emits_release():
|
||||
"""When an engaged hand disappears, a release edge is synthesised."""
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0)
|
||||
det.step([_pinch_hand(*_OPEN)], 0.0)
|
||||
engage = det.step([_pinch_hand(*_PINCH)], 0.1) # engage
|
||||
release = det.step([], 0.2) # hand gone
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
engage = det.step([_pinch_hand(*_PINCH), None], 0.1) # engage
|
||||
release = det.step([None, None], 0.2) # hand gone
|
||||
assert len(engage) == 1 and engage[0].state == 1
|
||||
assert len(release) == 1 and release[0].state == 0
|
||||
|
||||
@@ -163,9 +163,9 @@ def _pinch_hand_multi(thumb_xy, tips):
|
||||
def test_pinch_closest_finger_wins():
|
||||
# index ratio ~0.067, middle ratio ~0.33 -> margin 0.27 >= 0.20 -> index only.
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0)
|
||||
det.step([_pinch_hand_multi((0.5, 0.5), {})], 0.0) # open
|
||||
det.step([_pinch_hand_multi((0.5, 0.5), {}), None], 0.0) # open
|
||||
ev = det.step([_pinch_hand_multi((0.5, 0.5),
|
||||
{0: (0.52, 0.5), 1: (0.60, 0.5)})], 0.1)
|
||||
{0: (0.52, 0.5), 1: (0.60, 0.5)}), None], 0.1)
|
||||
assert len(ev) == 1
|
||||
assert ev[0].finger == 1 and ev[0].state == 1 # index only
|
||||
|
||||
@@ -174,9 +174,9 @@ def test_pinch_ambiguous_adjacent_fires_nothing():
|
||||
# index ratio ~0.067, middle ratio ~0.10 -> margin 0.033 < 0.20 -> no winner.
|
||||
# (both are below ratio_on, so the OLD per-finger logic would fire both.)
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0)
|
||||
det.step([_pinch_hand_multi((0.5, 0.5), {})], 0.0) # open
|
||||
det.step([_pinch_hand_multi((0.5, 0.5), {}), None], 0.0) # open
|
||||
ev = det.step([_pinch_hand_multi((0.5, 0.5),
|
||||
{0: (0.52, 0.5), 1: (0.53, 0.5)})], 0.1)
|
||||
{0: (0.52, 0.5), 1: (0.53, 0.5)}), None], 0.1)
|
||||
assert ev == []
|
||||
|
||||
|
||||
@@ -201,7 +201,7 @@ def test_relaxed_hand_rejected_by_extension_gate():
|
||||
ext_min=2)
|
||||
out = []
|
||||
for i in range(10):
|
||||
out += det.step([_relaxed_hand()], i * 0.033)
|
||||
out += det.step([_relaxed_hand(), None], i * 0.033)
|
||||
assert out == []
|
||||
|
||||
|
||||
@@ -210,8 +210,8 @@ def test_open_hand_pinch_passes_extension_gate():
|
||||
# (ext 1.67 / 1.49 / 1.37, all >= 1.35) -> 3 extended >= ext_min 2.
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0,
|
||||
ext_min=2)
|
||||
det.step([_pinch_hand(*_OPEN)], 0.0)
|
||||
ev = det.step([_pinch_hand(*_PINCH)], 0.1)
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.0)
|
||||
ev = det.step([_pinch_hand(*_PINCH), None], 0.1)
|
||||
assert len(ev) == 1 and ev[0].state == 1 and ev[0].finger == 1
|
||||
|
||||
|
||||
@@ -219,17 +219,17 @@ def test_extension_gate_defeat_reproduces_old_behavior():
|
||||
# ext_min=0 disables the gate: the relaxed hand fires like today.
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0,
|
||||
ext_min=0)
|
||||
ev = det.step([_relaxed_hand()], 0.1)
|
||||
ev = det.step([_relaxed_hand(), None], 0.1)
|
||||
assert len(ev) == 1 and ev[0].state == 1
|
||||
|
||||
|
||||
def test_debounce_delays_engage_to_nth_frame():
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0,
|
||||
debounce_frames=3)
|
||||
det.step([_pinch_hand(*_OPEN)], 0.00)
|
||||
a = det.step([_pinch_hand(*_PINCH)], 0.10) # qualifying 1
|
||||
b = det.step([_pinch_hand(*_PINCH)], 0.13) # qualifying 2
|
||||
c = det.step([_pinch_hand(*_PINCH)], 0.16) # qualifying 3 -> engage
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.00)
|
||||
a = det.step([_pinch_hand(*_PINCH), None], 0.10) # qualifying 1
|
||||
b = det.step([_pinch_hand(*_PINCH), None], 0.13) # qualifying 2
|
||||
c = det.step([_pinch_hand(*_PINCH), None], 0.16) # qualifying 3 -> engage
|
||||
assert a == [] and b == []
|
||||
assert len(c) == 1 and c[0].state == 1 and c[0].finger == 1
|
||||
|
||||
@@ -237,20 +237,20 @@ def test_debounce_delays_engage_to_nth_frame():
|
||||
def test_debounce_jitter_resets_counter():
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0,
|
||||
debounce_frames=3)
|
||||
det.step([_pinch_hand(*_PINCH)], 0.00) # qualifying 1
|
||||
det.step([_pinch_hand(*_PINCH)], 0.03) # qualifying 2
|
||||
det.step([_pinch_hand(*_OPEN)], 0.06) # jitter -> counter resets
|
||||
a = det.step([_pinch_hand(*_PINCH)], 0.09) # qualifying 1 again
|
||||
b = det.step([_pinch_hand(*_PINCH)], 0.12) # qualifying 2
|
||||
det.step([_pinch_hand(*_PINCH), None], 0.00) # qualifying 1
|
||||
det.step([_pinch_hand(*_PINCH), None], 0.03) # qualifying 2
|
||||
det.step([_pinch_hand(*_OPEN), None], 0.06) # jitter -> counter resets
|
||||
a = det.step([_pinch_hand(*_PINCH), None], 0.09) # qualifying 1 again
|
||||
b = det.step([_pinch_hand(*_PINCH), None], 0.12) # qualifying 2
|
||||
assert a == [] and b == []
|
||||
|
||||
|
||||
def test_release_immediate_after_debounced_engage():
|
||||
det = PinchDetector(ratio_on=0.45, ratio_off=0.65, refractory_ms=0,
|
||||
debounce_frames=3)
|
||||
det.step([_pinch_hand(*_PINCH)], 0.00)
|
||||
det.step([_pinch_hand(*_PINCH)], 0.03)
|
||||
eng = det.step([_pinch_hand(*_PINCH)], 0.06) # 3rd frame -> engage
|
||||
rel = det.step([_pinch_hand(*_OPEN)], 0.09) # very next frame
|
||||
det.step([_pinch_hand(*_PINCH), None], 0.00)
|
||||
det.step([_pinch_hand(*_PINCH), None], 0.03)
|
||||
eng = det.step([_pinch_hand(*_PINCH), None], 0.06) # 3rd frame -> 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
|
||||
|
||||
Reference in New Issue
Block a user