diff --git a/scripts/calibrazione.py b/scripts/calibrazione.py new file mode 100644 index 0000000..76abd81 --- /dev/null +++ b/scripts/calibrazione.py @@ -0,0 +1,144 @@ +"""Calibrazione a scacchiera dell'acquisitore (TieMeasureFlow, #83). + +Cosa fa: da una foto di scacchiera piana calcola la scala (px/mm) con +TieVision-Core (visionsuite.calibration), e conserva un risultato per sorgente +camera in un JSON (scrittura atomica). +Uso: importato da scripts/camera_service.py (rotte /camera/calibrate e +/camera/calibration); `calibra_scacchiera(gray, cols, rows, square_mm)` e' +logica pura, senza FastAPI, e si prova da sola. +Dipende da: numpy, opencv, visionsuite (vs-core). Niente altro. +""" +from __future__ import annotations + +import base64 +import hashlib +import json +import math +import os +import re +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +import cv2 +import numpy as np + +from visionsuite.calibration import calibrate_planar, find_chessboard +from visionsuite.core.image import Image + + +class CalibrazioneError(ValueError): + """Parametri invalidi o scacchiera non trovata: input sbagliato, non un guasto (422).""" + + +def _gray_u8(pixels: np.ndarray) -> np.ndarray: + """Immagine -> grigio uint8 2D.""" + px = np.asarray(pixels) + if px.ndim == 3: + px = cv2.cvtColor(px, cv2.COLOR_BGRA2GRAY if px.shape[2] == 4 else cv2.COLOR_BGR2GRAY) + if px.dtype != np.uint8: + # float/uint16 -> scala sul massimo, come fa vs-core + m = float(px.max()) if px.size else 1.0 + px = np.clip(px.astype(np.float32) / max(m, 1e-9) * 255, 0, 255).astype(np.uint8) + return np.ascontiguousarray(px) + + +def calibra_scacchiera(pixels, cols: int, rows: int, square_mm: float) -> dict: + """Scala (px/mm) da una foto di scacchiera piana. cols/rows = angoli INTERNI. + + Ritorna {px_per_mm, mm_per_px, rms_px, n_corners, cols, rows, square_mm, + image_width, image_height, scale_spread_pct, overlay_png (b64)}. + """ + if cols < 2 or rows < 2: + raise CalibrazioneError("cols e rows (angoli interni) devono essere almeno 2") + if not (square_mm > 0 and math.isfinite(square_mm)): + raise CalibrazioneError("square_mm deve essere > 0") + + frame = _gray_u8(pixels) + image = Image(frame) + try: + image_pts, world_pts = find_chessboard(image, (cols, rows), square_mm) + except RuntimeError as exc: + # find_chessboard solleva RuntimeError quando la scacchiera non c'e' + raise CalibrazioneError( + f"scacchiera {cols}x{rows} non trovata nell'immagine: {exc}") from exc + model = calibrate_planar(image, image_pts, world_pts) # omografia pura + + # Scala al centro della scacchiera (puo' non stare al centro immagine) + x0, y0 = image_pts.min(axis=0) + x1, y1 = image_pts.max(axis=0) + mm_per_px = model.pixel_size((x0 + x1) / 2.0, (y0 + y1) / 2.0) + if not (mm_per_px > 0 and math.isfinite(mm_per_px)): + raise CalibrazioneError("calibrazione degenere: scala non valida") + # Scarto massimo % della scala ai quattro angoli dell'area della scacchiera + # rispetto al centro: ottica telecentrica/piana ~ 0, prospettiva = grande. + spread = max(abs(model.pixel_size(x, y) - mm_per_px) / mm_per_px * 100.0 + for x in (x0, x1) for y in (y0, y1)) + + # Overlay: angoli disegnati sull'immagine (PNG base64) + bgr = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR) + cv2.drawChessboardCorners( + bgr, (cols, rows), image_pts.reshape(-1, 1, 2).astype(np.float32), True) + ok, png = cv2.imencode(".png", bgr) + if not ok: + raise RuntimeError("PNG dell'overlay non codificabile") + + h, w = frame.shape + return { + "px_per_mm": 1.0 / mm_per_px, + "mm_per_px": mm_per_px, + "rms_px": float(model.rms_reprojection), + "n_corners": int(len(image_pts)), + "cols": int(cols), + "rows": int(rows), + "square_mm": float(square_mm), + "image_width": int(w), + "image_height": int(h), + "scale_spread_pct": float(spread), + "overlay_png": base64.b64encode(png.tobytes()).decode("ascii"), + } + + +def ora_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +class Archivio: + """Un JSON per sorgente in `cartella`; nome file dalla sorgente ripulito.""" + + def __init__(self, cartella: Path | str): + self.cartella = Path(cartella) + + def _file(self, sorgente: str) -> Path: + # caratteri sicuri + hash dell'originale: due sorgenti diverse non si pestano + pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", sorgente).strip("._") or "sorgente" + sigla = hashlib.sha1(sorgente.encode("utf-8")).hexdigest()[:8] + return self.cartella / f"{pulito[:60]}-{sigla}.json" + + def leggi(self, sorgente: str) -> dict | None: + try: + return json.loads(self._file(sorgente).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None # assente o illeggibile = non calibrata + + def salva(self, sorgente: str, dati: dict) -> None: + destinazione = self._file(sorgente) + self.cartella.mkdir(parents=True, exist_ok=True) + # scrittura atomica: file temporaneo nella stessa cartella, poi replace + fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(dati, f, ensure_ascii=False, indent=2) + os.replace(tmp, destinazione) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + def elimina(self, sorgente: str) -> None: + try: + self._file(sorgente).unlink() + except FileNotFoundError: + pass diff --git a/scripts/calibrazione_pagina.py b/scripts/calibrazione_pagina.py new file mode 100644 index 0000000..33a9409 --- /dev/null +++ b/scripts/calibrazione_pagina.py @@ -0,0 +1,218 @@ +"""Pagina HTML della calibrazione a scacchiera (servita da camera_service su /calibrazione, #83). + +Cosa fa: una pagina sola, autonoma e offline (niente CDN, JS e CSS inline), in +italiano. Parla con le rotte dello stesso servizio (/camera/sources, +/camera/calibration, /camera/calibrate). +Dipende da: niente. E' una stringa, cosi' il pacchetto compilato non ha file dati. +""" + +PAGINA = """ + + + + +Calibrazione camera + + + +
+

Calibrazione camera con scacchiera

+ +
+
+
+
+
+
+
+

+ Gli angoli interni sono quelli dove si toccano quattro quadrati: una scacchiera di 10 x 7 quadrati ha 9 x 6 angoli. + Tenere la scacchiera piana e ferma, ben illuminata.

+
+ +
+
+ +
+ +
+
+
+ + + +
+ Calibrazione salvata +
-
+
+ +
+
+
+ + + + +""" diff --git a/scripts/camera_service.py b/scripts/camera_service.py index 15b8f50..ad4e894 100644 --- a/scripts/camera_service.py +++ b/scripts/camera_service.py @@ -6,17 +6,33 @@ OpenCV, cartella di immagini) con lo stesso contratto di `vs-gui`, ma solo GET /camera/sources -> {"sources": [{name, kind, label}]} POST /camera/grab json {source, trigger="continuo", timeout_ms=5000} -> PNG Errori: 400 parametri, 404 sorgente inesistente, 503 guasto camera, 504 timeout. + +Calibrazione a scacchiera per sorgente (#83, logica in scripts/calibrazione.py): + GET /camera/calibration?source= -> 200 {source, px_per_mm, mm_per_px, + rms_px, n_corners, cols, rows, square_mm, image_width, image_height, + scale_spread_pct, calibrated_at} | 404 "sorgente non calibrata" + POST /camera/calibrate multipart: source, cols, rows (angoli interni), + square_mm, image (file, opzionale: senza, grab dalla sorgente come + /camera/grab), save=false -> campi di sopra + overlay_png (b64) + saved; + 422 scacchiera non trovata o parametri invalidi + DELETE /camera/calibration?source= -> 204 + GET /calibrazione -> pagina HTML autonoma (offline) per fare tutto a mano +Il risultato salvato sta in un JSON per sorgente in --config-dir (default +%ProgramData%/Tielogic/TieMeasureFlow/acquisitore su Windows, ./acquisitore altrove). + Serve perche' vs-gui ascolta solo su 127.0.0.1 e il container `client` non lo raggiunge; questo servizio permette di scegliere l'host di ascolto. Uso (dalla radice del repo): vendor/visionsuite/.venv/bin/python scripts/camera_service.py \ - [--host 127.0.0.1] [--port 8765] [--folder ] + [--host 127.0.0.1] [--port 8765] [--folder ] \ + [--config-dir ] --folder abilita la sorgente finta `camera:folder` (loop sulle immagini). Da Docker su Linux: --host 172.17.0.1 (gateway di docker0). Dipende da: venv di VisionSuite (`uv sync --all-packages` in vendor/visionsuite): -vs-camera, fastapi, uvicorn, opencv. +vs-camera, fastapi, uvicorn, opencv, python-multipart (rotta /camera/calibrate), +visionsuite.calibration (vs-core). AVVERTENZA: nessuna autenticazione. Un --host diverso dal loopback espone la camera a chiunque raggiunga la porta: solo su rete isolata. @@ -24,17 +40,26 @@ camera a chiunque raggiunga la porta: solo su rete isolata. from __future__ import annotations import argparse +import os +import sys import threading from contextlib import asynccontextmanager from pathlib import Path import cv2 -from fastapi import Body, FastAPI, HTTPException -from fastapi.responses import Response +import numpy as np +from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile +from fastapi.concurrency import run_in_threadpool # #83: il grab bloccante fuori dall'event loop +from fastapi.responses import HTMLResponse, Response from visionsuite.camera.camera_interface import CameraError, CameraGrabError from visionsuite.camera.sources import SourceNotFound, list_sources, open_source +# #83: calibrazione e pagina stanno accanto a questo file (che non e' un pacchetto) +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import calibrazione # noqa: E402 +from calibrazione_pagina import PAGINA # noqa: E402 + _TRIGGER = ("continuo",) # errori noti di vs-camera (inglese) -> italiano; esatti, poi per prefisso @@ -61,7 +86,18 @@ def _traduci(messaggio: str) -> str: return messaggio -def create_app(folder: Path | str | None = None) -> FastAPI: +def _config_dir_default() -> Path: + """#83: cartella delle calibrazioni salvate.""" + if os.name == "nt": + base = os.environ.get("ProgramData") or "C:/ProgramData" + return Path(base) / "Tielogic" / "TieMeasureFlow" / "acquisitore" + return Path("acquisitore") + + +def create_app(folder: Path | str | None = None, + config_dir: Path | str | None = None) -> FastAPI: + archivio = calibrazione.Archivio(config_dir if config_dir is not None + else _config_dir_default()) # #83 aperte: dict = {} lock = threading.Lock() # protegge SOLO i dizionari, mai un'apertura lock_sorgenti: dict = {} # nome sorgente -> Lock che serializza apertura e grab @@ -95,6 +131,24 @@ def create_app(folder: Path | str | None = None) -> FastAPI: return {"sources": [{"name": v.name, "kind": v.kind, "label": v.label} for v in voci if v.kind in ("camera", "virtual")]} + def _grab(nome: str, timeout: int): + """Un frame dalla sorgente (apre e tiene aperta); errori camera -> HTTP.""" + try: + with _lock_di(nome): # solo questa sorgente; le altre non aspettano + with lock: + sorgente = aperte.get(nome) + if sorgente is None: + sorgente = open_source(nome, None, config=_config()) + with lock: + aperte[nome] = sorgente + return sorgente.grab(timeout) + except SourceNotFound as e: + raise HTTPException(404, _traduci(str(e))) + except TimeoutError as e: + raise HTTPException(504, _traduci(str(e)) or "timeout") + except (CameraGrabError, CameraError) as e: + raise HTTPException(503, _traduci(str(e))) + @app.post("/camera/grab") def camera_grab(corpo: dict = Body(default=None)): nome = corpo.get("source") if isinstance(corpo, dict) else None @@ -107,26 +161,63 @@ def create_app(folder: Path | str | None = None) -> FastAPI: timeout = corpo.get("timeout_ms", 5000) if isinstance(timeout, bool) or not isinstance(timeout, int) or not 1 <= timeout <= 10000: raise HTTPException(400, "«timeout_ms» deve essere un intero fra 1 e 10000") - try: - with _lock_di(nome): # solo questa sorgente; le altre non aspettano - with lock: - sorgente = aperte.get(nome) - if sorgente is None: - sorgente = open_source(nome, None, config=_config()) - with lock: - aperte[nome] = sorgente - frame = sorgente.grab(timeout) - except SourceNotFound as e: - raise HTTPException(404, _traduci(str(e))) - except TimeoutError as e: - raise HTTPException(504, _traduci(str(e)) or "timeout") - except (CameraGrabError, CameraError) as e: - raise HTTPException(503, _traduci(str(e))) + frame = _grab(nome, timeout) # #83: corpo estratto in _grab, invariato ok, buf = cv2.imencode(".png", frame) if not ok: raise HTTPException(500, "codifica PNG fallita") return Response(buf.tobytes(), media_type="image/png") + # ---- #83: calibrazione a scacchiera per sorgente + + def _sorgente(valore) -> str: + if not isinstance(valore, str) or not valore.strip(): + raise HTTPException(400, "manca «source»") + return valore.strip() + + @app.get("/camera/calibration") + def calibrazione_leggi(source: str = ""): + dati = archivio.leggi(_sorgente(source)) + if dati is None: + raise HTTPException(404, "sorgente non calibrata") + return dati + + @app.delete("/camera/calibration", status_code=204) + def calibrazione_elimina(source: str = ""): + archivio.elimina(_sorgente(source)) + return Response(status_code=204) + + @app.post("/camera/calibrate") + async def calibrazione_calcola( + source: str = Form(...), + cols: int = Form(...), + rows: int = Form(...), + square_mm: float = Form(...), + image: UploadFile | None = File(None), + save: bool = Form(False), + ): + nome = _sorgente(source) + contenuto = await image.read() if image is not None else b"" + if contenuto: + frame = cv2.imdecode(np.frombuffer(contenuto, np.uint8), cv2.IMREAD_UNCHANGED) + if frame is None: + raise HTTPException(422, "l'immagine caricata non è leggibile") + else: + frame = await run_in_threadpool(_grab, nome, 5000) # come /camera/grab (404/503/504) + try: + risultato = calibrazione.calibra_scacchiera(frame, cols, rows, square_mm) + except calibrazione.CalibrazioneError as e: + raise HTTPException(422, str(e)) + corpo = {"source": nome, + **{k: v for k, v in risultato.items() if k != "overlay_png"}, + "calibrated_at": calibrazione.ora_iso()} + if save: + archivio.salva(nome, corpo) + return {**corpo, "overlay_png": risultato["overlay_png"], "saved": bool(save)} + + @app.get("/calibrazione", response_class=HTMLResponse) + def calibrazione_pagina(): + return HTMLResponse(PAGINA) + return app @@ -136,8 +227,10 @@ def main() -> None: p.add_argument("--host", default="127.0.0.1") p.add_argument("--port", type=int, default=8765) p.add_argument("--folder", default=None, help="cartella immagini per camera:folder") + p.add_argument("--config-dir", default=None, + help="cartella delle calibrazioni salvate (default: ProgramData su Windows)") # #83 a = p.parse_args() - uvicorn.run(create_app(a.folder), host=a.host, port=a.port) + uvicorn.run(create_app(a.folder, a.config_dir), host=a.host, port=a.port) if __name__ == "__main__": diff --git a/scripts/tests/test_calibrazione.py b/scripts/tests/test_calibrazione.py new file mode 100644 index 0000000..726ec1a --- /dev/null +++ b/scripts/tests/test_calibrazione.py @@ -0,0 +1,184 @@ +"""Test della calibrazione dell'acquisitore (#83): logica pura e rotte di camera_service. + +Uso: dalla radice, `uv run --python 3.13 --extra vision-worker --extra dev +--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q` +(o il venv di VisionSuite come per test_camera_service.py). +""" +import base64 +import importlib.util +import json +from pathlib import Path + +import cv2 +import numpy as np +import pytest +from fastapi.testclient import TestClient + +SCRIPTS = Path(__file__).resolve().parents[1] +MODULO = SCRIPTS / "camera_service.py" + + +def _carica(): + spec = importlib.util.spec_from_file_location("camera_service", MODULO) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def una_scacchiera(square_px: int = 40, nx: int = 10, ny: int = 7, margin: int = 60) -> np.ndarray: + """nx x ny quadrati => (nx-1) x (ny-1) angoli interni, su fondo bianco.""" + img = np.full((ny * square_px + 2 * margin, nx * square_px + 2 * margin), 255, np.uint8) + for j in range(ny): + for i in range(nx): + if (i + j) % 2 == 0: + y, x = margin + j * square_px, margin + i * square_px + img[y:y + square_px, x:x + square_px] = 0 + return img + + +def _png(img: np.ndarray) -> bytes: + ok, buf = cv2.imencode(".png", img) + assert ok + return buf.tobytes() + + +@pytest.fixture +def cartella(tmp_path): + # la sorgente camera:folder fa il loop su una scacchiera: il grab la trova + cartella = tmp_path / "immagini" + cartella.mkdir() + cv2.imwrite(str(cartella / "scacchiera.png"), una_scacchiera()) + return cartella + + +@pytest.fixture +def client(cartella, tmp_path): + mod = _carica() + with TestClient(mod.create_app(cartella, config_dir=tmp_path / "config")) as c: + c.config = tmp_path / "config" + yield c + + +CAMPI = {"cols": "9", "rows": "6", "square_mm": "2"} +CHIAVI_C1 = {"source", "px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows", + "square_mm", "image_width", "image_height", "scale_spread_pct", "calibrated_at"} +NON_IMMAGINE = b"questo non e una immagine" + + +# ---- logica pura + +def test_scacchiera_sintetica_da_la_scala(): + _carica() # mette scripts/ in sys.path + import calibrazione + r = calibrazione.calibra_scacchiera(una_scacchiera(), 9, 6, 2.0) + assert r["px_per_mm"] == pytest.approx(20.0, rel=0.005) + assert r["mm_per_px"] == pytest.approx(0.05, rel=0.005) + assert r["rms_px"] < 0.5 + assert r["n_corners"] == 54 + assert (r["image_width"], r["image_height"]) == (520, 400) + assert r["scale_spread_pct"] < 1.0 + assert base64.b64decode(r["overlay_png"])[:8] == b"\x89PNG\r\n\x1a\n" + + +def test_immagine_bianca_solleva(): + _carica() + import calibrazione + with pytest.raises(calibrazione.CalibrazioneError, match="non trovata"): + calibrazione.calibra_scacchiera(np.full((300, 300), 255, np.uint8), 9, 6, 2.0) + + +@pytest.mark.parametrize("cols,rows,quadrato", [(1, 6, 2.0), (9, 6, 0.0), (9, 6, -1.0)]) +def test_parametri_invalidi_sollevano(cols, rows, quadrato): + _carica() + import calibrazione + with pytest.raises(calibrazione.CalibrazioneError): + calibrazione.calibra_scacchiera(una_scacchiera(), cols, rows, quadrato) + + +# ---- rotte + +def test_c1_non_calibrata_404(client): + r = client.get("/camera/calibration", params={"source": "camera:folder"}) + assert r.status_code == 404 + assert r.json()["detail"] == "sorgente non calibrata" + + +def test_c1_senza_source_400(client): + assert client.get("/camera/calibration").status_code == 400 + + +def test_c2_immagine_caricata_senza_salvare(client): + r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI}, + files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 200, r.text + corpo = r.json() + assert CHIAVI_C1 | {"overlay_png", "saved"} == set(corpo) + assert corpo["saved"] is False + assert corpo["px_per_mm"] == pytest.approx(20.0, rel=0.005) + # senza save non si scrive nulla + assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404 + assert not client.config.exists() or not list(client.config.glob("*.json")) + + +def test_c2_c1_c3_ciclo_con_grab_dalla_sorgente(client): + # nessun file: grab da camera:folder (la scacchiera in cartella), con save + r = client.post("/camera/calibrate", + data={"source": "camera:folder", "save": "true", **CAMPI}) + assert r.status_code == 200, r.text + assert r.json()["saved"] is True + + r1 = client.get("/camera/calibration", params={"source": "camera:folder"}) + assert r1.status_code == 200 + assert set(r1.json()) == CHIAVI_C1 # nessun overlay nel file salvato + assert r1.json()["px_per_mm"] == pytest.approx(20.0, rel=0.005) + assert r1.json()["source"] == "camera:folder" + file_salvati = list(client.config.glob("*.json")) + assert len(file_salvati) == 1 + assert json.loads(file_salvati[0].read_text(encoding="utf-8"))["cols"] == 9 + + assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204 + assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404 + # cancellare di nuovo e' idempotente + assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204 + + +def test_c2_scacchiera_non_trovata_422(client): + bianca = np.full((300, 300), 255, np.uint8) + r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI}, + files={"image": ("b.png", _png(bianca), "image/png")}) + assert r.status_code == 422 + assert "non trovata" in r.json()["detail"] + + +def test_c2_parametri_invalidi_422(client): + r = client.post("/camera/calibrate", + data={"source": "camera:folder", "cols": "1", "rows": "6", "square_mm": "2"}, + files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 422 + + +def test_c2_immagine_illeggibile_422(client): + r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI}, + files={"image": ("x.png", NON_IMMAGINE, "image/png")}) + assert r.status_code == 422 + + +def test_c2_errori_camera_come_grab(client): + r = client.post("/camera/calibrate", data={"source": "camera:non-esiste", **CAMPI}) + assert r.status_code == 404 + + +def test_due_sorgenti_hanno_file_diversi(client): + for nome in ("camera:a b", "camera:a_b"): + r = client.post("/camera/calibrate", data={"source": nome, "save": "true", **CAMPI}, + files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 200 + assert len(list(client.config.glob("*.json"))) == 2 + + +def test_c4_pagina_html(client): + r = client.get("/calibrazione") + assert r.status_code == 200 + assert r.headers["content-type"].startswith("text/html") + assert "Acquisisci e calcola" in r.text + assert "http://" not in r.text and "https://" not in r.text # offline, niente CDN diff --git a/src/vision/costruttori.py b/src/vision/costruttori.py new file mode 100644 index 0000000..1475be2 --- /dev/null +++ b/src/vision/costruttori.py @@ -0,0 +1,228 @@ +"""Costruttore di ricette DXF compare dal campo (immagine + DXF + scala -> ricetta v2). + +Logica pura, niente FastAPI ne' database: il worker (src/vision_worker/main.py) +lo espone come /recipes/dxf-compare, i test lo chiamano direttamente. Tutta la +visione sta in TieVision-Core (visionsuite.*). La calibrazione a scacchiera NON +sta qui: vive nell'acquisitore (scripts/calibrazione.py, #83). + +Errori di input (pezzo non trovato, DXF illeggibile, parametri fuori range) +sono `CostruttoreError`: il worker li mappa a 422. +""" +from __future__ import annotations + +import base64 +import hashlib +import json +import math +import tempfile +from pathlib import Path + +import numpy as np +from visionsuite.core.image import Image + +from src.vision.runner import RunOutcome, run_graph + + +class CostruttoreError(ValueError): + """Input che non permette di costruire (422 per il worker), non un guasto.""" + + +# Nomi/id fissi della ricetta (contratto #83 con backend e frontend). +DXF_FILE = "disegno.dxf" +CALIBRATION_FILE = "calibrazione.json" +_POLARITIES = ("dark", "light") +# Pose iniziali del disegno provate (gradi): il verso del pezzo non e' noto. +_OUTPUT_KEYS = ("dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max") +_START_ANGLES = (0.0, 90.0, 180.0, 270.0) + + +def _b64(data: bytes) -> str: + return base64.b64encode(data).decode("ascii") + + +def _gray_u8(pixels: np.ndarray) -> np.ndarray: + """Immagine -> grigio uint8 2D (LocateBlob accetta solo uint8).""" + px = np.asarray(pixels) + if px.ndim == 3: + px = px[..., 0] # multicanale: il primo, come fa LocateBlob + if px.dtype != np.uint8: + # float/uint16 -> scala sul massimo, come _to_gray_uint8 di vs-core + m = float(px.max()) if px.size else 1.0 + px = np.clip(px.astype(np.float32) / max(m, 1e-9) * 255, 0, 255).astype(np.uint8) + return np.ascontiguousarray(px) + + +# ------------------------------------------------------------------ W2 + +def _graph(dxf_sha: str, cal_sha: str, band_mm: float, min_points_pct: float, polarity: str, + reference_pose: dict, anchor: tuple[float, float, float] | None) -> dict: + """Ricetta v2 fissa: LocateBlob "pezzo" + CompareDxf "dxf" agganciato a `align`. + + `anchor` = (riga, colonna, angolo) dell'origine del disegno nell'immagine di + riferimento; None = solo il localizzatore (passo intermedio). + """ + tools: list[dict] = [{ + "id": "pezzo", "name": "Pezzo", "kind": "LocateBlob", + "values": {"polarity": polarity, "reference_pose": reference_pose}, + "links": {}, + }] + if anchor is not None: + row, col, angle = anchor + tools.append({ + "id": "dxf", "name": "Confronto DXF", "kind": "CompareDxf", + "values": { + "dxf": "disegno", + "anchor": "align", + "anchor_offset_row_px": float(row), + "anchor_offset_col_px": float(col), + "offset_angle_deg": float(angle), + "band_plus": float(band_mm), + "band_minus": float(band_mm), + "nominal_unit": "mm", + # tolleranza di points_in_pct: nominale = limite, plus = 100 - limite + "points_in_pct_nominal": float(min_points_pct), + "points_in_pct_plus": float(100.0 - min_points_pct), + "points_in_pct_minus": 0.0, + }, + "links": {"align": {"tool": "pezzo", "property": "poses"}}, + }) + return { + "schema_version": 2, + "name": "Confronto DXF", + # sha256 obbligatorio per vs-task (verifica i file della cartella ricetta) + "meta": {"dxf": {"disegno": {"path": DXF_FILE, "sha256": dxf_sha}}, + "calibration": {"path": CALIBRATION_FILE, "sha256": cal_sha}}, + "tools": tools, + } + + +def _locate_reference(gray: np.ndarray, polarity: str) -> dict: + """Posa del pezzo sull'immagine di riferimento (LocateBlob, soglia Otsu).""" + from visionsuite.task import Task, load_tools + + load_tools() + graph = _graph('', '', 1.0, 95.0, polarity, {"row": 0.0, "col": 0.0, "angle_deg": 0.0}, None) + task = Task.from_dict(graph) + task.run(context={"image": Image(gray)}) + pezzo = task["Pezzo"] + if not pezzo.found_poses: + raise CostruttoreError( + "pezzo non trovato nell'immagine di riferimento " + f"(polarity={polarity}): controllare luce, sfondo e polarita'" + ) + return pezzo.found_poses[0].to_dict() + + +def _candidate_anchors(gray: np.ndarray, dxf_path: Path, px_per_mm: float) -> list[tuple[float, float, float]]: + """Pose (riga, col, angolo) del disegno sull'immagine: ICP da 4 orientamenti. + + Il disegno e' in px (origine DXF, y capovolta); si parte da ognuno dei 4 + angoli con i centroidi sovrapposti e l'ICP rifinisce. Ritorna le posa trovate + (anche per orientamenti che non convergono: le giudica la ricetta eseguita). + """ + from visionsuite.calibration import ScaleCalibration + from visionsuite.dxf.compare import DXFCompare, _stack + from visionsuite.task.pose import Pose + + cmp_ = DXFCompare(dxf_path, ScaleCalibration(px_per_mm=px_per_mm)) + drawing = _stack(cmp_.drawing_contours()) + measured = _stack(cmp_.measured_contours(Image(gray))) + if len(drawing) < 3: + raise CostruttoreError("DXF senza geometria utilizzabile") + if len(measured) < 3: + raise CostruttoreError("pezzo non trovato: nessun bordo nell'immagine") + dc, mc = drawing.mean(axis=0), measured.mean(axis=0) + + anchors = [] + for theta in _START_ANGLES: + # traslazione che porta il centroide del disegno ruotato sul misurato + rr, cc = Pose(0.0, 0.0, theta).apply(dc[0], dc[1]) + start = Pose(float(mc[0] - rr), float(mc[1] - cc), theta) + try: + result = cmp_.compare(Image(gray), initial_pose=start) + except Exception: # noqa: BLE001 - un orientamento che cade non ferma gli altri + continue + # transform: misurato -> disegno; la posa del disegno e' l'inversa + m = np.linalg.inv(result.transform) + angle = math.degrees(math.atan2(m[1, 0], m[0, 0])) + anchors.append((float(m[0, 2]), float(m[1, 2]), angle)) + return anchors + + +def _run(gray, graph, px_per_mm, folder) -> RunOutcome: + return run_graph(gray, graph, calibration={"px_per_mm": px_per_mm}, recipe_dir=folder) + + +def build_dxf_compare( + gray, + dxf_bytes: bytes, + px_per_mm: float, + band_mm: float = 0.1, + min_points_pct: float = 95.0, + polarity: str = "dark", +) -> dict: + """Ricetta DXF compare dall'immagine di riferimento + disegno + scala. + + Ritorna {graph, files, outputs, overlay, outcome}. `files` sono i due file + della cartella ricetta (b64): disegno.dxf e calibrazione.json. + """ + if not (px_per_mm > 0 and math.isfinite(px_per_mm)): + raise CostruttoreError("px_per_mm deve essere > 0") + if not (band_mm > 0 and math.isfinite(band_mm)): + raise CostruttoreError("band_mm deve essere > 0") + if not (0 < min_points_pct <= 100): + raise CostruttoreError("min_points_pct deve stare in (0, 100]") + if polarity not in _POLARITIES: + raise CostruttoreError(f"polarity deve essere una di {list(_POLARITIES)}") + if not dxf_bytes: + raise CostruttoreError("DXF vuoto") + + frame = _gray_u8(gray) + calibration_bytes = json.dumps({"px_per_mm": px_per_mm}).encode("utf-8") + dxf_sha = hashlib.sha256(dxf_bytes).hexdigest() + cal_sha = hashlib.sha256(calibration_bytes).hexdigest() + + with tempfile.TemporaryDirectory() as tmp: + folder = Path(tmp) + dxf_path = folder / DXF_FILE + dxf_path.write_bytes(dxf_bytes) + (folder / CALIBRATION_FILE).write_bytes(calibration_bytes) + + # DXF leggibile? (ezdxf solleva su file corrotto/non DXF) + from visionsuite.io.dxf import read_xld_dxf + try: + drawing = list(read_xld_dxf(str(dxf_path), sampling_distance=0.1, + scale=px_per_mm, flip_y=True)) + except Exception as exc: # noqa: BLE001 - qualunque errore di lettura = DXF illeggibile + raise CostruttoreError(f"DXF illeggibile: {type(exc).__name__}: {exc}") from exc + if not drawing: + raise CostruttoreError("DXF illeggibile: nessuna entita' geometrica") + + reference_pose = _locate_reference(frame, polarity) + anchors = _candidate_anchors(frame, dxf_path, px_per_mm) + if not anchors: + raise CostruttoreError("disegno non sovrapponibile al pezzo (ICP fallito)") + + # Verita' = eseguire davvero la ricetta sul riferimento: vince il maggior points_in_pct + best: tuple[float, dict, RunOutcome] | None = None + for anchor in anchors: + graph = _graph(dxf_sha, cal_sha, band_mm, min_points_pct, polarity, reference_pose, anchor) + outcome = _run(frame, graph, px_per_mm, folder) + pct = outcome.outputs.get("dxf.points_in_pct") + if pct is not None and (best is None or pct > best[0]): + best = (pct, graph, outcome) + if best is None: + raise CostruttoreError( + "disegno non sovrapponibile al pezzo: il confronto non ha prodotto punti" + ) + _, graph, outcome = best + + # solo le quattro uscite del contratto (n_points resta fuori) + outputs = {k: v for k, v in outcome.outputs.items() if k in _OUTPUT_KEYS} + return { + "graph": graph, + "files": {DXF_FILE: _b64(dxf_bytes), CALIBRATION_FILE: _b64(calibration_bytes)}, + "outputs": outputs, + "overlay": outcome.overlay, + "outcome": outcome.outcome, + } diff --git a/src/vision/tests/test_costruttori.py b/src/vision/tests/test_costruttori.py new file mode 100644 index 0000000..0365abc --- /dev/null +++ b/src/vision/tests/test_costruttori.py @@ -0,0 +1,114 @@ +"""Costruttore di ricette dal campo: immagine + DXF + scala -> ricetta v2.""" +import pytest + +# Come test_runner.py: l'albero vision richiede Python >=3.13; su 3.11 e' skipped. +pytest.importorskip( + "visionsuite.task", + reason=( + "the vision tree requires Python >=3.13; run it with " + "`uv run --python 3.13 --extra vision --extra dev pytest src/vision/tests`" + ), +) + +import base64 +import json +from pathlib import Path + +import cv2 +import numpy as np + +from src.vision.costruttori import ( + CALIBRATION_FILE, + DXF_FILE, + CostruttoreError, + build_dxf_compare, +) +from src.vision.runner import run_graph + +SAMPLE_DXF = ( + Path(__file__).resolve().parents[3] + / "vendor/visionsuite/packages/vs-dxf/tests/fixtures/sample.dxf" +) +DXF_WIDTH_MM, DXF_HEIGHT_MM, DXF_HOLE_R_MM = 200.0, 100.0, 10.0 +PX_PER_MM = 4.0 + + +def a_scene(angle_deg: float = 0.0, shift_px: tuple[float, float] = (0.0, 0.0), + margin_mm: float = 60.0) -> np.ndarray: + """Il pezzo del sample.dxf (scuro su bianco) a 4 px/mm, ruotato e traslato.""" + w = int((DXF_WIDTH_MM + 2 * margin_mm) * PX_PER_MM) + h = int((DXF_HEIGHT_MM + 2 * margin_mm) * PX_PER_MM) + img = np.full((h, w), 255, np.uint8) + cx, cy = w // 2, h // 2 + hw, hh = int(DXF_WIDTH_MM * PX_PER_MM / 2), int(DXF_HEIGHT_MM * PX_PER_MM / 2) + cv2.rectangle(img, (cx - hw, cy - hh), (cx + hw, cy + hh), 0, -1) + cv2.circle(img, (cx, cy), int(DXF_HOLE_R_MM * PX_PER_MM), 255, -1) + if angle_deg or any(shift_px): + m = cv2.getRotationMatrix2D((cx, cy), angle_deg, 1.0) # >0 = antiorario a video + m[0, 2] += shift_px[0] + m[1, 2] += shift_px[1] + img = cv2.warpAffine(img, m, (w, h), flags=cv2.INTER_LINEAR, borderValue=255) + return img + + +# ------------------------------------------------------------------ ricetta DXF + +def test_ricetta_dxf_gira_su_riferimento_e_su_scena_ruotata(tmp_path): + dxf = SAMPLE_DXF.read_bytes() + risultato = build_dxf_compare(a_scene(0.0), dxf, px_per_mm=PX_PER_MM, band_mm=2.0) + + assert risultato["outputs"]["dxf.points_in_pct"] > 95.0 # sul riferimento + assert set(risultato["outputs"]) == { + "dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max"} + assert set(risultato["files"]) == {DXF_FILE, CALIBRATION_FILE} + assert json.loads(base64.b64decode(risultato["files"][CALIBRATION_FILE])) == { + "px_per_mm": PX_PER_MM} + assert risultato["overlay"] and risultato["outcome"] is not None + + graph = risultato["graph"] + assert graph["schema_version"] == 2 + # path come da contratto; lo sha256 lo esige vs-task per ogni file in meta + assert graph["meta"]["dxf"]["disegno"]["path"] == "disegno.dxf" + assert graph["meta"]["calibration"]["path"] == "calibrazione.json" + assert len(graph["meta"]["dxf"]["disegno"]["sha256"]) == 64 + pezzo, dxf_tool = graph["tools"] + assert (pezzo["id"], pezzo["kind"], pezzo["name"]) == ("pezzo", "LocateBlob", "Pezzo") + assert (dxf_tool["id"], dxf_tool["kind"], dxf_tool["name"]) == ("dxf", "CompareDxf", "Confronto DXF") + assert dxf_tool["links"] == {"align": {"tool": "pezzo", "property": "poses"}} + + # La cartella ricetta come la scriverebbe il backend, poi la scena ruotata di 5 gradi + for nome, b64 in risultato["files"].items(): + (tmp_path / nome).write_bytes(base64.b64decode(b64)) + ruotata = a_scene(5.0, shift_px=(30.0, 20.0)) + esito = run_graph(ruotata, graph, calibration={"px_per_mm": PX_PER_MM}, recipe_dir=tmp_path) + assert not esito.failures, esito.failures + assert esito.outputs["dxf.points_in_pct"] > 95.0 + + +def test_ricetta_dxf_con_riferimento_gia_ruotato(tmp_path): + # il riferimento non e' a 0 gradi: la ricetta si costruisce lo stesso + risultato = build_dxf_compare(a_scene(20.0, shift_px=(-15.0, 10.0)), + SAMPLE_DXF.read_bytes(), PX_PER_MM, band_mm=2.0) + assert risultato["outputs"]["dxf.points_in_pct"] > 95.0 + + +def test_immagine_senza_pezzo_solleva(): + bianca = np.full((600, 800), 255, np.uint8) + with pytest.raises(CostruttoreError, match="pezzo"): + build_dxf_compare(bianca, SAMPLE_DXF.read_bytes(), PX_PER_MM) + + +def test_dxf_illeggibile_solleva(): + with pytest.raises(CostruttoreError, match="DXF"): + build_dxf_compare(a_scene(), b"questo non e' un dxf", PX_PER_MM) + + +@pytest.mark.parametrize("kwargs", [ + {"px_per_mm": 0.0}, {"px_per_mm": PX_PER_MM, "band_mm": 0.0}, + {"px_per_mm": PX_PER_MM, "min_points_pct": 0.0}, + {"px_per_mm": PX_PER_MM, "min_points_pct": 101.0}, + {"px_per_mm": PX_PER_MM, "polarity": "grigio"}, +]) +def test_parametri_ricetta_invalidi(kwargs): + with pytest.raises(CostruttoreError): + build_dxf_compare(a_scene(), SAMPLE_DXF.read_bytes(), **kwargs) diff --git a/src/vision_worker/main.py b/src/vision_worker/main.py index c86d665..ef805ab 100644 --- a/src/vision_worker/main.py +++ b/src/vision_worker/main.py @@ -17,6 +17,7 @@ import numpy as np from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status from PIL import Image, UnidentifiedImageError +from src.vision.costruttori import CostruttoreError, build_dxf_compare # #83 from src.vision.runner import ( GraphBuildError, RunOutcome, @@ -223,3 +224,67 @@ async def validate( validate_graph, parsed_graph, resolved_dir ) return {"ok": ok, "errors": errors, "outputs": outputs} + + +def _decode_and_build_dxf(content: bytes, dxf: bytes, px_per_mm: float, + band_mm: float, min_points_pct: float, + polarity: str) -> dict: + """#83: meta' sincrona di /recipes/dxf-compare (decodifica + costruzione), + in un thread come `_decode_and_run`.""" + try: + frame = np.array(Image.open(io.BytesIO(content)).convert("L")) + except (UnidentifiedImageError, OSError) as exc: + raise _ImageDecodeError(str(exc)) from exc + return build_dxf_compare(frame, dxf, px_per_mm, band_mm, min_points_pct, polarity) + + +@app.post("/recipes/dxf-compare") +async def recipes_dxf_compare( + image: UploadFile = File(...), + dxf: UploadFile = File(...), + px_per_mm: float = Form(...), + band_mm: float = Form(0.1), + min_points_pct: float = Form(95.0), + polarity: str = Form("dark"), +) -> dict: + """#83: costruisce una ricetta DXF compare dall'immagine di riferimento, il + disegno e la scala. Stessi semaforo, timeout e mappa errori di /run.""" + content = await image.read() + dxf_bytes = await dxf.read() + + if _run_slots.locked(): + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=f"worker busy: already running {_MAX_CONCURRENCY} job(s)", + ) + + async with _run_slots: + try: + return await asyncio.wait_for( + asyncio.to_thread( + _decode_and_build_dxf, content, dxf_bytes, px_per_mm, + band_mm, min_points_pct, polarity, + ), + timeout=_RUN_TIMEOUT_S, + ) + except asyncio.TimeoutError as exc: + raise HTTPException( + status_code=status.HTTP_504_GATEWAY_TIMEOUT, + detail=f"vision run exceeded {_RUN_TIMEOUT_S}s", + ) from exc + except _ImageDecodeError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=f"image is not decodable: {exc}", + ) from exc + except (CostruttoreError, GraphBuildError) as exc: + # pezzo non trovato, DXF illeggibile, parametri invalidi: input, non guasto + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) from exc + except Exception as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"recipe build failed: {type(exc).__name__}: {exc}", + ) from exc diff --git a/src/vision_worker/tests/test_dxf_compare.py b/src/vision_worker/tests/test_dxf_compare.py new file mode 100644 index 0000000..877311c --- /dev/null +++ b/src/vision_worker/tests/test_dxf_compare.py @@ -0,0 +1,82 @@ +"""POST /recipes/dxf-compare del worker (#83): forma della risposta ed errori.""" +import pytest + +# Come test_worker.py: serve l'albero vision (Python >=3.13), altrimenti skipped. +pytest.importorskip( + "visionsuite.task", + reason="the vision tree requires Python >=3.13; see test_worker.py", +) + +import base64 +import io +import time + +import numpy as np +from httpx import ASGITransport, AsyncClient +from PIL import Image + +from src.vision.tests.test_costruttori import PX_PER_MM, SAMPLE_DXF, a_scene +from src.vision_worker.main import app + + +def _png(array: np.ndarray) -> bytes: + buffer = io.BytesIO() + Image.fromarray(array).save(buffer, format="PNG") + return buffer.getvalue() + + +async def _post(image: bytes, dxf: bytes, **data): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://worker") as client: + return await client.post( + "/recipes/dxf-compare", + files={"image": ("riferimento.png", image, "image/png"), + "dxf": ("disegno.dxf", dxf, "application/octet-stream")}, + data={"px_per_mm": str(PX_PER_MM), "band_mm": "2.0", **data}, + ) + + +@pytest.mark.asyncio +async def test_dxf_compare_200_con_le_chiavi_del_contratto(): + inizio = time.monotonic() + response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes()) + durata = time.monotonic() - inizio + + assert response.status_code == 200, response.text + body = response.json() + assert {"graph", "files", "outputs", "overlay", "outcome"} <= set(body) + assert set(body["files"]) == {"disegno.dxf", "calibrazione.json"} + assert base64.b64decode(body["files"]["disegno.dxf"]) == SAMPLE_DXF.read_bytes() + assert set(body["outputs"]) == { + "dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max"} + assert body["outputs"]["dxf.points_in_pct"] > 95.0 + assert body["graph"]["schema_version"] == 2 + print(f"\nW2 sulla scena di prova: {durata:.2f}s") # riportato nel rapporto + + +@pytest.mark.asyncio +async def test_dxf_compare_immagine_senza_pezzo_e_422(): + bianca = np.full((600, 800), 255, np.uint8) + response = await _post(_png(bianca), SAMPLE_DXF.read_bytes()) + assert response.status_code == 422 + assert "pezzo" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_dxf_compare_dxf_illeggibile_e_422(): + response = await _post(_png(a_scene(0.0)), b"non e' un dxf") + assert response.status_code == 422 + assert "DXF" in response.json()["detail"] + + +@pytest.mark.asyncio +async def test_dxf_compare_immagine_non_decodificabile_e_422(): + response = await _post(b"non e' un'immagine", SAMPLE_DXF.read_bytes()) + assert response.status_code == 422 + assert "not decodable" in response.json()["detail"].lower() + + +@pytest.mark.asyncio +async def test_dxf_compare_parametro_fuori_range_e_422(): + response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes(), polarity="grigio") + assert response.status_code == 422