Spaces:
Running
Running
File size: 9,105 Bytes
d646dd7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | """Small shared helpers for stable two-hand slot assignment.
MediaPipe does not promise that the order of ``hand_landmarks`` is stable
between frames. Keeping that order as ``hand_0``/``hand_1`` lets the
per-slot filters consume the other hand's history when the hands cross or
one hand is briefly lost. This module assigns detections to persistent
slots using handedness when available and image-space proximity as a
fallback.
"""
from __future__ import annotations
from itertools import combinations, permutations
from typing import Any
import numpy as np
def clip_normalized_xy(points: Any) -> np.ndarray:
"""Return keypoints with normalized image x/y constrained to [0, 1].
The detector may legitimately return a point just outside the image
after smoothing or tracker translation. Such a point cannot project
onto the source image and makes the overlay appear detached from the
hand, so only x/y are clipped; the model's z value is preserved.
"""
array = np.asarray(points, dtype=np.float32).copy()
if array.ndim == 2 and array.shape[1] >= 2:
array[:, :2] = np.clip(array[:, :2], 0.0, 1.0)
return array
class StableHandSlotAssigner:
"""Assign up to ``max_slots`` detections to persistent hand slots.
Candidates are dictionaries with ``label``, ``center`` and ``payload``
keys. The payload is returned untouched, so each caller can keep its
own landmark representation. Previous slots are deliberately retained
briefly when a hand is missing; this lets the next visible frame match
by identity without inventing a keypoint row for the missing hand.
"""
def __init__(self, max_slots: int = 2):
self.max_slots = max(1, int(max_slots))
self.labels: list[str | None] = [None] * self.max_slots
self.centers: list[np.ndarray | None] = [None] * self.max_slots
self.velocities: list[np.ndarray | None] = [None] * self.max_slots
self._last_seen: list[int | None] = [None] * self.max_slots
self._tick = 0
@staticmethod
def _distance(a: np.ndarray | None, b: np.ndarray | None) -> float:
if a is None or b is None:
return float("inf")
return float(np.linalg.norm(np.asarray(a) - np.asarray(b)))
def assign(self, candidates: list[dict[str, Any]]) -> list[dict[str, Any] | None]:
self._tick += 1
candidates = list(candidates[:self.max_slots])
# MediaPipe handedness is a per-detection estimate, not a guaranteed
# physical identity. In a non-mirrored overhead RGB view it can
# occasionally report both hands as the same side (for example
# [Right, Right]). On the first two-hand frame, use the separated
# image-space positions to recover the missing semantic distinction.
# This keeps both the 3D preview and exported hand_left/right fields
# correct; subsequent frames remain slot-latched.
if (not any(self.labels) and len(candidates) == 2
and all(str(c.get("label") or "").lower() in
{"", "left", "right"} for c in candidates)):
centers = [c.get("center") for c in candidates]
try:
x_values = [float(np.asarray(center, dtype=np.float32)[0])
for center in centers]
if all(np.isfinite(x) for x in x_values) \
and abs(x_values[0] - x_values[1]) > 0.05:
order = np.argsort(np.asarray(x_values))
normalized = list(candidates)
for rank, candidate_index in enumerate(order):
normalized[int(candidate_index)] = {
**candidates[int(candidate_index)],
"label": "Left" if rank == 0 else "Right",
}
candidates = normalized
except (TypeError, ValueError, IndexError):
pass
result: list[dict[str, Any] | None] = [None] * self.max_slots
used: set[int] = set()
def predicted_center(slot: int) -> np.ndarray | None:
center = self.centers[slot]
if center is None:
return None
velocity = self.velocities[slot]
last_seen = self._last_seen[slot]
if velocity is None or last_seen is None:
return center
gap = max(1, self._tick - last_seen)
# A short motion prediction is important when two hands cross:
# the nearest point after the crossing is often the other hand's
# old position. Do not extrapolate indefinitely over long gaps.
return center + velocity * min(gap, 3)
def assignment_cost(slot: int, index: int) -> float:
predicted = predicted_center(slot)
distance = self._distance(
predicted, candidates[index].get("center"))
if not np.isfinite(distance):
return 1e6
# Handedness is only a weak tie-breaker after a track exists. A
# hard label preference makes slots swap exactly when MediaPipe's
# label briefly flips during a crossing or occlusion.
label = str(candidates[index].get("label") or "").lower()
stable = str(self.labels[slot] or "").lower()
if label in {"left", "right"} and stable in {"left", "right"} \
and label != stable:
distance += 0.01
return distance
# Existing tracks are matched by predicted image position. Enumerate
# the small (at most two-hand) assignment space so the two hands are
# assigned jointly rather than greedily. This preserves identity when
# they cross in the image.
active_slots = [slot for slot in range(self.max_slots)
if self.centers[slot] is not None]
if candidates and active_slots:
count = min(len(active_slots), len(candidates))
best = None
for slot_subset in combinations(active_slots, count):
for candidate_order in permutations(range(len(candidates)), count):
cost = sum(
assignment_cost(slot, index)
for slot, index in zip(slot_subset, candidate_order)
)
key = (cost, tuple(slot_subset), tuple(candidate_order))
if best is None or key < best[0]:
best = (key, slot_subset, candidate_order)
if best is not None:
_, slot_subset, candidate_order = best
for slot, index in zip(slot_subset, candidate_order):
result[slot] = candidates[index]
used.add(index)
# Fill never-seen/empty slots deterministically, preferring the
# semantic label only during initialization (not during tracking).
for slot in range(self.max_slots):
if result[slot] is not None:
continue
remaining = [i for i in range(len(candidates)) if i not in used]
if not remaining:
continue
expected = "left" if slot == 0 else "right"
index = next(
(i for i in remaining
if str(candidates[i].get("label") or "").lower() == expected),
remaining[0],
)
result[slot] = candidates[index]
used.add(index)
for slot, candidate in enumerate(result):
if candidate is None:
continue
label = str(candidate.get("label") or "")
center = candidate.get("center")
new_center = (
np.asarray(center, dtype=np.float32).copy()
if center is not None else None
)
if self.labels[slot] is None and label:
self.labels[slot] = label
if new_center is not None:
old_center = self.centers[slot]
last_seen = self._last_seen[slot]
if old_center is not None and last_seen is not None:
gap = max(1, self._tick - last_seen)
self.velocities[slot] = (new_center - old_center) / gap
else:
self.velocities[slot] = None
self.centers[slot] = new_center
self._last_seen[slot] = self._tick
return result
def stable_label(self, slot: int, fallback: str | None = None) -> str:
"""Return the label latched when a slot was initialized.
The detector's per-frame handedness value is useful for initialization
but must not be allowed to rename an already tracked slot.
"""
index = max(0, min(int(slot), self.max_slots - 1))
if self.labels[index] is None and fallback:
value = str(fallback).strip()
if value:
self.labels[index] = value
return str(self.labels[index] or fallback or "")
|