diff --git a/src/presa_mano/api/pose_3d_routes.py b/src/presa_mano/api/pose_3d_routes.py index f8a7651..5f41c66 100644 --- a/src/presa_mano/api/pose_3d_routes.py +++ b/src/presa_mano/api/pose_3d_routes.py @@ -2,13 +2,15 @@ from __future__ import annotations import logging +import re from typing import Any, Literal, Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field +from .. import config as _config_mod from ..clients import G1Unavailable -from ..config import get_config +from ..config import get_config, update_config from ..geometry import normalize_quaternion from ..timing import Stopwatch from . import _deps @@ -16,6 +18,31 @@ from . import _deps router = APIRouter() log = logging.getLogger(__name__) +_NAME_RE = re.compile(r"^[A-Za-z0-9_-]{1,40}$") + + +def _validate_name(name: str) -> None: + if not _NAME_RE.match(name): + raise HTTPException( + status_code=422, + detail=f"name must match {_NAME_RE.pattern}", + ) + + +def _replace_named_poses(poses: dict[str, Any]) -> None: + """Wholesale replace pose_3d.named_poses. + update_config does a recursive deep-merge so a dict patch can't remove + keys from a nested dict. We bypass it by mutating the in-memory state + directly, then call update_config with an empty patch to persist.""" + with _config_mod._LOCK: + if _config_mod._STATE["config"] is None: + _config_mod.load_config() + cfg = _config_mod._STATE["config"] + if "pose_3d" not in cfg or not isinstance(cfg["pose_3d"], dict): + cfg["pose_3d"] = {} + cfg["pose_3d"]["named_poses"] = poses + update_config({}) # persist to disk + class WristPose(BaseModel): position: list[float] @@ -89,3 +116,42 @@ def post_apply(req: ApplyReq) -> dict[str, Any]: "quaternion": quat_norm, "timings_ms": sw.as_dict(), } + + +@router.get("/poses") +def get_poses() -> dict[str, Any]: + cfg = get_config().get("pose_3d", {}) or {} + return cfg.get("named_poses") or {} + + +@router.put("/poses/{name}") +def put_pose(name: str, req: NamedPose) -> dict[str, Any]: + _validate_name(name) + try: + quat_norm = normalize_quaternion(req.quaternion) + except ValueError as e: + raise HTTPException(status_code=422, detail=str(e)) + pose: dict[str, Any] = { + "side": req.side, + "position": list(req.position), + "quaternion": quat_norm, + } + if req.note is not None: + pose["note"] = req.note + cfg = get_config().get("pose_3d", {}) or {} + poses = dict(cfg.get("named_poses") or {}) + poses[name] = pose + _replace_named_poses(poses) + return {"name": name, "pose": pose} + + +@router.delete("/poses/{name}") +def delete_pose(name: str) -> dict[str, Any]: + _validate_name(name) + cfg = get_config().get("pose_3d", {}) or {} + poses = dict(cfg.get("named_poses") or {}) + if name not in poses: + raise HTTPException(status_code=404, detail=f"pose '{name}' non trovata") + del poses[name] + _replace_named_poses(poses) + return {"deleted": name} diff --git a/tests/test_pose_3d_routes.py b/tests/test_pose_3d_routes.py index 0f89c34..d3f01ab 100644 --- a/tests/test_pose_3d_routes.py +++ b/tests/test_pose_3d_routes.py @@ -117,3 +117,46 @@ def test_apply_normalizes_quaternion_before_forwarding(): q = captured["quaternion"] norm = sum(x*x for x in q) ** 0.5 assert abs(norm - 1.0) < 1e-6 + + +def test_put_pose_persists_in_config(): + payload = { + "side": "right", + "position": [0.30, -0.20, 0.30], + "quaternion": [0, 0, 0, 1], + "note": "test pose", + } + r = client.put("/api/pose_3d/poses/test1", json=payload) + assert r.status_code == 200 + body = r.json() + assert body["pose"]["side"] == "right" + assert body["pose"]["note"] == "test pose" + # Verifica che sia anche in GET /poses + r2 = client.get("/api/pose_3d/poses") + assert "test1" in r2.json() + + +def test_put_pose_422_bad_name(): + r = client.put("/api/pose_3d/poses/with spaces!", json={ + "side": "right", "position": [0,0,0], "quaternion": [0,0,0,1], + }) + assert r.status_code == 422 + assert "name" in r.json()["detail"].lower() + + +def test_delete_pose_removes_it(): + # prima salva + client.put("/api/pose_3d/poses/todelete", json={ + "side": "left", "position": [0,0,0], "quaternion": [0,0,0,1], + }) + # poi cancella + r = client.delete("/api/pose_3d/poses/todelete") + assert r.status_code == 200 + # verifica non presente + r2 = client.get("/api/pose_3d/poses") + assert "todelete" not in r2.json() + + +def test_delete_pose_404_when_missing(): + r = client.delete("/api/pose_3d/poses/nonexistent") + assert r.status_code == 404