"""Work-plane calibration endpoints: fit (preview), save, get, delete.""" from __future__ import annotations import logging 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 log = logging.getLogger(__name__) 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(), ) # Fetch the FULL-frame point cloud so the user can see the whole scene in # 3D (not just the ROI), with each point colored by signed distance to # the fitted plane. try: with sw.step("pointcloud_full"): pc_full, _ = nuwa.pointcloud_with_meta(step=4, max_depth_mm=4000) except Exception as e: log.warning("Nuwa full pointcloud failed: %s — falling back to ROI cloud", e) pc_full = pc scene_cv = nuwa_to_cv_frame(pc_full[:, :3]) if len(pc_full) else np.empty((0, 3)) if len(scene_cv) > req.max_preview_points: idx = np.random.choice(len(scene_cv), req.max_preview_points, replace=False) scene_cv = scene_cv[idx] return { "plane": asdict(wp), "preview": True, "points_camera": scene_cv.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}