From 32a722e2810033ca4be379c133b7d79465e789b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?L=27=C3=A9lectron=20rare?= <108685187+electron-rare@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:16:14 +0200 Subject: [PATCH] feat(viz): gesture status panel indicators Add gesture_slot_status [0..3] to State (absent/detected/armed/engaged). PinchDetector.engaged_slots() exposes per-slot pinch state. Publisher writes status after each tick via _update_gesture_slot_status(). Renderer draws panel frame always: dim (pid7 conf=0.3) when absent, normal (pid7) detected, pid8 armed, pid9 double-stroke engaged. --- data_only_viz/action_head_pub.py | 29 ++++++++++++++ data_only_viz/finger_strike.py | 7 ++++ data_only_viz/renderer.py | 43 +++++++++++++++------ data_only_viz/state.py | 4 ++ data_only_viz/tests/test_action_head_pub.py | 43 +++++++++++++++++++++ data_only_viz/tests/test_finger_strike.py | 28 ++++++++++++++ 6 files changed, 142 insertions(+), 12 deletions(-) diff --git a/data_only_viz/action_head_pub.py b/data_only_viz/action_head_pub.py index cef5591..f53dd5b 100644 --- a/data_only_viz/action_head_pub.py +++ b/data_only_viz/action_head_pub.py @@ -161,6 +161,7 @@ class ActionHeadPublisher(threading.Thread): self._emit_hands(t_now) self._emit_fingers(t_now) self._emit_pinch(t_now) + self._update_gesture_slot_status() def _emit_hands(self, t_now: float) -> None: """Lock state, extract hand features, emit /pose/hands once per tick.""" @@ -241,6 +242,7 @@ class ActionHeadPublisher(threading.Thread): hold_frames=cfg.hand_hold_frames, near_on=cfg.hand_near_min, near_off=cfg.hand_near_off, + face_min=cfg.hand_face_min, ) self._gesture_conf_min = cfg.hand_conf_min # Gesture-path gate stepped at ~30 Hz (publisher tick) — correctly @@ -251,6 +253,7 @@ class ActionHeadPublisher(threading.Thread): self._tick_hands: list | None = None self._tick_chir: list = [] self._tick_mirror: bool = False + self._tick_raw_slotted: list = [None, None] def _pick_hands(self) -> tuple[list, list, str, bool, bool]: """Return (hands, chirality, source_label, ip_fresh, mirror). @@ -311,6 +314,7 @@ class ActionHeadPublisher(threading.Thread): near_min=0.0, mirror=mirror, ) + self._tick_raw_slotted = list(raw) # save pre-stabilizer routing for status return self._stab.step(raw) def _emit_fingers(self, t_now: float) -> None: @@ -345,6 +349,31 @@ class ActionHeadPublisher(threading.Thread): used, ev.hand, ev.finger) self.bridge.send_pinch(ev) + def _update_gesture_slot_status(self) -> None: + """Compute per-slot gesture status and write it to state. + + Status values: 0=absent, 1=detected(plausible+established, not armed), + 2=armed(near+facing), 3=pinch engaged. + Called once per tick after all emitters have run. + """ + stab = getattr(self, "_stab", None) + if stab is None: + return + pinch_det = getattr(self, "_pinch_det", None) + raw = getattr(self, "_tick_raw_slotted", [None, None]) + active = stab.active_flags() + engaged = pinch_det.engaged_slots() if pinch_det is not None else (False, False) + status = [0, 0] + for slot in range(2): + if engaged[slot]: + status[slot] = 3 + elif active[slot]: + status[slot] = 2 + elif raw[slot] is not None: + status[slot] = 1 + with self.state.lock(): + self.state.gesture_slot_status = status + def _read_sources(self) -> tuple[ list[tuple[int, np.ndarray, np.ndarray, float, np.ndarray]] | None, float, str, bool, diff --git a/data_only_viz/finger_strike.py b/data_only_viz/finger_strike.py index c8331b5..ac90c08 100644 --- a/data_only_viz/finger_strike.py +++ b/data_only_viz/finger_strike.py @@ -161,6 +161,13 @@ class PinchDetector: self._state = [[_PinchState() for _ in range(4)] for _ in range(history_slots)] + def engaged_slots(self) -> tuple[bool, bool]: + """Return (slot0_has_engaged_pinch, slot1_has_engaged_pinch).""" + return ( + any(self._state[0][i].engaged for i in range(4)), + any(self._state[1][i].engaged for i in range(4)), + ) + def step(self, slotted: list, t_now: float) -> list[PinchEvent]: """Process a pre-routed 2-slot hand list [left|None, right|None]. diff --git a/data_only_viz/renderer.py b/data_only_viz/renderer.py index 1efb550..8d7771f 100644 --- a/data_only_viz/renderer.py +++ b/data_only_viz/renderer.py @@ -528,20 +528,39 @@ class MetalRenderer(NSObject): ) left_kp, right_kp = slotted - # Draw frame (pid 7) then wireframe (pid 5 left / 6 right) - for side_name, h_kp, pid_s in ( + # Panel frame colored by gesture status (read from state, already under lock): + # pid 7 conf 0.3 = status 0 (absent, dim) + # pid 7 conf 1.0 = status 1 (detected, normal) + # pid 8 conf 1.0 = status 2 (armed: near+facing) + # pid 9 conf 1.0 = status 3 (pinch engaged, double-stroke) + _slot_status = getattr(s, "gesture_slot_status", [0, 0]) + for _si, (side_name, h_kp, pid_s) in enumerate(( ("left", left_kp, 5), ("right", right_kp, 6), - ): - if h_kp is None: - 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, conf_min=self._hand_conf_min, - ): - push_panel(*seg, 1.0, pid_s) + )): + _st = _slot_status[_si] if _si < len(_slot_status) else 0 + _frame_segs = panel_frame(side_name, asp) + if _st == 0: + for seg in _frame_segs: + push_panel(*seg, 0.3, 7) # absent: dim + elif _st == 1: + for seg in _frame_segs: + push_panel(*seg, 1.0, 7) # detected: normal + elif _st == 2: + for seg in _frame_segs: + push_panel(*seg, 1.0, 8) # armed: distinct palette color + else: # 3: pinch engaged + _off = 0.001 # tiny offset for bold double-stroke + for seg in _frame_segs: + push_panel(*seg, 1.0, 9) + ax, ay, bx, by = seg + push_panel(ax + _off, ay, bx + _off, by, 1.0, 9) + if h_kp is not None: + 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: return 0 diff --git a/data_only_viz/state.py b/data_only_viz/state.py index 5c3d030..96a33de 100644 --- a/data_only_viz/state.py +++ b/data_only_viz/state.py @@ -176,6 +176,10 @@ class State: latest_bgr_id: int = 0 latest_bgr_t: float = 0.0 + # Gesture slot status per hand slot (written by action_head_pub, read by renderer): + # 0=absent, 1=detected(plausible+established not armed), 2=armed(near+facing), 3=pinch engaged + gesture_slot_status: list[int] = field(default_factory=lambda: [0, 0]) + # Renderer width: int = 1280 height: int = 720 diff --git a/data_only_viz/tests/test_action_head_pub.py b/data_only_viz/tests/test_action_head_pub.py index a18ae68..fc8444b 100644 --- a/data_only_viz/tests/test_action_head_pub.py +++ b/data_only_viz/tests/test_action_head_pub.py @@ -342,3 +342,46 @@ def test_emit_hands_features_require_establishment(monkeypatch) -> None: pub._tick(t_now=3.0) feats_est, _ = bridge.send_hands.call_args_list[-1][0] assert feats_est["L"] is not None # now established + + +# --------------------------------------------------------------------------- +# Gesture slot status +# --------------------------------------------------------------------------- + +def test_gesture_status_defaults_to_zero_with_no_hands() -> None: + """With no hands, _update_gesture_slot_status() writes [0, 0].""" + from data_only_viz.action_head_pub import ActionHeadPublisher + state = _FakeState() + state.gesture_slot_status = [99, 99] # sentinel: should be overwritten + bridge = MagicMock() + pub = ActionHeadPublisher(state, bridge, ckpt_path=None) + pub._update_gesture_slot_status() + assert state.gesture_slot_status == [0, 0] + + +def test_gesture_status_armed_when_stab_active() -> None: + """gesture_slot_status[slot]=2 when stabilizer output is non-None (arm state).""" + from data_only_viz.action_head_pub import ActionHeadPublisher + state = _FakeState() + state.gesture_slot_status = [0, 0] + bridge = MagicMock() + pub = ActionHeadPublisher(state, bridge, ckpt_path=None) + # Directly mark slot 0 as active in the stabilizer's output cache + pub._stab._out[0] = object() + pub._update_gesture_slot_status() + assert state.gesture_slot_status[0] == 2, "armed slot must report status 2" + assert state.gesture_slot_status[1] == 0, "idle slot must report status 0" + + +def test_gesture_status_engaged_when_pinch_active() -> None: + """gesture_slot_status[slot]=3 when a pinch is engaged in that slot.""" + from data_only_viz.action_head_pub import ActionHeadPublisher + state = _FakeState() + state.gesture_slot_status = [0, 0] + bridge = MagicMock() + pub = ActionHeadPublisher(state, bridge, ckpt_path=None) + pub._stab._out[1] = object() # slot 1 armed + pub._pinch_det._state[1][0].engaged = True # slot 1 finger engaged + pub._update_gesture_slot_status() + assert state.gesture_slot_status[0] == 0, "slot 0: idle" + assert state.gesture_slot_status[1] == 3, "slot 1: engaged pinch = status 3" diff --git a/data_only_viz/tests/test_finger_strike.py b/data_only_viz/tests/test_finger_strike.py index ae1938a..0ec9256 100644 --- a/data_only_viz/tests/test_finger_strike.py +++ b/data_only_viz/tests/test_finger_strike.py @@ -335,3 +335,31 @@ def test_reset_slot_does_not_affect_other_slot(): # slot 1 unchanged assert det._state[1][1].prev_rel is not None, "slot 1 must be unaffected" + + +# --------------------------------------------------------------------------- +# PinchDetector.engaged_slots +# --------------------------------------------------------------------------- + +def test_engaged_slots_both_false_initially(): + """Fresh PinchDetector: no pinch engaged in either slot.""" + det = PinchDetector() + assert det.engaged_slots() == (False, False) + + +def test_engaged_slots_true_for_engaged_slot(): + """engaged_slots() reflects per-slot engaged state correctly.""" + det = PinchDetector() + det._state[0][0].engaged = True # slot 0, index finger + s0, s1 = det.engaged_slots() + assert s0 is True, "slot 0 has an engaged finger" + assert s1 is False, "slot 1 has no engaged finger" + + +def test_engaged_slots_false_after_release(): + """Clearing engaged flag → engaged_slots returns False for that slot.""" + det = PinchDetector() + det._state[1][2].engaged = True # slot 1, ring finger + assert det.engaged_slots()[1] is True + det._state[1][2].engaged = False + assert det.engaged_slots() == (False, False)