feat(pose_3d): named poses GET/PUT/DELETE
Adds CRUD for named wrist poses persisted in config.yaml under
pose_3d.named_poses. PUT validates the name against ^[A-Za-z0-9_-]{1,40}$
and normalizes the quaternion. DELETE returns 404 if the pose is missing.
Bypasses update_config's deep-merge for wholesale named_poses replacement
(deep-merge cannot remove nested keys) via a small helper that mutates
the in-memory singleton then persists with an empty patch.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
df949c6989
commit
fb05409403
2 files changed
+110
-1
No files matched your search
@@ -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}
|
||||
Reference in new issue
Block a user