Acquisitore: anteprima, parametri camera e settaggi di default (#87, piano A)
Rotte nuove nel servizio camera: POST /camera/anteprima (JPEG ridotto con nitidezza, saturati e media sul frame pieno), GET/PUT /camera/parametri (esposizione e guadagno con i relativi auto), POST /camera/parametri/salva. I settaggi salvati vengono ricaricati alla prima apertura della camera. GET /acquisizione serve la stessa pagina di /calibrazione. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
This commit is contained in:
+167
-6
@@ -26,6 +26,23 @@ Calibrazione a scacchiera per sorgente (#83, logica in scripts/calibrazione.py):
|
||||
422 scacchiera non trovata o parametri invalidi
|
||||
DELETE /camera/calibration?source=<src> -> 204 (toglie JSON e mappa)
|
||||
GET /calibrazione -> pagina HTML autonoma (offline) per fare tutto a mano
|
||||
GET /acquisizione -> stessa pagina (#87)
|
||||
Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default:
|
||||
POST /camera/anteprima json {source, grezzo=false, lato_max=1280 (320..4096),
|
||||
timeout_ms=5000} -> JPEG ridotto (lato lungo <= lato_max, mai ingrandito).
|
||||
Header: X-Raddrizzato si|no, X-Larghezza/X-Altezza (frame PIENO),
|
||||
X-Nitidezza (varianza del Laplaciano), X-Saturati-Pct (% pixel >= 250),
|
||||
X-Media (livello medio 0-255), tutti sul frame pieno in grigio.
|
||||
GET /camera/parametri?source=<src> -> {source, regolabile, parametri:
|
||||
{esposizione, esposizione_auto, guadagno, guadagno_auto: dominio}};
|
||||
dominio = GenicamSource.dominio (valore, tipo, min/max/passo o scelte).
|
||||
Esposizione in us, guadagno in dB. Sorgente senza capacita': regolabile=false.
|
||||
PUT /camera/parametri json {source, valori: {nome: valore}} -> come GET.
|
||||
Scrive prima gli *_auto, poi esposizione/guadagno (assesta=3).
|
||||
400 nome ignoto o corpo errato, 409 non regolabile, 422 camera rifiuta
|
||||
POST /camera/parametri/salva json {source} -> {source, salvati, file}
|
||||
scrive <config-dir>/settaggi/<nome>-<sha1>.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
|
||||
%ProgramData%/Tielogic/TieMeasureFlow/acquisitore su Windows, ./acquisitore altrove).
|
||||
|
||||
@@ -49,6 +66,7 @@ camera a chiunque raggiunga la porta: solo su rete isolata.
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
@@ -61,7 +79,7 @@ from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile
|
||||
from fastapi.concurrency import run_in_threadpool # #83: il grab bloccante fuori dall'event loop
|
||||
from fastapi.responses import HTMLResponse, Response
|
||||
|
||||
from visionsuite.camera.camera_interface import CameraError, CameraGrabError
|
||||
from visionsuite.camera.camera_interface import CameraError, CameraGrabError, CameraStateError
|
||||
from visionsuite.camera.sources import SourceNotFound, list_sources, open_source
|
||||
|
||||
# #83: calibrazione e pagina stanno accanto a questo file (che non e' un pacchetto)
|
||||
@@ -70,6 +88,11 @@ import calibrazione # noqa: E402
|
||||
from calibrazione_pagina import PAGINA # noqa: E402
|
||||
|
||||
_TRIGGER = ("continuo",)
|
||||
_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")
|
||||
# #87: in scrittura prima gli automatismi (un valore manuale con auto acceso viene riscritto)
|
||||
_ORDINE_SCRITTURA = ("esposizione_auto", "guadagno_auto", "esposizione", "guadagno")
|
||||
|
||||
# errori noti di vs-camera (inglese) -> italiano; esatti, poi per prefisso
|
||||
_MESSAGGI = {
|
||||
@@ -140,17 +163,35 @@ def create_app(folder: Path | str | None = None,
|
||||
return {"sources": [{"name": v.name, "kind": v.kind, "label": v.label}
|
||||
for v in voci if v.kind in ("camera", "virtual")]}
|
||||
|
||||
def _grab(nome: str, timeout: int):
|
||||
"""Un frame dalla sorgente (apre e tiene aperta); errori camera -> HTTP."""
|
||||
try:
|
||||
with _lock_di(nome): # solo questa sorgente; le altre non aspettano
|
||||
def _apri(nome: str):
|
||||
"""#87: sorgente aperta (o da aprire); da chiamare SOLO sotto _lock_di(nome)."""
|
||||
with lock:
|
||||
sorgente = aperte.get(nome)
|
||||
if sorgente is None:
|
||||
sorgente = open_source(nome, None, config=_config())
|
||||
_carica_settaggi(nome, sorgente) # #87: predefiniti salvati, se ci sono
|
||||
with lock:
|
||||
aperte[nome] = sorgente
|
||||
return sorgente.grab(timeout)
|
||||
return sorgente
|
||||
|
||||
def _file_settaggi(nome: str) -> Path:
|
||||
# #87: stesso schema di nome della calibrazione, in una sottocartella
|
||||
return archivio.cartella / "settaggi" / archivio._file(nome).name
|
||||
|
||||
def _carica_settaggi(nome: str, sorgente) -> None:
|
||||
percorso = _file_settaggi(nome)
|
||||
if not percorso.is_file() or not hasattr(sorgente, "carica_settaggi"):
|
||||
return
|
||||
try:
|
||||
sorgente.carica_settaggi(percorso)
|
||||
except Exception as e: # noqa: BLE001 - un file rotto non blocca l'apertura
|
||||
_log.warning("settaggi di %s non caricati (%s): %s", nome, percorso, e)
|
||||
|
||||
def _con_sorgente(nome: str, fn):
|
||||
"""#87: fn(sorgente) sotto il lock della sorgente; errori camera -> HTTP."""
|
||||
try:
|
||||
with _lock_di(nome): # solo questa sorgente; le altre non aspettano
|
||||
return fn(_apri(nome))
|
||||
except SourceNotFound as e:
|
||||
raise HTTPException(404, _traduci(str(e)))
|
||||
except TimeoutError as e:
|
||||
@@ -158,6 +199,10 @@ def create_app(folder: Path | str | None = None,
|
||||
except (CameraGrabError, CameraError) as e:
|
||||
raise HTTPException(503, _traduci(str(e)))
|
||||
|
||||
def _grab(nome: str, timeout: int):
|
||||
"""Un frame dalla sorgente (apre e tiene aperta); errori camera -> HTTP."""
|
||||
return _con_sorgente(nome, lambda s: s.grab(timeout)) # #87: corpo in _con_sorgente
|
||||
|
||||
@app.post("/camera/grab")
|
||||
def camera_grab(corpo: dict = Body(default=None)):
|
||||
nome = corpo.get("source") if isinstance(corpo, dict) else None
|
||||
@@ -241,7 +286,123 @@ def create_app(folder: Path | str | None = None,
|
||||
archivio.salva(nome, corpo, mappa) # #84: la mappa (se densa) va in .npz accanto
|
||||
return {**corpo, "overlay_png": risultato["overlay_png"], "saved": bool(save)}
|
||||
|
||||
# ---- #87: acquisizione (anteprima, parametri, settaggi)
|
||||
|
||||
def _regolabile(s) -> bool:
|
||||
# #87: ha il vocabolario delle capacita' e ne supporta almeno una delle quattro
|
||||
if not (hasattr(s, "supporta") and hasattr(s, "dominio")):
|
||||
return False
|
||||
return any(s.supporta(n) for n in _PARAMETRI)
|
||||
|
||||
def _parametri(nome: str, s) -> dict:
|
||||
"""#87: forma di GET /camera/parametri; da chiamare sotto lock."""
|
||||
parametri: dict = {}
|
||||
if _regolabile(s):
|
||||
for n in _PARAMETRI:
|
||||
if not s.supporta(n):
|
||||
continue
|
||||
try:
|
||||
parametri[n] = s.dominio(n)
|
||||
except CameraStateError as e: # capacita' dichiarata ma non leggibile: si salta
|
||||
_log.warning("dominio di %s su %s non letto: %s", n, nome, e)
|
||||
return {"source": nome, "regolabile": bool(parametri), "parametri": parametri}
|
||||
|
||||
def _grigio8(frame):
|
||||
g = frame if frame.ndim == 2 else cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
if g.dtype == np.uint16: # #87: 16 bit -> 8 bit, per soglia e media coerenti
|
||||
g = (g >> 8).astype(np.uint8)
|
||||
elif g.dtype != np.uint8:
|
||||
g = cv2.convertScaleAbs(g, alpha=255.0 / max(float(g.max()), 1.0))
|
||||
return g
|
||||
|
||||
@app.post("/camera/anteprima")
|
||||
def camera_anteprima(corpo: dict = Body(default=None)):
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
nome = _sorgente(corpo.get("source"))
|
||||
grezzo = corpo.get("grezzo", False)
|
||||
if not isinstance(grezzo, bool):
|
||||
raise HTTPException(400, "«grezzo» deve essere true o false")
|
||||
lato = corpo.get("lato_max", 1280)
|
||||
if isinstance(lato, bool) or not isinstance(lato, int) or not 320 <= lato <= 4096:
|
||||
raise HTTPException(400, "«lato_max» deve essere un intero fra 320 e 4096")
|
||||
timeout = corpo.get("timeout_ms", 5000)
|
||||
if isinstance(timeout, bool) or not isinstance(timeout, int) or not 1 <= timeout <= 10000:
|
||||
raise HTTPException(400, "«timeout_ms» deve essere un intero fra 1 e 10000")
|
||||
frame = _grab(nome, timeout)
|
||||
raddrizzato = False
|
||||
if not grezzo: # come /camera/grab
|
||||
mappa = archivio.leggi_mappa(nome)
|
||||
if mappa is not None:
|
||||
corretto = calibrazione.raddrizza(frame, mappa)
|
||||
if corretto is not None:
|
||||
frame, raddrizzato = corretto, True
|
||||
alt, larg = frame.shape[:2]
|
||||
grigio = _grigio8(frame) # misure sul frame PIENO, non sul ridotto
|
||||
nitidezza = float(cv2.Laplacian(grigio, cv2.CV_64F).var())
|
||||
saturati = float((grigio >= 250).mean() * 100.0)
|
||||
media = float(grigio.mean())
|
||||
if frame.dtype != np.uint8: # il JPEG vuole 8 bit
|
||||
frame = grigio
|
||||
scala = lato / max(alt, larg)
|
||||
if scala < 1.0: # mai ingrandita
|
||||
frame = cv2.resize(frame, (max(1, round(larg * scala)), max(1, round(alt * scala))),
|
||||
interpolation=cv2.INTER_AREA)
|
||||
ok, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
|
||||
if not ok:
|
||||
raise HTTPException(500, "codifica JPEG fallita")
|
||||
return Response(buf.tobytes(), media_type="image/jpeg", headers={
|
||||
"X-Raddrizzato": "si" if raddrizzato else "no",
|
||||
"X-Larghezza": str(larg), "X-Altezza": str(alt),
|
||||
"X-Nitidezza": f"{nitidezza:.1f}", "X-Saturati-Pct": f"{saturati:.2f}",
|
||||
"X-Media": f"{media:.1f}"})
|
||||
|
||||
@app.get("/camera/parametri")
|
||||
def parametri_leggi(source: str = ""):
|
||||
nome = _sorgente(source)
|
||||
return _con_sorgente(nome, lambda s: _parametri(nome, s))
|
||||
|
||||
@app.put("/camera/parametri")
|
||||
def parametri_scrivi(corpo: dict = Body(default=None)):
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
nome = _sorgente(corpo.get("source"))
|
||||
valori = corpo.get("valori")
|
||||
if not isinstance(valori, dict) or not valori:
|
||||
raise HTTPException(400, "manca «valori»")
|
||||
for n, v in valori.items():
|
||||
if n not in _PARAMETRI:
|
||||
raise HTTPException(400, f"parametro non riconosciuto: «{n}»")
|
||||
if isinstance(v, bool) or not isinstance(v, (int, float, str)):
|
||||
raise HTTPException(400, f"valore non valido per «{n}»")
|
||||
|
||||
def scrivi(s):
|
||||
if not _regolabile(s):
|
||||
raise HTTPException(409, "questa sorgente non ha parametri regolabili")
|
||||
for n in _ORDINE_SCRITTURA: # prima gli *_auto
|
||||
if n in valori:
|
||||
try:
|
||||
s.scrivi(n, valori[n], assesta=3)
|
||||
except CameraStateError as e:
|
||||
raise HTTPException(422, _traduci(str(e)))
|
||||
return _parametri(nome, s) # valori riletti dalla camera
|
||||
|
||||
return _con_sorgente(nome, scrivi)
|
||||
|
||||
@app.post("/camera/parametri/salva")
|
||||
def parametri_salva(corpo: dict = Body(default=None)):
|
||||
nome = _sorgente(corpo.get("source") if isinstance(corpo, dict) else None)
|
||||
|
||||
def salva(s):
|
||||
if not _regolabile(s) or not hasattr(s, "salva_settaggi"):
|
||||
raise HTTPException(409, "questa sorgente non ha parametri regolabili")
|
||||
percorso = _file_settaggi(nome)
|
||||
percorso.parent.mkdir(parents=True, exist_ok=True)
|
||||
s.salva_settaggi(percorso)
|
||||
return {"source": nome, "salvati": True, "file": percorso.name}
|
||||
|
||||
return _con_sorgente(nome, salva)
|
||||
|
||||
@app.get("/calibrazione", response_class=HTMLResponse)
|
||||
@app.get("/acquisizione", response_class=HTMLResponse) # #87: stessa pagina
|
||||
def calibrazione_pagina():
|
||||
return HTMLResponse(PAGINA)
|
||||
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
"""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
|
||||
|
||||
|
||||
# ---- 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
|
||||
Reference in New Issue
Block a user