feat(data-only-viz): pose_bridge action+kin OSC
Ajoute 4 methodes OSC pour action_head et localisation cinetique: send_action(pid, label_idx, probs, t_now, force) envoie /pose/action avec [pid, label_idx, prob_0, prob_1, prob_2] pour les 3 classes. send_kin(pid, kin, t_now, force) envoie /pose/kin avec [pid, kin[0], kin[1], kin[2]] pour angles de poignets/coudes. send_enter/send_leave envoient /pose/enter et /pose/leave pour cycle vie des personnes. Throttle reuse _period/_last_t existants; force=True bypass throttle.
This commit is contained in:
@@ -113,3 +113,35 @@ class PoseSoundBridge:
|
||||
cli.send_message("/pose/limb_span", [pid, float(span)])
|
||||
try: self._avbody.send_message("/pose/limb_span", [pid, float(span)])
|
||||
except OSError: pass
|
||||
|
||||
def send_action(self, pid: int, label_idx: int,
|
||||
probs, t_now: float, force: bool = False) -> None:
|
||||
"""Send action classification result via /pose/action OSC route.
|
||||
|
||||
Sends: [pid (int), label_idx (int), prob_0 (float), prob_1 (float), prob_2 (float)]
|
||||
"""
|
||||
if not force and (t_now - self._last_t) < self._period:
|
||||
return
|
||||
p = [float(probs[0]), float(probs[1]), float(probs[2])]
|
||||
self._client.send_message("/pose/action", [int(pid), int(label_idx), *p])
|
||||
|
||||
def send_kin(self, pid: int, kin,
|
||||
t_now: float, force: bool = False) -> None:
|
||||
"""Send kinematic angles via /pose/kin OSC route.
|
||||
|
||||
Sends: [pid (int), kin_0 (float), kin_1 (float), kin_2 (float)]
|
||||
"""
|
||||
if not force and (t_now - self._last_t) < self._period:
|
||||
return
|
||||
self._client.send_message(
|
||||
"/pose/kin",
|
||||
[int(pid), float(kin[0]), float(kin[1]), float(kin[2])],
|
||||
)
|
||||
|
||||
def send_enter(self, pid: int) -> None:
|
||||
"""Send lifecycle event when person enters frame."""
|
||||
self._client.send_message("/pose/enter", [int(pid)])
|
||||
|
||||
def send_leave(self, pid: int) -> None:
|
||||
"""Send lifecycle event when person leaves frame."""
|
||||
self._client.send_message("/pose/leave", [int(pid)])
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Tests for /pose/action and /pose/kin OSC routes."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
def test_send_action_formats_5_args() -> None:
|
||||
from data_only_viz.pose_bridge import PoseSoundBridge
|
||||
b = PoseSoundBridge()
|
||||
b._client = MagicMock()
|
||||
b.send_action(pid=7, label_idx=2,
|
||||
probs=np.array([0.1, 0.2, 0.7], dtype=np.float32),
|
||||
t_now=0.0, force=True)
|
||||
b._client.send_message.assert_called_once()
|
||||
address, args = b._client.send_message.call_args.args
|
||||
assert address == "/pose/action"
|
||||
assert args[0] == 7
|
||||
assert args[1] == 2
|
||||
assert all(isinstance(v, float) for v in args[2:5])
|
||||
assert abs(sum(args[2:5]) - 1.0) < 1e-5
|
||||
|
||||
|
||||
def test_send_kin_formats_4_args() -> None:
|
||||
from data_only_viz.pose_bridge import PoseSoundBridge
|
||||
b = PoseSoundBridge()
|
||||
b._client = MagicMock()
|
||||
b.send_kin(pid=3, kin=np.array([0.5, 1.2, -0.3], dtype=np.float32),
|
||||
t_now=0.0, force=True)
|
||||
b._client.send_message.assert_called_once()
|
||||
address, args = b._client.send_message.call_args.args
|
||||
assert address == "/pose/kin"
|
||||
assert args[0] == 3
|
||||
assert len(args) == 4
|
||||
assert abs(args[1] - 0.5) < 1e-6
|
||||
assert abs(args[2] - 1.2) < 1e-6
|
||||
assert abs(args[3] - (-0.3)) < 1e-6
|
||||
|
||||
|
||||
def test_send_lifecycle_enter_leave() -> None:
|
||||
from data_only_viz.pose_bridge import PoseSoundBridge
|
||||
b = PoseSoundBridge()
|
||||
b._client = MagicMock()
|
||||
b.send_enter(pid=4)
|
||||
b.send_leave(pid=4)
|
||||
calls = [c.args[0] for c in b._client.send_message.call_args_list]
|
||||
assert "/pose/enter" in calls
|
||||
assert "/pose/leave" in calls
|
||||
Reference in New Issue
Block a user