acquisitore: rotte per calibrazioni con nome, caricamento con controllo risoluzione (Refs #91)

/camera/calibrate non salva piu' (calcolo_id in memoria, ultimi 8); nuove rotte
GET/POST/DELETE /camera/calibrazioni e PUT /camera/calibrazione-attiva (409 se la
risoluzione non combacia col frame della sorgente); DELETE /camera/calibration ora
scollega. Test: rotte nuove e test esistenti aggiornati (save=true -> salva con nome).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
This commit is contained in:
2026-10-24 16:10:10 +02:00
parent e7d5489438
commit a5e8472e88
3 changed files with 434 additions and 59 deletions
+250
View File
@@ -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()