Adriano
552f96bb47
POST /camera/calibrazioni/manuale salva una voce senza mappa (metodo manuale); anteprima/calibrata senza mappa risponde il frame grezzo (X-Raddrizzato: no); scheda «Calibrazione manuale» nella tab Calibrazione. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""Test di #92: immagini caricate come sorgenti e anteprima calibrata (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
|
||
Solo tmp_path: niente ProgramData, niente camera vera. Le immagini le scrive il test.
|
||
"""
|
||
from urllib.parse import quote, unquote
|
||
|
||
import cv2
|
||
import numpy as np
|
||
import pytest
|
||
|
||
from test_calibrazione import ( # fixture e helper della calibrazione (#83/#84)
|
||
CAMPI, _png, carica_su, cartella, client, salva_con_nome, una_scacchiera,
|
||
una_scacchiera_densa,
|
||
)
|
||
|
||
__all__ = ["cartella", "client"] # le fixture importate restano usate
|
||
|
||
DENSA = "densa.png"
|
||
PIANA = "piana.png"
|
||
|
||
|
||
def _modulo(c) -> dict:
|
||
"""Il dict globale di camera_service (per sostituire open_source con una spia)."""
|
||
return next(r.endpoint for r in c.app.routes if r.path == "/camera/grab").__globals__
|
||
|
||
|
||
def _carica(c, nome: str, img: np.ndarray, **extra):
|
||
"""POST /camera/immagini con un'immagine codificata PNG (o il contenuto dato)."""
|
||
return c.post("/camera/immagini", files={"image": (nome, _png(img), "image/png")}, **extra)
|
||
|
||
|
||
def _densa(**kw) -> np.ndarray:
|
||
return una_scacchiera_densa(k=1e-7, **kw)
|
||
|
||
|
||
def _calibra_e_attiva(c, nome_immagine: str, nome_calibrazione: str, **campi):
|
||
"""Calcola dalla sorgente, salva con nome e la rende attiva su quella sorgente."""
|
||
r = c.post("/camera/calibrate", data={"source": nome_immagine, "square_mm": "2", **campi})
|
||
assert r.status_code == 200, r.text
|
||
assert salva_con_nome(c, r.json(), nome_calibrazione).status_code == 201
|
||
assert carica_su(c, nome_immagine, nome_calibrazione).status_code == 200
|
||
|
||
|
||
def _anteprima(c, nome: str, **extra):
|
||
return c.post("/camera/anteprima", json={"source": nome, "grezzo": True, **extra})
|
||
|
||
|
||
def _calibrata(c, nome: str, **extra):
|
||
return c.post("/camera/anteprima/calibrata", json={"source": nome, **extra})
|
||
|
||
|
||
# ---- POST /camera/immagini
|
||
|
||
def test_carica_201_salva_in_cartella_immagini(client):
|
||
r = _carica(client, "calibration_135_1.bmp", una_scacchiera())
|
||
assert r.status_code == 201, r.text
|
||
assert r.json() == {"source": "calibration_135_1.bmp",
|
||
"label": "immagine — calibration_135_1.bmp", "kind": "image"}
|
||
assert (client.config / "immagini" / "calibration_135_1.bmp").is_file()
|
||
|
||
|
||
def test_carica_ripulisce_il_nome_e_toglie_il_percorso(client):
|
||
r = _carica(client, "C:\\foto\\ban co è1.PNG", una_scacchiera())
|
||
assert r.status_code == 201, r.text
|
||
assert r.json()["source"] == "ban_co_1.PNG" # niente percorso, caratteri sicuri, estensione intatta
|
||
assert (client.config / "immagini" / "ban_co_1.PNG").is_file()
|
||
|
||
|
||
def test_carica_stesso_nome_sostituisce(client):
|
||
_carica(client, "a.png", np.full((48, 64), 10, np.uint8))
|
||
assert _carica(client, "a.png", np.full((30, 40), 200, np.uint8)).status_code == 201
|
||
assert [f.name for f in (client.config / "immagini").iterdir()] == ["a.png"]
|
||
img = cv2.imread(str(client.config / "immagini" / "a.png"), cv2.IMREAD_UNCHANGED)
|
||
assert img.shape[:2] == (30, 40)
|
||
|
||
|
||
def test_carica_senza_file_400(client):
|
||
assert client.post("/camera/immagini").status_code == 400
|
||
|
||
|
||
def test_carica_estensione_non_ammessa_400(client):
|
||
r = client.post("/camera/immagini", files={"image": ("note.txt", b"ciao", "text/plain")})
|
||
assert r.status_code == 400
|
||
assert not (client.config / "immagini").exists() or not list((client.config / "immagini").iterdir())
|
||
|
||
|
||
def test_carica_file_illeggibile_422_e_niente_su_disco(client):
|
||
r = client.post("/camera/immagini", files={"image": ("rotta.png", b"non e una immagine", "image/png")})
|
||
assert r.status_code == 422
|
||
assert not (client.config / "immagini").exists() or not list((client.config / "immagini").iterdir())
|
||
|
||
|
||
# ---- GET /camera/sources
|
||
|
||
def test_sources_con_immagini_e_senza_camera_file(client):
|
||
_carica(client, "b.png", una_scacchiera())
|
||
_carica(client, "a.png", una_scacchiera())
|
||
voci = client.get("/camera/sources").json()["sources"]
|
||
nomi = [v["name"] for v in voci]
|
||
assert "camera:file" not in nomi
|
||
assert "camera:folder" in nomi
|
||
assert nomi[-2:] == ["a.png", "b.png"] # le immagini stanno in coda
|
||
assert voci[-1] == {"name": "b.png", "kind": "image", "label": "immagine — b.png"}
|
||
assert nomi.index("camera:folder") < nomi.index("a.png")
|
||
|
||
|
||
def test_sources_camere_in_cache_e_aggiorna(client):
|
||
# #92: la ricerca camere (~10 s) si fa una volta e resta 60 s; aggiorna=true la rilancia; le immagini sono sempre fresche
|
||
g = _modulo(client)
|
||
originale = g["list_sources"]
|
||
chiamate = []
|
||
|
||
def finto(cartella, cameras=True, **kw):
|
||
if cameras:
|
||
chiamate.append(1)
|
||
return originale(cartella, cameras=cameras, **kw)
|
||
|
||
g["list_sources"] = finto
|
||
try:
|
||
client.get("/camera/sources") # se la ricerca d'avvio e' ancora in corso aspetta il suo lock
|
||
base = len(chiamate)
|
||
_carica(client, "nuova.png", una_scacchiera())
|
||
voci = client.get("/camera/sources").json()["sources"]
|
||
client.get("/camera/sources")
|
||
assert len(chiamate) == base # seconda e terza chiamata: nessuna nuova ricerca
|
||
assert voci[-1]["name"] == "nuova.png" # ma l'immagine appena caricata c'e'
|
||
client.get("/camera/sources", params={"aggiorna": "true"})
|
||
assert len(chiamate) == base + 1 # aggiorna=true la rilancia
|
||
client.get("/camera/sources")
|
||
assert len(chiamate) == base + 1 # e il risultato nuovo torna in cache
|
||
finally:
|
||
g["list_sources"] = originale
|
||
|
||
|
||
def test_sources_senza_immagini_non_crea_la_cartella(client):
|
||
voci = client.get("/camera/sources").json()["sources"]
|
||
assert all(v["kind"] != "image" for v in voci)
|
||
|
||
|
||
# ---- grab / anteprima / calibrate su un'immagine caricata
|
||
|
||
def test_grab_anteprima_e_grab_ripetuti_su_immagine(client):
|
||
_carica(client, "foto.png", una_scacchiera())
|
||
for _ in range(3): # l'anteprima continua non deve esaurire un'immagine sola
|
||
r = client.post("/camera/grab", json={"source": "foto.png"})
|
||
assert r.status_code == 200 and r.headers["content-type"] == "image/png"
|
||
for _ in range(3):
|
||
r = _anteprima(client, "foto.png")
|
||
assert r.status_code == 200 and r.headers["content-type"] == "image/jpeg"
|
||
assert (r.headers["X-Larghezza"], r.headers["X-Altezza"]) == ("520", "400")
|
||
|
||
|
||
def test_grab_immagine_inesistente_404(client):
|
||
assert client.post("/camera/grab", json={"source": "non_c_e.png"}).status_code == 404
|
||
|
||
|
||
def test_calibrate_su_immagine_origine_e_il_nome(client):
|
||
_carica(client, "calibration_135_1.bmp", una_scacchiera())
|
||
r = client.post("/camera/calibrate", data={"source": "calibration_135_1.bmp", **CAMPI})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["origine"] == "calibration_135_1.bmp"
|
||
|
||
|
||
# ---- DELETE /camera/immagini
|
||
|
||
def test_elimina_204_poi_404_e_sparisce_dalle_sorgenti(client):
|
||
_carica(client, "foto.png", una_scacchiera())
|
||
assert client.delete("/camera/immagini", params={"nome": "foto.png"}).status_code == 204
|
||
assert not (client.config / "immagini" / "foto.png").exists()
|
||
assert client.delete("/camera/immagini", params={"nome": "foto.png"}).status_code == 404
|
||
assert "foto.png" not in [v["name"] for v in client.get("/camera/sources").json()["sources"]]
|
||
assert client.post("/camera/grab", json={"source": "foto.png"}).status_code == 404
|
||
|
||
|
||
def test_elimina_nome_con_percorso_404(client):
|
||
_carica(client, "foto.png", una_scacchiera())
|
||
assert client.delete("/camera/immagini", params={"nome": "../immagini/foto.png"}).status_code == 404
|
||
assert client.delete("/camera/immagini").status_code == 404
|
||
assert (client.config / "immagini" / "foto.png").is_file()
|
||
|
||
|
||
def test_elimina_chiude_la_sorgente_aperta(client, monkeypatch):
|
||
chiuse = []
|
||
modulo = _modulo(client)
|
||
reale = modulo["open_source"]
|
||
|
||
def spia(*a, **k):
|
||
s = reale(*a, **k)
|
||
chiusura = s.close
|
||
s.close = lambda: (chiuse.append(a[0]), chiusura())[1]
|
||
return s
|
||
|
||
monkeypatch.setitem(modulo, "open_source", spia)
|
||
_carica(client, "foto.png", una_scacchiera())
|
||
assert client.post("/camera/grab", json={"source": "foto.png"}).status_code == 200
|
||
assert client.delete("/camera/immagini", params={"nome": "foto.png"}).status_code == 204
|
||
assert chiuse == ["foto.png"]
|
||
|
||
|
||
# ---- POST /camera/anteprima/calibrata
|
||
|
||
def test_calibrata_senza_frame_404(client):
|
||
_carica(client, DENSA, _densa())
|
||
r = _calibrata(client, DENSA)
|
||
assert r.status_code == 404 and r.json()["detail"] == "nessun frame: avviare l'anteprima o scattare"
|
||
|
||
|
||
def test_calibrata_senza_attiva_409(client):
|
||
_carica(client, DENSA, _densa())
|
||
assert _anteprima(client, DENSA).status_code == 200
|
||
r = _calibrata(client, DENSA)
|
||
assert r.status_code == 409 and r.json()["detail"] == "nessuna calibrazione attiva su questa sorgente"
|
||
|
||
|
||
def test_calibrata_validazione_ingresso(client):
|
||
assert client.post("/camera/anteprima/calibrata", json={}).status_code == 400
|
||
assert _calibrata(client, DENSA, lato_max=100).status_code == 400
|
||
assert _calibrata(client, DENSA, lato_max=True).status_code == 400
|
||
|
||
|
||
def test_calibrata_200_densa_senza_nuova_lettura(client, monkeypatch):
|
||
modulo = _modulo(client)
|
||
reale = modulo["open_source"]
|
||
letture = []
|
||
|
||
def conta(*a, **k):
|
||
s = reale(*a, **k)
|
||
originale = s.grab
|
||
s.grab = lambda t=1000: (letture.append(1), originale(t))[1]
|
||
return s
|
||
|
||
monkeypatch.setitem(modulo, "open_source", conta)
|
||
_carica(client, DENSA, _densa())
|
||
_calibra_e_attiva(client, DENSA, "Banco àè 1")
|
||
assert _anteprima(client, DENSA).status_code == 200
|
||
prima = len(letture)
|
||
for _ in range(3):
|
||
r = _calibrata(client, DENSA)
|
||
assert r.status_code == 200, r.text
|
||
assert r.headers["content-type"] == "image/jpeg"
|
||
assert r.headers["X-Raddrizzato"] == "si"
|
||
assert (r.headers["X-Larghezza"], r.headers["X-Altezza"]) == ("640", "480")
|
||
assert unquote(r.headers["X-Calibrazione"]) == "Banco àè 1"
|
||
assert r.headers["X-Calibrazione"] == quote("Banco àè 1", safe="") # percent-encoded UTF-8
|
||
img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_UNCHANGED)
|
||
assert img.shape[:2] == (480, 640)
|
||
assert len(letture) == prima # la calibrata non legge mai un frame nuovo
|
||
|
||
|
||
def test_calibrata_ridotta_a_lato_max(client):
|
||
_carica(client, DENSA, _densa())
|
||
_calibra_e_attiva(client, DENSA, "Banco")
|
||
assert _anteprima(client, DENSA).status_code == 200
|
||
r = _calibrata(client, DENSA, lato_max=320)
|
||
assert r.status_code == 200
|
||
img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_UNCHANGED)
|
||
assert max(img.shape[:2]) == 320
|
||
assert (r.headers["X-Larghezza"], r.headers["X-Altezza"]) == ("640", "480") # misure del frame pieno
|
||
|
||
|
||
def test_anteprima_grezza_e_calibrata_usano_lo_stesso_frame_grezzo(client):
|
||
# anche con anteprima NON grezza la copia tenuta e' quella prima di raddrizzare
|
||
_carica(client, DENSA, _densa())
|
||
_calibra_e_attiva(client, DENSA, "Banco")
|
||
r = client.post("/camera/anteprima", json={"source": DENSA}) # calibrata: raddrizzata
|
||
assert r.status_code == 200 and r.headers["X-Raddrizzato"] == "si"
|
||
r = _calibrata(client, DENSA)
|
||
assert r.status_code == 200 and r.headers["X-Raddrizzato"] == "si" # raddrizzata una volta sola
|
||
|
||
|
||
def test_calibrata_risoluzione_diversa_409(client):
|
||
_carica(client, DENSA, _densa())
|
||
_calibra_e_attiva(client, DENSA, "Banco")
|
||
# stessa sorgente, immagine sostituita con un'altra risoluzione: la calibrazione resta attiva
|
||
assert _carica(client, DENSA, _densa(w=320, h=240)).status_code == 201
|
||
assert _calibrata(client, DENSA).status_code == 404 # sostituire l'immagine scorda il frame vecchio
|
||
assert _anteprima(client, DENSA).status_code == 200
|
||
r = _calibrata(client, DENSA)
|
||
assert r.status_code == 409
|
||
assert r.json()["detail"] == "la calibrazione «Banco» è per immagini 640×480, il frame è 320×240"
|
||
|
||
|
||
def test_calibrata_planare_senza_mappa_200_grezza(client):
|
||
# #95: senza mappa (planare) non e' piu' 409: esce il frame grezzo, X-Raddrizzato: no
|
||
_carica(client, PIANA, una_scacchiera())
|
||
_calibra_e_attiva(client, PIANA, "Piana", cols="9", rows="6")
|
||
assert _anteprima(client, PIANA).status_code == 200
|
||
r = _calibrata(client, PIANA)
|
||
assert r.status_code == 200
|
||
assert r.headers["X-Raddrizzato"] == "no" and unquote(r.headers["X-Calibrazione"]) == "Piana"
|