Spaces:
Running
Running
| """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 | |
| 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 "") | |