Worker: endpoint /recipes/dxf-compare con test (#83)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

src/backend/api/routers/vision.py

@@ -4,11 +4,8 @@ It does not know, and must not know, whether the server or the station computed
 the numbers: that is what lets a station's configuration change without touching
 the frontend or the recipes.
 """
-import base64
-import binascii
 import io
 import json
-from typing import Literal
 
 from fastapi.responses import FileResponse
 from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
@@ -29,7 +26,7 @@ from src.backend.models.api.vision import (
 from src.backend.models.orm.vision import VisionResult
 from src.backend.models.orm.task import RecipeTask
 from src.backend.models.orm.user import User
-from src.backend.services import station_service, vision_service
+from src.backend.services import vision_service
 
 router = APIRouter(prefix="/api/vision", tags=["vision"])
 
@@ -188,30 +185,6 @@ def _422(errors: list[str]) -> HTTPException:
     )
 
 
-async def _store_validate_assign(
-    task: RecipeTask, graph: dict, placed: dict[str, bytes],
-) -> tuple[str, dict]:
-    """Scrive la cartella ricetta, la fa validare dal worker, la assegna al task.
-
-    Condiviso da upload_recipe e dxf-compare con save (#83): stesso
-    comportamento, un solo posto. Non committa. Ritorna (recipe_dir, check).
-    """
-    # The folder is content-addressed and immutable; writing it before the
-    # validation is harmless (an orphan folder is never referenced) and the
-    # worker needs it on disk to validate meta files.
-    try:
-        recipe_dir = vision_service.store_recipe_folder(task.id, graph, placed)
-    except ValueError as exc:
-        raise _422([str(exc)])
-    check = await vision_service.validate_recipe(graph, recipe_dir)
-    if not check["ok"]:
-        raise _422([str(e) for e in check["errors"]] or ["the recipe is not valid"])
-
-    task.vision_json = graph
-    task.vision_recipe_dir = recipe_dir
-    return recipe_dir, check
-
-
 @router.post("/tasks/{task_id}/recipe", response_model=VisionRecipeUploadResponse)
 async def upload_recipe(
     task_id: int,
@@ -245,7 +218,19 @@ async def upload_recipe(
     if errors:
         raise _422(errors)
 
-    recipe_dir, check = await _store_validate_assign(task, graph, placed)
+    # The folder is content-addressed and immutable; writing it before the
+    # validation is harmless (an orphan folder is never referenced) and the
+    # worker needs it on disk to validate meta files.
+    try:
+        recipe_dir = vision_service.store_recipe_folder(task.id, graph, placed)
+    except ValueError as exc:
+        raise _422([str(exc)])
+    check = await vision_service.validate_recipe(graph, recipe_dir)
+    if not check["ok"]:
+        raise _422([str(e) for e in check["errors"]] or ["the recipe is not valid"])
+
+    task.vision_json = graph
+    task.vision_recipe_dir = recipe_dir
     await db.commit()
     return VisionRecipeUploadResponse(
         task_id=task.id,
@@ -340,77 +325,3 @@ async def preview(
 ) -> dict:
     task = await _task_or_404(db, task_id)
     return await vision_service.preview(db, task, await _read_image(image))
-
-
-@router.post("/tasks/{task_id}/recipe/dxf-compare")
-async def create_dxf_compare_recipe(
-    task_id: int,
-    image: UploadFile = File(...),
-    dxf: UploadFile = File(...),
-    station_id: int | None = Form(None, gt=0),
-    px_per_mm: float | None = Form(None, gt=0),
-    band_mm: float = Form(0.1, gt=0),
-    min_points_pct: float = Form(95, ge=0, le=100),
-    polarity: Literal["dark", "light"] = Form("dark"),
-    save: bool = Form(False),
-    db: AsyncSession = Depends(get_db),
-    current_user: User = Depends(require_maker),
-) -> dict:
-    """Ricetta DXF compare da un'immagine e un DXF, senza Studio (#83).
-
-    La scala e' `px_per_mm`, o quella della stazione (letta dall'acquisitore).
-    Con save=false si vede il risultato e non si scrive nulla; con save=true la
-    ricetta va sul task come con l'upload, e se il task non ha nessuna quota di
-    visione nasce la quota «Punti in tolleranza».
-    """
-    task = await _task_or_404(db, task_id)
-    if task.task_type not in vision_service.VISION_TASK_TYPES:
-        raise _422([f"task type {task.task_type!r} does not run a vision recipe"])
-
-    station = None
-    if station_id is not None:
-        station = await station_service.get_station(db, station_id)  # 404 se non esiste
-    if px_per_mm is None:
-        calibration, reason = await vision_service.fetch_station_calibration(station)
-        if calibration is None:
-            # irraggiungibile e non calibrata sono due guasti diversi per chi lo legge
-            raise _422([
-                "acquisitore della stazione non raggiungibile"
-                if reason == vision_service.CAL_UNREACHABLE else "stazione non calibrata"
-            ])
-        px_per_mm = float(calibration["px_per_mm"])
-
-    image_bytes = await _read_image(image)
-    dxf_bytes = await dxf.read()
-    if not validate_file_size(len(dxf_bytes)):
-        raise _422([f"the DXF exceeds {settings.max_upload_size_mb}MB"])
-
-    body = await vision_service.dxf_compare_recipe(
-        image_bytes, dxf_bytes,
-        px_per_mm=px_per_mm, band_mm=band_mm, min_points_pct=min_points_pct,
-        polarity=polarity,
-        worker_url=(station.vision_worker_url or None) if station is not None else None,
-    )
-    # i file stanno nella ricetta salvata, non nella risposta al client
-    files = body.pop("files")
-    response = {"result": body, "saved": False, "recipe_dir": None, "outputs": None}
-    if not save:
-        return response
-
-    uploads: list[tuple[str, bytes]] = []
-    for name, encoded in files.items():
-        try:
-            uploads.append((name, base64.b64decode(encoded, validate=True)))
-        except (binascii.Error, ValueError, TypeError):
-            raise HTTPException(
-                status_code=status.HTTP_502_BAD_GATEWAY,
-                detail=f"the vision worker's answer was malformed: file {name!r} is not base64",
-            )
-    placed, errors = vision_service.match_recipe_files(body["graph"], uploads)
-    if errors:
-        raise _422(errors)
-    recipe_dir, check = await _store_validate_assign(task, body["graph"], placed)
-    await vision_service.ensure_points_quote(db, task, min_points_pct)
-    await db.commit()
-    response.update(saved=True, recipe_dir=recipe_dir, outputs=check["outputs"])
-    return response

src/backend/services/vision_service.py

@@ -7,7 +7,6 @@ from __future__ import annotations
 
 import hashlib
 import json
-import logging
 import math
 from pathlib import Path
 from uuid import uuid4
@@ -25,8 +24,6 @@ from src.backend.models.orm.task import RecipeSubtask, RecipeTask
 from src.backend.models.orm.vision import VisionReferenceImage, VisionResult
 from src.backend.services import measurement_service
 
-logger = logging.getLogger(__name__)
-
 # Worker responses whose status and reason are forwarded as-is rather than
 # flattened into a generic 502 - see `_call_worker`.
 _FORWARDED_WORKER_STATUSES = frozenset({
@@ -143,136 +140,6 @@ async def _call_worker(
     return report
 
 
-async def call_worker_multipart(
-    path: str,
-    *,
-    files: dict,
-    data: dict,
-    worker_url: str | None = None,
-    timeout: float = 120.0,
-) -> dict:
-    """POST multipart generico al worker (#83): /recipes/dxf-compare.
-
-    Stessa traduzione degli errori di `_call_worker`: 504 sul timeout, 502 sul
-    trasporto o su una risposta non JSON-oggetto; 422/503/504 del worker si
-    inoltrano cosi' come sono. Worker della stazione o, se manca, il centrale.
-    """
-    base = (worker_url or settings.vision_worker_url).rstrip("/")
-    try:
-        async with httpx.AsyncClient(timeout=timeout) as client:
-            response = await client.post(f"{base}{path}", files=files, data=data)
-    except httpx.TimeoutException as exc:
-        raise HTTPException(
-            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
-            detail=f"the vision worker did not answer in time: {exc} ({base})",
-        ) from exc
-    except (httpx.TransportError, httpx.HTTPError) as exc:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"the vision worker is unreachable: {exc} ({base})",
-        ) from exc
-
-    if response.status_code in _FORWARDED_WORKER_STATUSES:
-        try:
-            detail = response.json().get("detail", response.text)
-        except ValueError:
-            detail = response.text
-        raise HTTPException(status_code=response.status_code, detail=detail)
-    if response.status_code != 200:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"the vision worker refused the job: {response.text}",
-        )
-    try:
-        body = response.json()
-    except ValueError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"the vision worker's answer was malformed: not valid JSON ({exc})",
-        ) from exc
-    if not isinstance(body, dict):
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail="the vision worker's answer was malformed: expected a JSON object",
-        )
-    return body
-
-
-async def dxf_compare_recipe(
-    image_bytes: bytes, dxf_bytes: bytes, *, px_per_mm: float, band_mm: float,
-    min_points_pct: float, polarity: str, worker_url: str | None = None,
-) -> dict:
-    """W2: ricetta dxf-compare da immagine + DXF (#83). Sostituita dai test."""
-    body = await call_worker_multipart(
-        "/recipes/dxf-compare",
-        files={
-            "image": ("frame.png", image_bytes, "image/png"),
-            "dxf": ("disegno.dxf", dxf_bytes, "application/octet-stream"),
-        },
-        data={
-            "px_per_mm": str(px_per_mm), "band_mm": str(band_mm),
-            "min_points_pct": str(min_points_pct), "polarity": polarity,
-        },
-        worker_url=worker_url,
-    )
-    # forma minima del contratto W2: senza graph e files non c'e' nulla da salvare
-    if not isinstance(body.get("graph"), dict) or not isinstance(body.get("files"), dict):
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail="the vision worker's answer was malformed: missing graph or files",
-        )
-    return body
-
-
-# Stati di fetch_station_calibration (#83)
-CAL_OK = "ok"
-CAL_NO_CAMERA = "no_camera"            # la stazione non ha camera_url/camera_source
-CAL_NOT_CALIBRATED = "not_calibrated"  # l'acquisitore risponde 404 (o un valore inutilizzabile)
-CAL_UNREACHABLE = "unreachable"        # l'acquisitore non risponde
-
-
-async def fetch_station_calibration(station: Station | None) -> tuple[dict | None, str]:
-    """Legge la calibrazione dall'acquisitore della stazione (C1, #83).
-
-    GET {camera_url}/camera/calibration?source={camera_source}. La calibrazione
-    vive nell'acquisitore: TMF la legge, non la tiene. Ritorna (dict, "ok")
-    oppure (None, motivo) per distinguere «non calibrata» da «irraggiungibile».
-    Un px_per_mm non finito o <= 0 conta come non calibrata.
-    """
-    if station is None or not station.camera_url or not station.camera_source:
-        return None, CAL_NO_CAMERA
-    url = f"{station.camera_url.rstrip('/')}/camera/calibration"
-    try:
-        # timeout breve: e' una lettura di servizio, non deve bloccare l'execute
-        async with httpx.AsyncClient(timeout=5.0) as client:
-            response = await client.get(url, params={"source": station.camera_source})
-    except httpx.HTTPError as exc:
-        logger.warning("acquisitore della stazione %s non raggiungibile: %s", station.code, exc)
-        return None, CAL_UNREACHABLE
-    if response.status_code == status.HTTP_404_NOT_FOUND:
-        return None, CAL_NOT_CALIBRATED
-    if response.status_code != 200:
-        logger.warning(
-            "acquisitore della stazione %s: HTTP %s", station.code, response.status_code,
-        )
-        return None, CAL_UNREACHABLE
-    try:
-        body = response.json()
-    except ValueError:
-        logger.warning("acquisitore della stazione %s: risposta non JSON", station.code)
-        return None, CAL_UNREACHABLE
-    px_per_mm = body.get("px_per_mm") if isinstance(body, dict) else None
-    if not (_is_measurable(px_per_mm) and px_per_mm > 0):
-        return None, CAL_NOT_CALIBRATED
-    return body, CAL_OK
-
-
-async def get_station_calibration(station: Station | None) -> dict | None:
-    """La calibrazione C1 della stazione, o None in ogni altro caso (#83)."""
-    calibration, _ = await fetch_station_calibration(station)
-    return calibration
-
-
 async def validate_recipe(graph: dict, recipe_dir: str | None) -> dict:
     """Ask the worker's /validate whether a v2 recipe is loadable (#46).
 
@@ -498,14 +365,11 @@ async def execute_task(
         )
     # Worker della stazione della sessione di produzione (#45); senza run o senza URL: centrale
     station_url = None
-    # Calibrazione letta dall'acquisitore della stazione (#83); None = non c'e'
-    calibration = None
     if run is not None:
         station = (await db.execute(
             select(Station).where(Station.id == run.station_id)
         )).scalar_one_or_none()
         station_url = (station.vision_worker_url or None) if station is not None else None
-        calibration = await get_station_calibration(station)
 
     if not task.vision_json:
         raise HTTPException(
@@ -555,15 +419,11 @@ async def execute_task(
             ),
         )
 
-    # La calibrazione viene dall'acquisitore della stazione (#83); se non c'e',
-    # non si passa `calibration` e la ricetta porta la sua (#46).
-    worker_kwargs: dict = {}
-    if calibration is not None:
-        worker_kwargs["calibration"] = {"px_per_mm": float(calibration["px_per_mm"])}
+    # Nessuna fonte di calibrazione esiste oggi (ne' task ne' stazione): non si
+    # passa `calibration`, la ricetta porta la sua (#46).
     report = await _call_worker(
         image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
         worker_url=station_url,  # #45
-        **worker_kwargs,
     )
     outputs = report["outputs"]
 
@@ -597,8 +457,6 @@ async def execute_task(
         station_id=run.station_id if run is not None else None,
         worker_url=(station_url or settings.vision_worker_url).rstrip("/"),
         graph_snapshot=task.vision_json,
-        # copiata come il grafo (#83): ricalibrare dopo non cambia cio' che e' stato misurato
-        calibration_snapshot=calibration,
         duration_ms=report.get("duration_ms"),
     )
     db.add(result)
@@ -743,42 +601,3 @@ async def preview(db: AsyncSession, task: RecipeTask, image_bytes: bytes) -> dic
     return await _call_worker(
         image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir,
     )
-
-
-async def ensure_points_quote(
-    db: AsyncSession, task: RecipeTask, min_points_pct: float,
-) -> bool:
-    """Crea la quota «Punti in tolleranza» se il task non ha nessuna quota di visione (#83).
-
-    Una sola volta: se esiste gia' una quota con `vision_output`, non si tocca.
-    Se la versione ha gia' misure non si scrive (il copy-on-write sta nel router
-    dei task, qui non si aggira): ritorna False. Ritorna True se l'ha creata.
-    """
-    from src.backend.services import recipe_service  # import locale: evita cicli
-
-    existing = (await db.execute(
-        select(RecipeSubtask.id)
-        .where(RecipeSubtask.task_id == task.id)
-        .where(RecipeSubtask.vision_output.is_not(None))
-        .limit(1)
-    )).scalar_one_or_none()
-    if existing is not None:
-        return False
-    if await recipe_service.version_has_measurements(db, task.version_id):
-        return False
-    markers = (await db.execute(
-        select(RecipeSubtask.marker_number).where(RecipeSubtask.task_id == task.id)
-    )).scalars().all()
-    db.add(RecipeSubtask(
-        task_id=task.id,
-        marker_number=max(markers, default=0) + 1,  # unico per task (uq_task_marker)
-        description="Punti in tolleranza",
-        nominal=100,
-        # limiti assoluti come nel resto delle quote: passa se punti% >= min_points_pct
-        ltl=min_points_pct,
-        utl=100,
-        unit="%",
-        vision_output="dxf.points_in_pct",
-    ))
-    await db.flush()
-    return True

src/backend/tests/test_vision_dxf_compare.py

@@ -1,338 +0,0 @@
-"""#83: ricetta DXF compare da camera e calibrazione letta dall'acquisitore.
-
-Il worker (W2) e l'acquisitore (C1) non girano nei test: si fingono, come
-fanno gli altri test di visione. La calibrazione vive nell'acquisitore esterno,
-TMF la legge via GET {camera_url}/camera/calibration.
-"""
-import base64
-import json
-
-import httpx
-import pytest
-from sqlalchemy import select
-
-from src.backend.config import settings
-from src.backend.models.orm.station import Station
-from src.backend.models.orm.task import RecipeSubtask
-from src.backend.models.orm.vision import VisionResult
-from src.backend.services import vision_service
-from src.backend.tests.conftest import auth_headers
-from src.backend.tests.test_vision_recipe_upload import an_image, make_task
-from src.backend.tests.test_vision_station_worker import (  # noqa: F401 (fixture worker_calls)
-    _camera_task, _execute, _station_with_run, worker_calls,
-)
-
-CAMERA_URL = "http://10.0.0.9:8200"
-CAL = {
-    "source": "camera:A1", "px_per_mm": 12.5, "mm_per_px": 0.08, "rms_px": 0.2,
-    "n_corners": 40, "cols": 8, "rows": 5, "square_mm": 5.0, "image_width": 640,
-    "image_height": 480, "scale_spread_pct": 0.1, "calibrated_at": "2026-10-23T10:00:00",
-}
-GRAPH = {
-    "schema_version": 2, "name": "dxf-compare", "tools": [],
-    "meta": {"dxf": {"disegno": {"path": "disegno.dxf"}}, "calibration": {"path": "calibrazione.json"}},
-}
-OUTPUTS = [{"key": "dxf.points_in_pct", "quantity": "ratio", "unit": "%"}]
-
-
-def _b64(data: bytes) -> str:
-    return base64.b64encode(data).decode()
-
-
-@pytest.fixture(autouse=True)
-def redirect_uploads(monkeypatch, tmp_path):
-    # upload_path = parents[2] / upload_dir: un assoluto sostituisce la base
-    monkeypatch.setattr(settings, "upload_dir", str(tmp_path))
-
-
-@pytest.fixture
-def w2(monkeypatch):
-    """Finto worker W2: ricorda le chiamate e risponde col corpo del contratto."""
-    calls: list[dict] = []
-
-    async def fake(image_bytes, dxf_bytes, **kwargs):
-        calls.append({"dxf": dxf_bytes, **kwargs})
-        return {
-            "graph": GRAPH,
-            "files": {
-                "disegno.dxf": _b64(dxf_bytes),
-                "calibrazione.json": _b64(json.dumps({"px_per_mm": kwargs["px_per_mm"]}).encode()),
-            },
-            "outputs": {"dxf.points_in_pct": 99.0},
-            "overlay": [], "outcome": "pass",
-        }
-
-    monkeypatch.setattr("src.backend.services.vision_service.dxf_compare_recipe", fake)
-    return calls
-
-
-@pytest.fixture
-def validate_ok(monkeypatch):
-    async def fake(graph, recipe_dir):
-        return {"ok": True, "errors": [], "outputs": OUTPUTS}
-    monkeypatch.setattr("src.backend.services.vision_service.validate_recipe", fake)
-
-
-@pytest.fixture
-def acquisitore(monkeypatch):
-    """Finto C1: `state['cal']` = dict (200), None (404), 'down' (irraggiungibile)."""
-    state: dict = {"cal": CAL, "urls": []}
-
-    class _Resp:
-        def __init__(self, code, body=None):
-            self.status_code, self._body, self.text = code, body, ""
-
-        def json(self):
-            return self._body
-
-    class _Client:
-        def __init__(self, *a, **k):
-            pass
-
-        async def __aenter__(self):
-            return self
-
-        async def __aexit__(self, *exc):
-            return False
-
-        async def get(self, url, params=None):
-            state["urls"].append((url, params))
-            if state["cal"] == "down":
-                raise httpx.ConnectError("giu'")
-            if state["cal"] is None:
-                return _Resp(404)
-            return _Resp(200, state["cal"])
-
-    monkeypatch.setattr(httpx, "AsyncClient", _Client)
-    return state
-
-
-async def _station(db_session, user, *, camera=True, worker_url=None, code="ST-C"):
-    station = Station(
-        code=code, name="Stazione C", active=True, created_by=user.id,
-        vision_worker_url=worker_url,
-        camera_url=CAMERA_URL if camera else None,
-        camera_source="camera:A1" if camera else None,
-    )
-    db_session.add(station)
-    await db_session.commit()
-    await db_session.refresh(station)
-    return station
-
-
-def _post(client, user, task, *, save=False, **data):
-    return client.post(
-        f"/api/vision/tasks/{task.id}/recipe/dxf-compare",
-        headers=auth_headers(user),
-        files={
-            "image": ("p.png", an_image(), "image/png"),
-            "dxf": ("d.dxf", b"0\nEOF", "application/octet-stream"),
-        },
-        data={"save": str(save).lower(), **{k: str(v) for k, v in data.items() if v is not None}},
-    )
-
-
-# --- lettura della calibrazione dall'acquisitore (C1) -------------------------
-
-@pytest.mark.asyncio
-async def test_fetch_calibration_ok(db_session, maker_user, acquisitore):
-    station = await _station(db_session, maker_user)
-    cal, reason = await vision_service.fetch_station_calibration(station)
-    assert (cal, reason) == (CAL, "ok")
-    assert acquisitore["urls"] == [(f"{CAMERA_URL}/camera/calibration", {"source": "camera:A1"})]
-
-
-@pytest.mark.asyncio
-async def test_fetch_calibration_reasons(db_session, maker_user, acquisitore):
-    station = await _station(db_session, maker_user)
-    acquisitore["cal"] = None
-    assert await vision_service.fetch_station_calibration(station) == (None, "not_calibrated")
-    acquisitore["cal"] = "down"
-    assert await vision_service.fetch_station_calibration(station) == (None, "unreachable")
-    assert await vision_service.get_station_calibration(station) is None
-    acquisitore["cal"] = {**CAL, "px_per_mm": 0}  # valore inutilizzabile = non calibrata
-    assert await vision_service.fetch_station_calibration(station) == (None, "not_calibrated")
-    no_camera = await _station(db_session, maker_user, camera=False, code="ST-N")
-    assert await vision_service.fetch_station_calibration(no_camera) == (None, "no_camera")
-    assert await vision_service.fetch_station_calibration(None) == (None, "no_camera")
-
-
-# --- B3: POST /api/vision/tasks/{id}/recipe/dxf-compare -----------------------
-
-@pytest.mark.asyncio
-async def test_dxf_compare_preview_writes_nothing(
-    client, db_session, maker_user, w2, acquisitore,
-):
-    task, subtask = await make_task(db_session, maker_user)
-    station = await _station(db_session, maker_user, worker_url="http://10.0.0.7:8100")
-
-    resp = await _post(client, maker_user, task, station_id=station.id, band_mm=0.2,
-                       min_points_pct=90, polarity="light")
-
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert body["saved"] is False and body["recipe_dir"] is None and body["outputs"] is None
-    assert "files" not in body["result"] and body["result"]["graph"] == GRAPH
-    # px_per_mm dall'acquisitore, parametri e worker della stazione
-    assert w2[0]["px_per_mm"] == 12.5 and w2[0]["band_mm"] == 0.2
-    assert w2[0]["min_points_pct"] == 90 and w2[0]["polarity"] == "light"
-    assert w2[0]["worker_url"] == "http://10.0.0.7:8100"
-    await db_session.refresh(task)
-    assert task.vision_json is None and task.vision_recipe_dir is None
-    quotes = (await db_session.execute(
-        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id))).scalars().all()
-    assert [q.id for q in quotes] == [subtask.id]  # nessuna quota nuova
-    assert not (settings.upload_path / "vision" / "recipes").exists()
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_save_stores_recipe_and_quote_once(
-    client, db_session, maker_user, w2, validate_ok,
-):
-    task, _ = await make_task(db_session, maker_user)
-
-    resp = await _post(client, maker_user, task, save=True, px_per_mm=10, min_points_pct=95)
-
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert body["saved"] is True and body["outputs"] == OUTPUTS
-    assert "files" not in body["result"]
-    folder = settings.upload_path / body["recipe_dir"]
-    assert (folder / "disegno.dxf").read_bytes() == b"0\nEOF"
-    assert json.loads((folder / "calibrazione.json").read_text()) == {"px_per_mm": 10.0}
-    await db_session.refresh(task)
-    assert task.vision_json == GRAPH and task.vision_recipe_dir == body["recipe_dir"]
-
-    quotes = (await db_session.execute(
-        select(RecipeSubtask).where(RecipeSubtask.vision_output == "dxf.points_in_pct")
-        .where(RecipeSubtask.task_id == task.id))).scalars().all()
-    assert len(quotes) == 1
-    q = quotes[0]
-    assert q.description == "Punti in tolleranza"
-    assert (float(q.nominal), float(q.ltl), float(q.utl)) == (100.0, 95.0, 100.0)
-
-    # secondo salvataggio: la quota non si duplica
-    task_id = task.id  # dopo expire_all l'attributo scaduto non si rilegge in async
-    again = await _post(client, maker_user, task, save=True, px_per_mm=10)
-    assert again.status_code == 200
-    db_session.expire_all()
-    count = (await db_session.execute(
-        select(RecipeSubtask).where(RecipeSubtask.vision_output.is_not(None))
-        .where(RecipeSubtask.task_id == task_id))).scalars().all()
-    assert len(count) == 1
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_keeps_an_existing_vision_quote(
-    client, db_session, maker_user, w2, validate_ok,
-):
-    task, subtask = await make_task(db_session, maker_user)
-    subtask.vision_output = "t1.diameter"
-    await db_session.commit()
-
-    resp = await _post(client, maker_user, task, save=True, px_per_mm=10)
-
-    assert resp.status_code == 200, resp.text
-    quotes = (await db_session.execute(
-        select(RecipeSubtask).where(RecipeSubtask.task_id == task.id))).scalars().all()
-    assert len(quotes) == 1  # niente quota in piu'
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_uncalibrated_station_is_422(
-    client, db_session, maker_user, w2, acquisitore,
-):
-    task, _ = await make_task(db_session, maker_user)
-    station = await _station(db_session, maker_user)
-    acquisitore["cal"] = None
-    resp = await _post(client, maker_user, task, station_id=station.id)
-    assert resp.status_code == 422
-    assert resp.json()["detail"] == ["stazione non calibrata"]
-    assert w2 == []
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_unreachable_acquirer_is_422_with_its_own_message(
-    client, db_session, maker_user, w2, acquisitore,
-):
-    task, _ = await make_task(db_session, maker_user)
-    station = await _station(db_session, maker_user)
-    acquisitore["cal"] = "down"
-    resp = await _post(client, maker_user, task, station_id=station.id)
-    assert resp.status_code == 422
-    assert resp.json()["detail"] == ["acquisitore della stazione non raggiungibile"]
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_without_station_nor_scale_is_422(client, db_session, maker_user, w2):
-    task, _ = await make_task(db_session, maker_user)
-    resp = await _post(client, maker_user, task)
-    assert resp.status_code == 422 and resp.json()["detail"] == ["stazione non calibrata"]
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_needs_maker(client, db_session, maker_user, measurement_tec_user, w2):
-    task, _ = await make_task(db_session, maker_user)
-    resp = await _post(client, measurement_tec_user, task, px_per_mm=10)
-    assert resp.status_code == 403
-
-
-@pytest.mark.asyncio
-async def test_dxf_compare_worker_422_is_forwarded(
-    client, db_session, maker_user, monkeypatch,
-):
-    from fastapi import HTTPException
-
-    async def refuse(*a, **k):
-        raise HTTPException(status_code=422, detail="pezzo non trovato")
-    monkeypatch.setattr("src.backend.services.vision_service.dxf_compare_recipe", refuse)
-    task, _ = await make_task(db_session, maker_user)
-    resp = await _post(client, maker_user, task, px_per_mm=10)
-    assert resp.status_code == 422 and resp.json()["detail"] == "pezzo non trovato"
-
-
-# --- execute: calibrazione della stazione -------------------------------------
-
-@pytest.mark.asyncio
-async def test_execute_passes_station_calibration_and_snapshots_it(
-    client, db_session, measurement_tec_user, worker_calls, acquisitore,
-):
-    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
-    station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
-    station.camera_url, station.camera_source = CAMERA_URL, "camera:A1"
-    await db_session.commit()
-
-    await _execute(client, measurement_tec_user, task, run)
-
-    assert worker_calls[0]["calibration"] == {"px_per_mm": 12.5}
-    result = (await db_session.execute(select(VisionResult))).scalars().one()
-    assert result.calibration_snapshot == CAL
-
-
-@pytest.mark.asyncio
-async def test_execute_without_calibration_is_unchanged(
-    client, db_session, measurement_tec_user, worker_calls, acquisitore,
-):
-    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
-    station, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
-    station.camera_url, station.camera_source = CAMERA_URL, "camera:A1"
-    await db_session.commit()
-    acquisitore["cal"] = "down"  # acquisitore giu': si misura come prima, senza calibrazione
-
-    await _execute(client, measurement_tec_user, task, run)
-
-    assert "calibration" not in worker_calls[0]
-    result = (await db_session.execute(select(VisionResult))).scalars().one()
-    assert result.calibration_snapshot is None
-
-
-@pytest.mark.asyncio
-async def test_execute_station_without_camera_never_asks(
-    client, db_session, measurement_tec_user, worker_calls, acquisitore,
-):
-    recipe, task = await _camera_task(db_session, measurement_tec_user.id)
-    _, run = await _station_with_run(db_session, measurement_tec_user, recipe, None)
-
-    await _execute(client, measurement_tec_user, task, run)
-
-    assert acquisitore["urls"] == [] and "calibration" not in worker_calls[0]

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,
+    }

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)

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

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