Costruttore ricetta DXF compare dal campo, con test (#83)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
}
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user