Spaces:
Running
Running
| """Unit tests for performance library and MovementManager exclusive mode.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import json | |
| import time | |
| from pathlib import Path | |
| from unittest.mock import MagicMock, call | |
| import numpy as np | |
| import pytest | |
| from reachy_dance_tutor_app.moves import MovementManager | |
| from reachy_dance_tutor_app.performance_library.catalog import ClipInfo, scan_data_dir, scan_local_performance_dir | |
| from reachy_dance_tutor_app.performance_library.lead_compensation import LeadCompensationModel | |
| from reachy_dance_tutor_app.performance_library.paths import LOCAL_REPO_ID, resolve_data_dir, slug_repo_id | |
| from reachy_dance_tutor_app.performance_library.resolve import find_clip_info, find_clip_json_path | |
| def test_slug_repo_id() -> None: | |
| assert slug_repo_id("foo/bar") == "foo__bar" | |
| def test_resolve_data_dir_prefers_data_subdir(tmp_path: Path) -> None: | |
| data = tmp_path / "data" | |
| data.mkdir() | |
| (data / "a.json").write_text("{}", encoding="utf-8") | |
| assert resolve_data_dir(tmp_path) == data | |
| def test_scan_data_dir(tmp_path: Path) -> None: | |
| data = tmp_path / "data" | |
| data.mkdir() | |
| move = { | |
| "description": "test", | |
| "time": [0.0, 0.5], | |
| "set_target_data": [{"head": [[1, 0, 0, 0]] * 4, "antennas": [0, 0], "body_yaw": 0.0}], | |
| } | |
| (data / "clip-one.json").write_text(json.dumps(move), encoding="utf-8") | |
| (data / "clip-one.wav").touch() | |
| clips = scan_data_dir(data, repo_id="test/repo") | |
| assert len(clips) == 1 | |
| assert clips[0].move_id == "clip-one" | |
| assert clips[0].has_audio is True | |
| assert clips[0].duration_sec == pytest.approx(0.5) | |
| def test_scan_local_performance_dir(tmp_path: Path) -> None: | |
| performance_dir = tmp_path / "custom" / "waka_waka" | |
| performance_dir.mkdir(parents=True) | |
| move = { | |
| "description": "test local clip", | |
| "time": [0.0, 1.5], | |
| "set_target_data": [{"head": [[1, 0, 0, 0]] * 4, "antennas": [0, 0], "body_yaw": 0.0}], | |
| } | |
| (performance_dir / "dance.json").write_text(json.dumps(move), encoding="utf-8") | |
| (performance_dir / "dance.wav").touch() | |
| (performance_dir / "preview.mp4").touch() | |
| (performance_dir / "manifest.json").write_text( | |
| json.dumps( | |
| { | |
| "id": "waka_waka", | |
| "title": "Waka Waka", | |
| "artist": "Shakira", | |
| "description": "Curated kids routine", | |
| "preview_video": "preview.mp4", | |
| "tags": ["kids", "energetic"], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| clips = scan_local_performance_dir(performance_dir) | |
| assert len(clips) == 1 | |
| clip = clips[0] | |
| assert clip.move_id == "waka_waka" | |
| assert clip.repo_id == LOCAL_REPO_ID | |
| assert clip.source_kind == "local" | |
| assert clip.title == "Waka Waka" | |
| assert clip.artist == "Shakira" | |
| assert clip.tags == ("kids", "energetic") | |
| assert clip.has_audio is True | |
| assert clip.preview_path is not None and clip.preview_path.name == "preview.mp4" | |
| assert clip.duration_sec == pytest.approx(1.5) | |
| def test_lead_compensation_preserves_length() -> None: | |
| n = 10 | |
| frames = [{"head": [[1, 0, 0, 0]] * 4, "antennas": [0.1, 0.2], "body_yaw": 0.05} for _ in range(n)] | |
| move = {"description": "x", "time": list(range(n)), "set_target_data": frames} | |
| out = LeadCompensationModel().transform_move(move) | |
| assert len(out["set_target_data"]) == n | |
| def test_find_clip_json_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| from reachy_dance_tutor_app import config as cfg | |
| cache = tmp_path / "perf_cache" | |
| cache.mkdir() | |
| repo_root = cache / slug_repo_id("acme/demo") | |
| data = repo_root / "data" | |
| data.mkdir(parents=True) | |
| (data / "hello.json").write_text('{"description":"d","time":[0,1],"set_target_data":[{"x":1}]}', encoding="utf-8") | |
| monkeypatch.setattr(cfg.config, "PERFORMANCE_CACHE_DIR", str(cache)) | |
| p = find_clip_json_path("hello", repo_id="acme/demo") | |
| assert p is not None and p.name == "hello.json" | |
| def test_find_clip_json_path_local_performance(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| from reachy_dance_tutor_app import config as cfg | |
| cache = tmp_path / "perf_cache" | |
| cache.mkdir() | |
| local_root = tmp_path / "performances" | |
| performance_dir = local_root / "custom" / "waka_waka" | |
| performance_dir.mkdir(parents=True) | |
| (performance_dir / "dance.json").write_text( | |
| '{"description":"local","time":[0,2],"set_target_data":[{"head":[[1,0,0,0],[0,1,0,0],[0,0,1,0],[0,0,0,1]],"antennas":[0,0],"body_yaw":0}]}', | |
| encoding="utf-8", | |
| ) | |
| (performance_dir / "dance.wav").touch() | |
| (performance_dir / "manifest.json").write_text( | |
| '{"id":"waka_waka","title":"Waka Waka","artist":"Shakira"}', | |
| encoding="utf-8", | |
| ) | |
| monkeypatch.setattr(cfg.config, "PERFORMANCE_CACHE_DIR", str(cache)) | |
| monkeypatch.setattr(cfg.config, "LOCAL_PERFORMANCE_DIR", str(local_root)) | |
| p = find_clip_json_path("waka_waka", repo_id=LOCAL_REPO_ID) | |
| assert p is not None and p.name == "dance.json" | |
| def test_find_clip_info_local_performance_custom_audio(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| from reachy_dance_tutor_app import config as cfg | |
| cache = tmp_path / "perf_cache" | |
| cache.mkdir() | |
| local_root = tmp_path / "performances" | |
| performance_dir = local_root / "custom" / "golden" | |
| performance_dir.mkdir(parents=True) | |
| (performance_dir / "dance.json").write_text( | |
| '{"description":"local","time":[0,2],"set_target_data":[{"head":[[1,0,0,0],[0,1,0,0],[0,0,1,0],[0,0,0,1]],"antennas":[0,0],"body_yaw":0}]}', | |
| encoding="utf-8", | |
| ) | |
| (performance_dir / "golden.wav").touch() | |
| (performance_dir / "manifest.json").write_text( | |
| '{"id":"golden_kpop_demon_hunters","title":"Golden","artist":"KPOP Demon Hunters","motion":"dance.json","audio":"golden.wav"}', | |
| encoding="utf-8", | |
| ) | |
| monkeypatch.setattr(cfg.config, "PERFORMANCE_CACHE_DIR", str(cache)) | |
| monkeypatch.setattr(cfg.config, "LOCAL_PERFORMANCE_DIR", str(local_root)) | |
| clip = find_clip_info("golden_kpop_demon_hunters", repo_id=LOCAL_REPO_ID) | |
| assert clip is not None | |
| assert clip.json_path.name == "dance.json" | |
| assert clip.audio_path is not None and clip.audio_path.name == "golden.wav" | |
| def test_movement_manager_exclusive_skips_set_target() -> None: | |
| robot = MagicMock() | |
| robot.get_current_head_pose.return_value = np.eye(4, dtype=np.float64) | |
| robot.get_current_joint_positions.return_value = ([0.0, 0.0, 0.0], np.array([0.0, 0.0])) | |
| robot.set_target = MagicMock() | |
| mm = MovementManager(current_robot=robot, camera_worker=None) | |
| mm.start() | |
| try: | |
| assert mm.enter_exclusive_playback(timeout_s=3.0) | |
| assert mm.exclusive_playback_active() is True | |
| robot.set_target.reset_mock() | |
| time.sleep(0.25) | |
| robot.set_target.assert_not_called() | |
| assert mm.exit_exclusive_playback(timeout_s=3.0) | |
| assert mm.exclusive_playback_active() is False | |
| finally: | |
| mm.stop() | |
| def test_movement_manager_exclusive_discards_new_commands_and_offsets() -> None: | |
| from reachy_mini.motion.move import Move | |
| class StaticMove(Move): | |
| def duration(self) -> float: | |
| return 0.15 | |
| def evaluate(self, t: float) -> tuple[np.ndarray, np.ndarray, float]: | |
| return ( | |
| np.eye(4, dtype=np.float64), | |
| np.array([0.25, -0.25], dtype=np.float64), | |
| 0.4, | |
| ) | |
| robot = MagicMock() | |
| robot.get_current_head_pose.return_value = np.eye(4, dtype=np.float64) | |
| robot.get_current_joint_positions.return_value = ([0.0, 0.0, 0.0], np.array([0.0, 0.0])) | |
| robot.set_target = MagicMock() | |
| mm = MovementManager(current_robot=robot, camera_worker=None) | |
| mm.idle_inactivity_delay = 10.0 | |
| mm.start() | |
| try: | |
| assert mm.enter_exclusive_playback(timeout_s=3.0) | |
| robot.set_target.reset_mock() | |
| mm.set_speech_offsets((0.1, 0.2, 0.3, 0.4, 0.5, 0.6)) | |
| mm.queue_move(StaticMove()) | |
| time.sleep(0.25) | |
| assert mm.state.speech_offsets == (0.0, 0.0, 0.0, 0.0, 0.0, 0.0) | |
| assert robot.set_target.call_count == 0 | |
| assert mm.exit_exclusive_playback(timeout_s=3.0) | |
| time.sleep(0.25) | |
| assert not any( | |
| abs(call.kwargs.get("body_yaw", 0.0) - 0.4) < 1e-9 for call in robot.set_target.call_args_list | |
| ) | |
| finally: | |
| mm.stop() | |
| def test_play_performance_clip_releases_exclusive_on_preflight_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| from reachy_dance_tutor_app.tools.play_performance_clip import PlayPerformanceClip | |
| import reachy_dance_tutor_app.tools.play_performance_clip as module | |
| json_path = tmp_path / "clip.json" | |
| json_path.write_text( | |
| '{"time":[0,1],"set_target_data":[{"head":[[1,0,0,0],[0,1,0,0],[0,0,1,0],[0,0,0,1]],"antennas":[0,0],"body_yaw":0}]}', | |
| encoding="utf-8", | |
| ) | |
| clip = ClipInfo( | |
| move_id="clip", | |
| duration_sec=1.0, | |
| has_audio=False, | |
| description="", | |
| json_path=json_path, | |
| repo_id=LOCAL_REPO_ID, | |
| ) | |
| monkeypatch.setattr(module, "find_clip_info", lambda move_id, repo_id=None: clip) | |
| monkeypatch.setattr(module, "sync_play_clip", lambda *args, **kwargs: {"ok": True}) | |
| mm = MagicMock() | |
| mm.clear_move_queue = MagicMock() | |
| mm.enter_exclusive_playback.return_value = True | |
| mm.exit_exclusive_playback.return_value = True | |
| hw = MagicMock() | |
| hw.reset.side_effect = RuntimeError("boom") | |
| deps = MagicMock() | |
| deps.movement_manager = mm | |
| deps.head_wobbler = hw | |
| deps.reachy_mini = MagicMock() | |
| out = asyncio.run(PlayPerformanceClip()(deps, move_id="clip")) | |
| assert out["ok"] is False | |
| assert "boom" in out["error"] | |
| assert mm.clear_move_queue.called | |
| assert mm.enter_exclusive_playback.called | |
| assert mm.exit_exclusive_playback.called | |
| assert hw.set_performance_mode.call_args_list == [call(True), call(False)] | |
| def test_play_performance_clip_uses_resolved_audio_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| from reachy_dance_tutor_app.tools.play_performance_clip import PlayPerformanceClip | |
| import reachy_dance_tutor_app.tools.play_performance_clip as module | |
| json_path = tmp_path / "dance.json" | |
| json_path.write_text( | |
| '{"time":[0,1],"set_target_data":[{"head":[[1,0,0,0],[0,1,0,0],[0,0,1,0],[0,0,0,1]],"antennas":[0,0],"body_yaw":0}]}', | |
| encoding="utf-8", | |
| ) | |
| audio_path = tmp_path / "golden.wav" | |
| audio_path.touch() | |
| clip = ClipInfo( | |
| move_id="golden_kpop_demon_hunters", | |
| duration_sec=1.0, | |
| has_audio=True, | |
| description="Golden", | |
| json_path=json_path, | |
| repo_id=LOCAL_REPO_ID, | |
| source_kind="local", | |
| audio_path=audio_path, | |
| ) | |
| captured: dict[str, object] = {} | |
| def fake_sync_play_clip(robot: object, resolved_json_path: Path, cancel_ev: object, *, sound_path: Path | None = None) -> dict[str, object]: | |
| captured["json_path"] = resolved_json_path | |
| captured["sound_path"] = sound_path | |
| return {"ok": True} | |
| monkeypatch.setattr(module, "find_clip_info", lambda move_id, repo_id=None: clip) | |
| monkeypatch.setattr(module, "sync_play_clip", fake_sync_play_clip) | |
| mm = MagicMock() | |
| mm.clear_move_queue = MagicMock() | |
| mm.enter_exclusive_playback.return_value = True | |
| mm.exit_exclusive_playback.return_value = True | |
| deps = MagicMock() | |
| deps.movement_manager = mm | |
| deps.head_wobbler = None | |
| deps.reachy_mini = MagicMock() | |
| out = asyncio.run(PlayPerformanceClip()(deps, move_id="golden_kpop_demon_hunters", repo_id=LOCAL_REPO_ID)) | |
| assert out["ok"] is True | |
| assert captured["json_path"] == json_path | |
| assert captured["sound_path"] == audio_path | |
| def test_move_head_uses_body_yaw_from_head_joints(monkeypatch: pytest.MonkeyPatch) -> None: | |
| import sys | |
| import types | |
| fake_pkg = types.ModuleType("reachy_mini_dances_library") | |
| fake_pkg.__path__ = [] | |
| fake_dance_move = types.ModuleType("reachy_mini_dances_library.dance_move") | |
| class FakeDanceMove: | |
| def __init__(self, move_name: str) -> None: | |
| self.duration = 1.0 | |
| def evaluate(self, t: float) -> tuple[np.ndarray, np.ndarray, float]: | |
| return np.eye(4, dtype=np.float64), np.array([0.0, 0.0], dtype=np.float64), 0.0 | |
| fake_dance_move.DanceMove = FakeDanceMove | |
| monkeypatch.setitem(sys.modules, "reachy_mini_dances_library", fake_pkg) | |
| monkeypatch.setitem(sys.modules, "reachy_mini_dances_library.dance_move", fake_dance_move) | |
| from reachy_dance_tutor_app.tools.move_head import MoveHead | |
| import reachy_dance_tutor_app.tools.move_head as module | |
| captured: dict[str, object] = {} | |
| class FakeGotoQueueMove: | |
| def __init__(self, **kwargs: object) -> None: | |
| captured.update(kwargs) | |
| monkeypatch.setattr(module, "GotoQueueMove", FakeGotoQueueMove) | |
| mm = MagicMock() | |
| deps = MagicMock() | |
| deps.movement_manager = mm | |
| deps.motion_duration_s = 1.5 | |
| deps.reachy_mini.get_current_head_pose.return_value = np.eye(4, dtype=np.float64) | |
| deps.reachy_mini.get_current_joint_positions.return_value = ( | |
| [1.23, 0.0, 0.0], | |
| np.array([0.11, 0.22], dtype=np.float64), | |
| ) | |
| out = asyncio.run(MoveHead()(deps, direction="left")) | |
| assert out == {"status": "looking left"} | |
| assert captured["start_body_yaw"] == 1.23 | |
| assert captured["start_antennas"] == (0.11, 0.22) | |