diff --git a/scripts/camera_service.py b/scripts/camera_service.py index 707d6e0..820fe7a 100644 --- a/scripts/camera_service.py +++ b/scripts/camera_service.py @@ -10,21 +10,35 @@ OpenCV, cartella di immagini) con lo stesso contratto di `vs-gui`, ma solo raddrizzato (header X-Raddrizzato: si|no); "grezzo": true lo dà com'è. 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= -> 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, square_mm, image (file, opzionale: - senza, grab GREZZO dalla sorgente), save=false, cols/rows (opzionali). +Calibrazione a scacchiera (#83, logica in scripts/calibrazione.py). +#91: archivio UNICO di calibrazioni con nome; una sorgente ne ha una attiva: + POST /camera/calibrate multipart: source (obbligatorio solo senza image), + square_mm, image (file, opzionale: senza, grab GREZZO dalla sorgente), + cols/rows (opzionali). NON salva mai (il vecchio campo save e' ignorato). #84: SENZA cols/rows = calibrazione densa (default): una foto di scacchiera che riempie il campo, griglia trovata da sola, mappa di - distorsione salvata accanto al JSON (.npz); mm/px e px/mm sono quelli - dell'immagine RADDRIZZATA. CON cols/rows (angoli interni) = metodo - planare di prima. Campi: quelli di C1 + overlay_png (b64) + saved + - metodo ("densa"|"planare"), distorsione (bool) e, per la densa, - residuo_rms_px, residuo_max_px, copertura_pct; + distorsione; mm/px e px/mm sono quelli dell'immagine RADDRIZZATA. CON + cols/rows (angoli interni) = metodo planare di prima. Risposta: i campi + del calcolo + overlay_png (b64), calibrated_at, origine ("camera:" | + "file:"), calcolo_id (il servizio tiene in memoria gli ultimi 8 + calcoli) e metodo ("densa"|"planare"), distorsione (bool) e, per la + densa, residuo_rms_px, residuo_max_px, copertura_pct; 422 scacchiera non trovata o parametri invalidi - DELETE /camera/calibration?source= -> 204 (toglie JSON e mappa) + GET /camera/calibrazioni[?source=S] -> {calibrazioni: [{nome, origine, + image_width, image_height, px_per_mm, mm_per_px, metodo, square_mm, + calibrated_at, attiva_su}], attiva: nome|null} (dalla piu' recente) + POST /camera/calibrazioni json {calcolo_id, nome, sovrascrivi=false} -> 201 + metadati. 404 calcolo scaduto, 400 nome non valido, 409 nome gia' usato + DELETE /camera/calibrazioni?nome=X -> 204 (toglie anche la mappa e la scollega + da ogni sorgente) | 404 + PUT /camera/calibrazione-attiva json {source, nome}: legge un frame dalla + sorgente e carica la calibrazione solo se la risoluzione (larghezza x + altezza) coincide: 409 altrimenti; 404 nome ignoto; 200 come GET sotto + GET /camera/calibration?source= -> 200 la calibrazione ATTIVA {source, + nome, px_per_mm, mm_per_px, ..., image_width, image_height, + calibrated_at} | 404 "sorgente non calibrata" (lo legge il backend) + DELETE /camera/calibration?source= -> 204 (SCOLLEGA la calibrazione attiva, + la libreria resta) GET /calibrazione -> pagina HTML autonoma (offline) per fare tutto a mano GET /acquisizione -> stessa pagina (#87) Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default: @@ -43,7 +57,7 @@ Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default: POST /camera/parametri/salva json {source} -> {source, salvati, file} scrive /settaggi/-.json; alla prima apertura della camera nel servizio il file, se c'e', viene ricaricato. -Il risultato salvato sta in un JSON per sorgente in --config-dir (default +Le calibrazioni stanno in /calibrazioni, le attive in attive.json (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 @@ -70,6 +84,8 @@ import logging import os import sys import threading +import uuid # #91 id dei calcoli in attesa di nome +from collections import OrderedDict from contextlib import asynccontextmanager from pathlib import Path @@ -88,6 +104,7 @@ import calibrazione # noqa: E402 from calibrazione_pagina import PAGINA # noqa: E402 _TRIGGER = ("continuo",) +_MAX_CALCOLI = 8 # #91: calcoli tenuti in memoria in attesa di un nome _log = logging.getLogger("camera_service") # #87: errori non bloccanti (settaggi) # #87: i quattro parametri regolabili, nell'ordine di presentazione _PARAMETRI = ("esposizione", "esposizione_auto", "guadagno", "guadagno_auto") @@ -133,6 +150,8 @@ def create_app(folder: Path | str | None = None, aperte: dict = {} lock = threading.Lock() # protegge SOLO i dizionari, mai un'apertura lock_sorgenti: dict = {} # nome sorgente -> Lock che serializza apertura e grab + calcoli: OrderedDict = OrderedDict() # #91: calcolo_id -> (dati, mappa); gli ultimi _MAX_CALCOLI + lock_calcoli = threading.Lock() def _config() -> dict: if folder is None: @@ -226,7 +245,7 @@ def create_app(folder: Path | str | None = None, raise HTTPException(400, "«timeout_ms» deve essere un intero fra 1 e 10000") frame = _grab(nome, timeout) # #83: corpo estratto in _grab, invariato raddrizzato = False - if not grezzo: # #84: sorgente calibrata (densa) -> frame raddrizzato dalla mappa in cache + if not grezzo: # #84: sorgente calibrata (densa) -> frame raddrizzato dalla mappa in cache (#91: quella attiva) mappa = archivio.leggi_mappa(nome) if mappa is not None: corretto = calibrazione.raddrizza(frame, mappa) @@ -247,33 +266,35 @@ def create_app(folder: Path | str | None = None, @app.get("/camera/calibration") def calibrazione_leggi(source: str = ""): - dati = archivio.leggi(_sorgente(source)) + dati = archivio.leggi(_sorgente(source)) # #91: la calibrazione ATTIVA della sorgente 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)) + archivio.scollega(_sorgente(source)) # #91: scollega soltanto, la libreria resta return Response(status_code=204) @app.post("/camera/calibrate") async def calibrazione_calcola( - source: str = Form(...), + source: str | None = Form(None), # #91: serve solo se manca l'immagine square_mm: float = Form(...), cols: int | None = Form(None), # #84: opzionali, senza = calibrazione densa rows: int | None = Form(None), image: UploadFile | None = File(None), - save: bool = Form(False), + save: bool = Form(False), # #91: ignorato, il calcolo non salva mai (si salva con nome) ): - 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") + origine = "file:" + (Path((image.filename or "").replace("\\", "/")).name or "immagine") else: + nome = _sorgente(source) frame = await run_in_threadpool(_grab, nome, 5000) # come /camera/grab (404/503/504) + origine = nome if (cols is None) != (rows is None): raise HTTPException(422, "cols e rows vanno indicati insieme (o nessuno dei due)") mappa = None @@ -285,12 +306,71 @@ def create_app(folder: Path | str | None = None, 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, mappa) # #84: la mappa (se densa) va in .npz accanto - return {**corpo, "overlay_png": risultato["overlay_png"], "saved": bool(save)} + corpo = {**{k: v for k, v in risultato.items() if k != "overlay_png"}, + "calibrated_at": calibrazione.ora_iso(), "origine": origine} + calcolo_id = uuid.uuid4().hex + with lock_calcoli: # #91: il calcolo resta in memoria finche' non gli si da' un nome + calcoli[calcolo_id] = (corpo, mappa) + while len(calcoli) > _MAX_CALCOLI: + calcoli.popitem(last=False) + return {**corpo, "calcolo_id": calcolo_id, "overlay_png": risultato["overlay_png"]} + + # ---- #91: archivio unico di calibrazioni con nome + + @app.get("/camera/calibrazioni") + def calibrazioni_elenco(source: str | None = None): + attiva = archivio.attiva_di(_sorgente(source)) if source is not None else None + return {"calibrazioni": archivio.elenco(), "attiva": attiva} + + @app.post("/camera/calibrazioni", status_code=201) + def calibrazioni_salva(corpo: dict = Body(default=None)): + corpo = corpo if isinstance(corpo, dict) else {} + calcolo_id = corpo.get("calcolo_id") + if not isinstance(calcolo_id, str) or not calcolo_id: + raise HTTPException(400, "manca «calcolo_id»") + sovrascrivi = corpo.get("sovrascrivi", False) + if not isinstance(sovrascrivi, bool): + raise HTTPException(400, "«sovrascrivi» deve essere true o false") + try: + nome = calibrazione.pulisci_nome(corpo.get("nome")) + except calibrazione.NomeNonValido as e: + raise HTTPException(400, str(e)) + with lock_calcoli: + voce = calcoli.get(calcolo_id) + if voce is None: + raise HTTPException(404, "calcolo non più disponibile: ricalcolare") + dati, mappa = voce + try: + return archivio.salva_nome(nome, dati, mappa, sovrascrivi) + except calibrazione.NomeGiaUsato as e: + raise HTTPException(409, str(e)) + + @app.delete("/camera/calibrazioni", status_code=204) + def calibrazioni_elimina(nome: str = ""): + if not archivio.elimina_nome(nome): + raise HTTPException(404, "calibrazione non trovata") + return Response(status_code=204) + + @app.put("/camera/calibrazione-attiva") + def calibrazione_attiva(corpo: dict = Body(default=None)): + corpo = corpo if isinstance(corpo, dict) else {} + sorgente = _sorgente(corpo.get("source")) + richiesto = corpo.get("nome") + dati = archivio.leggi_nome(richiesto) if isinstance(richiesto, str) else None + if dati is None: + raise HTTPException(404, "calibrazione non trovata") + nome = dati["nome"] + frame = _grab(sorgente, 5000) # stesso lock e stessi errori di /camera/grab (404/503/504) + alt, larg = frame.shape[:2] + cw, ch = dati.get("image_width"), dati.get("image_height") + if not isinstance(cw, int) or not isinstance(ch, int): + raise HTTPException(409, f"la calibrazione «{nome}» non ha la risoluzione registrata: " + "non si può caricare") + if (cw, ch) != (larg, alt): # la mappa vale solo per frame della stessa risoluzione + raise HTTPException(409, f"la calibrazione «{nome}» è per immagini {cw}×{ch}, " + f"la sorgente dà {larg}×{alt}: non si può caricare") + archivio.attiva(sorgente, nome) + return archivio.leggi(sorgente) # ---- #87: acquisizione (anteprima, parametri, settaggi) diff --git a/scripts/tests/test_calibrazione.py b/scripts/tests/test_calibrazione.py index b2a0d62..9af7a5f 100644 --- a/scripts/tests/test_calibrazione.py +++ b/scripts/tests/test_calibrazione.py @@ -82,12 +82,24 @@ def client(cartella, tmp_path): CAMPI = {"cols": "9", "rows": "6", "square_mm": "2"} -CHIAVI_C1 = {"source", "px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows", +CHIAVI_C1 = {"px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows", "square_mm", "image_width", "image_height", "scale_spread_pct", "calibrated_at"} # #84: campi in piu' per metodo planare (C2 con cols/rows) e per la densa -CHIAVI_PLANARE = CHIAVI_C1 | {"metodo", "distorsione"} +CHIAVI_PLANARE = CHIAVI_C1 | {"metodo", "distorsione", "origine"} # #91: + origine ("camera:X" | "file:Y") CHIAVI_DENSA = CHIAVI_PLANARE | {"residuo_rms_px", "residuo_max_px", "copertura_pct"} SORGENTE = "camera:folder" +# #91: GET /camera/calibration = voce di libreria + "nome" + "source" (la sorgente che la usa) +CHIAVI_ATTIVA = {"source", "nome"} + + +def salva_con_nome(client, calcolo: dict, nome: str, **extra): + """#91: POST /camera/calibrazioni sul calcolo_id di una risposta di /camera/calibrate.""" + return client.post("/camera/calibrazioni", json={"calcolo_id": calcolo["calcolo_id"], "nome": nome, **extra}) + + +def carica_su(client, sorgente: str, nome: str): + """#91: PUT /camera/calibrazione-attiva.""" + return client.put("/camera/calibrazione-attiva", json={"source": sorgente, "nome": nome}) @pytest.fixture @@ -157,34 +169,49 @@ def test_c2_immagine_caricata_senza_salvare(client): files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) assert r.status_code == 200, r.text corpo = r.json() - assert CHIAVI_PLANARE | {"overlay_png", "saved"} == set(corpo) + assert CHIAVI_PLANARE | {"overlay_png", "calcolo_id"} == set(corpo) # #91: niente "saved", niente "source" assert corpo["metodo"] == "planare" and corpo["distorsione"] is False # #84: con cols/rows nulla cambia - assert corpo["saved"] is False + assert corpo["origine"] == "file:s.png" assert corpo["px_per_mm"] == pytest.approx(20.0, rel=0.005) - # senza save non si scrive nulla + # il calcolo non salva mai assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404 - assert not client.config.exists() or not list(client.config.glob("*.json")) + assert not client.config.exists() or not list(client.config.iterdir()) + + +def test_c2_save_true_viene_ignorato(client): + # #91: il vecchio campo save non salva piu' nulla + r = client.post("/camera/calibrate", data={"source": "camera:folder", "save": "true", **CAMPI}) + assert r.status_code == 200, r.text + assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404 + assert client.get("/camera/calibrazioni").json()["calibrazioni"] == [] + assert not client.config.exists() or not list(client.config.iterdir()) 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}) + # nessun file: grab da camera:folder (la scacchiera in cartella) + r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI}) assert r.status_code == 200, r.text - assert r.json()["saved"] is True + assert r.json()["origine"] == "camera:folder" + assert salva_con_nome(client, r.json(), "Banco").status_code == 201 # #91: si salva con nome, dopo + + # salvata ma non ancora caricata sulla sorgente + assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404 + assert carica_su(client, "camera:folder", "Banco").status_code == 200 r1 = client.get("/camera/calibration", params={"source": "camera:folder"}) assert r1.status_code == 200 - assert set(r1.json()) == CHIAVI_PLANARE # nessun overlay nel file salvato + assert set(r1.json()) == CHIAVI_PLANARE | CHIAVI_ATTIVA # 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 r1.json()["source"] == "camera:folder" and r1.json()["nome"] == "Banco" + file_salvati = list((client.config / "calibrazioni").glob("*.json")) assert len(file_salvati) == 1 assert json.loads(file_salvati[0].read_text(encoding="utf-8"))["cols"] == 9 + # DELETE /camera/calibration scollega: la libreria resta 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 len(list((client.config / "calibrazioni").glob("*.json"))) == 1 + # scollegare di nuovo e' idempotente assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204 @@ -214,12 +241,14 @@ def test_c2_errori_camera_come_grab(client): 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_due_nomi_simili_hanno_file_diversi(client): + # #91: i nomi si ripuliscono come prima (hash sul nome originale): due nomi diversi non si pestano + r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI}, + files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 200 + for nome in ("banco a b", "banco a_b"): + assert salva_con_nome(client, r.json(), nome).status_code == 201 + assert len(list((client.config / "calibrazioni").glob("*.json"))) == 2 def test_c4_pagina_html(client): @@ -308,11 +337,10 @@ def test_c2_senza_cols_rows_e_densa(client_denso): r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}) assert r.status_code == 200, r.text corpo = r.json() - assert CHIAVI_DENSA | {"overlay_png", "saved"} == set(corpo) + assert CHIAVI_DENSA | {"overlay_png", "calcolo_id"} == set(corpo) assert corpo["metodo"] == "densa" and corpo["distorsione"] is True assert (corpo["cols"], corpo["rows"]) == (15, 11) assert corpo["px_per_mm"] == pytest.approx(1.0 / corpo["mm_per_px"]) - assert corpo["saved"] is False assert not client_denso.config.exists() or not list(client_denso.config.iterdir()) @@ -327,14 +355,17 @@ def test_grab_raddrizzato_se_calibrata_e_grezzo_a_richiesta(client_denso, cartel assert prima.headers["x-raddrizzato"] == "no" # non calibrata: frame com'e' assert np.array_equal(_grigio(prima), originale) - r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"}) + r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}) assert r.status_code == 200, r.text - assert r.json()["saved"] is True - json_salvati, mappe = list(client_denso.config.glob("*.json")), list(client_denso.config.glob("*.npz")) + assert salva_con_nome(client_denso, r.json(), "Densa").status_code == 201 # #91: salvata con nome + libreria = client_denso.config / "calibrazioni" + json_salvati, mappe = list(libreria.glob("*.json")), list(libreria.glob("*.npz")) assert len(json_salvati) == 1 and len(mappe) == 1 assert mappe[0].stem == json_salvati[0].stem # stesso nome del JSON + assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no" # salvata != attiva + assert carica_su(client_denso, SORGENTE, "Densa").status_code == 200 saved = client_denso.get("/camera/calibration", params={"source": SORGENTE}).json() - assert set(saved) == CHIAVI_DENSA and saved["metodo"] == "densa" + assert set(saved) == CHIAVI_DENSA | CHIAVI_ATTIVA and saved["metodo"] == "densa" dopo = client_denso.post("/camera/grab", json={"source": SORGENTE}) assert dopo.status_code == 200 and dopo.headers["x-raddrizzato"] == "si" @@ -359,19 +390,33 @@ def test_grab_grezzo_non_booleano_400(client_denso): assert r.status_code == 400 -def test_delete_toglie_anche_la_mappa(client_denso): - client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"}) - assert list(client_denso.config.glob("*.npz")) +def test_scollega_lascia_la_libreria_ed_elimina_toglie_la_mappa(client_denso): + r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}) + salva_con_nome(client_denso, r.json(), "Densa") + carica_su(client_denso, SORGENTE, "Densa") + libreria = client_denso.config / "calibrazioni" + assert list(libreria.glob("*.npz")) + # scollegare: il frame torna com'e', ma la voce (e la mappa) restano in libreria assert client_denso.delete("/camera/calibration", params={"source": SORGENTE}).status_code == 204 - assert not list(client_denso.config.glob("*.npz")) and not list(client_denso.config.glob("*.json")) + assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no" + assert list(libreria.glob("*.npz")) and list(libreria.glob("*.json")) + # eliminare la voce toglie JSON e mappa + carica_su(client_denso, SORGENTE, "Densa") + assert client_denso.delete("/camera/calibrazioni", params={"nome": "Densa"}).status_code == 204 + assert not list(libreria.glob("*.npz")) and not list(libreria.glob("*.json")) assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no" def test_ricalibrare_planare_toglie_la_mappa_densa(client_denso): - client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"}) - assert list(client_denso.config.glob("*.npz")) - planare = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "save": "true", **CAMPI}, + densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}) + salva_con_nome(client_denso, densa.json(), "Cal") + carica_su(client_denso, SORGENTE, "Cal") + libreria = client_denso.config / "calibrazioni" + assert list(libreria.glob("*.npz")) + planare = client_denso.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI}, files={"image": ("s.png", _png(una_scacchiera()), "image/png")}) assert planare.status_code == 200 and planare.json()["metodo"] == "planare" - assert not list(client_denso.config.glob("*.npz")) # niente mappa vecchia che raddrizza a vuoto + assert salva_con_nome(client_denso, planare.json(), "Cal").status_code == 409 # nome gia' preso + assert salva_con_nome(client_denso, planare.json(), "Cal", sovrascrivi=True).status_code == 201 + assert not list(libreria.glob("*.npz")) # niente mappa vecchia che raddrizza a vuoto assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no" diff --git a/scripts/tests/test_calibrazioni_rotte.py b/scripts/tests/test_calibrazioni_rotte.py new file mode 100644 index 0000000..666a2cf --- /dev/null +++ b/scripts/tests/test_calibrazioni_rotte.py @@ -0,0 +1,250 @@ +"""Test delle rotte dell'archivio unico di calibrazioni con nome (#91): scripts/camera_service.py. + +Uso: uv run --python 3.13 --extra vision-worker --extra dev \ + --with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q +Sorgenti: camera:folder su cartelle temporanee (520x400 scacchiera piana, 640x480 densa). +Solo tmp_path: niente ProgramData, niente camera vera. +""" +import json + +import cv2 +import numpy as np +import pytest +from fastapi.testclient import TestClient + +from test_calibrazione import ( # fixture e helper della calibrazione (#83/#84) + CAMPI, CHIAVI_ATTIVA, CHIAVI_DENSA, CHIAVI_PLANARE, SORGENTE, _carica, _png, + carica_su, cartella, cartella_densa, client, client_denso, salva_con_nome, + una_scacchiera, una_scacchiera_densa, +) + +__all__ = ["cartella", "cartella_densa", "client", "client_denso"] # le fixture importate restano usate + + +def _calcola_file(c, **extra): + """Calcolo planare da un'immagine (520x400): ritorna la risposta JSON.""" + r = c.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI, **extra}, + files={"image": ("scacco.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 200, r.text + return r.json() + + +# ---- POST /camera/calibrate + +def test_calibrate_senza_source_con_immagine_ok(client): + r = client.post("/camera/calibrate", data=CAMPI, + files={"image": ("C:\\foto\\scacco.png", _png(una_scacchiera()), "image/png")}) + assert r.status_code == 200, r.text + assert r.json()["origine"] == "file:scacco.png" # solo il nome del file, non il percorso + + +def test_calibrate_senza_source_e_senza_immagine_400(client): + r = client.post("/camera/calibrate", data=CAMPI) + assert r.status_code == 400 and "source" in r.json()["detail"] + + +def test_calibrate_da_sorgente_ha_origine_camera_e_id(client): + r = client.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI}) + assert r.status_code == 200, r.text + corpo = r.json() + assert corpo["origine"] == SORGENTE and corpo["calibrated_at"] + assert len(corpo["calcolo_id"]) == 32 and int(corpo["calcolo_id"], 16) >= 0 # esadecimale + assert corpo["calcolo_id"] != client.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI}).json()["calcolo_id"] + + +# ---- POST /camera/calibrazioni + +def test_salva_201_con_metadati(client): + calcolo = _calcola_file(client) + r = salva_con_nome(client, calcolo, " Banco 1 ") + assert r.status_code == 201, r.text + meta = r.json() + assert meta["nome"] == "Banco 1" and meta["origine"] == "file:scacco.png" + assert (meta["image_width"], meta["image_height"]) == (520, 400) + assert meta["metodo"] == "planare" and meta["attiva_su"] == [] + assert meta["px_per_mm"] == pytest.approx(20.0, rel=0.005) + # lo stesso calcolo si puo' salvare anche con un altro nome + assert salva_con_nome(client, calcolo, "Banco 2").status_code == 201 + + +def test_salva_calcolo_sconosciuto_404(client): + r = client.post("/camera/calibrazioni", json={"calcolo_id": "0" * 32, "nome": "X"}) + assert r.status_code == 404 and r.json()["detail"] == "calcolo non più disponibile: ricalcolare" + + +def test_salva_calcolo_scaduto_dopo_otto_calcoli_404(client): + primo = _calcola_file(client) + ultimo = primo + for _ in range(8): # il nono calcolo spinge fuori il primo + ultimo = _calcola_file(client) + assert salva_con_nome(client, primo, "Vecchio").status_code == 404 + assert salva_con_nome(client, ultimo, "Nuovo").status_code == 201 + + +@pytest.mark.parametrize("nome", ["", " ", "x" * 61, "a\x01b", None, 7]) +def test_salva_nome_non_valido_400(client, nome): + r = client.post("/camera/calibrazioni", json={"calcolo_id": _calcola_file(client)["calcolo_id"], "nome": nome}) + assert r.status_code == 400, r.text + assert client.get("/camera/calibrazioni").json()["calibrazioni"] == [] + + +def test_salva_corpo_incompleto_400(client): + assert client.post("/camera/calibrazioni", json={"nome": "X"}).status_code == 400 + calcolo = _calcola_file(client) + r = client.post("/camera/calibrazioni", json={"calcolo_id": calcolo["calcolo_id"], "nome": "X", "sovrascrivi": "si"}) + assert r.status_code == 400 + + +def test_salva_nome_gia_usato_409_e_sovrascrivi(client): + a = _calcola_file(client) + assert salva_con_nome(client, a, "Banco").status_code == 201 + r = salva_con_nome(client, a, "BANCO") # senza distinguere maiuscole + assert r.status_code == 409 and r.json()["detail"] == "esiste già una calibrazione «Banco»" + assert len(client.get("/camera/calibrazioni").json()["calibrazioni"]) == 1 + r = salva_con_nome(client, a, "BANCO", sovrascrivi=True) + assert r.status_code == 201 and r.json()["nome"] == "BANCO" + voci = client.get("/camera/calibrazioni").json()["calibrazioni"] + assert [v["nome"] for v in voci] == ["BANCO"] + + +# ---- GET /camera/calibrazioni + +def test_elenco_vuoto_e_ordine(client): + assert client.get("/camera/calibrazioni").json() == {"calibrazioni": [], "attiva": None} + prima = _calcola_file(client) + salva_con_nome(client, prima, "prima") + seconda = _calcola_file(client) + seconda_ts = seconda["calibrated_at"] + salva_con_nome(client, seconda, "seconda") + voci = client.get("/camera/calibrazioni").json()["calibrazioni"] + assert {v["nome"] for v in voci} == {"prima", "seconda"} + assert [v["calibrated_at"] for v in voci] == sorted((v["calibrated_at"] for v in voci), reverse=True) + assert set(voci[0]) == {"nome", "origine", "image_width", "image_height", "px_per_mm", "mm_per_px", + "metodo", "square_mm", "calibrated_at", "attiva_su"} + assert seconda_ts >= prima["calibrated_at"] + + +def test_elenco_con_source_dice_l_attiva(client): + salva_con_nome(client, _calcola_file(client), "Banco") + salva_con_nome(client, _calcola_file(client), "Altra") + assert carica_su(client, SORGENTE, "banco").status_code == 200 + r = client.get("/camera/calibrazioni", params={"source": SORGENTE}).json() + assert r["attiva"] == "Banco" + attiva_su = {v["nome"]: v["attiva_su"] for v in r["calibrazioni"]} + assert attiva_su == {"Banco": [SORGENTE], "Altra": []} + assert client.get("/camera/calibrazioni", params={"source": "camera:altra"}).json()["attiva"] is None + assert client.get("/camera/calibrazioni", params={"source": " "}).status_code == 400 + + +# ---- DELETE /camera/calibrazioni + +def test_elimina_204_404_e_toglie_dalle_sorgenti(client): + salva_con_nome(client, _calcola_file(client), "Banco") + carica_su(client, SORGENTE, "Banco") + assert client.delete("/camera/calibrazioni", params={"nome": "BANCO"}).status_code == 204 + assert client.get("/camera/calibrazioni").json()["calibrazioni"] == [] + assert client.get("/camera/calibration", params={"source": SORGENTE}).status_code == 404 # non piu' attiva + assert json.loads((client.config / "attive.json").read_text(encoding="utf-8")) == {} + assert client.delete("/camera/calibrazioni", params={"nome": "Banco"}).status_code == 404 + assert client.delete("/camera/calibrazioni").status_code == 404 # senza nome + + +# ---- PUT /camera/calibrazione-attiva + +def test_carica_risoluzione_uguale_200_come_get(client): + salva_con_nome(client, _calcola_file(client), "Banco") # 520x400 = la scacchiera in cartella + r = carica_su(client, SORGENTE, "banco") + assert r.status_code == 200, r.text + assert set(r.json()) == CHIAVI_PLANARE | CHIAVI_ATTIVA + assert r.json()["nome"] == "Banco" and r.json()["source"] == SORGENTE + assert client.get("/camera/calibration", params={"source": SORGENTE}).json() == r.json() + + +def test_carica_risoluzione_diversa_409_e_anteprima_continua(client_denso): + # calibrazione da file 520x400, sorgente 640x480 + salva_con_nome(client_denso, _calcola_file(client_denso), "Piccola") + r = carica_su(client_denso, SORGENTE, "Piccola") + assert r.status_code == 409 + assert r.json()["detail"] == ("la calibrazione «Piccola» è per immagini 520×400, " + "la sorgente dà 640×480: non si può caricare") + assert client_denso.get("/camera/calibration", params={"source": SORGENTE}).status_code == 404 # nulla attivato + # l'acquisizione continua: grab e anteprima rispondono, frame non raddrizzato + g = client_denso.post("/camera/grab", json={"source": SORGENTE}) + assert g.status_code == 200 and g.headers["x-raddrizzato"] == "no" + a = client_denso.post("/camera/anteprima", json={"source": SORGENTE}) + assert a.status_code == 200 and a.headers["x-raddrizzato"] == "no" + assert (a.headers["x-larghezza"], a.headers["x-altezza"]) == ("640", "480") + + +def test_carica_risoluzione_diversa_non_scalza_l_attiva_precedente(client_denso): + densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}).json() + salva_con_nome(client_denso, densa, "Giusta") + salva_con_nome(client_denso, _calcola_file(client_denso), "Sbagliata") + assert carica_su(client_denso, SORGENTE, "Giusta").status_code == 200 + assert carica_su(client_denso, SORGENTE, "Sbagliata").status_code == 409 + assert client_denso.get("/camera/calibration", params={"source": SORGENTE}).json()["nome"] == "Giusta" + assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "si" + + +def test_carica_densa_su_sorgente_stessa_risoluzione_raddrizza(client_denso): + densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"}).json() + assert salva_con_nome(client_denso, densa, "Densa").json()["metodo"] == "densa" + r = carica_su(client_denso, SORGENTE, "Densa") + assert r.status_code == 200 and set(r.json()) == CHIAVI_DENSA | CHIAVI_ATTIVA + assert client_denso.post("/camera/anteprima", json={"source": SORGENTE}).headers["x-raddrizzato"] == "si" + + +def test_carica_nome_sconosciuto_404(client): + r = carica_su(client, SORGENTE, "nessuna") + assert r.status_code == 404 + assert client.put("/camera/calibrazione-attiva", json={"source": SORGENTE}).status_code == 404 # senza nome + assert client.put("/camera/calibrazione-attiva", json={"nome": "x"}).status_code == 400 # senza source + + +def test_carica_su_sorgente_inesistente_404(client): + salva_con_nome(client, _calcola_file(client), "Banco") + assert carica_su(client, "camera:non-esiste", "Banco").status_code == 404 + + +def test_carica_senza_risoluzione_registrata_409(cartella, tmp_path): + _carica() + import calibrazione + config = tmp_path / "config_senza" + calibrazione.Archivio(config).salva_nome("Senza", {"px_per_mm": 10.0, "calibrated_at": "2026-01-01T00:00:00+00:00"}) + with TestClient(_carica().create_app(cartella, config_dir=config)) as c: + r = carica_su(c, SORGENTE, "Senza") + assert r.status_code == 409 + assert r.json()["detail"] == "la calibrazione «Senza» non ha la risoluzione registrata: non si può caricare" + + +# ---- contratto verso il backend e migrazione all'avvio + +def test_get_calibration_contratto_per_il_backend(client): + # src/backend/services/vision_service.py legge questi campi + salva_con_nome(client, _calcola_file(client), "Banco") + carica_su(client, SORGENTE, "Banco") + dati = client.get("/camera/calibration", params={"source": SORGENTE}).json() + for campo in ("px_per_mm", "mm_per_px", "image_width", "image_height", "calibrated_at", "metodo"): + assert campo in dati + assert client.get("/camera/calibration", params={"source": "camera:altra"}).json() == { + "detail": "sorgente non calibrata"} + + +def test_avvio_migra_il_formato_vecchio(cartella, tmp_path): + _carica() + import calibrazione + config = tmp_path / "config_vecchia" + config.mkdir() + nudo = calibrazione.Archivio.__new__(calibrazione.Archivio) # per avere il nome file del formato vecchio + nudo.cartella = config + vecchio = nudo._file(SORGENTE) + vecchio.write_text(json.dumps({"source": SORGENTE, "px_per_mm": 20.0, "mm_per_px": 0.05, + "image_width": 520, "image_height": 400, "metodo": "planare", + "calibrated_at": "2026-10-01T10:00:00+00:00"}), encoding="utf-8") + with TestClient(_carica().create_app(cartella, config_dir=config)) as c: + dati = c.get("/camera/calibration", params={"source": SORGENTE}).json() + assert dati["nome"] == SORGENTE and dati["source"] == SORGENTE and dati["px_per_mm"] == 20.0 + voci = c.get("/camera/calibrazioni", params={"source": SORGENTE}).json() + assert [v["nome"] for v in voci["calibrazioni"]] == [SORGENTE] and voci["attiva"] == SORGENTE + assert voci["calibrazioni"][0]["attiva_su"] == [SORGENTE] + assert not vecchio.exists()