Acquisitore: calibrazione a scacchiera per sorgente, rotte e pagina /calibrazione (#83)

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

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

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 = """<!doctype html>
+<html lang="it">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>Calibrazione camera</title>
+<style>
+  body { font-family: system-ui, sans-serif; margin: 0; background: #f3f4f6; color: #1f2937; }
+  main { max-width: 960px; margin: 0 auto; padding: 16px; }
+  h1 { font-size: 1.4rem; margin: 0 0 12px; }
+  .scheda { background: #fff; border: 1px solid #d1d5db; border-radius: 8px; padding: 14px; margin-bottom: 14px; }
+  label { display: block; font-size: .85rem; margin-bottom: 4px; color: #4b5563; }
+  select, input[type=number] { padding: 6px 8px; font-size: 1rem; border: 1px solid #9ca3af; border-radius: 6px; }
+  .riga { display: flex; flex-wrap: wrap; gap: 12px; align-items: flex-end; }
+  button { padding: 8px 14px; font-size: 1rem; border: 0; border-radius: 6px; background: #2563eb; color: #fff; cursor: pointer; }
+  button.secondario { background: #6b7280; }
+  button.pericolo { background: #b91c1c; }
+  button:disabled { opacity: .5; cursor: default; }
+  .messaggio { margin-top: 10px; font-weight: 600; }
+  .errore { color: #b91c1c; }
+  .ok { color: #15803d; }
+  table { border-collapse: collapse; }
+  td { padding: 3px 14px 3px 0; }
+  td:first-child { color: #4b5563; }
+  img { max-width: 100%; border: 1px solid #d1d5db; border-radius: 6px; margin-top: 8px; }
+  .avviso { color: #92400e; }
+</style>
+</head>
+<body>
+<main>
+  <h1>Calibrazione camera con scacchiera</h1>
+
+  <div class="scheda">
+    <div class="riga">
+      <div><label for="sorgente">Sorgente</label><select id="sorgente"></select></div>
+      <div><label for="cols">Angoli interni in orizzontale</label><input id="cols" type="number" min="2" value="9"></div>
+      <div><label for="rows">Angoli interni in verticale</label><input id="rows" type="number" min="2" value="6"></div>
+      <div><label for="quadrato">Lato del quadrato (mm)</label><input id="quadrato" type="number" min="0.001" step="any" value="2"></div>
+    </div>
+    <p style="font-size:.85rem;color:#4b5563;margin:10px 0 0">
+      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.</p>
+  </div>
+
+  <div class="scheda">
+    <div class="riga">
+      <button id="acquisisci">Acquisisci e calcola</button>
+      <div><label for="file">oppure un file immagine</label><input id="file" type="file" accept="image/*"></div>
+      <button id="carica" class="secondario">Carica immagine e calcola</button>
+    </div>
+    <div id="messaggio" class="messaggio"></div>
+  </div>
+
+  <div class="scheda" id="risultato" hidden>
+    <strong>Risultato del calcolo</strong>
+    <table id="tabella"></table>
+    <div id="avviso" class="avviso"></div>
+    <div class="riga" style="margin-top:10px">
+      <button id="salva">Salva per questa sorgente</button>
+      <span style="font-size:.85rem;color:#4b5563">Da camera, salvando si acquisisce e si calcola di nuovo: tenere la scacchiera ferma.</span>
+    </div>
+    <img id="overlay" alt="Angoli trovati sulla scacchiera">
+  </div>
+
+  <div class="scheda">
+    <strong>Calibrazione salvata</strong>
+    <div id="stato">-</div>
+    <div class="riga" style="margin-top:10px">
+      <button id="elimina" class="pericolo" disabled>Elimina</button>
+    </div>
+  </div>
+</main>
+
+<script>
+"use strict";
+var $ = function (id) { return document.getElementById(id); };
+var ultimo = null;  // ultimo calcolo, per «Salva»
+
+function messaggio(testo, classe) {
+  var m = $("messaggio");
+  m.textContent = testo || "";
+  m.className = "messaggio " + (classe || "");
+}
+
+function numero(v, cifre) { return Number(v).toFixed(cifre); }
+
+function righe(d) {
+  return [
+    ["Scala", numero(d.px_per_mm, 4) + " px/mm (" + numero(d.mm_per_px, 5) + " mm/px)"],
+    ["Errore di riproiezione (RMS)", numero(d.rms_px, 3) + " px"],
+    ["Variazione di scala nell'area", numero(d.scale_spread_pct, 2) + " %"],
+    ["Angoli trovati", d.n_corners + " (" + d.cols + " x " + d.rows + ")"],
+    ["Lato del quadrato", d.square_mm + " mm"],
+    ["Immagine", d.image_width + " x " + d.image_height + " px"]
+  ];
+}
+
+function tabella(el, d) {
+  el.textContent = "";
+  righe(d).forEach(function (r) {
+    var tr = document.createElement("tr");
+    r.forEach(function (c) {
+      var td = document.createElement("td");
+      td.textContent = c;
+      tr.appendChild(td);
+    });
+    el.appendChild(tr);
+  });
+}
+
+async function dettaglio(risposta) {
+  try { var j = await risposta.json(); return j.detail || JSON.stringify(j); }
+  catch (e) { return "errore " + risposta.status; }
+}
+
+async function caricaSorgenti() {
+  var sel = $("sorgente");
+  try {
+    var r = await fetch("/camera/sources");
+    var j = await r.json();
+    (j.sources || []).forEach(function (s) {
+      var o = document.createElement("option");
+      o.value = s.name;
+      o.textContent = s.label || s.name;
+      sel.appendChild(o);
+    });
+  } catch (e) { messaggio("Elenco delle sorgenti non disponibile.", "errore"); }
+  if (!sel.options.length) messaggio("Nessuna sorgente camera trovata.", "errore");
+  aggiornaStato();
+}
+
+async function aggiornaStato() {
+  var sorgente = $("sorgente").value;
+  var stato = $("stato");
+  $("elimina").disabled = true;
+  if (!sorgente) { stato.textContent = "-"; return; }
+  var r = await fetch("/camera/calibration?source=" + encodeURIComponent(sorgente));
+  if (r.status === 404) { stato.textContent = "Sorgente non calibrata."; return; }
+  if (!r.ok) { stato.textContent = await dettaglio(r); return; }
+  var d = await r.json();
+  stato.textContent = "";
+  var t = document.createElement("table");
+  tabella(t, d);
+  var quando = document.createElement("div");
+  quando.textContent = "Calibrata il " + d.calibrated_at;
+  stato.appendChild(t);
+  stato.appendChild(quando);
+  $("elimina").disabled = false;
+}
+
+function parametri() {
+  return {
+    cols: $("cols").value, rows: $("rows").value, quadrato: $("quadrato").value
+  };
+}
+
+async function calcola(conFile, salva) {
+  var sorgente = $("sorgente").value;
+  if (!sorgente) { messaggio("Scegliere una sorgente.", "errore"); return; }
+  var p = parametri();
+  var f = new FormData();
+  f.append("source", sorgente);
+  f.append("cols", p.cols);
+  f.append("rows", p.rows);
+  f.append("square_mm", p.quadrato);
+  f.append("save", salva ? "true" : "false");
+  if (conFile) {
+    var file = $("file").files[0];
+    if (!file) { messaggio("Scegliere un file immagine.", "errore"); return; }
+    f.append("image", file);
+  }
+  messaggio("Calcolo in corso...", "");
+  ["acquisisci", "carica", "salva"].forEach(function (i) { $(i).disabled = true; });
+  try {
+    var r = await fetch("/camera/calibrate", { method: "POST", body: f });
+    if (!r.ok) { messaggio(await dettaglio(r), "errore"); return; }
+    var d = await r.json();
+    ultimo = { conFile: conFile, dati: d };
+    tabella($("tabella"), d);
+    $("overlay").src = "data:image/png;base64," + d.overlay_png;
+    $("avviso").textContent = d.scale_spread_pct > 1
+      ? "Attenzione: la scala varia di " + numero(d.scale_spread_pct, 2) + " % nell'area: scacchiera inclinata o ottica non telecentrica."
+      : "";
+    $("risultato").hidden = false;
+    messaggio(d.saved ? "Calibrazione salvata." : "Calcolo riuscito: controllare e salvare.", "ok");
+    if (d.saved) aggiornaStato();
+  } catch (e) {
+    messaggio("Servizio non raggiungibile.", "errore");
+  } finally {
+    ["acquisisci", "carica", "salva"].forEach(function (i) { $(i).disabled = false; });
+  }
+}
+
+$("acquisisci").onclick = function () { calcola(false, false); };
+$("carica").onclick = function () { calcola(true, false); };
+// «Salva» rifa il calcolo con save=true sulla stessa origine (altra acquisizione se da camera)
+$("salva").onclick = function () { calcola(ultimo ? ultimo.conFile : false, true); };
+$("elimina").onclick = async function () {
+  var sorgente = $("sorgente").value;
+  if (!confirm("Eliminare la calibrazione salvata di " + sorgente + "?")) return;
+  var r = await fetch("/camera/calibration?source=" + encodeURIComponent(sorgente), { method: "DELETE" });
+  messaggio(r.ok ? "Calibrazione eliminata." : await dettaglio(r), r.ok ? "ok" : "errore");
+  aggiornaStato();
+};
+$("sorgente").onchange = aggiornaStato;
+caricaSorgenti();
+</script>
+</body>
+</html>
+"""

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=<src> -> 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=<src> -> 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 <cartella immagini>]
+      [--host 127.0.0.1] [--port 8765] [--folder <cartella immagini>] \
+      [--config-dir <cartella calibrazioni>]
   --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__":

scripts/seed_fiera.py

@@ -1,458 +0,0 @@
-"""Seed per il PC della fiera: utenti, stazione, tre ricette e quattro settimane di misure.
-
-Cosa fa
-  Popola un database TieMeasureFlow (SQLite del pacchetto Windows) cosi' che al primo
-  avvio NON compaia la procedura di setup e la demo abbia gia' qualcosa da mostrare:
-
-  - utenti adriano (admin), maker, capoturno, operatore (password nella costante
-    UTENTI qui sotto, decise da Adriano per la fiera);
-  - la stazione ST-DEFAULT (quella del pacchetto, vedi packaging/windows/avvio.py)
-    con camera_url http://127.0.0.1:8765 e camera_source camera:601020 (Itala);
-  - FIERA-FLANGIA «Flangia Ø80» (5 quote) e FIERA-ALBERO «Albero rettificato»
-    (4 quote), con circa 4 settimane di storico fino a ieri (lun-ven 8-17, una
-    produzione chiusa per giorno, lotti e seriali coerenti, misure con Cpk diversi
-    per quota, una deriva lenta nell'ultima settimana, una quota con 1-3% fuori
-    tolleranza);
-  - FIERA-DXF «Profilo DXF (visione)»: un task dxf_compare senza ricetta di visione
-    (si crea in fiera dal pannello «Crea ricetta da DXF»).
-
-  «Setup fatto» per il backend vuol dire: esiste almeno un utente admin
-  (api/routers/setup.py::_admin_exists), quindi basta creare adriano.
-
-Come si usa (da daitarn, TMF fermo sul PC, su una COPIA del file SQLite)
-    uv run python scripts/seed_fiera.py --db C:/percorso/tmf.db [--replace] [--dry-run]
-                                        [--fine AAAA-MM-GG]
-
-  Il file deve avere gia' lo schema (alembic upgrade head: lo fa l'exe al primo
-  avvio). --replace toglie solo i dati FIERA-* (ricette, misure, produzioni) e li
-  rifa'; senza, un secondo giro non duplica le ricette. Gli utenti e la stazione
-  sono idempotenti: se esistono si aggiornano password, ruoli e camera.
-  --dry-run fa tutto e annulla (stampa i numeri). Il seme casuale e' fisso: stesso
-  comando e stessa data di fine, stessi dati.
-
-Da cosa dipende
-  src.backend (modelli ORM, auth_service, measurement_service), SQLAlchemy +
-  aiosqlite, tzdata su Windows (gia' in pyproject). Non usa MySQL.
-"""
-import argparse
-import asyncio
-import random
-import sys
-from datetime import date, datetime, time, timedelta, timezone
-from pathlib import Path
-from zoneinfo import ZoneInfo
-
-from sqlalchemy import delete, event, select
-from sqlalchemy.ext.asyncio import (
-    AsyncSession, async_sessionmaker, create_async_engine,
-)
-
-# La radice del repo nel path, cosi' "src.backend" si importa lanciando lo script da ovunque
-sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
-
-from src.backend.database import async_session_factory  # noqa: E402
-from src.backend.models.orm.measurement import Measurement  # noqa: E402
-from src.backend.models.orm.production import (  # noqa: E402
-    ProductionEvent, ProductionRun,
-)
-from src.backend.models.orm.recipe import Recipe, RecipeVersion  # noqa: E402
-from src.backend.models.orm.setting import RecipeVersionAudit  # noqa: E402
-from src.backend.models.orm.station import (  # noqa: E402
-    Station, StationRecipeAssignment,
-)
-from src.backend.models.orm.task import RecipeSubtask, RecipeTask  # noqa: E402
-from src.backend.models.orm.task_result import TaskResult  # noqa: E402
-from src.backend.models.orm.user import User  # noqa: E402
-from src.backend.services import auth_service  # noqa: E402
-from src.backend.services.measurement_service import calculate_pass_fail  # noqa: E402
-
-CODES = ("FIERA-FLANGIA", "FIERA-ALBERO", "FIERA-DXF")
-STATION_CODE = "ST-DEFAULT"  # STATION_CODE del pacchetto Windows (avvio.py PREDEFINITI)
-CAMERA_URL = "http://127.0.0.1:8765"
-CAMERA_SOURCE = "camera:601020"  # Itala sul PC fiera
-SEED = 20261026  # seme fisso: dati ripetibili
-FUSO = ZoneInfo("Europe/Rome")  # le misure sono salvate in UTC naive, l'orario di turno e' locale
-GIORNI_STORICO = 28  # circa 4 settimane
-
-# (username, password, nome, ruoli, admin). Password "123" per tutti, chiesta da Adriano per la fiera (#85); il login non controlla la lunghezza.
-UTENTI = [
-    ("adriano", "123", "Adriano Dal Pastro",
-     ["Maker", "MeasurementTec", "Metrologist", "Supervisor"], True),
-    ("maker", "123", "Marco Ferrari (Maker)", ["Maker"], False),
-    ("capoturno", "123", "Luca Bianchi (Capoturno)",
-     ["Supervisor", "MeasurementTec"], False),
-    ("operatore", "123", "Giulia Rossi (Operatore)", ["MeasurementTec"], False),
-]
-
-
-def _quota(marker, descrizione, nominale, tol, unita="mm", dec=3):
-    """Quota simmetrica: tolleranza = nominale +/- tol, warning al 70% della tolleranza."""
-    return dict(
-        marker_number=marker, description=descrizione, nominal=nominale,
-        ltl=round(nominale - tol, dec), lwl=round(nominale - 0.7 * tol, dec),
-        uwl=round(nominale + 0.7 * tol, dec), utl=round(nominale + tol, dec),
-        unit=unita,
-    )
-
-
-# Profilo statistico di una quota: scostamento medio (in frazioni di tol), sigma (in
-# frazioni di tol) e deriva (frazione di tol aggiunta linearmente nell'ultima settimana).
-# Cpk ~ (1 - |scostamento|) / (3 * sigma): sigma 0.2 -> 1,67; 0.33 -> 1,0.
-FLANGIA = [
-    # (task, quota, scostamento, sigma, deriva)
-    (0, _quota(1, "Diametro esterno Ø80", 80.0, 0.05), 0.0, 0.20, 0.0),      # Cpk ~1,67 capace
-    (0, _quota(2, "Diametro foro centrale Ø25", 25.0, 0.02), 0.0, 0.33, 0.0),  # Cpk ~1,0 al limite
-    (1, _quota(3, "Spessore flangia", 12.0, 0.10), 0.0, 0.22, 0.55),         # deriva ultima settimana
-    (1, _quota(4, "Interasse fori Ø60", 60.0, 0.10), 0.10, 0.25, 0.0),       # Cpk ~1,2
-    (1, _quota(5, "Profondita' lamatura", 8.0, 0.15), 0.50, 0.244, 0.0),     # ~2% fuori tolleranza
-]
-ALBERO = [
-    (0, _quota(1, "Diametro cuscinetto Ø20 h6", 20.0, 0.010), 0.0, 0.20, 0.0),  # Cpk ~1,67
-    (0, _quota(2, "Diametro sede Ø25 j5", 25.0, 0.008), 0.0, 0.25, 0.0),         # Cpk ~1,33
-    (1, _quota(3, "Lunghezza totale", 150.0, 0.10), 0.0, 0.33, 0.0),             # Cpk ~1,0
-    (1, _quota(4, "Rugosita' Ra", 0.40, 0.20, unita="µm"), 0.20, 0.25, 0.0),     # Cpk ~1,07
-]
-
-
-def _ora_utc(giorno: date, minuti_da_mezzanotte: int) -> datetime:
-    """Ora locale di officina -> datetime UTC naive, come le salva il backend."""
-    locale = datetime.combine(giorno, time(0, 0), tzinfo=FUSO) + timedelta(minutes=minuti_da_mezzanotte)
-    return locale.astimezone(timezone.utc).replace(tzinfo=None)
-
-
-def _giorni_lavorativi(fine: date) -> list[date]:
-    """Lun-ven degli ultimi GIORNI_STORICO giorni, fino a `fine` compreso."""
-    giorni = [fine - timedelta(days=i) for i in range(GIORNI_STORICO)]
-    return sorted(g for g in giorni if g.weekday() < 5)
-
-
-def _slot_giornalieri(rng: random.Random, n: int) -> list[int]:
-    """n orari (minuti dalla mezzanotte) in 8-12 e 13-17, ogni ~30 min con un po' di scarto."""
-    mezz_ore = [8 * 60 + 30 * i for i in range(8)] + [13 * 60 + 30 * i for i in range(8)]
-    scelti = sorted(rng.sample(mezz_ore, n))
-    return [m + rng.randint(0, 9) for m in scelti]
-
-
-async def _upsert_utenti(session: AsyncSession) -> dict[str, User]:
-    """Crea gli utenti o, se esistono, ne aggiorna password, nome, ruoli e flag admin."""
-    utenti = {}
-    for username, password, nome, ruoli, admin in UTENTI:
-        esistente = (await session.execute(
-            select(User).where(User.username == username)
-        )).scalar_one_or_none()
-        if esistente is None:
-            utenti[username] = await auth_service.create_user(
-                session, username=username, password=password, display_name=nome,
-                roles=ruoli, is_admin=admin,
-            )
-        else:
-            esistente.password_hash = auth_service.hash_password(password)
-            esistente.display_name = nome
-            esistente.roles = ruoli
-            esistente.is_admin = admin
-            esistente.active = True
-            utenti[username] = esistente
-    await session.flush()
-    return utenti
-
-
-async def _upsert_stazione(session: AsyncSession, owner: User) -> Station:
-    """La stazione del pacchetto, con la camera Itala; se c'e' gia' ne aggiorna la camera."""
-    stazione = (await session.execute(
-        select(Station).where(Station.code == STATION_CODE)
-    )).scalar_one_or_none()
-    if stazione is None:
-        stazione = Station(
-            code=STATION_CODE, name="Postazione fiera", location="Stand",
-            created_by=owner.id,
-        )
-        session.add(stazione)
-    stazione.camera_url = CAMERA_URL
-    stazione.camera_source = CAMERA_SOURCE
-    await session.flush()
-    return stazione
-
-
-async def _conta_e_cancella(session: AsyncSession, ricette: list[Recipe]) -> dict:
-    """Toglie le ricette FIERA-* e tutto cio' che e' misurato contro di esse. Ritorna i conteggi."""
-    conti = {"misure": 0, "produzioni": 0, "eventi": 0}
-    for ricetta in ricette:
-        versioni = (await session.execute(
-            select(RecipeVersion.id).where(RecipeVersion.recipe_id == ricetta.id)
-        )).scalars().all()
-        runs = (await session.execute(
-            select(ProductionRun.id).where(ProductionRun.recipe_id == ricetta.id)
-        )).scalars().all()
-        if versioni:
-            conti["misure"] += (await session.execute(
-                delete(Measurement).where(Measurement.version_id.in_(versioni))
-            )).rowcount
-            await session.execute(  # esiti dei task di queste versioni (FK sulle versioni)
-                delete(TaskResult).where(TaskResult.version_id.in_(versioni))
-            )
-        if runs:
-            conti["eventi"] += (await session.execute(
-                delete(ProductionEvent).where(ProductionEvent.run_id.in_(runs))
-            )).rowcount
-            conti["produzioni"] += (await session.execute(
-                delete(ProductionRun).where(ProductionRun.id.in_(runs))
-            )).rowcount
-        await session.execute(
-            delete(StationRecipeAssignment).where(StationRecipeAssignment.recipe_id == ricetta.id)
-        )
-        await session.execute(
-            delete(RecipeVersionAudit).where(RecipeVersionAudit.recipe_id == ricetta.id)
-        )
-        # Versioni, task e quote se ne vanno con la cascade ORM della ricetta.
-        await session.delete(ricetta)
-    await session.flush()
-    return conti
-
-
-async def _nuova_ricetta(session, owner, stazione, *, code, name, description,
-                         interval, requires_serial, note_versione):
-    """Ricetta + versione 1 corrente + assegnazione alla stazione."""
-    ricetta = Recipe(
-        code=code, name=name, description=description, created_by=owner.id, active=True,
-        measurement_interval_minutes=interval, requires_lot=True,
-        requires_serial=requires_serial, allow_manual_input=True,
-    )
-    session.add(ricetta)
-    await session.flush()
-    versione = RecipeVersion(
-        recipe_id=ricetta.id, version_number=1, is_current=True,
-        created_by=owner.id, change_notes=note_versione,
-    )
-    session.add(versione)
-    session.add(StationRecipeAssignment(
-        station_id=stazione.id, recipe_id=ricetta.id, assigned_by=owner.id,
-    ))
-    await session.flush()
-    return ricetta, versione
-
-
-async def _task(session, versione, ordine, titolo, tipo, direttiva=None,
-                descrizione=None, quote=()):
-    task = RecipeTask(
-        version_id=versione.id, order_index=ordine, title=titolo, task_type=tipo,
-        directive=direttiva, description=descrizione,
-    )
-    session.add(task)
-    await session.flush()
-    sottotask = []
-    for q in quote:
-        st = RecipeSubtask(task_id=task.id, **q)
-        session.add(st)
-        sottotask.append(st)
-    await session.flush()
-    return task, sottotask
-
-
-async def _ricetta_meccanica(session, owner, stazione, *, code, name, description,
-                             interval, requires_serial, profilo, titoli_task):
-    """Ricetta con una nota iniziale e un task di misura per ogni gruppo di quote del profilo."""
-    ricetta, versione = await _nuova_ricetta(
-        session, owner, stazione, code=code, name=name, description=description,
-        interval=interval, requires_serial=requires_serial,
-        note_versione="Ricetta dimostrativa per la fiera",
-    )
-    await _task(session, versione, 0, "Preparazione del pezzo", "note",
-                direttiva="Prima di misurare",
-                descrizione="Pulire il pezzo e controllare che sia a **temperatura ambiente**.")
-    sottotask_per_quota = {}
-    for indice, titolo in enumerate(titoli_task):
-        quote = [q for (t, q, *_r) in profilo if t == indice]
-        _t, sottotask = await _task(
-            session, versione, indice + 1, titolo, "measure",
-            direttiva="Misurare le quote in sequenza", quote=quote,
-        )
-        for st in sottotask:
-            sottotask_per_quota[st.marker_number] = st
-    return ricetta, versione, sottotask_per_quota
-
-
-def _valore(rng, quota, scost, sigma, deriva, frazione_tempo):
-    """Un valore: normale attorno a nominale + scostamento (+ deriva nell'ultima settimana)."""
-    tol = quota["utl"] - quota["nominal"]
-    media = quota["nominal"] + tol * (scost + deriva * max(0.0, (frazione_tempo - 0.75) / 0.25))
-    return round(rng.gauss(media, tol * sigma), 6)
-
-
-async def _storico(session, rng, fine, owner_ricetta, stazione, utenti, ricetta, versione,
-                   sottotask, profilo, pezzi_giorno, prefisso_lotto, prefisso_seriale):
-    """Una produzione chiusa per giorno lavorativo, con i suoi pezzi, eventi e misure."""
-    giorni = _giorni_lavorativi(fine)
-    operatore, capoturno = utenti["operatore"], utenti["capoturno"]
-    task_ids = {}  # marker -> id del task di misura
-    for marker, st in sottotask.items():
-        task_ids[marker] = st.task_id
-    ordine_task = sorted(set(task_ids.values()))
-    ultimo_task = ordine_task[-1]
-    contatore = 0
-    totale = 0
-    for indice_giorno, giorno in enumerate(giorni):
-        frazione = indice_giorno / max(1, len(giorni) - 1)  # 0 = primo giorno, 1 = ieri
-        n = rng.randint(*pezzi_giorno)
-        slot = _slot_giornalieri(rng, n)
-        lotto = f"{prefisso_lotto}{giorno:%y%m%d}"
-        inizio, fine_run = _ora_utc(giorno, slot[0] - 5), _ora_utc(giorno, slot[-1] + 10)
-        run = ProductionRun(
-            station_id=stazione.id, recipe_id=ricetta.id, version_id=versione.id,
-            operator_id=operatore.id, lot_number=lotto, status="closed",
-            measurement_interval_minutes=ricetta.measurement_interval_minutes,
-            cycle_count=n, started_at=inizio, closed_at=fine_run, closed_by=capoturno.id,
-            active_station_id=None,
-        )
-        session.add(run)
-        await session.flush()
-        session.add(ProductionEvent(run_id=run.id, event_type="start", user_id=operatore.id,
-                                    created_at=inizio))
-        for numero, minuti in enumerate(slot, start=1):
-            contatore += 1
-            seriale = f"{prefisso_seriale}{contatore:05d}"
-            istante = _ora_utc(giorno, minuti)
-            for marker in sorted(sottotask):
-                st = sottotask[marker]
-                _t, quota, scost, sigma, deriva = next(p for p in profilo if p[1]["marker_number"] == marker)
-                valore = _valore(rng, quota, scost, sigma, deriva, frazione)
-                esito, scarto = calculate_pass_fail(valore, st)
-                m = Measurement(
-                    subtask_id=st.id, version_id=versione.id, measured_by=operatore.id,
-                    value=valore, pass_fail=esito, deviation=scarto, lot_number=lotto,
-                    serial_number=seriale,
-                    input_method="usb_caliper" if rng.random() < 0.75 else "manual",
-                    input_duration_ms=rng.randint(2500, 9000),
-                    production_run_id=run.id,
-                    # un fuori tolleranza passa solo con l'ok del capoturno (come in produzione)
-                    supervisor_id=capoturno.id if esito == "fail" else None,
-                    authorised_at=istante + timedelta(minutes=1) if esito == "fail" else None,
-                    measured_at=istante + timedelta(seconds=20 * marker),
-                )
-                session.add(m)
-                totale += 1
-            # un task_measured per ogni task di misura non finale, poi il ciclo completato
-            for tid in ordine_task:
-                tipo = "cycle_completed" if tid == ultimo_task else "task_measured"
-                session.add(ProductionEvent(
-                    run_id=run.id, event_type=tipo, user_id=operatore.id, task_id=tid,
-                    created_at=istante + timedelta(minutes=1),
-                ))
-        session.add(ProductionEvent(run_id=run.id, event_type="close", user_id=capoturno.id,
-                                    supervisor_id=capoturno.id, created_at=fine_run))
-        await session.flush()
-    return contatore, totale
-
-
-async def _ricetta_dxf(session, owner, stazione):
-    """Demo visione: un task dxf_compare, senza grafo (si crea in fiera dal pannello DXF)."""
-    ricetta, versione = await _nuova_ricetta(
-        session, owner, stazione, code="FIERA-DXF", name="Profilo DXF (visione)",
-        description="Confronto di un profilo con il disegno DXF tramite la camera. "
-                    "La ricetta di visione si crea in fiera dal pannello «Crea ricetta da DXF».",
-        interval=None, requires_serial=False, note_versione="Ricetta dimostrativa per la fiera",
-    )
-    await _task(session, versione, 0, "Preparazione del pezzo", "note",
-                direttiva="Prima di misurare",
-                descrizione="Appoggiare il pezzo sul piano della camera, **in piano**.")
-    await _task(
-        session, versione, 1, "Confronto profilo con DXF", "dxf_compare",
-        direttiva="Acquisire e confrontare",
-        descrizione="Lo scostamento massimo del profilo rispetto al DXF deve restare entro la tolleranza.",
-        quote=[_quota(1, "Scostamento massimo profilo", 0.0, 0.15)],
-    )
-    return ricetta
-
-
-async def seed(replace: bool, dry_run: bool, fine: date) -> int:
-    """Esegue il seed in una sola transazione; con dry_run annulla tutto alla fine."""
-    rng = random.Random(SEED)
-    async with async_session_factory() as session:
-        utenti = await _upsert_utenti(session)
-        owner = utenti["adriano"]
-        stazione = await _upsert_stazione(session, owner)
-
-        esistenti = list((await session.execute(
-            select(Recipe).where(Recipe.code.in_(CODES))
-        )).scalars().all())
-        if esistenti and not replace:
-            print(f"Ricette FIERA gia' presenti ({', '.join(r.code for r in esistenti)}): "
-                  "non duplico. Usa --replace per rifarle.")
-        else:
-            if esistenti:
-                conti = await _conta_e_cancella(session, esistenti)
-                print(f"--replace: rimosse {', '.join(r.code for r in esistenti)} "
-                      f"({conti['misure']} misure, {conti['produzioni']} produzioni, "
-                      f"{conti['eventi']} eventi)")
-
-            r1, v1, sub1 = await _ricetta_meccanica(
-                session, owner, stazione, code="FIERA-FLANGIA", name="Flangia Ø80",
-                description="Flangia tornita in alluminio: diametri, spessore, interasse fori "
-                            "e profondita' della lamatura. Lotto e seriale obbligatori.",
-                interval=30, requires_serial=True, profilo=FLANGIA,
-                titoli_task=["Misure diametrali", "Spessore e posizione fori"],
-            )
-            n1, m1 = await _storico(session, rng, fine, owner, stazione, utenti, r1, v1, sub1,
-                                    FLANGIA, (10, 14), "LF", "FL")
-            r2, v2, sub2 = await _ricetta_meccanica(
-                session, owner, stazione, code="FIERA-ALBERO", name="Albero rettificato",
-                description="Albero in acciaio rettificato: sedi dei cuscinetti, lunghezza e "
-                            "rugosita'. Tolleranze strette.",
-                interval=30, requires_serial=False, profilo=ALBERO,
-                titoli_task=["Diametri rettificati", "Lunghezza e rugosita'"],
-            )
-            n2, m2 = await _storico(session, rng, fine, owner, stazione, utenti, r2, v2, sub2,
-                                    ALBERO, (8, 11), "LA", "AL")
-            await _ricetta_dxf(session, owner, stazione)
-            print(f"FIERA-FLANGIA: {n1} pezzi, {m1} misure; FIERA-ALBERO: {n2} pezzi, {m2} misure")
-
-        if dry_run:
-            await session.rollback()
-            print("--dry-run: nessuna modifica salvata.")
-        else:
-            await session.commit()
-            print(f"Seed fiera completato (stazione {stazione.code}, utenti: "
-                  f"{', '.join(u[0] for u in UTENTI)}).")
-    return 0
-
-
-def _factory_per_file(db: Path):
-    """Sessioni async su un file SQLite qualsiasi, con le stesse PRAGMA dell'app (FK attive)."""
-    engine = create_async_engine(f"sqlite+aiosqlite:///{db.resolve().as_posix()}", echo=False)
-
-    @event.listens_for(engine.sync_engine, "connect")
-    def _pragma(dbapi_connection, _record) -> None:
-        cur = dbapi_connection.cursor()
-        cur.execute("PRAGMA foreign_keys=ON")
-        cur.execute("PRAGMA busy_timeout=5000")
-        cur.close()
-
-    return engine, async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-
-async def _main_async(args) -> int:
-    global async_session_factory
-    engine, async_session_factory = _factory_per_file(args.db)
-    try:
-        fine = date.fromisoformat(args.fine) if args.fine else date.today() - timedelta(days=1)
-        return await seed(args.replace, args.dry_run, fine)
-    finally:
-        await engine.dispose()
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__,
-                                     formatter_class=argparse.RawDescriptionHelpFormatter)
-    parser.add_argument("--db", required=True, type=Path, help="file SQLite con lo schema gia' creato")
-    parser.add_argument("--replace", action="store_true",
-                        help="rimuove i dati FIERA-* (ricette, misure, produzioni) e li rifa'")
-    parser.add_argument("--dry-run", action="store_true", help="esegue tutto e annulla")
-    parser.add_argument("--fine", default=None, help="ultimo giorno dello storico AAAA-MM-GG (default: ieri)")
-    args = parser.parse_args()
-    if not args.db.is_file():
-        print(f"File non trovato: {args.db}")
-        return 2
-    return asyncio.run(_main_async(args))
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

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

src/backend/tests/test_seed_fiera.py

@@ -1,194 +0,0 @@
-"""Test di scripts/seed_fiera.py (issue #85): il seed del PC della fiera.
-
-Lo script non e' un modulo del pacchetto: si carica da file con importlib, come
-test_seed_collaudo. La parte "dato" gira sul DB in memoria dei test; l'ultimo test
-percorre invece la strada vera (file SQLite con `alembic upgrade head` + --db).
-"""
-import importlib.util
-import sqlite3
-from argparse import Namespace
-from datetime import date
-from pathlib import Path
-
-import pytest
-from alembic import command
-from alembic.config import Config
-from sqlalchemy import func, select
-
-from src.backend.config import Settings, settings
-from src.backend.models.orm.measurement import Measurement
-from src.backend.models.orm.recipe import Recipe
-from src.backend.models.orm.station import Station, StationRecipeAssignment
-from src.backend.models.orm.user import User
-from src.backend.services.auth_service import authenticate_user
-from src.backend.tests.conftest import TestSessionFactory, test_engine
-
-_SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "seed_fiera.py"
-_ALEMBIC_INI = Path(__file__).resolve().parents[1] / "migrations" / "alembic.ini"
-FINE = date(2026, 10, 22)  # data fissa: il test non dipende dal giorno in cui gira
-
-
-@pytest.fixture
-def seed_fiera(monkeypatch):
-    spec = importlib.util.spec_from_file_location("seed_fiera", _SCRIPT_PATH)
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    monkeypatch.setattr(module, "async_session_factory", TestSessionFactory)
-    return module
-
-
-async def _utente(db_session, username) -> User:
-    return (await db_session.execute(
-        select(User).where(User.username == username)
-    )).scalar_one()
-
-
-class TestSeedFiera:
-    async def test_utenti_ruoli_e_login(self, seed_fiera, db_session):
-        assert await seed_fiera.seed(False, False, FINE) == 0
-
-        attesi = {
-            "adriano": ("123", True),
-            "maker": ("123", False),
-            "capoturno": ("123", False),
-            "operatore": ("123", False),
-        }
-        for username, (password, admin) in attesi.items():
-            utente = await authenticate_user(db_session, username, password)
-            assert utente is not None, username
-            assert utente.is_admin is admin
-        assert (await _utente(db_session, "maker")).roles == ["Maker"]
-        assert "Supervisor" in (await _utente(db_session, "capoturno")).roles
-        assert (await _utente(db_session, "operatore")).roles == ["MeasurementTec"]
-        assert {"Maker", "Metrologist"} <= set((await _utente(db_session, "adriano")).roles)
-
-    async def test_setup_risulta_completato(self, seed_fiera, client, monkeypatch):
-        import src.backend.api.routers.setup as setup_mod
-        monkeypatch.setattr(settings, "setup_password", "test-setup-password")
-        monkeypatch.setattr(setup_mod, "engine", test_engine)
-        monkeypatch.setattr(setup_mod, "async_session_factory", TestSessionFactory)
-
-        # prima del seed il frontend vede il setup da fare, dopo no
-        assert (await client.get("/api/setup/status")).json() == {"initialised": False}
-        await seed_fiera.seed(False, False, FINE)
-        assert (await client.get("/api/setup/status")).json() == {"initialised": True}
-
-    async def test_stazione_e_ricette(self, seed_fiera, db_session):
-        await seed_fiera.seed(False, False, FINE)
-
-        stazione = (await db_session.execute(
-            select(Station).where(Station.code == "ST-DEFAULT")
-        )).scalar_one()
-        assert stazione.camera_url == "http://127.0.0.1:8765"
-        assert stazione.camera_source == "camera:601020"
-
-        assegnate = (await db_session.execute(
-            select(Recipe.code)
-            .join(StationRecipeAssignment, StationRecipeAssignment.recipe_id == Recipe.id)
-            .where(StationRecipeAssignment.station_id == stazione.id)
-        )).scalars().all()
-        assert sorted(assegnate) == ["FIERA-ALBERO", "FIERA-DXF", "FIERA-FLANGIA"]
-
-        # la ricetta DXF ha un task dxf_compare e nessun grafo di visione
-        from src.backend.models.orm.task import RecipeTask
-        dxf = (await db_session.execute(
-            select(RecipeTask).where(RecipeTask.task_type == "dxf_compare")
-        )).scalars().all()
-        assert len(dxf) == 1 and dxf[0].vision_json is None
-
-    async def test_misure_nel_numero_atteso(self, seed_fiera, db_session):
-        await seed_fiera.seed(False, False, FINE)
-
-        # ogni pezzo ha una misura per quota: pezzi 150-300 per ricetta
-        for codice, quote in (("FIERA-FLANGIA", 5), ("FIERA-ALBERO", 4)):
-            misure = (await db_session.execute(
-                select(func.count(Measurement.id))
-                .where(Measurement.version_id.in_(
-                    select(seed_fiera.RecipeVersion.id)
-                    .join(Recipe, Recipe.id == seed_fiera.RecipeVersion.recipe_id)
-                    .where(Recipe.code == codice)
-                ))
-            )).scalar_one()
-            assert misure % quote == 0
-            assert 150 <= misure // quote <= 300, (codice, misure // quote)
-
-        # lun-ven soltanto, e le misure compaiono nelle produzioni
-        giorni = (await db_session.execute(
-            select(seed_fiera.ProductionRun.started_at)
-        )).scalars().all()
-        assert giorni and all(g.weekday() < 5 for g in giorni)
-        senza_run = (await db_session.execute(
-            select(func.count(Measurement.id)).where(Measurement.production_run_id.is_(None))
-        )).scalar_one()
-        assert senza_run == 0
-
-    async def test_statistiche_numeriche(self, seed_fiera, client, db_session):
-        await seed_fiera.seed(False, False, FINE)
-        adriano = await _utente(db_session, "adriano")
-        from src.backend.services.auth_service import issue_api_key
-        chiave = await issue_api_key(db_session, adriano)
-        intestazioni = {"X-API-Key": chiave}
-
-        flangia = (await db_session.execute(
-            select(Recipe).where(Recipe.code == "FIERA-FLANGIA")
-        )).scalar_one()
-        quote = (await client.get(
-            "/api/statistics/subtasks", headers=intestazioni,
-            params={"recipe_id": flangia.id},
-        )).json()
-        assert len(quote) == 5
-        cpk = []
-        for quota in quote:
-            risposta = await client.get(
-                "/api/statistics/capability", headers=intestazioni,
-                params={"recipe_id": flangia.id, "subtask_id": quota["id"]},
-            )
-            assert risposta.status_code == 200
-            d = risposta.json()
-            for campo in ("mean", "std_dev", "cp", "cpk"):
-                assert isinstance(d[campo], (int, float)), (quota["id"], campo)
-            cpk.append(d["cpk"])
-        # Cpk diversi per quota: una capace, una al limite o sotto
-        assert max(cpk) > 1.4
-        assert min(cpk) < 1.1
-
-    async def test_secondo_giro_non_duplica_e_replace_rifa(self, seed_fiera, db_session):
-        async def conta():
-            return [
-                (await db_session.execute(select(func.count(t.id)))).scalar_one()
-                for t in (Recipe, Measurement, User)
-            ]
-
-        await seed_fiera.seed(False, False, FINE)
-        primo = await conta()
-        await seed_fiera.seed(False, False, FINE)
-        assert await conta() == primo  # utenti aggiornati, ricette e misure non duplicate
-
-        await seed_fiera.seed(True, False, FINE)
-        assert await conta() == primo  # stessi numeri: il seme e' fisso
-
-    async def test_dry_run_non_salva(self, seed_fiera, db_session):
-        await seed_fiera.seed(False, True, FINE)
-        assert (await db_session.execute(select(func.count(User.id)))).scalar_one() == 0
-        assert (await db_session.execute(select(func.count(Measurement.id)))).scalar_one() == 0
-
-
-async def test_file_sqlite_con_alembic(seed_fiera, tmp_path, monkeypatch):
-    """La strada vera: schema da `alembic upgrade head`, poi il seed con --db."""
-    db = tmp_path / "tmf.db"
-    url = f"sqlite+aiosqlite:///{db.as_posix()}"
-    monkeypatch.setattr(Settings, "database_url", property(lambda self: url))
-    # alembic fa asyncio.run(): fuori dal loop di pytest-asyncio, in un thread
-    import asyncio
-    await asyncio.to_thread(command.upgrade, Config(str(_ALEMBIC_INI)), "head")
-
-    argomenti = Namespace(db=db, replace=False, dry_run=False, fine=FINE.isoformat())
-    assert await seed_fiera._main_async(argomenti) == 0
-
-    con = sqlite3.connect(db)
-    try:
-        assert con.execute("SELECT COUNT(*) FROM users WHERE is_admin=1").fetchone()[0] == 1
-        assert con.execute("SELECT COUNT(*) FROM recipes").fetchone()[0] == 3
-        assert con.execute("SELECT COUNT(*) FROM measurements").fetchone()[0] > 1000
-    finally:
-        con.close()

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