feat(pose_3d): POST /poses/{name}/go
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
d124a7c896
commit
8d40ebaf89
2 files changed
+59
No files matched your search
@@ -48,6 +48,11 @@ class ApplyReq(BaseModel):
|
|||||||
precise: bool = True
|
precise: bool = True
|
||||||
|
|
||||||
|
|
||||||
|
class GoReq(BaseModel):
|
||||||
|
dry_run: bool = True
|
||||||
|
precise: bool = True
|
||||||
|
|
||||||
|
|
||||||
@router.get("/state")
|
@router.get("/state")
|
||||||
def get_state() -> dict[str, Any]:
|
def get_state() -> dict[str, Any]:
|
||||||
sw = Stopwatch()
|
sw = Stopwatch()
|
||||||
@@ -129,6 +134,33 @@ def put_pose(name: str, req: NamedPose) -> dict[str, Any]:
|
|||||||
return {"name": name, "pose": pose}
|
return {"name": name, "pose": pose}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/poses/{name}/go")
|
||||||
|
def post_go_pose(name: str, req: GoReq) -> dict[str, Any]:
|
||||||
|
_validate_name(name)
|
||||||
|
cfg = get_config().get("pose_3d", {}) or {}
|
||||||
|
poses = cfg.get("named_poses") or {}
|
||||||
|
pose = poses.get(name)
|
||||||
|
if pose is None:
|
||||||
|
raise HTTPException(status_code=404, detail=f"pose '{name}' non trovata")
|
||||||
|
sw = Stopwatch()
|
||||||
|
try:
|
||||||
|
with sw.step("wrist_absolute"):
|
||||||
|
res = _deps.g1(dry_run=req.dry_run).wrist_absolute(
|
||||||
|
side=pose["side"],
|
||||||
|
position=list(pose["position"]),
|
||||||
|
quaternion=list(pose["quaternion"]),
|
||||||
|
precise=req.precise,
|
||||||
|
)
|
||||||
|
except G1Unavailable as e:
|
||||||
|
raise HTTPException(status_code=502, detail=f"G1 unavailable: {e}")
|
||||||
|
return {
|
||||||
|
"name": name,
|
||||||
|
"result": res,
|
||||||
|
"applied": not req.dry_run,
|
||||||
|
"timings_ms": sw.as_dict(),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@router.delete("/poses/{name}")
|
@router.delete("/poses/{name}")
|
||||||
def delete_pose(name: str) -> dict[str, Any]:
|
def delete_pose(name: str) -> dict[str, Any]:
|
||||||
_validate_name(name)
|
_validate_name(name)
|
||||||
|
|||||||
@@ -160,3 +160,30 @@ def test_delete_pose_removes_it():
|
|||||||
def test_delete_pose_404_when_missing():
|
def test_delete_pose_404_when_missing():
|
||||||
r = client.delete("/api/pose_3d/poses/nonexistent")
|
r = client.delete("/api/pose_3d/poses/nonexistent")
|
||||||
assert r.status_code == 404
|
assert r.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
def test_go_pose_404_when_missing():
|
||||||
|
r = client.post("/api/pose_3d/poses/missing/go", json={"dry_run": True})
|
||||||
|
assert r.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
def test_go_pose_executes_via_g1_with_stored_values():
|
||||||
|
captured = {}
|
||||||
|
def _abs(**kw):
|
||||||
|
captured.update(kw)
|
||||||
|
return {"ok": True}
|
||||||
|
# salva una posa
|
||||||
|
client.put("/api/pose_3d/poses/test_go", json={
|
||||||
|
"side": "left",
|
||||||
|
"position": [0.11, 0.22, 0.33],
|
||||||
|
"quaternion": [0, 0, 0, 1],
|
||||||
|
})
|
||||||
|
with patch("presa_mano.api.pose_3d_routes._deps.g1") as g1_factory:
|
||||||
|
g1 = MagicMock(); g1.wrist_absolute.side_effect = _abs
|
||||||
|
g1_factory.return_value = g1
|
||||||
|
r = client.post("/api/pose_3d/poses/test_go/go",
|
||||||
|
json={"dry_run": False, "precise": True})
|
||||||
|
assert r.status_code == 200
|
||||||
|
assert captured["side"] == "left"
|
||||||
|
assert captured["position"] == [0.11, 0.22, 0.33]
|
||||||
|
assert captured["precise"] is True
|
||||||
Reference in new issue
Block a user