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)