321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""Test delle rotte di acquisizione di scripts/camera_service.py (#87).
|
|
|
|
Uso: uv run --python 3.13 --extra vision-worker --extra dev \
|
|
--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q
|
|
Copre: POST /camera/anteprima, GET/PUT /camera/parametri, POST /camera/parametri/salva,
|
|
caricamento dei settaggi alla riapertura, alias GET /acquisizione.
|
|
Sorgente regolabile finta (iniettata su open_source) e FolderCamera su PNG temporanei.
|
|
"""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from visionsuite.camera.camera_interface import CameraStateError
|
|
|
|
MODULO = Path(__file__).resolve().parents[1] / "camera_service.py"
|
|
SORGENTE = "camera:finta"
|
|
|
|
|
|
def _carica():
|
|
spec = importlib.util.spec_from_file_location("camera_service_acq", MODULO)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
class Finta:
|
|
"""Camera regolabile finta: stesso vocabolario di GenicamSource."""
|
|
|
|
DOMINI = {
|
|
"esposizione": {"tipo": "numero", "min": 10.0, "max": 100000.0, "passo": 1.0},
|
|
"esposizione_auto": {"tipo": "scelta", "scelte": ["Off", "Once", "Continuous"]},
|
|
"guadagno": {"tipo": "numero", "min": 0.0, "max": 24.0},
|
|
"guadagno_auto": {"tipo": "scelta", "scelte": ["Off", "Once", "Continuous"]},
|
|
}
|
|
|
|
def __init__(self, supportati=None, rifiuta=()):
|
|
self.valori = {"esposizione": 5000.0, "esposizione_auto": "Continuous",
|
|
"guadagno": 1.0, "guadagno_auto": "Off"}
|
|
self.supportati = set(supportati if supportati is not None else self.DOMINI)
|
|
self.rifiuta = set(rifiuta)
|
|
self.scritture = []
|
|
self.caricato = None
|
|
self.chiusa = False
|
|
# metà bianca (255) e metà a 100: saturati 50%, media 177,5
|
|
frame = np.full((1000, 2000, 3), 100, np.uint8)
|
|
frame[:, :1000] = 255
|
|
self.frame = frame
|
|
|
|
def supporta(self, nome):
|
|
return nome in self.supportati
|
|
|
|
def dominio(self, nome):
|
|
return {"capacita": nome, "valore": self.valori[nome], **self.DOMINI[nome]}
|
|
|
|
def leggi(self, nome):
|
|
return self.valori[nome]
|
|
|
|
def scrivi(self, nome, valore, assesta=0):
|
|
if nome in self.rifiuta:
|
|
raise CameraStateError(f"non posso impostare {nome}: nodo in sola lettura")
|
|
self.scritture.append((nome, valore, assesta))
|
|
self.valori[nome] = valore
|
|
|
|
def salva_settaggi(self, percorso):
|
|
Path(percorso).write_text(json.dumps(self.valori), encoding="utf-8")
|
|
return dict(self.valori)
|
|
|
|
def carica_settaggi(self, percorso, forza=False):
|
|
self.caricato = Path(percorso)
|
|
self.valori.update(json.loads(self.caricato.read_text(encoding="utf-8")))
|
|
return []
|
|
|
|
def grab(self, timeout=0):
|
|
return self.frame.copy()
|
|
|
|
def close(self):
|
|
self.chiusa = True
|
|
|
|
|
|
@pytest.fixture
|
|
def cartella(tmp_path):
|
|
img = tmp_path / "foto"
|
|
img.mkdir()
|
|
for nome in ("a.png", "b.png"):
|
|
cv2.imwrite(str(img / nome), np.full((48, 64, 3), 128, np.uint8))
|
|
return img
|
|
|
|
|
|
def _client(cartella, config, monkeypatch, finta):
|
|
mod = _carica()
|
|
monkeypatch.setattr(mod, "open_source", lambda *a, **k: finta)
|
|
c = TestClient(mod.create_app(cartella, config))
|
|
c.mod = mod
|
|
return c
|
|
|
|
|
|
@pytest.fixture
|
|
def finta():
|
|
return Finta()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(cartella, tmp_path, monkeypatch, finta):
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, finta) as c:
|
|
c.finta = finta
|
|
c.config = tmp_path / "cfg"
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def client_cartella(cartella, tmp_path):
|
|
mod = _carica()
|
|
with TestClient(mod.create_app(cartella, tmp_path / "cfg")) as c:
|
|
yield c
|
|
|
|
|
|
# ---- anteprima
|
|
|
|
def test_anteprima_jpeg_ridotta_e_header(client):
|
|
r = client.post("/camera/anteprima", json={"source": SORGENTE, "lato_max": 500})
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"] == "image/jpeg"
|
|
img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_COLOR)
|
|
assert img.shape[:2] == (250, 500) # lato lungo 2000 -> 500
|
|
assert r.headers["x-raddrizzato"] == "no"
|
|
assert (r.headers["x-larghezza"], r.headers["x-altezza"]) == ("2000", "1000") # frame pieno
|
|
assert r.headers["x-saturati-pct"] == "50.00"
|
|
assert r.headers["x-media"] == "177.5"
|
|
assert float(r.headers["x-nitidezza"]) > 0 # bordo netto fra le due metà
|
|
|
|
|
|
def test_anteprima_mai_ingrandita(client_cartella):
|
|
r = client_cartella.post("/camera/anteprima", json={"source": "camera:folder"})
|
|
assert r.status_code == 200
|
|
img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_COLOR)
|
|
assert img.shape[:2] == (48, 64) # lato_max 1280 > 64: resta com'è
|
|
assert r.headers["x-nitidezza"] == "0.0" # grigio uniforme
|
|
assert r.headers["x-saturati-pct"] == "0.00"
|
|
assert r.headers["x-media"] == "128.0"
|
|
|
|
|
|
def test_anteprima_parametri_errati_400(client):
|
|
for corpo in ({}, {"source": SORGENTE, "lato_max": 100}, {"source": SORGENTE, "lato_max": 5000},
|
|
{"source": SORGENTE, "timeout_ms": 0}, {"source": SORGENTE, "grezzo": "si"}):
|
|
assert client.post("/camera/anteprima", json=corpo).status_code == 400, corpo
|
|
|
|
|
|
def test_anteprima_sorgente_inesistente_404(client_cartella):
|
|
assert client_cartella.post("/camera/anteprima",
|
|
json={"source": "camera:non-esiste"}).status_code == 404
|
|
|
|
|
|
def test_anteprima_errore_camera_503_e_timeout_504(client):
|
|
from visionsuite.camera.camera_interface import CameraGrabError
|
|
|
|
def rotta(timeout=0):
|
|
raise CameraGrabError("Camera is not open")
|
|
|
|
client.finta.grab = rotta
|
|
r = client.post("/camera/anteprima", json={"source": SORGENTE})
|
|
assert r.status_code == 503 and r.json()["detail"] == "La camera non è aperta"
|
|
|
|
def lenta(timeout=0):
|
|
raise TimeoutError("scaduto")
|
|
|
|
client.finta.grab = lenta
|
|
assert client.post("/camera/anteprima", json={"source": SORGENTE}).status_code == 504
|
|
|
|
|
|
# ---- parametri
|
|
|
|
def test_parametri_regolabile(client):
|
|
r = client.get("/camera/parametri", params={"source": SORGENTE})
|
|
assert r.status_code == 200
|
|
j = r.json()
|
|
assert j["source"] == SORGENTE and j["regolabile"] is True
|
|
assert list(j["parametri"]) == ["esposizione", "esposizione_auto", "guadagno", "guadagno_auto"]
|
|
assert j["parametri"]["esposizione"]["valore"] == 5000.0
|
|
assert j["parametri"]["esposizione_auto"]["scelte"] == ["Off", "Once", "Continuous"]
|
|
|
|
|
|
def test_parametri_solo_quelli_supportati(cartella, tmp_path, monkeypatch):
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, Finta(supportati=["esposizione"])) as c:
|
|
j = c.get("/camera/parametri", params={"source": SORGENTE}).json()
|
|
assert j["regolabile"] is True and list(j["parametri"]) == ["esposizione"]
|
|
|
|
|
|
def test_parametri_sorgente_non_regolabile(client_cartella):
|
|
j = client_cartella.get("/camera/parametri", params={"source": "camera:folder"}).json()
|
|
assert j == {"source": "camera:folder", "regolabile": False, "parametri": {}}
|
|
|
|
|
|
def test_parametri_senza_source_400(client):
|
|
assert client.get("/camera/parametri").status_code == 400
|
|
|
|
|
|
def test_put_scrive_prima_gli_auto(client):
|
|
r = client.put("/camera/parametri", json={"source": SORGENTE, "valori": {
|
|
"esposizione": 8000, "guadagno": 3.5, "esposizione_auto": "Off", "guadagno_auto": "Off"}})
|
|
assert r.status_code == 200
|
|
assert client.finta.scritture == [
|
|
("esposizione_auto", "Off", 3), ("guadagno_auto", "Off", 3),
|
|
("esposizione", 8000, 3), ("guadagno", 3.5, 3)]
|
|
j = r.json() # valori riletti dalla camera
|
|
assert j["parametri"]["esposizione"]["valore"] == 8000
|
|
assert j["parametri"]["esposizione_auto"]["valore"] == "Off"
|
|
|
|
|
|
def test_put_nome_ignoto_400(client):
|
|
r = client.put("/camera/parametri", json={"source": SORGENTE, "valori": {"frame_rate": 10}})
|
|
assert r.status_code == 400
|
|
assert client.finta.scritture == []
|
|
for corpo in ({"source": SORGENTE}, {"source": SORGENTE, "valori": {}},
|
|
{"source": SORGENTE, "valori": {"guadagno": True}}):
|
|
assert client.put("/camera/parametri", json=corpo).status_code == 400, corpo
|
|
|
|
|
|
def test_put_non_regolabile_409(client_cartella):
|
|
r = client_cartella.put("/camera/parametri",
|
|
json={"source": "camera:folder", "valori": {"esposizione": 1000}})
|
|
assert r.status_code == 409
|
|
assert r.json()["detail"] == "questa sorgente non ha parametri regolabili"
|
|
|
|
|
|
def test_put_rifiuto_della_camera_422(cartella, tmp_path, monkeypatch):
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, Finta(rifiuta=["guadagno"])) as c:
|
|
r = c.put("/camera/parametri", json={"source": SORGENTE, "valori": {"guadagno": 5}})
|
|
assert r.status_code == 422
|
|
assert "guadagno" in r.json()["detail"]
|
|
|
|
|
|
# ---- salvataggio e riapertura
|
|
|
|
def test_salva_crea_il_file(client):
|
|
r = client.post("/camera/parametri/salva", json={"source": SORGENTE})
|
|
assert r.status_code == 200
|
|
j = r.json()
|
|
assert j["source"] == SORGENTE and j["salvati"] is True
|
|
atteso = client.config / "settaggi" / j["file"]
|
|
assert atteso.is_file()
|
|
assert j["file"].startswith("camera_finta-") and j["file"].endswith(".json")
|
|
assert json.loads(atteso.read_text(encoding="utf-8"))["esposizione"] == 5000.0
|
|
|
|
|
|
def test_salva_non_regolabile_409(client_cartella):
|
|
r = client_cartella.post("/camera/parametri/salva", json={"source": "camera:folder"})
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_riapertura_carica_i_settaggi(cartella, tmp_path, monkeypatch):
|
|
config = tmp_path / "cfg"
|
|
prima = Finta()
|
|
with _client(cartella, config, monkeypatch, prima) as c:
|
|
c.put("/camera/parametri", json={"source": SORGENTE, "valori": {"esposizione": 7777}})
|
|
file = c.post("/camera/parametri/salva", json={"source": SORGENTE}).json()["file"]
|
|
seconda = Finta() # servizio riavviato: camera nuova, valori di fabbrica
|
|
with _client(cartella, config, monkeypatch, seconda) as c:
|
|
assert seconda.caricato is None # niente apertura finché nessuno la chiede
|
|
c.get("/camera/parametri", params={"source": SORGENTE})
|
|
assert seconda.caricato == config / "settaggi" / file
|
|
assert seconda.valori["esposizione"] == 7777
|
|
|
|
|
|
def test_settaggi_rotti_non_bloccano_l_apertura(cartella, tmp_path, monkeypatch):
|
|
class Rotta(Finta):
|
|
def carica_settaggi(self, percorso, forza=False):
|
|
raise ValueError("file illeggibile")
|
|
|
|
config = tmp_path / "cfg"
|
|
with _client(cartella, config, monkeypatch, Finta()) as c:
|
|
file = c.post("/camera/parametri/salva", json={"source": SORGENTE}).json()["file"]
|
|
assert (config / "settaggi" / file).is_file()
|
|
with _client(cartella, config, monkeypatch, Rotta()) as c:
|
|
assert c.get("/camera/parametri", params={"source": SORGENTE}).status_code == 200
|
|
|
|
|
|
def test_apertura_chiede_il_frame_piu_recente(cartella, tmp_path, monkeypatch):
|
|
# #89: solo_ultimo_frame() una volta sola all'apertura, non a ogni richiesta
|
|
class ConUltimoFrame(Finta):
|
|
chiamate = 0
|
|
|
|
def solo_ultimo_frame(self, attivo=True):
|
|
self.chiamate += 1
|
|
|
|
finta = ConUltimoFrame()
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, finta) as c:
|
|
for _ in range(3):
|
|
assert c.get("/camera/parametri", params={"source": SORGENTE}).status_code == 200
|
|
assert finta.chiamate == 1
|
|
|
|
|
|
def test_frame_piu_recente_non_impostabile_non_blocca_l_apertura(cartella, tmp_path, monkeypatch, caplog):
|
|
# #89: producer senza il nodo -> avviso nel log, apertura riuscita
|
|
class SenzaNodo(Finta):
|
|
def solo_ultimo_frame(self, attivo=True):
|
|
raise CameraStateError("questo producer non espone StreamBufferHandlingMode")
|
|
|
|
with caplog.at_level("WARNING"):
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, SenzaNodo()) as c:
|
|
assert c.get("/camera/parametri", params={"source": SORGENTE}).status_code == 200
|
|
assert "frame piu' recente non impostabile" in caplog.text
|
|
|
|
|
|
def test_sorgente_senza_solo_ultimo_frame_si_apre(cartella, tmp_path, monkeypatch):
|
|
# #89: FolderCamera e simili non hanno il metodo: nessun errore
|
|
assert not hasattr(Finta(), "solo_ultimo_frame")
|
|
with _client(cartella, tmp_path / "cfg", monkeypatch, Finta()) as c:
|
|
assert c.get("/camera/parametri", params={"source": SORGENTE}).status_code == 200
|
|
|
|
|
|
# ---- pagina
|
|
|
|
def test_acquisizione_alias_della_pagina(client):
|
|
a = client.get("/acquisizione")
|
|
c = client.get("/calibrazione")
|
|
assert a.status_code == 200 and a.text == c.text
|