TieMeasureFlow/scripts/calibrazione.py

"""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