Adriano/Presa-Mano
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Presa-Mano/docs/superpowers/plans/2026-05-11-find-model.md
T
Adriano Dal PastroandClaude Opus 4.7 edbd5b5136 docs: implementation plan for Cerca modello 3D pose-fit overlay 
13 bite-sized TDD tasks covering: numpy-stl dep + config, partition_by_plane,
stl_sampler, projection, Open3DModel client + LRU mesh cache, sam21_bridge
refactor, _deps factories, GET /api/models/recipes proxy, POST
/api/grasp/find_model handler, GUI rebuild, manual integration smoke test.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 18:42:33 +02:00

2177 lines
74 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Cerca modello: 3D pose-fit overlay — Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Aggiungere a presa-mano un endpoint `POST /api/grasp/find_model` che integra `Open3DModel /api/pose` per il fit 6-DoF di un modello 3D sul cluster SAM21, e ridisegnare il tab "Cerca modello" del GUI con un picker ricetta + overlay risultato (punti modello blu + punti accettati verdi + punti esclusi rossi togglabili).
**Architecture:** Riusa la pipeline YOLO+SAM21 esistente (refactorizzata per esporre i punti pre-plane-filter e l'RGB snapshot). Aggiunge un client HTTP per Open3DModel con cache LRU del mesh STL, un sampler per il mesh, e un proiettore camera→pixel. Il fit vive interamente in frame camera OpenCV — nessuna trasformazione in base frame. Il GUI fa una sola chiamata, riceve RGB base64 + 3 liste di pixel, e disegna con toggle client-side.
**Tech Stack:** Python 3.10+ (FastAPI, httpx, numpy, opencv-python), `numpy-stl` (nuova dep, ~50KB), vanilla JS + Canvas 2D.
**Spec di riferimento:** `docs/superpowers/specs/2026-05-11-find-model-design.md`.
---
## File structure
**Nuovi (backend):**
- `src/presa_mano/clients/open3dmodel.py` — HTTP client per Open3DModel
- `src/presa_mano/clients/open3dmodel_cache.py` — LRU mesh cache
- `src/presa_mano/perception/stl_sampler.py` — parse STL + sample surface
- `src/presa_mano/perception/projection.py` — proiezione camera→pixel
- `src/presa_mano/api/models_routes.py` — proxy `GET /api/models/recipes`
**Modificati (backend):**
- `src/presa_mano/perception/work_plane.py` — aggiunta `partition_by_plane`
- `src/presa_mano/perception/sam21_bridge.py` — refactor: estrae `segment_with_sam21` + `Sam21Segmentation`
- `src/presa_mano/clients/__init__.py` — export Open3DModel*
- `src/presa_mano/api/_deps.py` — factory `open3dmodel()` + singleton `recipe_cache()`
- `src/presa_mano/api/grasp_routes.py` — nuovo handler `find_model`
- `src/presa_mano/main.py` — mount `models_routes`
- `config.yaml`, `.env.example`, `pyproject.toml`
**Modificati (frontend):**
- `web/index.html` — rimuove sezioni vecchie, aggiunge picker + canvas risultato
- `web/app.js` — rimuove handler `planBtn`/`executeBtn` + crop/scatters, aggiunge blocco "fit modello"
**Test nuovi:**
- `tests/test_partition_plane.py`
- `tests/test_stl_sampler.py`
- `tests/test_projection.py`
- `tests/test_open3dmodel_client.py`
- `tests/test_recipe_mesh_cache.py`
- `tests/test_sam21_bridge_segment.py`
---
## Task 1: Setup dipendenza + config
**Files:**
- Modify: `pyproject.toml:6-15`
- Modify: `config.yaml:1-22` (sezione `services`) e fine file (sezione `grasp`)
- Modify: `.env.example`
- [ ] **Step 1: Aggiungi numpy-stl a pyproject.toml**
```toml
dependencies = [
"fastapi>=0.110.0",
"uvicorn[standard]>=0.27.0",
"httpx>=0.27.0",
"numpy>=1.24.0",
"numpy-stl>=3.1.0",
"opencv-contrib-python-headless>=4.9.0.80",
"pyyaml>=6.0",
"pydantic>=2.5.0",
"python-multipart>=0.0.9",
]
```
- [ ] **Step 2: Aggiorna lockfile**
Run: `uv sync`
Expected: `Resolved … Installed numpy-stl …` senza errori.
- [ ] **Step 3: Aggiungi `services.open3dmodel` in config.yaml**
Sotto `services.sam21:` aggiungi:
```yaml
open3dmodel:
base_url: http://localhost:8878
timeout_s: 30.0
api_key_env: OPEN3D_API_KEY
max_stl_bytes: 20000000
```
- [ ] **Step 4: Aggiungi `grasp.find_model` in config.yaml**
Dentro la sezione `grasp:` esistente, aggiungi:
```yaml
find_model:
sample_points: 2000
min_fitness: 0.5
max_rmse_mm: 5.0
```
- [ ] **Step 5: Aggiorna .env.example**
Aggiungi in fondo al file:
```
# Chiave API per Open3DModel (pose-fit 6DOF su STL).
# Generala con la procedura del README di Open3DModel; deve combaciare
# con OPEN3D_API_KEY in ./Open3DModel/.env.
OPEN3D_API_KEY=
```
- [ ] **Step 6: Verifica import-time**
Run: `uv run python -c "import stl; print(stl.__version__)"`
Expected: stampa una versione (es. `3.1.x`).
- [ ] **Step 7: Commit**
```bash
git add pyproject.toml uv.lock config.yaml .env.example
git commit -m "build: add numpy-stl + Open3DModel config block for find_model
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 2: `partition_by_plane` in perception/work_plane.py
**Files:**
- Test: `tests/test_partition_plane.py`
- Modify: `src/presa_mano/perception/work_plane.py:125-146`
- [ ] **Step 1: Scrivi il test che fallisce**
Crea `tests/test_partition_plane.py`:
```python
"""Unit tests for partition_by_plane."""
from __future__ import annotations
import numpy as np
import pytest
from presa_mano.perception.work_plane import WorkPlane, partition_by_plane
def _wp(offset_mm: float = 0.0) -> WorkPlane:
"""Plane z=0 with normal pointing toward camera (-Z)."""
return WorkPlane(
normal=[0.0, 0.0, -1.0],
offset=0.0,
centroid=[0.0, 0.0, 0.0],
n_inliers=100,
fit_rms_mm=0.5,
roi=[0, 0, 100, 100],
offset_mm=offset_mm,
created_at=0.0,
)
def test_partition_no_plane_returns_all_accepted():
pts = np.array([[0.0, 0.0, 1.0], [1.0, 2.0, 3.0]])
accepted, excluded, info = partition_by_plane(pts, None)
assert np.array_equal(accepted, pts)
assert excluded.shape == (0, 3)
assert info == {"plane_applied": False, "n_points_before_plane": 2}
def test_partition_empty_input():
pts = np.empty((0, 3), dtype=np.float64)
accepted, excluded, info = partition_by_plane(pts, _wp())
assert accepted.shape == (0, 3)
assert excluded.shape == (0, 3)
assert info["plane_applied"] is False
assert info["n_points_before_plane"] == 0
def test_partition_splits_by_offset():
"""Plane at z=0, normal=-Z, offset_mm=5. With OpenCV sign convention
(camera looks +Z, surfaces in front have z>0), points with z<-0.005 m are
below the plane and get excluded."""
pts = np.array([
[0.0, 0.0, 0.10], # 100mm sopra → accepted
[0.0, 0.0, 0.00], # sul piano → accepted (boundary)
[0.0, 0.0, -0.004], # 4mm sotto → accepted (entro offset 5mm)
[0.0, 0.0, -0.010], # 10mm sotto → excluded
[0.0, 0.0, -0.500], # 500mm sotto → excluded
])
accepted, excluded, info = partition_by_plane(pts, _wp(offset_mm=5.0))
assert accepted.shape[0] == 3
assert excluded.shape[0] == 2
assert info["plane_applied"] is True
assert info["n_points_before_plane"] == 5
# partition: nessuna sovrapposizione
union = np.vstack([accepted, excluded])
assert sorted(union[:, 2].tolist()) == sorted(pts[:, 2].tolist())
def test_partition_consistent_with_apply_below_plane_filter():
"""`accepted` di partition_by_plane deve essere bit-identico a quello
ritornato da apply_below_plane_filter (l'API legacy non deve cambiare)."""
from presa_mano.perception.work_plane import apply_below_plane_filter
rng = np.random.default_rng(42)
pts = rng.normal(scale=0.1, size=(200, 3))
wp = _wp(offset_mm=10.0)
accepted_new, _excluded, _info = partition_by_plane(pts, wp)
accepted_legacy, info_legacy = apply_below_plane_filter(pts, wp)
np.testing.assert_array_equal(accepted_new, accepted_legacy)
assert info_legacy["n_points_before_plane"] == 200
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_partition_plane.py -v`
Expected: `ImportError: cannot import name 'partition_by_plane'` o `AttributeError`.
- [ ] **Step 3: Implementa `partition_by_plane` e refactor `apply_below_plane_filter`**
Sostituisci il blocco `apply_below_plane_filter` (linee 125-146) in `src/presa_mano/perception/work_plane.py` con:
```python
def partition_by_plane(
pts_cv: np.ndarray,
wp: WorkPlane | None,
) -> tuple[np.ndarray, np.ndarray, dict]:
"""Split points into (accepted_above_plane, excluded_below_plane).
Same threshold as `apply_below_plane_filter`: keep iff
`signed >= -offset_mm/1000`. When `wp is None` or `pts_cv` is empty,
`excluded` is empty and `plane_applied=False`.
"""
info = {"plane_applied": False,
"n_points_before_plane": int(len(pts_cv))}
empty = np.empty((0, 3), dtype=pts_cv.dtype if pts_cv.size else np.float64)
if wp is None or len(pts_cv) == 0:
return pts_cv, empty, info
n_vec = np.asarray(wp.normal, dtype=np.float64)
d_off = float(wp.offset)
offset_m = float(wp.offset_mm) / 1000.0
signed = pts_cv @ n_vec + d_off
keep = signed >= -offset_m
info["plane_applied"] = True
return pts_cv[keep], pts_cv[~keep], info
def apply_below_plane_filter(
pts_cv: np.ndarray,
wp: WorkPlane | None,
) -> tuple[np.ndarray, dict]:
"""Thin wrapper around partition_by_plane that drops the excluded set.
Kept for backwards compatibility with existing callers."""
accepted, _excluded, info = partition_by_plane(pts_cv, wp)
return accepted, info
```
- [ ] **Step 4: Esegui i nuovi test**
Run: `uv run pytest tests/test_partition_plane.py -v`
Expected: 4 passed.
- [ ] **Step 5: Esegui i test esistenti per assicurarsi che non rompiamo nulla**
Run: `uv run pytest tests/test_work_plane.py -v`
Expected: tutti i test esistenti passano.
- [ ] **Step 6: Commit**
```bash
git add tests/test_partition_plane.py src/presa_mano/perception/work_plane.py
git commit -m "feat(perception): partition_by_plane returns accepted + excluded sets
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 3: `perception/stl_sampler.py`
**Files:**
- Create: `src/presa_mano/perception/stl_sampler.py`
- Test: `tests/test_stl_sampler.py`
- [ ] **Step 1: Scrivi i test che falliscono**
Crea `tests/test_stl_sampler.py`:
```python
"""Unit tests for stl_sampler."""
from __future__ import annotations
import io
import struct
import numpy as np
import pytest
from presa_mano.perception.stl_sampler import parse_stl_bytes, sample_surface
def _make_cube_stl_bytes(side_mm: float = 10.0) -> bytes:
"""Synthesize a binary STL of an axis-aligned cube centered at origin,
from -side/2 to +side/2 on each axis. 12 triangles, no normals needed."""
s = side_mm / 2.0
v = [
(-s, -s, -s), ( s, -s, -s), ( s, s, -s), (-s, s, -s),
(-s, -s, s), ( s, -s, s), ( s, s, s), (-s, s, s),
]
# 6 faces × 2 tris (vertex order: any orientation works for our use)
tris = [
(0, 1, 2), (0, 2, 3), # z=-s
(4, 6, 5), (4, 7, 6), # z=+s
(0, 4, 5), (0, 5, 1), # y=-s
(3, 2, 6), (3, 6, 7), # y=+s
(0, 3, 7), (0, 7, 4), # x=-s
(1, 5, 6), (1, 6, 2), # x=+s
]
buf = io.BytesIO()
buf.write(b"\x00" * 80) # header
buf.write(struct.pack("<I", len(tris))) # n_triangles
for (i, j, k) in tris:
buf.write(struct.pack("<fff", 0.0, 0.0, 0.0)) # normal (unused)
for idx in (i, j, k):
buf.write(struct.pack("<fff", *v[idx]))
buf.write(struct.pack("<H", 0)) # attribute byte count
return buf.getvalue()
def test_parse_stl_cube_bbox():
stl_bytes = _make_cube_stl_bytes(side_mm=10.0)
verts, tris = parse_stl_bytes(stl_bytes)
assert verts.ndim == 2 and verts.shape[1] == 3
assert tris.ndim == 2 and tris.shape[1] == 3
assert tris.dtype.kind == "i"
# BB del cubo: [-5, +5] su ogni asse
np.testing.assert_allclose(verts.min(axis=0), [-5.0, -5.0, -5.0], atol=1e-6)
np.testing.assert_allclose(verts.max(axis=0), [ 5.0, 5.0, 5.0], atol=1e-6)
def test_sample_surface_in_bbox():
stl_bytes = _make_cube_stl_bytes(side_mm=10.0)
verts, tris = parse_stl_bytes(stl_bytes)
pts = sample_surface(verts, tris, n=1000, seed=0)
assert pts.shape == (1000, 3)
# tutti i punti devono cadere dentro la BB del cubo (più epsilon
# per il rumore di floating point bary)
np.testing.assert_array_less(pts.max(axis=0), 5.0 + 1e-4)
np.testing.assert_array_less(-pts.min(axis=0), 5.0 + 1e-4)
def test_sample_surface_distribution_uniform():
"""8 octanti del cubo, ognuno ≥ 8% dei punti (perfetto = 100/6 ≈ 16.7%
per faccia, distribuita su 4 octanti adiacenti)."""
stl_bytes = _make_cube_stl_bytes(side_mm=10.0)
verts, tris = parse_stl_bytes(stl_bytes)
pts = sample_surface(verts, tris, n=2000, seed=1)
counts = np.zeros(8, dtype=np.int64)
for p in pts:
# bin (x>0, y>0, z>0) → 0..7
idx = (int(p[0] > 0) << 2) | (int(p[1] > 0) << 1) | int(p[2] > 0)
counts[idx] += 1
frac = counts / pts.shape[0]
assert (frac > 0.08).all(), f"octant fractions too uneven: {frac}"
def test_sample_surface_seed_deterministic():
stl_bytes = _make_cube_stl_bytes(side_mm=10.0)
verts, tris = parse_stl_bytes(stl_bytes)
a = sample_surface(verts, tris, n=100, seed=42)
b = sample_surface(verts, tris, n=100, seed=42)
np.testing.assert_array_equal(a, b)
def test_parse_stl_bad_payload_raises():
with pytest.raises(Exception): # numpy-stl raises various; we just want it to not silently pass
parse_stl_bytes(b"not a real STL")
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_stl_sampler.py -v`
Expected: `ImportError: No module named 'presa_mano.perception.stl_sampler'`.
- [ ] **Step 3: Implementa `stl_sampler.py`**
Crea `src/presa_mano/perception/stl_sampler.py`:
```python
"""Parse binary/ASCII STL bytes and uniformly sample the surface.
Output unit follows the STL convention (no unit conversion — STL files for
robotics are by convention in millimetres, but we don't enforce it here).
"""
from __future__ import annotations
import io
import numpy as np
from stl import mesh as stl_mesh
def parse_stl_bytes(stl_bytes: bytes) -> tuple[np.ndarray, np.ndarray]:
"""Parse STL bytes (binary or ASCII) → (vertices (M,3) float64,
triangles (T,3) int32).
Vertices are deduplicated so the same point used by multiple triangles
appears once. Raises numpy_stl exceptions on malformed input.
"""
m = stl_mesh.Mesh.from_file(filename="dummy.stl", fh=io.BytesIO(stl_bytes))
# m.vectors shape: (T, 3, 3) — three vertices per triangle, raw.
tris_raw = np.asarray(m.vectors, dtype=np.float64).reshape(-1, 3)
unique, inverse = np.unique(tris_raw, axis=0, return_inverse=True)
triangles = inverse.reshape(-1, 3).astype(np.int32)
return unique, triangles
def sample_surface(
vertices: np.ndarray,
triangles: np.ndarray,
n: int,
seed: int = 0,
) -> np.ndarray:
"""Sample n points uniformly on the mesh surface, weighted by triangle area.
Algorithm:
1. Compute area for each triangle.
2. Cumulative area → pick triangle index via inverse CDF.
3. For each pick, sample a uniform barycentric coordinate:
`(1-√r1) A + √r1 (1-r2) B + √r1 r2 C` with r1, r2 ~ U(0,1).
"""
if triangles.size == 0:
return np.empty((0, 3), dtype=np.float64)
A = vertices[triangles[:, 0]]
B = vertices[triangles[:, 1]]
C = vertices[triangles[:, 2]]
areas = 0.5 * np.linalg.norm(np.cross(B - A, C - A), axis=1)
total = areas.sum()
if total <= 0.0:
# degenerate mesh: fall back to vertex sampling
rng = np.random.default_rng(seed)
idx = rng.integers(0, vertices.shape[0], size=n)
return vertices[idx]
cdf = np.cumsum(areas) / total
rng = np.random.default_rng(seed)
u = rng.random(n)
tri_idx = np.searchsorted(cdf, u, side="right")
tri_idx = np.clip(tri_idx, 0, triangles.shape[0] - 1)
r1 = rng.random(n)
r2 = rng.random(n)
sqrt_r1 = np.sqrt(r1)
w_a = (1.0 - sqrt_r1)[:, None]
w_b = (sqrt_r1 * (1.0 - r2))[:, None]
w_c = (sqrt_r1 * r2)[:, None]
return (w_a * A[tri_idx] + w_b * B[tri_idx] + w_c * C[tri_idx]).astype(np.float64)
```
- [ ] **Step 4: Esegui i test**
Run: `uv run pytest tests/test_stl_sampler.py -v`
Expected: 5 passed.
- [ ] **Step 5: Commit**
```bash
git add tests/test_stl_sampler.py src/presa_mano/perception/stl_sampler.py
git commit -m "feat(perception): parse STL + uniform surface sampling
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 4: `perception/projection.py`
**Files:**
- Create: `src/presa_mano/perception/projection.py`
- Test: `tests/test_projection.py`
- [ ] **Step 1: Scrivi i test che falliscono**
Crea `tests/test_projection.py`:
```python
"""Unit tests for camera-to-pixel projection."""
from __future__ import annotations
import numpy as np
from presa_mano.perception.projection import K_from_intrinsics, project_camera_to_pixels
def _K():
return np.array([[600.0, 0.0, 320.0],
[ 0.0, 600.0, 240.0],
[ 0.0, 0.0, 1.0]])
def test_K_from_intrinsics_matches_manual():
intr = {"fx": 600.0, "fy": 600.0, "cx": 320.0, "cy": 240.0,
"width": 640, "height": 480}
K = K_from_intrinsics(intr)
np.testing.assert_array_equal(K, _K())
def test_project_principal_point():
"""Punto sull'asse ottico a 1m → cx,cy."""
pts = np.array([[0.0, 0.0, 1000.0]]) # mm
out = project_camera_to_pixels(pts, _K(), (640, 480))
assert out.shape == (1, 2)
assert out[0, 0] == 320 and out[0, 1] == 240
def test_project_behind_camera_dropped():
"""Z<=0 deve essere scartato."""
pts = np.array([[0.0, 0.0, -1000.0], [10.0, 20.0, 0.0]])
out = project_camera_to_pixels(pts, _K(), (640, 480))
assert out.shape == (0, 2)
def test_project_outside_frame_dropped():
"""Punto che proietta fuori bbox immagine viene scartato."""
pts = np.array([
[0.0, 0.0, 1000.0], # → (320, 240): inside
[10000.0, 0.0, 1000.0], # → u molto fuori bbox: dropped
[0.0, -10000.0, 1000.0], # → v molto fuori bbox (sopra): dropped
])
out = project_camera_to_pixels(pts, _K(), (640, 480))
assert out.shape == (1, 2)
assert out[0, 0] == 320 and out[0, 1] == 240
def test_project_known_offset():
"""Punto in (0.1, 0.0, 1.0) m → u = cx + fx*(0.1/1.0) = 320 + 60 = 380."""
pts = np.array([[100.0, 0.0, 1000.0]]) # mm
out = project_camera_to_pixels(pts, _K(), (640, 480))
assert out[0, 0] == 380 and out[0, 1] == 240
def test_project_unit_invariance():
"""Stesso punto in m vs mm proietta sullo stesso pixel (K@P/P.z elide unità)."""
pts_mm = np.array([[100.0, 0.0, 1000.0]])
pts_m = pts_mm / 1000.0
out_mm = project_camera_to_pixels(pts_mm, _K(), (640, 480))
out_m = project_camera_to_pixels(pts_m, _K(), (640, 480))
np.testing.assert_array_equal(out_mm, out_m)
def test_project_empty_input():
out = project_camera_to_pixels(np.empty((0, 3)), _K(), (640, 480))
assert out.shape == (0, 2)
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_projection.py -v`
Expected: `ImportError`.
- [ ] **Step 3: Implementa `projection.py`**
Crea `src/presa_mano/perception/projection.py`:
```python
"""Project 3D points (camera frame) to 2D pixels using OpenCV intrinsics."""
from __future__ import annotations
from typing import Any
import numpy as np
def K_from_intrinsics(intr: dict[str, Any]) -> np.ndarray:
"""Build a 3x3 K from a {fx,fy,cx,cy,width,height} dict.
Width/height are not used here but must be present so callers can pass
the same dict to `project_camera_to_pixels`."""
return np.array([
[float(intr["fx"]), 0.0, float(intr["cx"])],
[0.0, float(intr["fy"]), float(intr["cy"])],
[0.0, 0.0, 1.0],
], dtype=np.float64)
def project_camera_to_pixels(
points_cam: np.ndarray,
K: np.ndarray,
image_size: tuple[int, int],
) -> np.ndarray:
"""Project (N,3) points in OpenCV camera frame to (M,2) int pixels.
Drops points with Z <= 0 (behind the camera) and points that project
outside `image_size` (W,H). The unit of `points_cam` is free: K@P/P.z
cancels it as long as K is in pixels.
"""
if points_cam.size == 0:
return np.empty((0, 2), dtype=np.int32)
W, H = image_size
pts = np.asarray(points_cam, dtype=np.float64)
z = pts[:, 2]
keep_z = z > 0.0
if not keep_z.any():
return np.empty((0, 2), dtype=np.int32)
pts = pts[keep_z]
proj = (K @ pts.T).T # (M, 3)
uv = proj[:, :2] / proj[:, 2:3]
u = np.round(uv[:, 0]).astype(np.int32)
v = np.round(uv[:, 1]).astype(np.int32)
in_frame = (u >= 0) & (u < W) & (v >= 0) & (v < H)
return np.stack([u[in_frame], v[in_frame]], axis=1)
```
- [ ] **Step 4: Esegui i test**
Run: `uv run pytest tests/test_projection.py -v`
Expected: 7 passed.
- [ ] **Step 5: Commit**
```bash
git add tests/test_projection.py src/presa_mano/perception/projection.py
git commit -m "feat(perception): project camera-frame 3D points to image pixels
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 5: `clients/open3dmodel.py` — HTTP client
**Files:**
- Create: `src/presa_mano/clients/open3dmodel.py`
- Modify: `src/presa_mano/clients/__init__.py`
- Test: `tests/test_open3dmodel_client.py`
- [ ] **Step 1: Scrivi i test che falliscono**
Crea `tests/test_open3dmodel_client.py`:
```python
"""Unit tests for Open3DModelClient."""
from __future__ import annotations
from unittest.mock import patch
import httpx
import numpy as np
import pytest
from presa_mano.clients.open3dmodel import (
Open3DModelClient,
Open3DModelError,
Open3DModelUnavailable,
)
def _intrinsics():
return {"fx": 600.0, "fy": 600.0, "cx": 320.0, "cy": 240.0,
"width": 640, "height": 480}
def _ok_pose_response():
body = {
"success": True,
"data": {
"symmetry_type": "cylindrical",
"translation_mm": {"x": 1.0, "y": 2.0, "z": 500.0},
"transform_4x4": [
[1.0, 0.0, 0.0, 1.0],
[0.0, 1.0, 0.0, 2.0],
[0.0, 0.0, 1.0, 500.0],
[0.0, 0.0, 0.0, 1.0],
],
"fitness": 0.87,
"inlier_rmse_mm": 1.8,
},
"error": None,
}
return httpx.Response(200, json=body)
def _envelope_err(status: int, code: str, message: str):
body = {"success": False, "data": None,
"error": {"code": code, "message": message}}
return httpx.Response(status, json=body)
def test_list_recipes_happy_path():
body = [{"id": "r1", "name": "pestello", "symmetry_type": "cylindrical"}]
response = httpx.Response(200, json=body)
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.get", return_value=response) as mock_get:
recs = client.list_recipes()
assert recs == body
posted_url = mock_get.call_args.args[0]
assert posted_url.endswith("/api/recipes")
# API key header
headers = mock_get.call_args.kwargs.get("headers", {})
assert headers.get("X-Api-Key") == "K"
def test_list_recipes_connect_error_raises_unavailable():
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.get", side_effect=httpx.ConnectError("nope")):
with pytest.raises(Open3DModelUnavailable):
client.list_recipes()
def test_estimate_pose_happy_path():
pts_m = np.array([[0.01, 0.02, 0.5], [0.015, 0.022, 0.49]])
response = _ok_pose_response()
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.post", return_value=response) as mock_post:
data = client.estimate_pose("r1", pts_m, _intrinsics())
assert data["fitness"] == 0.87
assert data["inlier_rmse_mm"] == 1.8
body = mock_post.call_args.kwargs["json"]
assert body["recipe_id"] == "r1"
assert body["unit"] == "m"
assert body["point_cloud"][0] == [0.01, 0.02, 0.5]
headers = mock_post.call_args.kwargs.get("headers", {})
assert headers.get("X-Api-Key") == "K"
def test_estimate_pose_envelope_failure_recipe_not_found():
pts_m = np.array([[0.0, 0.0, 0.5]])
response = _envelope_err(200, "RECIPE_NOT_FOUND", "Ricetta ... non esiste")
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.post", return_value=response):
with pytest.raises(Open3DModelError) as exc:
client.estimate_pose("r-bogus", pts_m, _intrinsics())
assert exc.value.code == "RECIPE_NOT_FOUND"
def test_estimate_pose_envelope_failure_registration_failed():
pts_m = np.array([[0.0, 0.0, 0.5]])
response = _envelope_err(200, "REGISTRATION_FAILED", "low fitness")
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.post", return_value=response):
with pytest.raises(Open3DModelError) as exc:
client.estimate_pose("r1", pts_m, _intrinsics())
assert exc.value.code == "REGISTRATION_FAILED"
def test_estimate_pose_http_403_raises_open3dmodel_error_auth():
"""403 con detail testuale (HTTPException di FastAPI, NON envelope)."""
pts_m = np.array([[0.0, 0.0, 0.5]])
response = httpx.Response(403, json={"detail": "API key non valida"})
client = Open3DModelClient("http://o3d:8878", api_key="WRONG", timeout_s=2.0)
with patch("httpx.Client.post", return_value=response):
with pytest.raises(Open3DModelError) as exc:
client.estimate_pose("r1", pts_m, _intrinsics())
assert exc.value.code == "HTTP_403"
def test_estimate_pose_connect_error_raises_unavailable():
pts_m = np.array([[0.0, 0.0, 0.5]])
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.post", side_effect=httpx.ConnectError("down")):
with pytest.raises(Open3DModelUnavailable):
client.estimate_pose("r1", pts_m, _intrinsics())
def test_get_stl_bytes_happy_path():
response = httpx.Response(200, content=b"BINARYSTL", headers={"ETag": "abc123"})
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.get", return_value=response):
body, etag = client.get_stl_bytes("r1")
assert body == b"BINARYSTL"
assert etag == "abc123"
def test_get_stl_bytes_oversize_raises():
big = b"x" * (25 * 1024 * 1024) # 25 MB
response = httpx.Response(200, content=big)
client = Open3DModelClient("http://o3d:8878", api_key="K",
timeout_s=2.0, max_stl_bytes=20_000_000)
with patch("httpx.Client.get", return_value=response):
with pytest.raises(Open3DModelError) as exc:
client.get_stl_bytes("r1")
assert exc.value.code == "STL_TOO_LARGE"
def test_get_stl_bytes_missing_etag_falls_back_to_hash():
response = httpx.Response(200, content=b"BINARYSTL") # no ETag header
client = Open3DModelClient("http://o3d:8878", api_key="K", timeout_s=2.0)
with patch("httpx.Client.get", return_value=response):
body, etag = client.get_stl_bytes("r1")
assert body == b"BINARYSTL"
assert len(etag) > 0 # qualunque hash deterministico
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_open3dmodel_client.py -v`
Expected: ImportError.
- [ ] **Step 3: Implementa `open3dmodel.py`**
Crea `src/presa_mano/clients/open3dmodel.py`:
```python
"""Client for the Open3DModel pose-estimation service.
Open3DModel API (see /opt/docker/robotserver/Open3DModel/README.md):
- GET /api/recipes → list
- GET /api/recipes/{id}/stl → STL bytes (HTTPException 404 if missing)
- POST /api/pose → envelope {success, data, error}
data has transform_4x4, fitness,
inlier_rmse_mm, translation_mm, ...
- Header X-Api-Key richiesto (403 con detail testuale altrimenti).
"""
from __future__ import annotations
import hashlib
from typing import Any
import httpx
import numpy as np
class Open3DModelUnavailable(RuntimeError):
"""Transport-level failure (connect refused, timeout)."""
class Open3DModelError(RuntimeError):
"""Semantic failure: envelope error code, HTTP 4xx/5xx with non-envelope
body, or client-side validation (e.g. STL oversize)."""
def __init__(self, code: str, message: str):
super().__init__(f"{code}: {message}")
self.code = code
self.message = message
class Open3DModelClient:
def __init__(self, base_url: str, api_key: str | None,
timeout_s: float = 30.0,
max_stl_bytes: int = 20_000_000):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout_s = timeout_s
self.max_stl_bytes = max_stl_bytes
def _headers(self) -> dict[str, str]:
h = {}
if self.api_key:
h["X-Api-Key"] = self.api_key
return h
def list_recipes(self) -> list[dict[str, Any]]:
url = f"{self.base_url}/api/recipes"
try:
with httpx.Client(timeout=self.timeout_s) as c:
r = c.get(url, headers=self._headers())
except httpx.RequestError as e:
raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
if r.status_code != 200:
raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))
try:
body = r.json()
except ValueError as e:
raise Open3DModelUnavailable(f"non-JSON response: {e}") from e
if not isinstance(body, list):
raise Open3DModelError("UNEXPECTED_PAYLOAD",
f"expected list, got {type(body).__name__}")
return body
def estimate_pose(self, recipe_id: str, points_cv_m: np.ndarray,
intrinsics: dict[str, Any]) -> dict[str, Any]:
body = {
"recipe_id": str(recipe_id),
"point_cloud": np.asarray(points_cv_m, dtype=float).tolist(),
"intrinsics": {
"fx": float(intrinsics["fx"]),
"fy": float(intrinsics["fy"]),
"cx": float(intrinsics["cx"]),
"cy": float(intrinsics["cy"]),
"width": int(intrinsics["width"]),
"height": int(intrinsics["height"]),
},
"unit": "m",
}
url = f"{self.base_url}/api/pose"
try:
with httpx.Client(timeout=self.timeout_s) as c:
r = c.post(url, json=body, headers=self._headers())
except httpx.RequestError as e:
raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
try:
payload = r.json()
except ValueError as e:
raise Open3DModelUnavailable(
f"non-JSON response (status {r.status_code}): {e}") from e
if isinstance(payload, dict) and "success" in payload:
if payload.get("success") is True:
return payload.get("data") or {}
err = payload.get("error") or {}
raise Open3DModelError(
str(err.get("code") or "UNKNOWN"),
str(err.get("message") or "no message"),
)
# Non-envelope response (e.g. FastAPI HTTPException 403/422)
raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))
def get_stl_bytes(self, recipe_id: str) -> tuple[bytes, str]:
url = f"{self.base_url}/api/recipes/{recipe_id}/stl"
try:
with httpx.Client(timeout=self.timeout_s) as c:
r = c.get(url, headers=self._headers())
except httpx.RequestError as e:
raise Open3DModelUnavailable(f"open3dmodel unreachable: {e}") from e
if r.status_code == 404:
raise Open3DModelError("RECIPE_NOT_FOUND",
f"STL for {recipe_id} not found")
if r.status_code != 200:
raise Open3DModelError(f"HTTP_{r.status_code}", self._detail_msg(r))
body = r.content
if len(body) > self.max_stl_bytes:
raise Open3DModelError("STL_TOO_LARGE",
f"{len(body)} bytes > {self.max_stl_bytes}")
etag = r.headers.get("ETag") or r.headers.get("Last-Modified") \
or hashlib.sha1(body).hexdigest()
return body, etag
@staticmethod
def _detail_msg(r: httpx.Response) -> str:
try:
body = r.json()
if isinstance(body, dict) and "detail" in body:
return str(body["detail"])
except ValueError:
pass
return r.text[:200] or f"http {r.status_code}"
```
- [ ] **Step 4: Esporta da `clients/__init__.py`**
Modifica `src/presa_mano/clients/__init__.py`:
```python
"""HTTP clients for upstream microservices."""
from .g1 import G1Client, G1StatusError, G1Unavailable
from .marker import MarkerClient
from .nuwa import NuwaClient, NuwaError
from .open3dmodel import Open3DModelClient, Open3DModelError, Open3DModelUnavailable
from .sam21 import Sam21Client, Sam21Error, Sam21Unavailable
from .yolo import YoloClient
__all__ = [
"G1Client", "G1StatusError", "G1Unavailable",
"MarkerClient", "NuwaClient", "NuwaError",
"Open3DModelClient", "Open3DModelError", "Open3DModelUnavailable",
"Sam21Client", "Sam21Error", "Sam21Unavailable",
"YoloClient",
]
```
- [ ] **Step 5: Esegui i test**
Run: `uv run pytest tests/test_open3dmodel_client.py -v`
Expected: 10 passed.
- [ ] **Step 6: Commit**
```bash
git add tests/test_open3dmodel_client.py src/presa_mano/clients/open3dmodel.py src/presa_mano/clients/__init__.py
git commit -m "feat(clients): Open3DModel client (list_recipes, estimate_pose, get_stl_bytes)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 6: `clients/open3dmodel_cache.py` — LRU mesh cache
**Files:**
- Create: `src/presa_mano/clients/open3dmodel_cache.py`
- Test: `tests/test_recipe_mesh_cache.py`
- [ ] **Step 1: Scrivi i test che falliscono**
Crea `tests/test_recipe_mesh_cache.py`:
```python
"""Unit tests for RecipeMeshCache."""
from __future__ import annotations
from unittest.mock import MagicMock
import numpy as np
import pytest
from presa_mano.clients.open3dmodel_cache import RecipeMesh, RecipeMeshCache
def _verts_tris():
verts = np.array([[0.,0.,0.],[1.,0.,0.],[0.,1.,0.]])
tris = np.array([[0,1,2]], dtype=np.int32)
return verts, tris
def _mock_client(stl_bytes_by_id: dict[str, tuple[bytes, str]]):
"""Mock client whose get_stl_bytes returns the registered (bytes, etag)
for each recipe_id, and counts calls."""
m = MagicMock()
m._call_count = {}
def _get(recipe_id: str):
m._call_count[recipe_id] = m._call_count.get(recipe_id, 0) + 1
return stl_bytes_by_id[recipe_id]
m.get_stl_bytes.side_effect = _get
return m
def test_cache_miss_then_hit(monkeypatch):
verts, tris = _verts_tris()
monkeypatch.setattr(
"presa_mano.clients.open3dmodel_cache.parse_stl_bytes",
lambda b: (verts, tris),
)
client = _mock_client({"r1": (b"STL1", "etag-A")})
cache = RecipeMeshCache(client=client, capacity=4)
m1 = cache.get("r1")
assert isinstance(m1, RecipeMesh)
np.testing.assert_array_equal(m1.vertices, verts)
np.testing.assert_array_equal(m1.triangles, tris)
assert m1.etag == "etag-A"
assert client._call_count["r1"] == 1
# secondo accesso → hit
m2 = cache.get("r1")
assert m2 is m1
assert client._call_count["r1"] == 1
def test_cache_etag_change_refetches(monkeypatch):
verts, tris = _verts_tris()
monkeypatch.setattr(
"presa_mano.clients.open3dmodel_cache.parse_stl_bytes",
lambda b: (verts, tris),
)
client = MagicMock()
seq = iter([(b"STL_OLD", "etag-A"), (b"STL_NEW", "etag-B")])
client.get_stl_bytes.side_effect = lambda rid: next(seq)
cache = RecipeMeshCache(client=client, capacity=4)
cache.get("r1")
m2 = cache.get("r1")
assert m2.etag == "etag-B"
assert client.get_stl_bytes.call_count == 2
def test_cache_lru_evicts_oldest(monkeypatch):
verts, tris = _verts_tris()
monkeypatch.setattr(
"presa_mano.clients.open3dmodel_cache.parse_stl_bytes",
lambda b: (verts, tris),
)
client = _mock_client({
f"r{i}": (f"S{i}".encode(), f"e{i}") for i in range(10)
})
cache = RecipeMeshCache(client=client, capacity=3)
cache.get("r1"); cache.get("r2"); cache.get("r3")
# r1 è il più vecchio
cache.get("r4") # evict r1
cache.get("r1") # miss → re-fetch (call_count va a 2)
assert client._call_count["r1"] == 2
def test_cache_get_uses_overrideable_client(monkeypatch):
"""`get(recipe_id, client=other)` permette di passare un client fresco
(utile quando PUT /api/config cambia base_url)."""
verts, tris = _verts_tris()
monkeypatch.setattr(
"presa_mano.clients.open3dmodel_cache.parse_stl_bytes",
lambda b: (verts, tris),
)
default_client = _mock_client({"r1": (b"S1", "e1")})
override_client = _mock_client({"r1": (b"S2", "e2")})
cache = RecipeMeshCache(client=default_client, capacity=2)
m = cache.get("r1", client=override_client)
assert m.etag == "e2"
assert default_client._call_count.get("r1", 0) == 0
assert override_client._call_count["r1"] == 1
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_recipe_mesh_cache.py -v`
Expected: ImportError.
- [ ] **Step 3: Implementa la cache**
Crea `src/presa_mano/clients/open3dmodel_cache.py`:
```python
"""In-memory LRU cache of parsed STL meshes, keyed by recipe_id + etag."""
from __future__ import annotations
from collections import OrderedDict
from dataclasses import dataclass
import numpy as np
from ..perception.stl_sampler import parse_stl_bytes
from .open3dmodel import Open3DModelClient
@dataclass(frozen=True)
class RecipeMesh:
vertices: np.ndarray # (M, 3) float64, mm in model frame
triangles: np.ndarray # (T, 3) int32
etag: str
class RecipeMeshCache:
"""LRU keyed by recipe_id. On hit, validates the etag against the client
and re-fetches if it changed. Singleton-friendly: thread-safe with a
simple lock is NOT implemented (presa-mano is single-process; if
concurrency becomes a need, wrap the OrderedDict ops in a lock)."""
def __init__(self, client: Open3DModelClient, capacity: int = 8):
self._client = client
self._capacity = max(1, int(capacity))
self._store: OrderedDict[str, RecipeMesh] = OrderedDict()
def get(self, recipe_id: str, client: Open3DModelClient | None = None
) -> RecipeMesh:
c = client or self._client
body, etag = c.get_stl_bytes(recipe_id)
cached = self._store.get(recipe_id)
if cached is not None and cached.etag == etag:
self._store.move_to_end(recipe_id)
return cached
verts, tris = parse_stl_bytes(body)
mesh = RecipeMesh(vertices=verts, triangles=tris, etag=etag)
self._store[recipe_id] = mesh
self._store.move_to_end(recipe_id)
while len(self._store) > self._capacity:
self._store.popitem(last=False)
return mesh
```
- [ ] **Step 4: Esegui i test**
Run: `uv run pytest tests/test_recipe_mesh_cache.py -v`
Expected: 4 passed.
- [ ] **Step 5: Commit**
```bash
git add tests/test_recipe_mesh_cache.py src/presa_mano/clients/open3dmodel_cache.py
git commit -m "feat(clients): LRU mesh cache for Open3DModel recipes (etag-validated)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 7: Refactor `perception/sam21_bridge.py` — estrai `segment_with_sam21`
**Files:**
- Modify: `src/presa_mano/perception/sam21_bridge.py:34-136`
- Test: `tests/test_sam21_bridge_segment.py`
- [ ] **Step 1: Scrivi il test di regressione per il refactor**
Crea `tests/test_sam21_bridge_segment.py`:
```python
"""Tests for the segment_with_sam21/points_via_sam21 decomposition.
We exercise the path that the find_model endpoint will take:
segment_with_sam21(nuwa, sam21, bbox, sw) → Sam21Segmentation
and the legacy wrapper still works.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import cv2
import numpy as np
import pytest
from presa_mano.perception.sam21_bridge import (
Sam21Segmentation,
segment_with_sam21,
points_via_sam21,
Sam21Fallback,
)
from presa_mano.timing import Stopwatch
def _fake_nuwa():
"""Returns a MagicMock Nuwa that yields a 4x4 black RGB jpeg and a 4x4
uint16 depth PNG (all 1000 mm) + a minimal Nuwa intrinsics dict."""
rgb = np.zeros((4, 4, 3), dtype=np.uint8)
ok, jpg = cv2.imencode(".jpg", rgb)
assert ok
depth = np.full((4, 4), 1000, dtype=np.uint16)
ok, dpng = cv2.imencode(".png", depth)
assert ok
intr = {
"rgb": {"fx": 600.0, "fy": 600.0, "cx": 2.0, "cy": 2.0},
"ir": {"fx": 600.0, "fy": 600.0, "cx": 2.0, "cy": 2.0,
"extrinsics": {"R": [[1,0,0],[0,1,0],[0,0,1]],
"t": [0.0, 0.0, 0.0]}},
}
n = MagicMock()
n.rgb_jpeg.return_value = bytes(jpg)
n.depth_png.return_value = bytes(dpng)
n.intrinsics.return_value = intr
return n, bytes(jpg)
def _fake_sam21(points_m: np.ndarray):
s = MagicMock()
s.segment_pointcloud.return_value = points_m
return s
def test_segment_with_sam21_returns_segmentation(monkeypatch):
nuwa, rgb_jpeg = _fake_nuwa()
pts_m = np.array([[0.01, 0.02, 0.5], [0.011, 0.022, 0.49]])
sam21 = _fake_sam21(pts_m)
# Niente plane filter qui — segment_with_sam21 NON deve applicarlo
monkeypatch.setattr(
"presa_mano.perception.sam21_bridge.get_work_plane", lambda: None
)
sw = Stopwatch()
seg = segment_with_sam21(nuwa, sam21, bbox=(0, 0, 4, 4), sw=sw)
assert isinstance(seg, Sam21Segmentation)
np.testing.assert_array_equal(seg.points_cv_m, pts_m)
assert seg.rgb_jpeg == rgb_jpeg
assert "width" in seg.intrinsics_rgb and "height" in seg.intrinsics_rgb
def test_points_via_sam21_still_applies_plane_filter(monkeypatch):
"""L'API legacy non deve cambiare: ritorna (pts, plane_info) e applica
apply_below_plane_filter come prima."""
nuwa, _rgb = _fake_nuwa()
pts_m = np.array([[0.0, 0.0, 0.5]] * 20)
sam21 = _fake_sam21(pts_m)
monkeypatch.setattr(
"presa_mano.perception.sam21_bridge.get_work_plane", lambda: None
)
sw = Stopwatch()
pts, info = points_via_sam21(nuwa, sam21, bbox=(0, 0, 4, 4), sw=sw)
assert pts.shape == (20, 3)
assert info["plane_applied"] is False
assert info["n_points_before_plane"] == 20
def test_points_via_sam21_too_few_points_raises_fallback(monkeypatch):
nuwa, _rgb = _fake_nuwa()
pts_m = np.array([[0.0, 0.0, 0.5]] * 5) # <10
sam21 = _fake_sam21(pts_m)
monkeypatch.setattr(
"presa_mano.perception.sam21_bridge.get_work_plane", lambda: None
)
sw = Stopwatch()
with pytest.raises(Sam21Fallback):
points_via_sam21(nuwa, sam21, bbox=(0, 0, 4, 4), sw=sw)
```
- [ ] **Step 2: Esegui i test per vedere il fallimento**
Run: `uv run pytest tests/test_sam21_bridge_segment.py -v`
Expected: ImportError per `segment_with_sam21` / `Sam21Segmentation`.
- [ ] **Step 3: Refactor `sam21_bridge.py`**
Sostituisci la funzione `points_via_sam21` in `src/presa_mano/perception/sam21_bridge.py` (linee 34-136) con questa decomposizione:
```python
@dataclass
class Sam21Segmentation:
"""Output di segment_with_sam21 — SAM21 puro senza plane filter."""
points_cv_m: np.ndarray # (N,3) OpenCV camera frame, metri
rgb_jpeg: bytes # snapshot Nuwa usato per la segmentazione
intrinsics_rgb: dict # con width/height iniettati dal decoded RGB
def segment_with_sam21(nuwa, sam21, bbox: tuple[int, int, int, int],
sw: Stopwatch) -> Sam21Segmentation:
"""SAM21 puro: nuwa_grab + depth_align + segment, SENZA plane filter.
Raises Sam21Fallback per failure semantiche (mask vuota, no depth, SAM21
unreachable, decode fail). Re-raises HTTPException(502) per failure Nuwa.
"""
try:
with sw.step("nuwa_grab"):
rgb_jpeg = nuwa.rgb_jpeg()
depth_png_in = nuwa.depth_png()
intr_full = nuwa.intrinsics()
except (NuwaError, httpx.RequestError) as e:
raise HTTPException(status_code=502, detail=f"Nuwa unreachable: {e}")
if not intr_full.get("rgb"):
log.warning("SAM21 path skipped: nuwa intrinsics missing 'rgb' block")
raise Sam21Fallback("nuwa intrinsics missing 'rgb' block")
try:
bgr = cv2.imdecode(np.frombuffer(rgb_jpeg, np.uint8), cv2.IMREAD_COLOR)
if bgr is None:
log.warning("SAM21 path skipped: could not decode Nuwa RGB JPEG")
raise Sam21Fallback("could not decode Nuwa RGB JPEG")
h_rgb, w_rgb = bgr.shape[:2]
ok, rgb_png_buf = cv2.imencode(".png", bgr)
if not ok:
log.warning("SAM21 path skipped: could not re-encode RGB as PNG")
raise Sam21Fallback("could not re-encode RGB as PNG")
rgb_png = bytes(rgb_png_buf)
with sw.step("depth_align"):
depth_ir = cv2.imdecode(np.frombuffer(depth_png_in, np.uint8),
cv2.IMREAD_UNCHANGED)
if depth_ir is None:
log.warning("SAM21 path skipped: could not decode Nuwa depth PNG")
raise Sam21Fallback("could not decode Nuwa depth PNG")
if depth_ir.dtype != np.uint16:
log.warning("SAM21 path skipped: depth PNG dtype %s (expected uint16)",
depth_ir.dtype)
raise Sam21Fallback(f"depth PNG dtype expected uint16, got {depth_ir.dtype}")
h_ir, w_ir = depth_ir.shape
intr_full = {
**intr_full,
"rgb": {**intr_full["rgb"], "width": w_rgb, "height": h_rgb},
"ir": {**intr_full.get("ir", {}), "width": w_ir, "height": h_ir},
}
reg = CameraRegistration.from_nuwa(intr_full)
depth_aligned = align_depth_to_rgb(depth_ir, reg)
ok, depth_png_buf = cv2.imencode(".png", depth_aligned)
if not ok:
log.warning("SAM21 path skipped: could not encode aligned depth as PNG")
raise Sam21Fallback("could not encode aligned depth as PNG")
depth_png_out = bytes(depth_png_buf)
except Sam21Fallback:
raise
except Exception as e:
log.error("Unexpected SAM21 prep error: %r", e)
raise Sam21Fallback(f"prep failed: {e!r}") from e
with sw.step("sam21"):
try:
pts_cv_m = sam21.segment_pointcloud(
rgb_png=rgb_png,
depth_png=depth_png_out,
intrinsics=intr_full["rgb"],
bbox=bbox,
depth_scale_mm=1.0,
)
except Sam21Unavailable as e:
log.warning("SAM21 unreachable, falling back to bbox: %s", e)
raise Sam21Fallback(f"unreachable: {e}") from e
except Sam21Error as e:
if e.code in SAM21_BUG_CODES:
log.error("SAM21 rejected payload (%s): %s", e.code, e.message)
else:
log.warning("SAM21 returned %s: %s — falling back to bbox", e.code, e.message)
raise Sam21Fallback(f"{e.code}: {e.message}") from e
except Exception as e:
log.error("Unexpected SAM21 error: %r", e)
raise Sam21Fallback(f"unexpected: {e!r}") from e
return Sam21Segmentation(
points_cv_m=pts_cv_m,
rgb_jpeg=rgb_jpeg,
intrinsics_rgb=intr_full["rgb"],
)
def points_via_sam21(nuwa, sam21, bbox: tuple[int, int, int, int],
sw: Stopwatch) -> tuple[np.ndarray, dict]:
"""Legacy wrapper: segment_with_sam21 + apply_below_plane_filter.
Returns (points_cv_m, plane_info). Raises Sam21Fallback if <10 points
after plane filter (unchanged semantics)."""
seg = segment_with_sam21(nuwa, sam21, bbox, sw)
pts_cv, plane_info = apply_below_plane_filter(seg.points_cv_m, get_work_plane())
if pts_cv.shape[0] < 10:
log.warning("SAM21 cloud has %d points after plane filter (<10), falling back to bbox",
pts_cv.shape[0])
raise Sam21Fallback(
f"too few points after plane filter: {pts_cv.shape[0]}"
f" (before plane: {plane_info['n_points_before_plane']})"
)
return pts_cv, plane_info
```
Aggiungi l'import per `dataclass` in cima al file se manca (`from dataclasses import dataclass`).
- [ ] **Step 4: Esegui i nuovi test**
Run: `uv run pytest tests/test_sam21_bridge_segment.py -v`
Expected: 3 passed.
- [ ] **Step 5: Verifica che i test esistenti passano ancora**
Run: `uv run pytest tests/ -v -k "not models and not open3d and not stl"`
Expected: tutti passano.
- [ ] **Step 6: Commit**
```bash
git add tests/test_sam21_bridge_segment.py src/presa_mano/perception/sam21_bridge.py
git commit -m "refactor(sam21_bridge): extract segment_with_sam21 (no plane filter)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 8: `api/_deps.py` — factory open3dmodel + singleton cache
**Files:**
- Modify: `src/presa_mano/api/_deps.py:1-40`
- [ ] **Step 1: Aggiungi le factory in `_deps.py`**
Modifica `src/presa_mano/api/_deps.py`. Sostituisci la sezione import in cima (linee 1-11) con:
```python
"""Build clients & helpers from current config — called per-request so runtime
updates (PUT /api/config) take effect without restart."""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from ..clients import (
G1Client, MarkerClient, NuwaClient,
Open3DModelClient, Sam21Client, YoloClient,
)
from ..clients.open3dmodel_cache import RecipeMeshCache
from ..config import data_dir, get_config
_recipe_cache_singleton: RecipeMeshCache | None = None
```
Poi, sotto la funzione `sam21()` (intorno a riga 37), aggiungi:
```python
def open3dmodel() -> Open3DModelClient:
cfg = get_config()
c = _svc(cfg, "open3dmodel")
api_key_env = c.get("api_key_env")
api_key = os.environ.get(api_key_env) if api_key_env else None
return Open3DModelClient(
base_url=c.get("base_url", "http://localhost:8878"),
api_key=api_key,
timeout_s=float(c.get("timeout_s", 30.0)),
max_stl_bytes=int(c.get("max_stl_bytes", 20_000_000)),
)
def recipe_cache() -> RecipeMeshCache:
"""Singleton mesh cache. Sopravvive a PUT /api/config; etag stale forza
refetch al prossimo accesso. Inizializzata lazy con il client corrente
al momento della prima get."""
global _recipe_cache_singleton
if _recipe_cache_singleton is None:
_recipe_cache_singleton = RecipeMeshCache(open3dmodel(), capacity=8)
return _recipe_cache_singleton
```
- [ ] **Step 2: Verifica che presa-mano si avvia ancora**
Run: `uv run python -c "from presa_mano.api import _deps; print(_deps.open3dmodel().base_url)"`
Expected: `http://localhost:8878` (o quello che è in config.yaml).
- [ ] **Step 3: Commit**
```bash
git add src/presa_mano/api/_deps.py
git commit -m "feat(api): _deps factories for Open3DModel client + recipe cache
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 9: `api/models_routes.py` — proxy GET /api/models/recipes
**Files:**
- Create: `src/presa_mano/api/models_routes.py`
- Modify: `src/presa_mano/main.py:12-24`
- [ ] **Step 1: Implementa il router**
Crea `src/presa_mano/api/models_routes.py`:
```python
"""Proxy minimale verso Open3DModel — evita di esporre l'api_key al browser.
Per ora un solo endpoint: GET /api/models/recipes → lista ricette caricate
in Open3DModel.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException
from ..clients import Open3DModelError, Open3DModelUnavailable
from . import _deps
router = APIRouter()
@router.get("/recipes")
def list_recipes() -> dict[str, Any]:
try:
recs = _deps.open3dmodel().list_recipes()
except Open3DModelUnavailable as e:
raise HTTPException(status_code=502, detail=f"Open3DModel unreachable: {e}")
except Open3DModelError as e:
raise HTTPException(status_code=502, detail=f"Open3DModel error: {e.code}: {e.message}")
return {"recipes": recs}
```
- [ ] **Step 2: Monta il router in main.py**
Modifica `src/presa_mano/main.py`. Nell'import block (linee 12-23), aggiungi `models_routes`:
```python
from .api import (
agent_routes,
calibration_routes,
config_routes,
grasp_routes,
home_routes,
models_routes,
perception_routes,
posture_routes,
robot_routes,
wrist_routes,
work_plane_routes,
)
```
Poi, dopo la linea `app.include_router(grasp_routes.router, prefix="/api/grasp", tags=["grasp"])` (linea 42), aggiungi:
```python
app.include_router(models_routes.router, prefix="/api/models", tags=["models"])
```
- [ ] **Step 3: Verifica che il router è registrato**
Run: `uv run python -c "from presa_mano.main import app; print([r.path for r in app.routes if 'models' in r.path])"`
Expected: stampa `['/api/models/recipes']`.
- [ ] **Step 4: Commit**
```bash
git add src/presa_mano/api/models_routes.py src/presa_mano/main.py
git commit -m "feat(api): GET /api/models/recipes proxy to Open3DModel
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 10: `api/grasp_routes.py` — POST /api/grasp/find_model
**Files:**
- Modify: `src/presa_mano/api/grasp_routes.py` (aggiungi handler dopo `execute`)
- [ ] **Step 1: Aggiungi i nuovi import in cima al file**
Modifica `src/presa_mano/api/grasp_routes.py`. Sostituisci le linee 1-22 (sezione import) con:
```python
"""Grasp planning + execution endpoints."""
from __future__ import annotations
import base64
from typing import Any, Optional
import numpy as np
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from ..calibration import CalibrationStore
from ..clients import G1Unavailable, Open3DModelError, Open3DModelUnavailable
from ..config import get_config
from ..geometry import transform_points
from ..grasp import GraspPlanner, compute_axes
from ..perception import ObjectDetector, filter_outliers, nuwa_to_cv_frame
from ..perception.projection import K_from_intrinsics, project_camera_to_pixels
from ..perception.sam21_bridge import (
Sam21Fallback,
points_via_sam21,
segment_with_sam21,
)
from ..perception.stl_sampler import sample_surface
from ..perception.work_plane import (
apply_below_plane_filter,
get_work_plane,
partition_by_plane,
)
from .perception_routes import _bbox_to_roi
from ..timing import Stopwatch
from . import _deps
router = APIRouter()
```
- [ ] **Step 2: Aggiungi il Pydantic model + l'handler `find_model`**
In fondo al file `src/presa_mano/api/grasp_routes.py` (dopo `execute`), aggiungi:
```python
class FindModelReq(BaseModel):
recipe_id: str = Field(..., min_length=1)
conf: Optional[float] = None
class_name: Optional[str] = None
_FIT_DOMAIN_CODES = {
"REGISTRATION_FAILED", "EMPTY_CLOUD", "MODEL_MISSING",
"INVALID_MESH", "NO_CORRESPONDENCES",
}
@router.post("/find_model")
def find_model(req: FindModelReq) -> dict[str, Any]:
sw = Stopwatch()
cfg = get_config()
fm_cfg = cfg.get("grasp", {}).get("find_model", {})
min_fitness = float(fm_cfg.get("min_fitness", 0.5))
max_rmse_mm = float(fm_cfg.get("max_rmse_mm", 5.0))
sample_n = int(fm_cfg.get("sample_points", 2000))
side = cfg.get("robot", {}).get("wrist_side", "right")
T_base_cam = CalibrationStore(_deps.calibration_file_path(side)).T_base_camera()
if T_base_cam is None:
raise HTTPException(status_code=412, detail=f"calibration missing for side='{side}'")
nuwa = _deps.nuwa()
yolo = _deps.yolo()
sam21 = _deps.sam21()
o3d = _deps.open3dmodel()
# 1. YOLO
detector = ObjectDetector(yolo, nuwa)
try:
with sw.step("yolo"):
dets = detector.detect(conf=req.conf)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=502, detail=f"YOLO detect failed: {e}")
if req.class_name:
dets = [d for d in dets if (d.get("class") or "").lower() == req.class_name.lower()]
if not dets:
raise HTTPException(status_code=404, detail="no object detected")
dets.sort(key=lambda d: d["confidence"], reverse=True)
top = dets[0]
bbox = _bbox_to_roi(top["bbox"])
# 2. SAM21 (no plane filter)
try:
seg = segment_with_sam21(nuwa, sam21, bbox, sw)
except Sam21Fallback as e:
raise HTTPException(status_code=422, detail=f"SAM21 fail: {e}")
# 3. Partition by plane
with sw.step("partition_plane"):
accepted_m, excluded_m, plane_info = partition_by_plane(
seg.points_cv_m, get_work_plane()
)
if accepted_m.shape[0] < 10:
raise HTTPException(
status_code=422,
detail=f"too few accepted points after plane filter: {accepted_m.shape[0]}"
f" (before plane: {plane_info['n_points_before_plane']})"
)
# 4. Open3DModel pose
fit_failed = False
fit_reason: str | None = None
pose_data: dict[str, Any] = {}
try:
with sw.step("open3d_pose"):
pose_data = o3d.estimate_pose(req.recipe_id, accepted_m, seg.intrinsics_rgb)
except Open3DModelUnavailable as e:
raise HTTPException(status_code=502, detail=f"Open3DModel unreachable: {e}")
except Open3DModelError as e:
if e.code == "RECIPE_NOT_FOUND":
raise HTTPException(status_code=404, detail=f"recipe_id not found: {req.recipe_id}")
if e.code in {"HTTP_401", "HTTP_403"}:
raise HTTPException(status_code=502, detail=f"Open3DModel auth: {e.message}")
if e.code in _FIT_DOMAIN_CODES:
fit_failed = True
fit_reason = f"{e.code}: {e.message}"
else:
raise HTTPException(status_code=502, detail=f"Open3DModel error: {e.code}: {e.message}")
fitness = float(pose_data.get("fitness", 0.0))
rmse_mm = float(pose_data.get("inlier_rmse_mm", 1e9))
transform_4x4 = pose_data.get("transform_4x4")
if not fit_failed:
if fitness < min_fitness:
fit_failed = True
fit_reason = f"low fitness {fitness:.2f} < {min_fitness}"
elif rmse_mm > max_rmse_mm:
fit_failed = True
fit_reason = f"rmse {rmse_mm:.2f}mm > {max_rmse_mm}"
# 5. Mesh + sample + project
W = int(seg.intrinsics_rgb["width"])
H = int(seg.intrinsics_rgb["height"])
K = K_from_intrinsics(seg.intrinsics_rgb)
model_px = np.empty((0, 2), dtype=np.int32)
if not fit_failed and transform_4x4 is not None:
try:
with sw.step("recipe_mesh"):
mesh = _deps.recipe_cache().get(req.recipe_id, client=o3d)
with sw.step("sample"):
model_pts_mm = sample_surface(mesh.vertices, mesh.triangles, sample_n)
with sw.step("transform"):
T = np.asarray(transform_4x4, dtype=np.float64)
# transform_points expects (4,4) acting on column vectors → use directly
model_pts_cam_mm = transform_points(T, model_pts_mm)
with sw.step("project_model"):
model_px = project_camera_to_pixels(model_pts_cam_mm, K, (W, H))
except (Open3DModelError, Open3DModelUnavailable) as e:
raise HTTPException(status_code=502, detail=f"STL fetch/parse failed: {e}")
with sw.step("project_points"):
accepted_px = project_camera_to_pixels(accepted_m, K, (W, H))
excluded_px = project_camera_to_pixels(excluded_m, K, (W, H))
with sw.step("encode_rgb"):
rgb_b64 = base64.b64encode(seg.rgb_jpeg).decode("ascii")
fit_ok = not fit_failed
return {
"ok": True,
"rgb_jpeg_b64": rgb_b64,
"rgb_size": [W, H],
"detection": top,
"recipe": {"id": req.recipe_id,
"name": pose_data.get("recipe_name"),
"symmetry_type": pose_data.get("symmetry_type")},
"model_points_px": model_px.tolist(),
"accepted_points_px": accepted_px.tolist(),
"excluded_points_px": excluded_px.tolist(),
"n_accepted": int(accepted_m.shape[0]),
"n_excluded": int(excluded_m.shape[0]),
"plane_applied": bool(plane_info["plane_applied"]),
"fit_ok": fit_ok,
"fit_failed": fit_failed,
"fit_reason": fit_reason,
"fitness": fitness if not fit_failed or pose_data else None,
"inlier_rmse_mm": rmse_mm if not fit_failed or pose_data else None,
"transform_camera_4x4": transform_4x4,
"timings_ms": sw.as_dict(),
}
```
- [ ] **Step 3: Smoke-test l'import**
Run: `uv run python -c "from presa_mano.api.grasp_routes import router; print([r.path for r in router.routes])"`
Expected: include `/find_model`.
- [ ] **Step 4: Commit**
```bash
git add src/presa_mano/api/grasp_routes.py
git commit -m "feat(api): POST /api/grasp/find_model with pose-fit + overlay projection
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 11: Frontend — `web/index.html`
**Files:**
- Modify: `web/index.html:229-282`
- [ ] **Step 1: Sostituisci la sezione tab-grasp**
In `web/index.html`, sostituisci interamente il blocco da `<section class="panel" id="tab-grasp">` fino al `</section>` corrispondente (linee 228-282) con:
```html
<!-- ============================== CERCA MODELLO ============================== -->
<section class="panel" id="tab-grasp">
<div class="grid-2">
<div>
<h2>RGB + detection — click per misurare</h2>
<div class="canvas-wrap">
<canvas id="graspCanvas" width="640" height="480"></canvas>
</div>
<div class="measure-strip" id="measureInfo">click sull'immagine per la distanza</div>
<div class="kv">
<label>Nuvola di <select id="graspObjectPick"><option value="">(top overall)</option></select></label>
<label>Confidenza <input type="number" id="graspConf" value="0.5" step="0.05" min="0" max="1" /></label>
</div>
<div id="graspLegend" class="grasp-legend"></div>
<div class="actions">
<button id="detectBtn">Rileva oggetto</button>
<button id="loopBtn">▶ Loop riconoscimento</button>
<label class="inline" style="margin-left:8px"><input type="number" id="loopRateMs" value="500" min="100" step="50" style="width:80px"> ms</label>
</div>
</div>
<div>
<h2>Fit modello 3D</h2>
<div class="kv">
<label>Modello
<select id="findModelPick"><option value="">— carico ricette… —</option></select>
</label>
<button id="findRecipesRefreshBtn" class="ghost" title="Ricarica ricette">↻</button>
</div>
<div class="actions">
<button id="findBtn" class="primary" disabled>Trova</button>
</div>
<div id="findStatus" class="status-line">—</div>
<div class="canvas-wrap">
<canvas id="findResultCanvas" width="640" height="480"></canvas>
</div>
<div class="kv">
<label class="inline"><input type="checkbox" id="findShowModel" checked> Punti modello (blu)</label>
<label class="inline"><input type="checkbox" id="findShowAccepted"> Punti accettati (verdi)</label>
<label class="inline"><input type="checkbox" id="findShowExcluded"> Punti esclusi (rossi)</label>
</div>
<pre id="findDetails" class="log small">—</pre>
<!-- DOM hooks dormienti consumati da app.js residuo. Tenuti per evitare null deref. -->
<div style="display:none">
<span id="statsWindow">5</span>
<table id="statsTable"><tbody>
<tr><td>x</td><td>—</td><td>—</td><td>—</td><td>—</td><td>—</td></tr>
<tr><td>y</td><td>—</td><td>—</td><td>—</td><td>—</td><td>—</td></tr>
<tr><td>z</td><td>—</td><td>—</td><td>—</td><td>—</td><td>—</td></tr>
</tbody></table>
<div id="statsMeta"></div>
<pre id="graspOut"></pre>
</div>
</div>
</div>
</section>
```
- [ ] **Step 2: Verifica HTML valido (smoke)**
Run: `grep -c 'id="findResultCanvas"' web/index.html`
Expected: `1`.
Run: `grep -E 'id="(planBtn|executeBtn|cropCanvas|scatterXY|scatterXZ|graspDryRun)"' web/index.html`
Expected: nessun output (tutti rimossi).
- [ ] **Step 3: Commit**
```bash
git add web/index.html
git commit -m "feat(web): rebuild Cerca modello tab — model picker + result canvas
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 12: Frontend — `web/app.js`
**Files:**
- Modify: `web/app.js` (sezione TROVA UTENSILE — circa righe 112-580)
- [ ] **Step 1: Rimuovi i blocchi morti**
Apri `web/app.js` e:
1. **Rimuovi** queste righe (riferimenti agli elementi rimossi):
```js
const cropCanvas = $('cropCanvas');
const cropCtx = cropCanvas.getContext('2d');
const scatterXY = $('scatterXY');
const scatterXZ = $('scatterXZ');
```
Sostituiscile con:
```js
// (rimossi cropCanvas/scatter — ora c'è findResultCanvas)
```
2. **Rimuovi** le variabili dichiarate intorno alla riga 121-124:
```js
let lastCropBbox = null;
let lastPoints = null;
```
(restano `lastDetections`, `lastSelectedClass`, `knownClasses`).
3. **Rimuovi** interamente le funzioni `drawCrop`, `drawScatter`, `drawScatters` (intorno alle righe 164-216).
4. Nella funzione `applyScanResponse` (intorno alla riga 264-274), **rimuovi** le righe che fanno riferimento ai morti:
```js
lastCropBbox = sel ? sel.bbox : null;
lastPoints = r.points_base || r.points_camera || null;
```
E rimuovi le chiamate `drawCrop(); drawScatters();` da quella funzione.
5. **Rimuovi** completamente gli handler di `$('planBtn').addEventListener(...)` e `$('executeBtn').addEventListener(...)` (intorno alle righe 298-490). Sono ~200 righe.
6. **Rimuovi** ogni altra chiamata residua a `drawCrop()`, `drawScatters()`, `lastCropBbox`, `lastPoints` ovunque nel file. Usa una ricerca grep per trovarle:
```bash
grep -n "drawCrop\|drawScatter\|lastCropBbox\|lastPoints\|cropCanvas\|cropCtx" web/app.js
```
Deve restituire 0 match dopo la pulizia.
- [ ] **Step 2: Aggiungi il blocco FIT MODELLO**
Dopo l'ultimo handler residuo del tab "Trova utensile" (subito prima del commento successivo che separa un'altra sezione, es. `// ==================== VISTA 3D ====================`), aggiungi questo blocco:
```js
// ==================== FIT MODELLO ====================
const findCanvas = $('findResultCanvas');
const findCtx = findCanvas.getContext('2d');
let findResult = null;
let findRgbImg = null;
async function loadRecipes() {
const sel = $('findModelPick');
try {
const r = await api('GET', '/api/models/recipes');
sel.innerHTML = '<option value="">— seleziona modello —</option>';
(r.recipes || []).forEach(rec => {
const opt = document.createElement('option');
opt.value = rec.id;
opt.textContent = `${rec.name} (${rec.symmetry_type})`;
sel.appendChild(opt);
});
sel.disabled = (r.recipes || []).length === 0;
$('findBtn').disabled = !sel.value;
} catch (e) {
sel.innerHTML = `<option value="">errore: ${e.message}</option>`;
}
}
$('findModelPick').addEventListener('change', () => {
$('findBtn').disabled = !$('findModelPick').value;
});
$('findRecipesRefreshBtn').addEventListener('click', loadRecipes);
$('findBtn').addEventListener('click', async () => {
const recipeId = $('findModelPick').value;
if (!recipeId) return;
$('findBtn').disabled = true;
$('findStatus').textContent = 'Cerco posa…';
$('findStatus').className = 'status-line';
try {
const body = {
recipe_id: recipeId,
conf: +$('graspConf').value || null,
class_name: pickedClass(),
};
const r = await api('POST', '/api/grasp/find_model', body);
findResult = r;
findRgbImg = new Image();
findRgbImg.onload = drawFindResult;
findRgbImg.src = 'data:image/jpeg;base64,' + r.rgb_jpeg_b64;
const fit = r.fit_ok
? `OK · fitness=${(r.fitness ?? 0).toFixed(2)} · rmse=${(r.inlier_rmse_mm ?? 0).toFixed(2)}mm`
: `Fit fallito: ${r.fit_reason}`;
$('findStatus').textContent = `${fit} · ${fmtTimings(r.timings_ms)}`;
$('findStatus').className = 'status-line ' + (r.fit_ok ? 'ok' : 'warn');
$('findDetails').textContent = JSON.stringify({
detection: r.detection,
recipe: r.recipe,
n_accepted: r.n_accepted,
n_excluded: r.n_excluded,
fitness: r.fitness,
rmse_mm: r.inlier_rmse_mm,
timings: r.timings_ms,
}, null, 2);
} catch (e) {
$('findStatus').textContent = `errore: ${e.message}`;
$('findStatus').className = 'status-line err';
} finally {
$('findBtn').disabled = !$('findModelPick').value;
}
});
['findShowModel', 'findShowAccepted', 'findShowExcluded'].forEach(id =>
$(id).addEventListener('change', drawFindResult));
function drawFindResult() {
if (!findResult || !findRgbImg || !findRgbImg.complete) return;
const [W, H] = findResult.rgb_size;
if (findCanvas.width !== W) { findCanvas.width = W; findCanvas.height = H; }
findCtx.drawImage(findRgbImg, 0, 0);
const dot = (pts, color, r) => {
findCtx.fillStyle = color;
for (const [u, v] of pts) findCtx.fillRect(u - r, v - r, r*2+1, r*2+1);
};
if ($('findShowExcluded').checked) dot(findResult.excluded_points_px, '#e74c3c', 1);
if ($('findShowAccepted').checked) dot(findResult.accepted_points_px, '#3ecf8e', 1);
if ($('findShowModel').checked) dot(findResult.model_points_px, '#4da3ff', 1);
}
function loadRecipesIfNeeded() {
if ($('tab-grasp').classList.contains('active') && !$('findModelPick').dataset.loaded) {
loadRecipes();
$('findModelPick').dataset.loaded = '1';
}
}
document.querySelectorAll('.tab').forEach(t =>
t.addEventListener('click', () => setTimeout(loadRecipesIfNeeded, 50)));
loadRecipesIfNeeded();
```
- [ ] **Step 3: Verifica nessun riferimento morto**
Run:
```bash
grep -nE "planBtn|executeBtn|drawCrop|drawScatter|cropCanvas|cropCtx|lastCropBbox|lastPoints" web/app.js
```
Expected: nessun output.
- [ ] **Step 4: Smoke-test il sito (server già up via docker compose)**
Apri `http://localhost:8890/` nel browser. Console JS deve essere pulita (nessun `Cannot read property of null`). Il tab "Cerca modello" deve mostrare il picker.
- [ ] **Step 5: Commit**
```bash
git add web/app.js
git commit -m "feat(web): Cerca modello — model picker + find_model overlay handlers
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
```
---
## Task 13: Integration test manuale
Questo non è un test automatico — segui i passi e annota cosa funziona.
**Files:** nessuna modifica al codice.
- [ ] **Step 1: Verifica che tutti i servizi sono up**
```bash
docker compose ps
```
Deve elencare in stato `Up`/`healthy`:
- `presa-mano`
- `nuwa-hp60c`
- `sam21` / `segmentation-sam21`
- `yolo`
- `open3dmodel`
Se manca `open3dmodel`: `docker compose up -d open3dmodel`.
- [ ] **Step 2: Verifica che il proxy ricette funziona**
```bash
curl -s http://localhost:8890/api/models/recipes | jq
```
Expected: `{"recipes": [{"id": "...", "name": "pestello", "symmetry_type": "cylindrical", ...}]}`.
Se 502 "auth": verifica che `OPEN3D_API_KEY` in `presa-mano/.env` combaci con `Open3DModel/.env`.
- [ ] **Step 3: Verifica calibrazione hand-eye + work_plane sono attivi**
```bash
curl -s http://localhost:8890/api/calibration/result?side=right | jq '.calibration | keys'
curl -s http://localhost:8890/api/work_plane | jq '.plane.offset_mm'
```
Devono restituire matrici/numeri, non null. Se mancano, calibrali tramite GUI prima di proseguire.
- [ ] **Step 4: Chiamata find_model felice**
Metti un pestello sul tavolo, poi:
```bash
RECIPE_ID=$(curl -s http://localhost:8890/api/models/recipes | jq -r '.recipes[0].id')
curl -s -X POST http://localhost:8890/api/grasp/find_model \
-H 'Content-Type: application/json' \
-d "{\"recipe_id\":\"$RECIPE_ID\",\"conf\":0.5}" \
| jq '{fit_ok, fit_failed, fit_reason, fitness, rmse:.inlier_rmse_mm, n_accepted, n_excluded, n_model:(.model_points_px|length), total_ms:.timings_ms.total}'
```
Expected (oggetto giusto in scena):
```json
{
"fit_ok": true,
"fit_failed": false,
"fit_reason": null,
"fitness": <≥0.5>,
"rmse": <≤5.0>,
"n_accepted": <>0>,
"n_excluded": <≥0>,
"n_model": <>0, intorno a 2000 entro il bbox>,
"total_ms": <<2000>
}
```
- [ ] **Step 5: Chiamata find_model con recipe_id sbagliato**
```bash
curl -s -X POST http://localhost:8890/api/grasp/find_model \
-H 'Content-Type: application/json' \
-d '{"recipe_id":"bogus-uuid","conf":0.5}' -w "\nHTTP: %{http_code}\n"
```
Expected: HTTP 404 con detail `recipe_id not found: bogus-uuid`.
- [ ] **Step 6: Open3DModel down → 502**
```bash
docker compose stop open3dmodel
curl -s -X POST http://localhost:8890/api/grasp/find_model \
-H 'Content-Type: application/json' \
-d "{\"recipe_id\":\"$RECIPE_ID\",\"conf\":0.5}" -w "\nHTTP: %{http_code}\n"
docker compose start open3dmodel
```
Expected: HTTP 502 con detail `Open3DModel unreachable: …`.
- [ ] **Step 7: Smoke-test GUI**
In browser su `http://localhost:8890/`:
1. Tab "Cerca modello" → dropdown popolato con almeno "pestello".
2. Click "Rileva oggetto" → bbox YOLO visibili a sinistra.
3. Seleziona "pestello" dal dropdown modello.
4. Click "Trova" → entro 2s appare RGB con punti blu sopra il pestello.
5. Toggle "Punti accettati" → comparsa di punti verdi.
6. Toggle "Punti esclusi" → comparsa di punti rossi (se work_plane è calibrato; altrimenti nessun rosso).
7. Nessun errore in console JS.
- [ ] **Step 8: Se tutto OK, annota il successo**
Nessun commit (nessuna modifica al codice). Annota nel file di sessione (o memoria) che l'integrazione end-to-end è verde.
---
## Self-review
Dopo aver completato tutti i task, controlla:
1. **Spec coverage:** Ogni sezione della spec ha un task corrispondente?
- `clients/open3dmodel.py` → Task 5 ✓
- `clients/open3dmodel_cache.py` → Task 6 ✓
- `perception/stl_sampler.py` → Task 3 ✓
- `perception/projection.py` → Task 4 ✓
- `perception/sam21_bridge.py` refactor → Task 7 ✓
- `perception/work_plane.py` (partition_by_plane) → Task 2 ✓
- `api/_deps.py` → Task 8 ✓
- `api/grasp_routes.py` (find_model) → Task 10 ✓
- `api/models_routes.py` + main.py mount → Task 9 ✓
- `config.yaml` + `.env.example` + `pyproject.toml` → Task 1 ✓
- `web/index.html` → Task 11 ✓
- `web/app.js` → Task 12 ✓
- Test unit (partition, sampler, projection, client, cache, bridge) → Task 2/3/4/5/6/7 ✓
- Integration manuale → Task 13 ✓
2. **Comandi e percorsi:** Tutti i path sono assoluti dalla repo root o relativi a essa. Tutti i comandi `uv run`/`curl`/`grep` sono completi.
3. **Tipi coerenti:** `RecipeMesh.vertices/triangles` usati identici tra Task 3, 6, 10. `Sam21Segmentation.points_cv_m/rgb_jpeg/intrinsics_rgb` identici tra Task 7 e 10.
Reference in new issue View Git Blame Copy Permalink