feat(viz): hand persistence gate + conf knobs

HandPersistenceGate (min_frames=3, radius=0.15) suppresses ghost
hands that flash for fewer than min_frames consecutive frames.
Gate is called once per frame in the renderer on persons_hands;
result filters both the video overlay loop and the side panels.

hand_plausible now rejects wrists outside [-0.1, 1.1] (out-of-frame
ghost anchor check).  panel_segments gains a conf_min param.

Env knobs read at renderer init (single parse, no per-frame cost):
  HAND_CONF_MIN (default 0.45, raised from 0.3)
  HAND_PERSIST_FRAMES (default 3)
  ARKIT_BONE_MAX (default 0.5, passed to arkit_segments)

New tests: 12 gate tests + 4 wrist-bounds tests.
This commit is contained in:
L'électron rare
2026-07-02 10:27:58 +02:00
parent ac6768d8a8
commit 5bfc63f86d
4 changed files with 230 additions and 22 deletions
+81 -7
View File
@@ -66,11 +66,15 @@ def hand_plausible(
Rejects:
- fewer than 21 landmarks
- wrist (kp[0]) outside [-0.1, 1.1] in x or y (out-of-frame ghost anchor)
- size (wrist->middle-MCP) outside [size_min, size_max]
- mean confidence below conf_min
"""
if len(kp) < 21:
return False
w = kp[0]
if not (-0.1 <= w.x <= 1.1 and -0.1 <= w.y <= 1.1):
return False
size = hand_size(kp)
if size < size_min or size > size_max:
return False
@@ -153,6 +157,7 @@ def panel_segments(
bones: list[tuple[int, int]],
aspect: float,
mirror: bool = True,
conf_min: float = 0.3,
) -> list[tuple[float, float, float, float]]:
"""Map hand landmarks into a side panel using uniform pixel-space scale.
@@ -173,17 +178,18 @@ def panel_segments(
the mirrored video background (i.e. the panel always shows a "front view").
Args:
kp: 21+ keypoints with .x, .y (normalized image coords, y down), .c.
side: "left" or "right".
bones: List of (index_a, index_b) pairs defining which keypoints to connect.
aspect: view width / view height.
mirror: If True, flip X within the panel.
kp: 21+ keypoints with .x, .y (normalized image coords, y down), .c.
side: "left" or "right".
bones: List of (index_a, index_b) pairs defining which keypoints to connect.
aspect: view width / view height.
mirror: If True, flip X within the panel.
conf_min: Minimum confidence for hand_plausible and segment_ok checks.
Returns:
List of (ax, ay, bx, by) in normalized [0,1] screen coords, y down.
Returns [] if hand_plausible(kp) fails.
"""
if not hand_plausible(kp):
if not hand_plausible(kp, conf_min=conf_min):
return []
x0, y0, x1, y1 = panel_rect(side, aspect)
@@ -234,10 +240,78 @@ def panel_segments(
continue
A = kp[a]
B = kp[b]
if not segment_ok(A, B, sz):
if not segment_ok(A, B, sz, conf_min=conf_min):
continue
ax, ay = _map(A)
bx, by = _map(B)
result.append((ax, ay, bx, by))
return result
# ---------------------------------------------------------------------------
# Hand temporal persistence gate
# ---------------------------------------------------------------------------
class HandPersistenceGate:
"""Suppress ghost hand detections that only flash for 1-2 frames.
Tracks each hand by wrist position (landmark 0). A hand is only
drawable once the same wrist position has been matched for
*min_frames* consecutive calls. A track that goes unmatched for even
one frame is immediately dropped (resets the consecutive count).
min_frames=1 disables gating: every incoming hand is drawable on the
first call (useful for the single-person MediaPipe fallback path).
"""
def __init__(self, min_frames: int = 3, radius: float = 0.15) -> None:
self._min_frames = min_frames
self._radius = radius
# Each track: [wrist_x, wrist_y, consecutive_count]
self._tracks: list[list] = []
def step(self, hands: list) -> list[bool]:
"""Update tracks and return a draw flag per hand.
Args:
hands: list of 21-landmark hand objects; landmark 0 is the wrist
and must expose .x and .y attributes.
Returns:
list[bool] of length len(hands). True = this hand has been
seen at the same position for at least min_frames consecutive
calls and should be drawn.
"""
result: list[bool] = [False] * len(hands)
new_tracks: list[list] = []
used: set[int] = set()
for h_idx, hand in enumerate(hands):
wx = hand[0].x
wy = hand[0].y
# Find the nearest existing track within radius (greedy).
best_dist = float("inf")
best_t = -1
for t_idx, (tx, ty, _count) in enumerate(self._tracks):
if t_idx in used:
continue
d = math.hypot(wx - tx, wy - ty)
if d < best_dist and d < self._radius:
best_dist = d
best_t = t_idx
if best_t >= 0:
used.add(best_t)
new_count = self._tracks[best_t][2] + 1
new_tracks.append([wx, wy, new_count])
result[h_idx] = new_count >= self._min_frames
else:
# New track — starts at count 1.
new_tracks.append([wx, wy, 1])
result[h_idx] = 1 >= self._min_frames
# Unmatched tracks are dropped (not carried into new_tracks).
self._tracks = new_tracks
return result
+43 -15
View File
@@ -44,7 +44,15 @@ from Metal import (
from MetalKit import MTKView # noqa: F401 (utilise par d'autres modules)
from .arkit_skeleton import arkit_segments
from .hand_display import arkit_2d_fresh, hand_plausible, hand_size, segment_ok, panel_frame, panel_segments
from .hand_display import (
HandPersistenceGate,
arkit_2d_fresh,
hand_plausible,
hand_size,
segment_ok,
panel_frame,
panel_segments,
)
from .mesh_topology import (
BODY_TRIANGLES,
FACE_TRIANGLES,
@@ -151,6 +159,11 @@ class MetalRenderer(NSObject):
self._mesh_buf = self._device.newBufferWithLength_options_(
MESH_MAX_TRIS * 3 * MESH_VERT_FLOATS * 4, MTLResourceStorageModeShared)
self._mp_bones = _mediapipe_bones() # None si pas dispo
self._hand_conf_min = float(os.environ.get("HAND_CONF_MIN", "0.45"))
self._arkit_bone_max = float(os.environ.get("ARKIT_BONE_MAX", "0.5"))
self._hand_gate = HandPersistenceGate(
min_frames=int(os.environ.get("HAND_PERSIST_FRAMES", "3"))
)
self._init_skel_cpu_buffer()
self._init_mesh_cpu_buffer()
self._build_pipelines()
@@ -383,6 +396,10 @@ class MetalRenderer(NSObject):
segs += 1
return True
# Gate called once per frame on the full persons_hands list.
# Used for both the overlay loop and the side panels below.
_hand_draw_flags = self._hand_gate.step(s.persons_hands)
if use_arkit:
parents = s.arkit_parents
for pid, arr2d in s.persons_arkit_2d.items():
@@ -390,7 +407,9 @@ class MetalRenderer(NSObject):
if ts is None or (_arkit_now - ts) >= 1.0:
continue # skip stale or unknown pid
valid = s.persons_arkit_2d_valid.get(pid)
for (ax, ay, bx, by) in arkit_segments(arr2d, valid, parents):
for (ax, ay, bx, by) in arkit_segments(
arr2d, valid, parents, max_len=self._arkit_bone_max
):
if not push_seg(ax, ay, bx, by, 1.0, pid):
break
@@ -418,14 +437,16 @@ class MetalRenderer(NSObject):
if not push(face_kp[a], face_kp[b], 1.0, pid): break
if segs >= SKEL_MAX_SEGS: break
for i, hand_kp in enumerate(s.persons_hands):
if not _hand_draw_flags[i]:
continue
pid = ids_h[i] if i < len(ids_h) else i
if not hand_plausible(hand_kp):
if not hand_plausible(hand_kp, conf_min=self._hand_conf_min):
continue
size = hand_size(hand_kp)
for a, b in lhand_bones:
if a >= len(hand_kp) or b >= len(hand_kp): continue
A = hand_kp[a]; B = hand_kp[b]
if not segment_ok(A, B, size): continue
if not segment_ok(A, B, size, conf_min=self._hand_conf_min): continue
# Decalage palette mains (+5) pour les distinguer
if not push(A, B, min(A.c, B.c), pid + 5): break
# ----- FALLBACK single-person si persons_* vides -----
@@ -444,13 +465,13 @@ class MetalRenderer(NSObject):
for kp_list in (s.left_hand_kp, s.right_hand_kp):
if not any(p.x != 0.0 or p.y != 0.0 for p in kp_list):
continue
if not hand_plausible(kp_list):
if not hand_plausible(kp_list, conf_min=self._hand_conf_min):
continue
size = hand_size(kp_list)
for a, b in lhand_bones:
if a >= len(kp_list) or b >= len(kp_list): continue
A = kp_list[a]; B = kp_list[b]
if not segment_ok(A, B, size): continue
if not segment_ok(A, B, size, conf_min=self._hand_conf_min): continue
if not push(A, B, min(A.c, B.c), 0): break
else:
# Fallback COCO 17 (YOLO legacy)
@@ -460,13 +481,18 @@ class MetalRenderer(NSObject):
if not push(A, B, min(A.c, B.c), 0): break
# ---- SIDE PANELS: left/right hand front-view zoomed wireframe ----
# Runs on every path (arkit + mediapipe + fallback COCO).
# Hands always come from s.persons_hands regardless of use_arkit.
# Gated on self._mp_bones (panels need _mp_bones for hand bone topology,
# 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
hands = s.persons_hands
chir = getattr(s, "persons_hands_chirality", None) or []
asp = float(s.width) / float(s.height) if s.height else 1.0
# Filter hands by persistence gate before routing to panels.
hands = [h for h, ok in zip(s.persons_hands, _hand_draw_flags) if ok]
chir_src = getattr(s, "persons_hands_chirality", None) or []
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]
else:
chir = []
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).
@@ -485,7 +511,7 @@ class MetalRenderer(NSObject):
plausible = [
(h_kp, sum(p.x for p in h_kp) / len(h_kp))
for h_kp in hands
if hand_plausible(h_kp)
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:
@@ -502,7 +528,10 @@ class MetalRenderer(NSObject):
continue
for seg in panel_frame(side_name, asp):
push_panel(*seg, 1.0, 7)
for seg in panel_segments(h_kp, side_name, lhand_bones_p, asp, mirror=mirror):
for seg in panel_segments(
h_kp, side_name, lhand_bones_p, asp,
mirror=mirror, conf_min=self._hand_conf_min,
):
push_panel(*seg, 1.0, pid_s)
if segs == 0:
@@ -513,7 +542,7 @@ class MetalRenderer(NSObject):
return segs
def _update_mesh(self, s: State) -> int:
"""Remplit self._mesh_buf avec des triangles face/hand/body.
"""Remplit self._mesh_buf avec des triangles face/body.
Retourne le nombre de triangles ecrits (chacun = 3 vertices).
Filtre les triangles dont au moins un sommet a confiance < 0.3.
@@ -563,7 +592,6 @@ class MetalRenderer(NSObject):
ids_b = s.persons_body_ids or list(range(len(s.persons_body)))
ids_f = s.persons_face_ids or list(range(len(s.persons_face)))
ids_h = s.persons_hands_ids or list(range(len(s.persons_hands)))
# Body
for i, body_kp in enumerate(s.persons_body):
+1
View File
@@ -109,6 +109,7 @@ class State:
persons_hands_iphone_t: float = 0.0
# Chirality for each entry in persons_hands_iphone: 0=left, 1=right.
# Aligned 1:1 with persons_hands_iphone (same length after each TAG_HANDS update).
# Not valid for MediaPipe-written persons_hands (stays empty on that path).
persons_hands_chirality: list[int] = field(default_factory=list)
hand_feats: dict | None = None
# MediaPipe pose_world_landmarks per person : 33 keypoints in meters,
+105
View File
@@ -10,6 +10,7 @@ from dataclasses import dataclass
import pytest
from data_only_viz.hand_display import (
HandPersistenceGate,
arkit_2d_fresh,
hand_plausible,
hand_size,
@@ -397,3 +398,107 @@ def test_arkit_2d_fresh_custom_max_age_stale() -> None:
"""max_age=0.5 makes a 0.6 s old timestamp stale."""
ts_by_pid = {5: 999.4}
assert arkit_2d_fresh(ts_by_pid, now=1000.0, max_age=0.5) is False
# ---------------------------------------------------------------------------
# hand_plausible — wrist out-of-frame rejection
# ---------------------------------------------------------------------------
def test_plausible_wrist_out_of_frame_x_neg_rejected():
"""Wrist x < -0.1 → ghost anchor, reject."""
kp = _make_hand(wrist_x=-0.2, wrist_y=0.5, size=0.12, c=1.0)
assert hand_plausible(kp) is False
def test_plausible_wrist_out_of_frame_x_pos_rejected():
"""Wrist x > 1.1 → ghost anchor, reject."""
kp = _make_hand(wrist_x=1.2, wrist_y=0.5, size=0.12, c=1.0)
assert hand_plausible(kp) is False
def test_plausible_wrist_out_of_frame_y_rejected():
"""Wrist y < -0.1 → ghost anchor, reject."""
kp = _make_hand(wrist_x=0.5, wrist_y=-0.15, size=0.12, c=1.0)
assert hand_plausible(kp) is False
def test_plausible_wrist_at_exact_boundary_accepted():
"""Wrist at x=-0.1 is at the boundary — still accepted (inclusive)."""
kp = _make_hand(wrist_x=-0.1, wrist_y=0.5, size=0.12, c=1.0)
assert hand_plausible(kp) is True
# ---------------------------------------------------------------------------
# HandPersistenceGate
# ---------------------------------------------------------------------------
def test_gate_ghost_two_frames_never_draws():
"""A hand appearing only 2 times is never drawn (min_frames=3)."""
gate = HandPersistenceGate(min_frames=3)
hand = _make_hand(wrist_x=0.5, wrist_y=0.5)
assert gate.step([hand]) == [False] # frame 1: count=1
assert gate.step([hand]) == [False] # frame 2: count=2
def test_gate_steady_hand_appears_at_third_frame():
"""A steady hand becomes drawable from the 3rd consecutive frame."""
gate = HandPersistenceGate(min_frames=3)
hand = _make_hand(wrist_x=0.5, wrist_y=0.5)
assert gate.step([hand]) == [False] # frame 1
assert gate.step([hand]) == [False] # frame 2
assert gate.step([hand]) == [True] # frame 3: count=3 >= min_frames
assert gate.step([hand]) == [True] # frame 4: count=4, stays drawable
def test_gate_teleporting_hand_never_draws():
"""A hand that jumps >radius every frame never accumulates a track."""
gate = HandPersistenceGate(min_frames=3, radius=0.15)
for px in (0.1, 0.5, 0.9): # each jump is 0.4 > radius=0.15
hand = _make_hand(wrist_x=px, wrist_y=0.5)
assert gate.step([hand]) == [False]
def test_gate_two_steady_hands_both_drawable():
"""Two hands that remain steady both become drawable at frame 3."""
gate = HandPersistenceGate(min_frames=3)
lh = _make_hand(wrist_x=0.3, wrist_y=0.5)
rh = _make_hand(wrist_x=0.7, wrist_y=0.5)
for _ in range(2):
gate.step([lh, rh])
result = gate.step([lh, rh])
assert result == [True, True]
def test_gate_min_frames_one_immediate():
"""min_frames=1 makes any hand drawable on first appearance."""
gate = HandPersistenceGate(min_frames=1)
hand = _make_hand(wrist_x=0.5, wrist_y=0.5)
assert gate.step([hand]) == [True]
def test_gate_track_resets_after_absence():
"""A track that disappears for one frame must re-accumulate from scratch."""
gate = HandPersistenceGate(min_frames=3)
hand = _make_hand(wrist_x=0.5, wrist_y=0.5)
gate.step([hand])
gate.step([hand]) # count=2, not yet drawable
gate.step([]) # hand absent → track dropped
# Reappears: back to count=1
assert gate.step([hand]) == [False]
def test_gate_empty_input_returns_empty():
"""step([]) always returns []."""
gate = HandPersistenceGate()
assert gate.step([]) == []
def test_gate_greedy_no_double_assign():
"""Two hands close together: each matches its own track, not the same one."""
gate = HandPersistenceGate(min_frames=2, radius=0.15)
h1 = _make_hand(wrist_x=0.3, wrist_y=0.5)
h2 = _make_hand(wrist_x=0.4, wrist_y=0.5) # 0.1 apart, both within radius
gate.step([h1, h2])
result = gate.step([h1, h2])
# Both should have count=2 >= min_frames=2
assert result == [True, True]