acquisitore: immagini caricate come sorgenti e anteprima calibrata (Refs #92)

POST/DELETE /camera/immagini (cartella <config>/immagini), elenco sorgenti con le immagini e senza camera:file, apertura per nome di file con loop, frame grezzo tenuto dall'anteprima e POST /camera/anteprima/calibrata che raddrizza senza leggere un frame nuovo. 22 test nuovi.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-24 16:26:30 +02:00
parent 8eb80e20a8
commit 632fd51050
2 changed files with 372 additions and 5 deletions
+263
View File
@@ -0,0 +1,263 @@
"""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_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_409(client):
_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 == 409
assert r.json()["detail"] == "la calibrazione «Piana» non ha una mappa di correzione (planare)"