feat(api): /api/work_plane/{fit,save,get,delete} endpoints

POST /fit returns a candidate WorkPlane + a downsampled CV-frame cloud
preview, without writing to disk. POST /save persists the plane. GET
returns the current saved plane (or null). DELETE clears it.

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-11 17:00:24 +02:00
1 parent 2478e9201c
commit 565c3df2f4
2 files changed
+125

No files matched your search

+123
View File
@@ -0,0 +1,123 @@
"""Work-plane calibration endpoints: fit (preview), save, get, delete."""
from __future__ import annotations
import time
from dataclasses import asdict
from typing import Any
import numpy as np
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from ..perception import filter_outliers, nuwa_to_cv_frame
from ..perception.work_plane import (
WorkPlane,
WorkPlaneError,
fit_plane_svd,
get_work_plane,
set_work_plane,
)
from ..timing import Stopwatch
from . import _deps
router = APIRouter()
class FitReq(BaseModel):
u_min: int = Field(..., ge=0)
v_min: int = Field(..., ge=0)
u_max: int = Field(..., gt=0)
v_max: int = Field(..., gt=0)
offset_mm: float = Field(5.0, ge=0.0, le=200.0)
max_preview_points: int = Field(5000, ge=100, le=20000)
class SaveReq(BaseModel):
plane: dict[str, Any] # accept a serialised WorkPlane
@router.post("/fit")
def fit(req: FitReq) -> dict[str, Any]:
"""Compute a candidate work plane from a rectangle on the RGB image.
Does NOT persist. Returns the plane + a downsampled CV-frame cloud for
the GUI 3D preview.
"""
sw = Stopwatch()
nuwa = _deps.nuwa()
try:
with sw.step("pointcloud_roi"):
pc, _meta = nuwa.pointcloud_roi(
req.u_min, req.v_min, req.u_max, req.v_max,
step=1, max_depth_mm=4000,
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"Nuwa pointcloud_roi failed: {e}")
if len(pc) < 10:
raise HTTPException(
status_code=422,
detail=f"too few depth points in ROI: {len(pc)}",
)
pts_cv = nuwa_to_cv_frame(pc[:, :3])
with sw.step("filter_outliers"):
pts_cv = filter_outliers(pts_cv, n_std=2.0)
if len(pts_cv) < 10:
raise HTTPException(
status_code=422,
detail=f"too few inliers after outlier filter: {len(pts_cv)}",
)
try:
with sw.step("fit_plane_svd"):
normal, offset, centroid, rms_mm = fit_plane_svd(pts_cv)
except WorkPlaneError as e:
raise HTTPException(status_code=422, detail=str(e))
wp = WorkPlane(
normal=[float(x) for x in normal],
offset=float(offset),
centroid=[float(x) for x in centroid],
n_inliers=int(len(pts_cv)),
fit_rms_mm=float(rms_mm),
roi=[int(req.u_min), int(req.v_min), int(req.u_max), int(req.v_max)],
offset_mm=float(req.offset_mm),
created_at=time.time(),
)
# Downsample for preview.
if len(pts_cv) > req.max_preview_points:
idx = np.random.choice(len(pts_cv), req.max_preview_points, replace=False)
preview = pts_cv[idx]
else:
preview = pts_cv
return {
"plane": asdict(wp),
"preview": True,
"points_camera": preview.tolist(),
"timings_ms": sw.as_dict(),
}
@router.post("/save")
def save(req: SaveReq) -> dict[str, Any]:
"""Persist a previously-fitted work plane to disk."""
try:
d = dict(req.plane)
d.setdefault("offset_mm", 5.0)
wp = WorkPlane(**d)
except Exception as e:
raise HTTPException(status_code=422, detail=f"invalid plane payload: {e}")
set_work_plane(wp)
return {"ok": True, "plane": asdict(wp), "saved": True}
@router.get("")
def get_current() -> dict[str, Any]:
wp = get_work_plane()
return {"plane": asdict(wp) if wp is not None else None}
@router.delete("")
def delete_current() -> dict[str, Any]:
set_work_plane(None)
return {"ok": True, "plane": None}
+2
View File
@@ -19,6 +19,7 @@ from .api import (
posture_routes,
robot_routes,
wrist_routes,
work_plane_routes,
)
from .clients import G1StatusError
from .config import load_config
@@ -42,6 +43,7 @@ app.include_router(grasp_routes.router, prefix="/api/grasp", tags=["grasp"])
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(work_plane_routes.router, prefix="/api/work_plane", tags=["work_plane"])
WEB_DIR = Path(__file__).resolve().parents[2] / "web"
if WEB_DIR.exists():