feat(right_arm): router skeleton con /state e /activate

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal PastroandClaude Opus 4.7 committed 2026-05-20 12:13:33 +02:00
1 parent 55e84778d8
commit dd2053cc47
3 files changed
+120

No files matched your search

+56
View File
@@ -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))
+2
View File
@@ -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"
+62
View File
@@ -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"