feat: prefer iphone vision hands for air-piano
iPhone Vision hands (on-device, .right upright, y already top-left, rotation-invariant) are stored in a dedicated state field and used by the strike detector when fresh (FINGER_SOURCE auto/iphone/mediapipe). More stable than MediaPipe on the rotated downscaled video. Adds FINGER_DEBUG strike logging.
This commit is contained in:
@@ -173,14 +173,41 @@ class ActionHeadPublisher(threading.Thread):
|
||||
self._finger_det = FingerStrikeDetector(
|
||||
vel_thresh=vel, refractory_ms=refr,
|
||||
)
|
||||
self._finger_dbg = os.environ.get("FINGER_DEBUG", "0") not in (
|
||||
"0", "", "false", "False",
|
||||
)
|
||||
# auto = iPhone Vision hands if fresh, else MediaPipe. Or force
|
||||
# "iphone" / "mediapipe".
|
||||
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 on persons_hands and emit /pose/finger."""
|
||||
"""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
|
||||
with self.state.lock():
|
||||
hands = list(getattr(self.state, "persons_hands", None) or [])
|
||||
for ev in self._finger_det.step(hands, t_now):
|
||||
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_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")
|
||||
events = self._finger_det.step(hands, t_now)
|
||||
if getattr(self, "_finger_dbg", False):
|
||||
self._fdbg_n += 1
|
||||
if events or self._fdbg_n % 30 == 0:
|
||||
LOG.info("fingers dbg: src=%s hands=%d strikes=%d (ip_fresh=%s)",
|
||||
used, len(hands), len(events), ip_fresh)
|
||||
for ev in events:
|
||||
self.bridge.send_finger(ev)
|
||||
|
||||
def _read_sources(self) -> tuple[
|
||||
|
||||
@@ -138,9 +138,15 @@ class IphoneUSBSource:
|
||||
self.state.persons_arkit_last_t[pid] = time.perf_counter()
|
||||
elif tag == TAG_HANDS:
|
||||
hands = _decode_hands(payload)
|
||||
if hands is not None and self._write_hands and self.state is not None:
|
||||
if hands is not None and self.state is not None:
|
||||
with self.state.lock():
|
||||
self.state.persons_hands = hands
|
||||
# Always expose iPhone Vision hands (stable,
|
||||
# rotation-invariant) for the air-piano.
|
||||
self.state.persons_hands_iphone = hands
|
||||
self.state.persons_hands_iphone_t = \
|
||||
time.perf_counter()
|
||||
if self._write_hands:
|
||||
self.state.persons_hands = hands
|
||||
except OSError as e:
|
||||
LOG.warning("iphone usb stream error: %s", e)
|
||||
finally:
|
||||
|
||||
@@ -102,6 +102,11 @@ class State:
|
||||
persons_body: list[list[PoseKp]] = field(default_factory=list)
|
||||
persons_face: list[list[PoseKp]] = field(default_factory=list)
|
||||
persons_hands: list[list[PoseKp]] = field(default_factory=list)
|
||||
# iPhone Vision hands (on-device, 21 kp MediaPipe order, .right upright,
|
||||
# y already top-left/down). Stored separately from MediaPipe persons_hands
|
||||
# so the air-piano can prefer this stabler, rotation-invariant source.
|
||||
persons_hands_iphone: list[list[PoseKp]] = field(default_factory=list)
|
||||
persons_hands_iphone_t: float = 0.0
|
||||
hand_feats: dict | None = None
|
||||
# MediaPipe pose_world_landmarks per person : 33 keypoints in meters,
|
||||
# relative to the hip-center. Optional companion of persons_body
|
||||
|
||||
Reference in New Issue
Block a user