feat(viz): gesture status panel indicators
CI build oscope-of / build-check (push) Has been cancelled
CI build oscope-of / build-check (push) Has been cancelled
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.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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].
|
||||
|
||||
|
||||
+31
-12
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user