diff --git a/src/presa_mano/api/right_arm_routes.py b/src/presa_mano/api/right_arm_routes.py new file mode 100644 index 0000000..2aac8b8 --- /dev/null +++ b/src/presa_mano/api/right_arm_routes.py @@ -0,0 +1,56 @@ +"""Endpoint per la tab Braccio DX (G1, polso destro, Nuwa eye-in-hand).""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException + +from ..clients import G1Unavailable +from ..config import update_config +from ..geometry import quaternion_xyzw_to_rpy_deg +from ..timing import Stopwatch +from . import _deps + +router = APIRouter() + + +def _state_payload(sw: Stopwatch) -> dict[str, Any]: + cfg = _deps.tab_right_arm_cfg() + with sw.step("g1_wrist"): + wrist = _deps.g1().wrist()["right_wrist"] + pos = wrist["position"] + quat = wrist["quaternion"] + rpy = quaternion_xyzw_to_rpy_deg(quat) + return { + "wrist": {"position": pos, "quaternion": quat, + "rpy_deg": [rpy[0], rpy[1], rpy[2]]}, + "robot_active": _deps.active_robot_key(), + "side": _deps.default_side(), + "home_pose": cfg.get("home_pose"), + "custom_pose": cfg.get("custom_pose"), + "offset_camera_mm": cfg.get("move_offset_camera_mm"), + "axis_signs": cfg.get("axis_sign_cam_to_wrist"), + "max_step_m": cfg.get("max_step_m"), + "z_floor_m": cfg.get("z_floor_m"), + "timings_ms": sw.as_dict(), + } + + +@router.get("/state") +def get_state() -> dict[str, Any]: + sw = Stopwatch() + try: + return _state_payload(sw) + except G1Unavailable as e: + raise HTTPException(status_code=502, detail=str(e)) + + +@router.post("/activate") +def post_activate() -> dict[str, Any]: + sw = Stopwatch() + with sw.step("config_update"): + update_config({"robot": {"active": "g1", "wrist_side": "right"}}) + try: + return _state_payload(sw) + except G1Unavailable as e: + raise HTTPException(status_code=502, detail=str(e)) diff --git a/src/presa_mano/main.py b/src/presa_mano/main.py index f11a1f5..9cf08a6 100644 --- a/src/presa_mano/main.py +++ b/src/presa_mano/main.py @@ -18,6 +18,7 @@ from .api import ( models_routes, perception_routes, posture_routes, + right_arm_routes, robot_routes, wrist_routes, work_plane_routes, @@ -45,6 +46,7 @@ app.include_router(models_routes.router, prefix="/api/models", tags=["models"]) app.include_router(wrist_routes.router, prefix="/api/wrist", tags=["wrist"]) app.include_router(posture_routes.router, prefix="/api/posture", tags=["posture"]) app.include_router(robot_routes.router, prefix="/api/robot", tags=["robot"]) +app.include_router(right_arm_routes.router, prefix="/api/right_arm", tags=["right_arm"]) app.include_router(work_plane_routes.router, prefix="/api/work_plane", tags=["work_plane"]) WEB_DIR = Path(__file__).resolve().parents[2] / "web" diff --git a/tests/test_right_arm_routes.py b/tests/test_right_arm_routes.py new file mode 100644 index 0000000..61c94f9 --- /dev/null +++ b/tests/test_right_arm_routes.py @@ -0,0 +1,62 @@ +"""Integration test del router /api/right_arm/*.""" +from __future__ import annotations + +from unittest.mock import patch, MagicMock + +from fastapi.testclient import TestClient + +from presa_mano.main import app + +client = TestClient(app) + + +def _fake_wrist_response(): + return { + "left_wrist": {"position": [0.10, 0.20, 0.30], "quaternion": [0, 0, 0, 1]}, + "right_wrist": {"position": [0.30, -0.20, 0.30], "quaternion": [0, 0, 0, 1]}, + } + + +def test_state_returns_wrist_pose_and_config(): + with patch("presa_mano.api.right_arm_routes._deps.g1") as g1_factory: + g1 = MagicMock() + g1.wrist.return_value = _fake_wrist_response() + g1_factory.return_value = g1 + + r = client.get("/api/right_arm/state") + assert r.status_code == 200 + body = r.json() + assert body["wrist"]["position"] == [0.30, -0.20, 0.30] + assert "rpy_deg" in body["wrist"] + assert body["side"] == "right" or body["side"] in ("left", "right") + assert "home_pose" in body + assert "custom_pose" in body + assert "offset_camera_mm" in body + assert "axis_signs" in body + assert "max_step_m" in body + assert "z_floor_m" in body + assert "robot_active" in body + assert "timings_ms" in body + + +def test_state_502_when_g1_unavailable(): + from presa_mano.clients import G1Unavailable + with patch("presa_mano.api.right_arm_routes._deps.g1") as g1_factory: + g1 = MagicMock() + g1.wrist.side_effect = G1Unavailable("connection refused") + g1_factory.return_value = g1 + r = client.get("/api/right_arm/state") + assert r.status_code == 502 + + +def test_activate_sets_robot_active_g1_and_side_right(): + # /activate รจ idempotente: chiama PUT /api/config e ritorna /state + with patch("presa_mano.api.right_arm_routes._deps.g1") as g1_factory: + g1 = MagicMock() + g1.wrist.return_value = _fake_wrist_response() + g1_factory.return_value = g1 + r = client.post("/api/right_arm/activate") + assert r.status_code == 200 + body = r.json() + assert body["robot_active"] == "g1" + assert body["side"] == "right"