Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 25f8b4d609 | |||
| 552f96bb47 | |||
| 71daab184f | |||
| 9beaf2ed67 | |||
| 800ec41254 | |||
| 9c45a51770 | |||
| 4a48af2ae3 | |||
| b879563738 | |||
| f504c86491 | |||
| 632fd51050 | |||
| 8eb80e20a8 | |||
| b2f82fe779 | |||
| 3c5c5f9bc5 | |||
| a5e8472e88 | |||
| e7d5489438 |
+249
-34
@@ -3,6 +3,9 @@
|
||||
Cosa fa: da una foto di scacchiera piana calcola la scala (px/mm) con
|
||||
TieVision-Core (visionsuite.calibration), e conserva un risultato per sorgente
|
||||
camera in un JSON (scrittura atomica).
|
||||
#91: le calibrazioni stanno in un archivio UNICO, salvate con nome (classe Archivio);
|
||||
una sorgente ne ha una attiva (attive.json) e il formato vecchio (una per sorgente)
|
||||
si migra da solo.
|
||||
#84: `calibra_scacchiera_densa(pixels, square_mm)` calibra da UNA foto di
|
||||
scacchiera che riempie tutto il campo (ottica telecentrica): trova da sola la
|
||||
griglia, costruisce la mappa di distorsione (DistortionMap) e il mm/px
|
||||
@@ -24,6 +27,7 @@ import os
|
||||
import re
|
||||
import tempfile
|
||||
import threading
|
||||
import unicodedata # #91 caratteri di controllo nei nomi
|
||||
from collections import deque
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
@@ -463,43 +467,87 @@ def ora_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||||
|
||||
|
||||
class Archivio:
|
||||
"""Un JSON per sorgente in `cartella`; nome file dalla sorgente ripulito.
|
||||
class NomeNonValido(ValueError):
|
||||
"""#91: nome di calibrazione vuoto, troppo lungo o con caratteri di controllo (400)."""
|
||||
|
||||
#84: se la calibrazione e' densa, accanto al JSON sta la mappa (stesso nome,
|
||||
.npz); `leggi_mappa` la tiene in memoria finche' il file non cambia.
|
||||
|
||||
class NomeGiaUsato(ValueError):
|
||||
"""#91: esiste gia' una calibrazione con questo nome (senza distinguere maiuscole) (409)."""
|
||||
|
||||
|
||||
class NomeSconosciuto(KeyError):
|
||||
"""#91: nessuna calibrazione con questo nome (404)."""
|
||||
|
||||
|
||||
def pulisci_nome(nome) -> str:
|
||||
"""#91: nome valido = strip, 1-60 caratteri, niente caratteri di controllo."""
|
||||
if not isinstance(nome, str):
|
||||
raise NomeNonValido("il nome della calibrazione deve essere un testo")
|
||||
nome = nome.strip()
|
||||
if not nome:
|
||||
raise NomeNonValido("il nome della calibrazione è vuoto")
|
||||
if len(nome) > 60:
|
||||
raise NomeNonValido("il nome della calibrazione supera i 60 caratteri")
|
||||
if any(unicodedata.category(c) == "Cc" for c in nome):
|
||||
raise NomeNonValido("il nome della calibrazione contiene caratteri di controllo")
|
||||
return nome
|
||||
|
||||
|
||||
# #91: campi che l'elenco espone per ogni calibrazione (il resto resta nel JSON)
|
||||
_CAMPI_ELENCO = ("nome", "origine", "image_width", "image_height", "px_per_mm",
|
||||
"mm_per_px", "metodo", "square_mm", "calibrated_at")
|
||||
|
||||
|
||||
class Archivio:
|
||||
"""#91: archivio UNICO di calibrazioni con nome, in `cartella`.
|
||||
|
||||
Libreria: <cartella>/calibrazioni/<nome ripulito>-<sha1 del nome>.json (+ .npz se
|
||||
densa, stesso nome). Attive: <cartella>/attive.json = {sorgente: nome}. Una
|
||||
calibrazione non appartiene a una sorgente: la sorgente ne sceglie una (`attiva`).
|
||||
#84: `leggi_mappa` tiene la mappa in memoria finche' il file non cambia.
|
||||
Il formato di prima (un JSON per sorgente, campo "source", in <cartella>) si
|
||||
migra da solo al costruttore (`migra`, idempotente).
|
||||
"""
|
||||
|
||||
def __init__(self, cartella: Path | str):
|
||||
self.cartella = Path(cartella)
|
||||
self._cache: dict = {} # percorso .npz -> (mtime_ns, MappaPronta)
|
||||
self._lock = threading.Lock()
|
||||
self._lock_dati = threading.RLock() # #91: libreria e attive.json (leggi-modifica-scrivi)
|
||||
self.migra()
|
||||
|
||||
# ---- percorsi
|
||||
|
||||
def _file(self, sorgente: str) -> Path:
|
||||
# caratteri sicuri + hash dell'originale: due sorgenti diverse non si pestano
|
||||
# (#91: oggi serve ai settaggi camera e al formato vecchio da migrare)
|
||||
pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", sorgente).strip("._") or "sorgente"
|
||||
sigla = hashlib.sha1(sorgente.encode("utf-8")).hexdigest()[:8]
|
||||
return self.cartella / f"{pulito[:60]}-{sigla}.json"
|
||||
|
||||
def _file_mappa(self, sorgente: str) -> Path:
|
||||
return self._file(sorgente).with_suffix(".npz") # #84 stesso nome del JSON
|
||||
@property
|
||||
def _libreria(self) -> Path:
|
||||
return self.cartella / "calibrazioni"
|
||||
|
||||
def leggi(self, sorgente: str) -> dict | None:
|
||||
try:
|
||||
return json.loads(self._file(sorgente).read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None # assente o illeggibile = non calibrata
|
||||
def _file_nome(self, nome: str) -> Path:
|
||||
# #91: stesso schema di _file, ma l'hash e' sul nome senza maiuscole/minuscole
|
||||
pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", nome).strip("._") or "calibrazione"
|
||||
sigla = hashlib.sha1(nome.casefold().encode("utf-8")).hexdigest()[:8]
|
||||
return self._libreria / f"{pulito[:60]}-{sigla}.json"
|
||||
|
||||
def salva(self, sorgente: str, dati: dict, mappa: DistortionMap | None = None) -> None:
|
||||
destinazione = self._file(sorgente)
|
||||
self.cartella.mkdir(parents=True, exist_ok=True)
|
||||
# #84: prima la mappa (o la rimozione di una vecchia), poi il JSON che la dichiara
|
||||
if mappa is not None:
|
||||
self._salva_mappa(sorgente, mappa)
|
||||
else:
|
||||
self._togli_mappa(sorgente) # ricalibrazione planare: la vecchia mappa non vale piu'
|
||||
def _file_mappa_nome(self, nome: str) -> Path:
|
||||
return self._file_nome(nome).with_suffix(".npz") # #84 stesso nome del JSON
|
||||
|
||||
@property
|
||||
def _file_attive(self) -> Path:
|
||||
return self.cartella / "attive.json"
|
||||
|
||||
# ---- scrittura atomica
|
||||
|
||||
def _scrivi_json(self, destinazione: Path, dati: dict) -> None:
|
||||
destinazione.parent.mkdir(parents=True, exist_ok=True)
|
||||
# scrittura atomica: file temporaneo nella stessa cartella, poi replace
|
||||
fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp")
|
||||
fd, tmp = tempfile.mkstemp(dir=destinazione.parent, suffix=".tmp")
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
json.dump(dati, f, ensure_ascii=False, indent=2)
|
||||
@@ -511,9 +559,128 @@ class Archivio:
|
||||
pass
|
||||
raise
|
||||
|
||||
def _salva_mappa(self, sorgente: str, mappa: DistortionMap) -> None:
|
||||
destinazione = self._file_mappa(sorgente)
|
||||
fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp.npz") # .npz: numpy non aggiunge nulla
|
||||
def _leggi_json(self, percorso: Path) -> dict | None:
|
||||
try:
|
||||
dati = json.loads(percorso.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None # assente o illeggibile
|
||||
return dati if isinstance(dati, dict) else None
|
||||
|
||||
# ---- attive
|
||||
|
||||
def _leggi_attive(self) -> dict:
|
||||
dati = self._leggi_json(self._file_attive) or {}
|
||||
return {k: v for k, v in dati.items() if isinstance(k, str) and isinstance(v, str)}
|
||||
|
||||
def _scrivi_attive(self, attive: dict) -> None:
|
||||
self._scrivi_json(self._file_attive, attive)
|
||||
|
||||
# ---- migrazione dal formato vecchio
|
||||
|
||||
def migra(self) -> None:
|
||||
"""#91: ogni <cartella>/*.json con "source" (una per sorgente) entra in libreria.
|
||||
|
||||
Nome = valore di "source" (gia' preso: "<source> (2)", "(3)"...); il .npz
|
||||
accanto segue; la sorgente resta calibrata con quella voce (attive.json).
|
||||
Idempotente: il vecchio file sparisce, un secondo giro non trova nulla.
|
||||
"""
|
||||
if not self.cartella.is_dir():
|
||||
return
|
||||
with self._lock_dati:
|
||||
for vecchio in sorted(self.cartella.glob("*.json")):
|
||||
if vecchio.name == self._file_attive.name:
|
||||
continue
|
||||
dati = self._leggi_json(vecchio)
|
||||
sorgente = dati.get("source") if dati else None
|
||||
if not isinstance(sorgente, str) or not sorgente.strip():
|
||||
continue
|
||||
try:
|
||||
self._migra_uno(vecchio, dati, sorgente)
|
||||
except OSError:
|
||||
continue # un file bloccato non ferma il servizio: riprova al prossimo avvio
|
||||
|
||||
def _migra_uno(self, vecchio: Path, dati: dict, sorgente: str) -> None:
|
||||
base = "".join(c for c in sorgente if unicodedata.category(c) != "Cc").strip()[:60].strip() or "calibrazione"
|
||||
nome, n = base, 1
|
||||
while True:
|
||||
esistente = self.leggi_nome(nome)
|
||||
if esistente is None:
|
||||
break
|
||||
if (esistente.get("calibrated_at") == dati.get("calibrated_at")
|
||||
and esistente.get("px_per_mm") == dati.get("px_per_mm")):
|
||||
break # gia' copiata da un giro interrotto: si completa e basta
|
||||
n += 1
|
||||
suffisso = f" ({n})"
|
||||
nome = base[:60 - len(suffisso)].rstrip() + suffisso
|
||||
nuovo = {k: v for k, v in dati.items() if k != "source"}
|
||||
nuovo["nome"] = nome
|
||||
nuovo.setdefault("origine", sorgente) # l'origine vera non era registrata
|
||||
vecchia_mappa = vecchio.with_suffix(".npz")
|
||||
if vecchia_mappa.is_file(): # prima la mappa, poi il JSON che la dichiara
|
||||
self._libreria.mkdir(parents=True, exist_ok=True)
|
||||
os.replace(vecchia_mappa, self._file_mappa_nome(nome))
|
||||
self._scrivi_json(self._file_nome(nome), nuovo)
|
||||
attive = self._leggi_attive()
|
||||
attive[sorgente.strip()] = nome
|
||||
self._scrivi_attive(attive)
|
||||
vecchio.unlink()
|
||||
|
||||
# ---- libreria
|
||||
|
||||
def leggi_nome(self, nome: str) -> dict | None:
|
||||
"""#91: la calibrazione con questo nome (senza distinguere maiuscole), o None."""
|
||||
try:
|
||||
nome = pulisci_nome(nome)
|
||||
except NomeNonValido:
|
||||
return None
|
||||
return self._leggi_json(self._file_nome(nome))
|
||||
|
||||
def _meta(self, dati: dict, attive: dict) -> dict:
|
||||
meta = {k: dati.get(k) for k in _CAMPI_ELENCO}
|
||||
chiave = str(dati.get("nome", "")).casefold()
|
||||
meta["attiva_su"] = sorted(s for s, n in attive.items() if n.casefold() == chiave)
|
||||
return meta
|
||||
|
||||
def elenco(self) -> list[dict]:
|
||||
"""#91: metadati di tutte le calibrazioni, dalla piu' recente."""
|
||||
with self._lock_dati:
|
||||
attive = self._leggi_attive()
|
||||
voci = []
|
||||
for percorso in self._libreria.glob("*.json") if self._libreria.is_dir() else []:
|
||||
dati = self._leggi_json(percorso)
|
||||
if dati and isinstance(dati.get("nome"), str):
|
||||
voci.append(self._meta(dati, attive))
|
||||
voci.sort(key=lambda m: str(m.get("calibrated_at") or ""), reverse=True)
|
||||
return voci
|
||||
|
||||
def metadati(self, nome: str) -> dict | None:
|
||||
"""#91: i metadati di una voce (come in `elenco`), o None."""
|
||||
dati = self.leggi_nome(nome)
|
||||
return self._meta(dati, self._leggi_attive()) if dati else None
|
||||
|
||||
def salva_nome(self, nome: str, dati: dict, mappa: DistortionMap | None = None,
|
||||
sovrascrivi: bool = False) -> dict:
|
||||
"""#91: salva `dati` (e la mappa, se densa) con questo nome; ritorna i metadati."""
|
||||
nome = pulisci_nome(nome)
|
||||
destinazione = self._file_nome(nome)
|
||||
with self._lock_dati:
|
||||
if destinazione.exists() and not sovrascrivi:
|
||||
esistente = self._leggi_json(destinazione) or {}
|
||||
raise NomeGiaUsato(f"esiste già una calibrazione «{esistente.get('nome', nome)}»")
|
||||
corpo = {k: v for k, v in dati.items() if k not in ("source", "overlay_png", "saved")}
|
||||
corpo["nome"] = nome
|
||||
self._libreria.mkdir(parents=True, exist_ok=True)
|
||||
# #84: prima la mappa (o la rimozione di una vecchia), poi il JSON che la dichiara
|
||||
if mappa is not None:
|
||||
self._salva_mappa(nome, mappa)
|
||||
else:
|
||||
self._togli_mappa(nome) # voce planare: una vecchia mappa non vale piu'
|
||||
self._scrivi_json(destinazione, corpo)
|
||||
return self._meta(corpo, self._leggi_attive())
|
||||
|
||||
def _salva_mappa(self, nome: str, mappa: DistortionMap) -> None:
|
||||
destinazione = self._file_mappa_nome(nome)
|
||||
fd, tmp = tempfile.mkstemp(dir=destinazione.parent, suffix=".tmp.npz") # .npz: numpy non aggiunge nulla
|
||||
os.close(fd)
|
||||
try:
|
||||
mappa.save(tmp)
|
||||
@@ -527,8 +694,8 @@ class Archivio:
|
||||
with self._lock:
|
||||
self._cache.pop(str(destinazione), None)
|
||||
|
||||
def _togli_mappa(self, sorgente: str) -> None:
|
||||
destinazione = self._file_mappa(sorgente)
|
||||
def _togli_mappa(self, nome: str) -> None:
|
||||
destinazione = self._file_mappa_nome(nome)
|
||||
try:
|
||||
destinazione.unlink()
|
||||
except FileNotFoundError:
|
||||
@@ -536,9 +703,64 @@ class Archivio:
|
||||
with self._lock:
|
||||
self._cache.pop(str(destinazione), None)
|
||||
|
||||
def elimina_nome(self, nome: str) -> bool:
|
||||
"""#91: toglie la voce (JSON e mappa) e la scollega da ogni sorgente. False se non c'era."""
|
||||
dati = self.leggi_nome(nome)
|
||||
if dati is None:
|
||||
return False
|
||||
canonico = str(dati.get("nome", nome))
|
||||
with self._lock_dati:
|
||||
try:
|
||||
self._file_nome(canonico).unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
self._togli_mappa(canonico)
|
||||
attive = self._leggi_attive()
|
||||
restanti = {s: n for s, n in attive.items() if n.casefold() != canonico.casefold()}
|
||||
if restanti != attive:
|
||||
self._scrivi_attive(restanti)
|
||||
return True
|
||||
|
||||
# ---- sorgenti
|
||||
|
||||
def attiva(self, sorgente: str, nome: str) -> str:
|
||||
"""#91: la sorgente usa questa calibrazione; ritorna il nome com'e' in libreria."""
|
||||
dati = self.leggi_nome(nome)
|
||||
if dati is None:
|
||||
raise NomeSconosciuto(nome)
|
||||
canonico = str(dati["nome"])
|
||||
with self._lock_dati:
|
||||
attive = self._leggi_attive()
|
||||
attive[sorgente] = canonico
|
||||
self._scrivi_attive(attive)
|
||||
return canonico
|
||||
|
||||
def scollega(self, sorgente: str) -> None:
|
||||
"""#91: la sorgente non ha piu' una calibrazione attiva; la libreria resta."""
|
||||
with self._lock_dati:
|
||||
attive = self._leggi_attive()
|
||||
if sorgente in attive:
|
||||
del attive[sorgente]
|
||||
self._scrivi_attive(attive)
|
||||
|
||||
def attiva_di(self, sorgente: str) -> str | None:
|
||||
"""#91: il nome della calibrazione attiva della sorgente (se esiste ancora)."""
|
||||
nome = self._leggi_attive().get(sorgente)
|
||||
dati = self.leggi_nome(nome) if nome else None
|
||||
return str(dati["nome"]) if dati else None
|
||||
|
||||
def leggi(self, sorgente: str) -> dict | None:
|
||||
"""#91: la calibrazione ATTIVA della sorgente, con "nome" e "source"; None se nessuna."""
|
||||
nome = self.attiva_di(sorgente)
|
||||
dati = self.leggi_nome(nome) if nome else None
|
||||
return {**dati, "source": sorgente} if dati else None
|
||||
|
||||
def leggi_mappa(self, sorgente: str) -> MappaPronta | None:
|
||||
"""La mappa salvata, pronta per `raddrizza`; None se la sorgente non ne ha una."""
|
||||
percorso = self._file_mappa(sorgente)
|
||||
"""La mappa della calibrazione attiva, pronta per `raddrizza`; None se non ne ha una."""
|
||||
nome = self.attiva_di(sorgente) # #91: la mappa e' della voce attiva, non della sorgente
|
||||
if nome is None:
|
||||
return None
|
||||
percorso = self._file_mappa_nome(nome)
|
||||
chiave = str(percorso)
|
||||
try:
|
||||
mtime = percorso.stat().st_mtime_ns
|
||||
@@ -557,10 +779,3 @@ class Archivio:
|
||||
with self._lock:
|
||||
self._cache[chiave] = (mtime, pronta)
|
||||
return pronta
|
||||
|
||||
def elimina(self, sorgente: str) -> None:
|
||||
try:
|
||||
self._file(sorgente).unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
self._togli_mappa(sorgente) # #84 toglie anche la mappa
|
||||
|
||||
+576
-196
File diff suppressed because it is too large
Load Diff
+311
-32
@@ -10,21 +10,39 @@ OpenCV, cartella di immagini) con lo stesso contratto di `vs-gui`, ma solo
|
||||
raddrizzato (header X-Raddrizzato: si|no); "grezzo": true lo dà com'è.
|
||||
Errori: 400 parametri, 404 sorgente inesistente, 503 guasto camera, 504 timeout.
|
||||
|
||||
Calibrazione a scacchiera per sorgente (#83, logica in scripts/calibrazione.py):
|
||||
GET /camera/calibration?source=<src> -> 200 {source, px_per_mm, mm_per_px,
|
||||
rms_px, n_corners, cols, rows, square_mm, image_width, image_height,
|
||||
scale_spread_pct, calibrated_at} | 404 "sorgente non calibrata"
|
||||
POST /camera/calibrate multipart: source, square_mm, image (file, opzionale:
|
||||
senza, grab GREZZO dalla sorgente), save=false, cols/rows (opzionali).
|
||||
Calibrazione a scacchiera (#83, logica in scripts/calibrazione.py).
|
||||
#91: archivio UNICO di calibrazioni con nome; una sorgente ne ha una attiva:
|
||||
POST /camera/calibrate multipart: source (obbligatorio solo senza image),
|
||||
square_mm, image (file, opzionale: senza, grab GREZZO dalla sorgente),
|
||||
cols/rows (opzionali). NON salva mai (il vecchio campo save e' ignorato).
|
||||
#84: SENZA cols/rows = calibrazione densa (default): una foto di
|
||||
scacchiera che riempie il campo, griglia trovata da sola, mappa di
|
||||
distorsione salvata accanto al JSON (.npz); mm/px e px/mm sono quelli
|
||||
dell'immagine RADDRIZZATA. CON cols/rows (angoli interni) = metodo
|
||||
planare di prima. Campi: quelli di C1 + overlay_png (b64) + saved +
|
||||
metodo ("densa"|"planare"), distorsione (bool) e, per la densa,
|
||||
residuo_rms_px, residuo_max_px, copertura_pct;
|
||||
distorsione; mm/px e px/mm sono quelli dell'immagine RADDRIZZATA. CON
|
||||
cols/rows (angoli interni) = metodo planare di prima. Risposta: i campi
|
||||
del calcolo + overlay_png (b64), calibrated_at, origine ("camera:<src>" |
|
||||
"file:<nome>"), calcolo_id (il servizio tiene in memoria gli ultimi 8
|
||||
calcoli) e metodo ("densa"|"planare"), distorsione (bool) e, per la
|
||||
densa, residuo_rms_px, residuo_max_px, copertura_pct;
|
||||
422 scacchiera non trovata o parametri invalidi
|
||||
DELETE /camera/calibration?source=<src> -> 204 (toglie JSON e mappa)
|
||||
GET /camera/calibrazioni[?source=S] -> {calibrazioni: [{nome, origine,
|
||||
image_width, image_height, px_per_mm, mm_per_px, metodo, square_mm,
|
||||
calibrated_at, attiva_su}], attiva: nome|null} (dalla piu' recente)
|
||||
POST /camera/calibrazioni json {calcolo_id, nome, sovrascrivi=false} -> 201
|
||||
metadati. 404 calcolo scaduto, 400 nome non valido, 409 nome gia' usato
|
||||
POST /camera/calibrazioni/manuale json {source, nome, mm_per_px, sovrascrivi=false}
|
||||
-> 201 metadati (#95: mm/px impostato a mano, metodo "manuale", nessuna
|
||||
mappa; non la attiva). 400 mm/px fuori 0.00001..100 o nome non valido,
|
||||
409 nome gia' usato; la risoluzione e' quella del frame tenuto o di un grab
|
||||
DELETE /camera/calibrazioni?nome=X -> 204 (toglie anche la mappa e la scollega
|
||||
da ogni sorgente) | 404
|
||||
PUT /camera/calibrazione-attiva json {source, nome}: legge un frame dalla
|
||||
sorgente e carica la calibrazione solo se la risoluzione (larghezza x
|
||||
altezza) coincide: 409 altrimenti; 404 nome ignoto; 200 come GET sotto
|
||||
GET /camera/calibration?source=<src> -> 200 la calibrazione ATTIVA {source,
|
||||
nome, px_per_mm, mm_per_px, ..., image_width, image_height,
|
||||
calibrated_at} | 404 "sorgente non calibrata" (lo legge il backend)
|
||||
DELETE /camera/calibration?source=<src> -> 204 (SCOLLEGA la calibrazione attiva,
|
||||
la libreria resta)
|
||||
GET /calibrazione -> pagina HTML autonoma (offline) per fare tutto a mano
|
||||
GET /acquisizione -> stessa pagina (#87)
|
||||
Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default:
|
||||
@@ -43,7 +61,7 @@ Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default:
|
||||
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
|
||||
Le calibrazioni stanno in <config-dir>/calibrazioni, le attive in attive.json (default
|
||||
%ProgramData%/Tielogic/TieMeasureFlow/acquisitore su Windows, ./acquisitore altrove).
|
||||
|
||||
Serve perche' vs-gui ascolta solo su 127.0.0.1 e il container `client` non lo
|
||||
@@ -54,6 +72,9 @@ Uso (dalla radice del repo):
|
||||
[--host 127.0.0.1] [--port 8765] [--folder <cartella immagini>] \
|
||||
[--config-dir <cartella calibrazioni>]
|
||||
--folder abilita la sorgente finta `camera:folder` (loop sulle immagini).
|
||||
Immagini caricate (#92): POST/DELETE /camera/immagini, in <config-dir>/immagini; sono sorgenti
|
||||
a pieno titolo (/camera/sources, grab, anteprima, calibrate). POST /camera/anteprima/calibrata
|
||||
raddrizza l'ultimo frame grezzo dell'anteprima con la calibrazione attiva, senza leggerne uno nuovo.
|
||||
Da Docker su Linux: --host 172.17.0.1 (gateway di docker0).
|
||||
|
||||
Dipende da: venv di VisionSuite (`uv sync --all-packages` in vendor/visionsuite):
|
||||
@@ -67,9 +88,15 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import math # #95: mm/px finito nella calibrazione manuale
|
||||
import os
|
||||
import re # #92: nome sicuro per le immagini caricate
|
||||
import sys
|
||||
import threading
|
||||
import time # #92: scadenza della cache delle camere
|
||||
import urllib.parse # #92: nome calibrazione nell'header X-Calibrazione
|
||||
import uuid # #91 id dei calcoli in attesa di nome
|
||||
from collections import OrderedDict
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
@@ -80,7 +107,7 @@ from fastapi.concurrency import run_in_threadpool # #83: il grab bloccante fuor
|
||||
from fastapi.responses import HTMLResponse, Response
|
||||
|
||||
from visionsuite.camera.camera_interface import CameraError, CameraGrabError, CameraStateError
|
||||
from visionsuite.camera.sources import SourceNotFound, list_sources, open_source
|
||||
from visionsuite.camera.sources import _ESTENSIONI, SourceNotFound, list_sources, open_source # #92: _ESTENSIONI = tipi ammessi per le immagini caricate
|
||||
|
||||
# #83: calibrazione e pagina stanno accanto a questo file (che non e' un pacchetto)
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
@@ -88,6 +115,8 @@ import calibrazione # noqa: E402
|
||||
from calibrazione_pagina import PAGINA # noqa: E402
|
||||
|
||||
_TRIGGER = ("continuo",)
|
||||
_MAX_CALCOLI = 8 # #91: calcoli tenuti in memoria in attesa di un nome
|
||||
_CACHE_CAMERE_S = 60.0 # #92: durata della cache della ricerca camere
|
||||
_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")
|
||||
@@ -131,8 +160,13 @@ def create_app(folder: Path | str | None = None,
|
||||
archivio = calibrazione.Archivio(config_dir if config_dir is not None
|
||||
else _config_dir_default()) # #83
|
||||
aperte: dict = {}
|
||||
grezzi: dict = {} # #92: sorgente -> ultimo frame pieno GREZZO dell'anteprima (per anteprima/calibrata)
|
||||
bloccati: dict = {} # #94: sorgente -> copia del frame grezzo pieno «bloccata» per il calcolo (stesso lock di grezzi)
|
||||
cartella_immagini = archivio.cartella / "immagini" # #92: immagini caricate dalla pagina (creata al bisogno)
|
||||
lock = threading.Lock() # protegge SOLO i dizionari, mai un'apertura
|
||||
lock_sorgenti: dict = {} # nome sorgente -> Lock che serializza apertura e grab
|
||||
calcoli: OrderedDict = OrderedDict() # #91: calcolo_id -> (dati, mappa); gli ultimi _MAX_CALCOLI
|
||||
lock_calcoli = threading.Lock()
|
||||
|
||||
def _config() -> dict:
|
||||
if folder is None:
|
||||
@@ -143,8 +177,32 @@ def create_app(folder: Path | str | None = None,
|
||||
with lock:
|
||||
return lock_sorgenti.setdefault(nome, threading.Lock())
|
||||
|
||||
# #92: la ricerca camere di list_sources dura ~10 s: il risultato (gia' filtrato) resta 60 s in cache
|
||||
cache_camere: dict = {"voci": None, "t": 0.0}
|
||||
lock_ricerca = threading.Lock() # due richieste insieme non lanciano due ricerche
|
||||
|
||||
def _camere(aggiorna: bool = False) -> list:
|
||||
with lock_ricerca:
|
||||
fresca = cache_camere["voci"] is not None and time.monotonic() - cache_camere["t"] < _CACHE_CAMERE_S
|
||||
if fresca and not aggiorna:
|
||||
return cache_camere["voci"]
|
||||
voci = list_sources(None, cameras=True, config=_config())
|
||||
# camera:file (replay) non e' mai configurato in TMF -> fuori dall'elenco
|
||||
cache_camere["voci"] = [{"name": v.name, "kind": v.kind, "label": v.label}
|
||||
for v in voci if v.kind in ("camera", "virtual") and v.name != "camera:file"]
|
||||
cache_camere["t"] = time.monotonic()
|
||||
return cache_camere["voci"]
|
||||
|
||||
def _prima_ricerca() -> None:
|
||||
try:
|
||||
_camere()
|
||||
except Exception: # noqa: BLE001 - e' solo un riscaldamento: la prima richiesta riprova
|
||||
_log.exception("prima ricerca camere fallita")
|
||||
|
||||
@asynccontextmanager
|
||||
async def _vita(_app):
|
||||
# #92: la prima ricerca parte all'avvio, cosi' la prima pagina non aspetta
|
||||
threading.Thread(target=_prima_ricerca, daemon=True, name="ricerca-camere").start()
|
||||
yield
|
||||
with lock: # alla chiusura rilascia le camere aperte
|
||||
sorgenti = list(aperte.values())
|
||||
@@ -158,17 +216,69 @@ def create_app(folder: Path | str | None = None,
|
||||
app = FastAPI(title="TMF camera service", lifespan=_vita)
|
||||
|
||||
@app.get("/camera/sources")
|
||||
def camera_sources():
|
||||
voci = list_sources(None, cameras=True, config=_config())
|
||||
return {"sources": [{"name": v.name, "kind": v.kind, "label": v.label}
|
||||
for v in voci if v.kind in ("camera", "virtual")]}
|
||||
def camera_sources(aggiorna: bool = False): # #92: aggiorna=true forza la ricerca (camera appena collegata)
|
||||
elenco = list(_camere(aggiorna))
|
||||
# #92: in coda le immagini caricate, sempre fresche dalla cartella (costano nulla)
|
||||
for v in list_sources(cartella_immagini, cameras=False):
|
||||
elenco.append({"name": v.name, "kind": "image", "label": f"immagine — {v.name}"})
|
||||
return {"sources": elenco}
|
||||
|
||||
def _chiudi(nome: str) -> None:
|
||||
"""#92: chiude la sorgente aperta e scorda il frame tenuto (immagine sostituita o tolta)."""
|
||||
with _lock_di(nome):
|
||||
with lock:
|
||||
s = aperte.pop(nome, None)
|
||||
grezzi.pop(nome, None)
|
||||
bloccati.pop(nome, None) # #94: tolta l'immagine/chiusa la sorgente, anche la copia bloccata
|
||||
if s is not None:
|
||||
try:
|
||||
s.close()
|
||||
except Exception: # noqa: BLE001 - la chiusura non deve far fallire la rotta
|
||||
pass
|
||||
|
||||
def _nome_immagine(grezzo: str) -> str:
|
||||
"""#92: solo il nome del file, coi caratteri sicuri di Archivio._file; "" se non ammesso."""
|
||||
base = Path((grezzo or "").replace("\\", "/")).name
|
||||
pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", base).strip("._")
|
||||
radice, punto, estensione = pulito.rpartition(".")
|
||||
if not punto or not radice or f".{estensione.lower()}" not in _ESTENSIONI:
|
||||
return ""
|
||||
return f"{radice[:100]}.{estensione}"
|
||||
|
||||
@app.post("/camera/immagini", status_code=201)
|
||||
async def immagini_carica(image: UploadFile | None = File(None)):
|
||||
if image is None:
|
||||
raise HTTPException(400, "manca il file «image»")
|
||||
nome = _nome_immagine(image.filename or "")
|
||||
if not nome:
|
||||
raise HTTPException(400, "estensione non ammessa: " + ", ".join(_ESTENSIONI))
|
||||
contenuto = await image.read()
|
||||
# #92: 422 se non e' un'immagine; la verifica sta PRIMA della scrittura, cosi' non resta nulla su disco
|
||||
if not contenuto or cv2.imdecode(np.frombuffer(contenuto, np.uint8), cv2.IMREAD_UNCHANGED) is None:
|
||||
raise HTTPException(422, "il file caricato non è leggibile come immagine")
|
||||
cartella_immagini.mkdir(parents=True, exist_ok=True)
|
||||
(cartella_immagini / nome).write_bytes(contenuto) # stesso nome = sostituisce
|
||||
await run_in_threadpool(_chiudi, nome) # se era aperta, si riapre sul file nuovo
|
||||
return {"source": nome, "label": f"immagine — {nome}", "kind": "image"}
|
||||
|
||||
@app.delete("/camera/immagini", status_code=204)
|
||||
def immagini_elimina(nome: str = ""):
|
||||
percorso = cartella_immagini / nome
|
||||
# #92: solo il nome di un file dentro la cartella, niente percorsi
|
||||
if not nome or Path(nome).name != nome or not percorso.is_file():
|
||||
raise HTTPException(404, "immagine non trovata")
|
||||
_chiudi(nome)
|
||||
percorso.unlink()
|
||||
return Response(status_code=204)
|
||||
|
||||
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())
|
||||
sorgente = open_source(nome, cartella_immagini, config=_config()) # #92: un nome di file = immagine caricata
|
||||
if hasattr(sorgente, "loop"): # #92: l'immagine singola non deve "finire" al secondo grab (anteprima continua)
|
||||
sorgente.loop = True
|
||||
# TieVision-Core#341 / #89: free-run, serve la scena di adesso, non il frame vecchio in coda
|
||||
if hasattr(sorgente, "solo_ultimo_frame"):
|
||||
try:
|
||||
@@ -226,7 +336,7 @@ def create_app(folder: Path | str | None = None,
|
||||
raise HTTPException(400, "«timeout_ms» deve essere un intero fra 1 e 10000")
|
||||
frame = _grab(nome, timeout) # #83: corpo estratto in _grab, invariato
|
||||
raddrizzato = False
|
||||
if not grezzo: # #84: sorgente calibrata (densa) -> frame raddrizzato dalla mappa in cache
|
||||
if not grezzo: # #84: sorgente calibrata (densa) -> frame raddrizzato dalla mappa in cache (#91: quella attiva)
|
||||
mappa = archivio.leggi_mappa(nome)
|
||||
if mappa is not None:
|
||||
corretto = calibrazione.raddrizza(frame, mappa)
|
||||
@@ -247,33 +357,43 @@ def create_app(folder: Path | str | None = None,
|
||||
|
||||
@app.get("/camera/calibration")
|
||||
def calibrazione_leggi(source: str = ""):
|
||||
dati = archivio.leggi(_sorgente(source))
|
||||
dati = archivio.leggi(_sorgente(source)) # #91: la calibrazione ATTIVA della sorgente
|
||||
if dati is None:
|
||||
raise HTTPException(404, "sorgente non calibrata")
|
||||
return dati
|
||||
|
||||
@app.delete("/camera/calibration", status_code=204)
|
||||
def calibrazione_elimina(source: str = ""):
|
||||
archivio.elimina(_sorgente(source))
|
||||
archivio.scollega(_sorgente(source)) # #91: scollega soltanto, la libreria resta
|
||||
return Response(status_code=204)
|
||||
|
||||
@app.post("/camera/calibrate")
|
||||
async def calibrazione_calcola(
|
||||
source: str = Form(...),
|
||||
source: str | None = Form(None), # #91: serve solo se manca l'immagine
|
||||
square_mm: float = Form(...),
|
||||
cols: int | None = Form(None), # #84: opzionali, senza = calibrazione densa
|
||||
rows: int | None = Form(None),
|
||||
image: UploadFile | None = File(None),
|
||||
save: bool = Form(False),
|
||||
save: bool = Form(False), # #91: ignorato, il calcolo non salva mai (si salva con nome)
|
||||
bloccata: bool = Form(False), # #94: true = calcola sulla copia bloccata, senza grab
|
||||
):
|
||||
nome = _sorgente(source)
|
||||
contenuto = await image.read() if image is not None else b""
|
||||
if contenuto:
|
||||
frame = cv2.imdecode(np.frombuffer(contenuto, np.uint8), cv2.IMREAD_UNCHANGED)
|
||||
if frame is None:
|
||||
raise HTTPException(422, "l'immagine caricata non è leggibile")
|
||||
origine = "file:" + (Path((image.filename or "").replace("\\", "/")).name or "immagine")
|
||||
elif bloccata: # #94: copia bloccata, nessun grab e nessun lock camera
|
||||
nome = _sorgente(source)
|
||||
with lock:
|
||||
frame = bloccati.get(nome)
|
||||
if frame is None:
|
||||
raise HTTPException(409, "nessuna immagine bloccata su questa sorgente")
|
||||
origine = nome
|
||||
else:
|
||||
nome = _sorgente(source)
|
||||
frame = await run_in_threadpool(_grab, nome, 5000) # come /camera/grab (404/503/504)
|
||||
origine = nome
|
||||
if (cols is None) != (rows is None):
|
||||
raise HTTPException(422, "cols e rows vanno indicati insieme (o nessuno dei due)")
|
||||
mappa = None
|
||||
@@ -285,12 +405,101 @@ def create_app(folder: Path | str | None = None,
|
||||
risultato = calibrazione.calibra_scacchiera(frame, cols, rows, square_mm)
|
||||
except calibrazione.CalibrazioneError as e:
|
||||
raise HTTPException(422, str(e))
|
||||
corpo = {"source": nome,
|
||||
**{k: v for k, v in risultato.items() if k != "overlay_png"},
|
||||
"calibrated_at": calibrazione.ora_iso()}
|
||||
if save:
|
||||
archivio.salva(nome, corpo, mappa) # #84: la mappa (se densa) va in .npz accanto
|
||||
return {**corpo, "overlay_png": risultato["overlay_png"], "saved": bool(save)}
|
||||
corpo = {**{k: v for k, v in risultato.items() if k != "overlay_png"},
|
||||
"calibrated_at": calibrazione.ora_iso(), "origine": origine}
|
||||
calcolo_id = uuid.uuid4().hex
|
||||
with lock_calcoli: # #91: il calcolo resta in memoria finche' non gli si da' un nome
|
||||
calcoli[calcolo_id] = (corpo, mappa)
|
||||
while len(calcoli) > _MAX_CALCOLI:
|
||||
calcoli.popitem(last=False)
|
||||
return {**corpo, "calcolo_id": calcolo_id, "overlay_png": risultato["overlay_png"],
|
||||
"bloccata": bool(bloccata and not contenuto)} # #94: solo nella risposta (non nel salvato): true se il frame e' la copia bloccata
|
||||
|
||||
# ---- #91: archivio unico di calibrazioni con nome
|
||||
|
||||
@app.get("/camera/calibrazioni")
|
||||
def calibrazioni_elenco(source: str | None = None):
|
||||
attiva = archivio.attiva_di(_sorgente(source)) if source is not None else None
|
||||
return {"calibrazioni": archivio.elenco(), "attiva": attiva}
|
||||
|
||||
@app.post("/camera/calibrazioni", status_code=201)
|
||||
def calibrazioni_salva(corpo: dict = Body(default=None)):
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
calcolo_id = corpo.get("calcolo_id")
|
||||
if not isinstance(calcolo_id, str) or not calcolo_id:
|
||||
raise HTTPException(400, "manca «calcolo_id»")
|
||||
sovrascrivi = corpo.get("sovrascrivi", False)
|
||||
if not isinstance(sovrascrivi, bool):
|
||||
raise HTTPException(400, "«sovrascrivi» deve essere true o false")
|
||||
try:
|
||||
nome = calibrazione.pulisci_nome(corpo.get("nome"))
|
||||
except calibrazione.NomeNonValido as e:
|
||||
raise HTTPException(400, str(e))
|
||||
with lock_calcoli:
|
||||
voce = calcoli.get(calcolo_id)
|
||||
if voce is None:
|
||||
raise HTTPException(404, "calcolo non più disponibile: ricalcolare")
|
||||
dati, mappa = voce
|
||||
try:
|
||||
return archivio.salva_nome(nome, dati, mappa, sovrascrivi)
|
||||
except calibrazione.NomeGiaUsato as e:
|
||||
raise HTTPException(409, str(e))
|
||||
|
||||
@app.post("/camera/calibrazioni/manuale", status_code=201)
|
||||
def calibrazioni_manuale(corpo: dict = Body(default=None)):
|
||||
"""#95: calibrazione con mm/px impostato a mano (nessuna scacchiera, nessuna mappa); NON la attiva."""
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
sorgente = _sorgente(corpo.get("source"))
|
||||
mm_px = corpo.get("mm_per_px")
|
||||
if (isinstance(mm_px, bool) or not isinstance(mm_px, (int, float))
|
||||
or not math.isfinite(mm_px) or not 0.00001 <= mm_px <= 100):
|
||||
raise HTTPException(400, "mm/px deve essere un numero positivo")
|
||||
sovrascrivi = corpo.get("sovrascrivi", False)
|
||||
if not isinstance(sovrascrivi, bool):
|
||||
raise HTTPException(400, "«sovrascrivi» deve essere true o false")
|
||||
try:
|
||||
nome = calibrazione.pulisci_nome(corpo.get("nome"))
|
||||
except calibrazione.NomeNonValido as e:
|
||||
raise HTTPException(400, str(e))
|
||||
with lock:
|
||||
frame = grezzi.get(sorgente)
|
||||
if frame is None: # nessun frame tenuto: un grab come /camera/grab (404/503/504)
|
||||
frame = _grab(sorgente, 5000)
|
||||
alt, larg = frame.shape[:2] # serve al controllo W×H di PUT /camera/calibrazione-attiva
|
||||
dati = {"px_per_mm": 1.0 / mm_px, "mm_per_px": float(mm_px), "metodo": "manuale",
|
||||
"image_width": larg, "image_height": alt, "origine": sorgente,
|
||||
"calibrated_at": calibrazione.ora_iso(), "rms_px": None, "square_mm": None}
|
||||
try:
|
||||
return archivio.salva_nome(nome, dati, None, sovrascrivi)
|
||||
except calibrazione.NomeGiaUsato as e:
|
||||
raise HTTPException(409, str(e))
|
||||
|
||||
@app.delete("/camera/calibrazioni", status_code=204)
|
||||
def calibrazioni_elimina(nome: str = ""):
|
||||
if not archivio.elimina_nome(nome):
|
||||
raise HTTPException(404, "calibrazione non trovata")
|
||||
return Response(status_code=204)
|
||||
|
||||
@app.put("/camera/calibrazione-attiva")
|
||||
def calibrazione_attiva(corpo: dict = Body(default=None)):
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
sorgente = _sorgente(corpo.get("source"))
|
||||
richiesto = corpo.get("nome")
|
||||
dati = archivio.leggi_nome(richiesto) if isinstance(richiesto, str) else None
|
||||
if dati is None:
|
||||
raise HTTPException(404, "calibrazione non trovata")
|
||||
nome = dati["nome"]
|
||||
frame = _grab(sorgente, 5000) # stesso lock e stessi errori di /camera/grab (404/503/504)
|
||||
alt, larg = frame.shape[:2]
|
||||
cw, ch = dati.get("image_width"), dati.get("image_height")
|
||||
if not isinstance(cw, int) or not isinstance(ch, int):
|
||||
raise HTTPException(409, f"la calibrazione «{nome}» non ha la risoluzione registrata: "
|
||||
"non si può caricare")
|
||||
if (cw, ch) != (larg, alt): # la mappa vale solo per frame della stessa risoluzione
|
||||
raise HTTPException(409, f"la calibrazione «{nome}» è per immagini {cw}×{ch}, "
|
||||
f"la sorgente dà {larg}×{alt}: non si può caricare")
|
||||
archivio.attiva(sorgente, nome)
|
||||
return archivio.leggi(sorgente)
|
||||
|
||||
# ---- #87: acquisizione (anteprima, parametri, settaggi)
|
||||
|
||||
@@ -334,7 +543,12 @@ def create_app(folder: Path | str | None = None,
|
||||
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)
|
||||
def leggi(s): # #92: il frame grezzo si tiene sotto il lock della sorgente
|
||||
f = s.grab(timeout)
|
||||
with lock:
|
||||
grezzi[nome] = f.copy()
|
||||
return f
|
||||
frame = _con_sorgente(nome, leggi)
|
||||
raddrizzato = False
|
||||
if not grezzo: # come /camera/grab
|
||||
mappa = archivio.leggi_mappa(nome)
|
||||
@@ -362,6 +576,71 @@ def create_app(folder: Path | str | None = None,
|
||||
"X-Nitidezza": f"{nitidezza:.1f}", "X-Saturati-Pct": f"{saturati:.2f}",
|
||||
"X-Media": f"{media:.1f}"})
|
||||
|
||||
@app.post("/camera/blocca")
|
||||
def camera_blocca(corpo: dict = Body(default=None)):
|
||||
"""#94: copia l'ultimo frame grezzo della sorgente; /camera/calibrate con bloccata=true calcola su quella."""
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
nome = _sorgente(corpo.get("source"))
|
||||
with lock:
|
||||
frame = grezzi.get(nome)
|
||||
if frame is None:
|
||||
raise HTTPException(404, "nessun frame: avviare l'anteprima o scattare")
|
||||
bloccati[nome] = frame.copy()
|
||||
alt, larg = frame.shape[:2]
|
||||
return {"source": nome, "larghezza": larg, "altezza": alt}
|
||||
|
||||
@app.delete("/camera/blocca", status_code=204)
|
||||
def camera_sblocca(source: str = ""):
|
||||
with lock: # #94: idempotente, 204 anche se non c'era nulla
|
||||
bloccati.pop(source.strip(), None)
|
||||
return Response(status_code=204)
|
||||
|
||||
@app.post("/camera/anteprima/calibrata")
|
||||
def camera_anteprima_calibrata(corpo: dict = Body(default=None)):
|
||||
"""#92: l'ultimo frame grezzo dell'anteprima, raddrizzato; NON legge un frame nuovo. #95: senza mappa esce grezzo (X-Raddrizzato: no)."""
|
||||
corpo = corpo if isinstance(corpo, dict) else {}
|
||||
nome = _sorgente(corpo.get("source"))
|
||||
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")
|
||||
with lock:
|
||||
frame = grezzi.get(nome)
|
||||
if frame is None:
|
||||
raise HTTPException(404, "nessun frame: avviare l'anteprima o scattare")
|
||||
attiva = archivio.attiva_di(nome)
|
||||
if attiva is None:
|
||||
raise HTTPException(409, "nessuna calibrazione attiva su questa sorgente")
|
||||
mappa = archivio.leggi_mappa(nome)
|
||||
if mappa is None: # #95: senza mappa (manuale o planare) solo scala: frame grezzo, nessuna correzione
|
||||
dati = archivio.leggi(nome) or {}
|
||||
cw, ch = dati.get("image_width"), dati.get("image_height")
|
||||
alt, larg = frame.shape[:2]
|
||||
if isinstance(cw, int) and isinstance(ch, int) and (cw, ch) != (larg, alt):
|
||||
raise HTTPException(409, f"la calibrazione «{attiva}» è per immagini {cw}×{ch}, "
|
||||
f"il frame è {larg}×{alt}")
|
||||
corretto, raddrizzato = frame, "no"
|
||||
else:
|
||||
corretto = calibrazione.raddrizza(frame, mappa)
|
||||
if corretto is None: # risoluzione diversa da quella calibrata
|
||||
ch, cw = mappa.shape[:2]
|
||||
alt, larg = frame.shape[:2]
|
||||
raise HTTPException(409, f"la calibrazione «{attiva}» è per immagini {cw}×{ch}, "
|
||||
f"il frame è {larg}×{alt}")
|
||||
raddrizzato = "si"
|
||||
alt, larg = corretto.shape[:2]
|
||||
if corretto.dtype != np.uint8: # il JPEG vuole 8 bit
|
||||
corretto = _grigio8(corretto)
|
||||
scala = lato / max(alt, larg)
|
||||
if scala < 1.0: # mai ingrandita
|
||||
corretto = cv2.resize(corretto, (max(1, round(larg * scala)), max(1, round(alt * scala))),
|
||||
interpolation=cv2.INTER_AREA)
|
||||
ok, buf = cv2.imencode(".jpg", corretto, [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": raddrizzato, "X-Larghezza": str(larg), "X-Altezza": str(alt),
|
||||
"X-Calibrazione": urllib.parse.quote(attiva, safe="")})
|
||||
|
||||
@app.get("/camera/parametri")
|
||||
def parametri_leggi(source: str = ""):
|
||||
nome = _sorgente(source)
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Test di #94: «Blocca immagine» (scripts/camera_service.py) e riquadro vivo nella tab Calibrazione (PAGINA).
|
||||
|
||||
Uso: uv run --python 3.13 --extra vision-worker --extra dev \
|
||||
--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q
|
||||
Sorgente finta che conta i grab: niente camera vera, niente ProgramData.
|
||||
"""
|
||||
import re
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from test_calibrazione import _png, cartella, client, una_scacchiera # fixture e helper della calibrazione
|
||||
from test_immagini_affiancate import _anteprima, _carica, _modulo
|
||||
from test_pagina_acquisizione import PAGINA
|
||||
|
||||
__all__ = ["cartella", "client"] # le fixture importate restano usate
|
||||
|
||||
FINTA = "finta"
|
||||
CAMPI = {"cols": "9", "rows": "6", "square_mm": "2"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def finta(client, monkeypatch):
|
||||
"""Sorgente «finta»: il primo grab da' la scacchiera, i successivi un frame bianco (non calibrabile)."""
|
||||
letture = []
|
||||
scacchiera = una_scacchiera()
|
||||
bianco = np.full(scacchiera.shape, 255, np.uint8)
|
||||
|
||||
class Finta:
|
||||
def grab(self, timeout=1000):
|
||||
letture.append(1)
|
||||
return scacchiera.copy() if len(letture) == 1 else bianco.copy()
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setitem(_modulo(client), "open_source", lambda *a, **k: Finta())
|
||||
client.letture = letture
|
||||
return client
|
||||
|
||||
|
||||
def _blocca(c, nome=FINTA):
|
||||
return c.post("/camera/blocca", json={"source": nome})
|
||||
|
||||
|
||||
def _calcola(c, **extra):
|
||||
return c.post("/camera/calibrate", data={"source": FINTA, **CAMPI, **extra})
|
||||
|
||||
|
||||
# ---- servizio
|
||||
|
||||
def test_blocca_senza_frame_404(finta):
|
||||
r = _blocca(finta)
|
||||
assert r.status_code == 404 and r.json()["detail"] == "nessun frame: avviare l'anteprima o scattare"
|
||||
|
||||
|
||||
def test_blocca_senza_source_400(finta):
|
||||
assert finta.post("/camera/blocca", json={}).status_code == 400
|
||||
assert finta.post("/camera/blocca").status_code == 400
|
||||
|
||||
|
||||
def test_anteprima_poi_blocca_200_con_misure(finta):
|
||||
assert _anteprima(finta, FINTA).status_code == 200
|
||||
r = _blocca(finta)
|
||||
alt, larg = una_scacchiera().shape[:2]
|
||||
assert r.status_code == 200 and r.json() == {"source": FINTA, "larghezza": larg, "altezza": alt}
|
||||
|
||||
|
||||
def test_calibrate_bloccata_usa_la_copia_senza_grab(finta):
|
||||
assert _anteprima(finta, FINTA).status_code == 200 # frame 1: la scacchiera
|
||||
assert _blocca(finta).status_code == 200
|
||||
assert _anteprima(finta, FINTA).status_code == 200 # frame 2: bianco, il grezzo cambia ma la copia no
|
||||
prima = len(finta.letture)
|
||||
r = _calcola(finta, bloccata="true")
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["bloccata"] is True and r.json()["origine"] == FINTA
|
||||
assert len(finta.letture) == prima # nessun grab per il calcolo
|
||||
# senza il blocco il calcolo legge un frame nuovo (bianco): non calibrabile
|
||||
assert _calcola(finta).status_code == 422
|
||||
assert len(finta.letture) == prima + 1
|
||||
|
||||
|
||||
def test_calibrate_normale_dice_bloccata_false(client):
|
||||
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["bloccata"] is False
|
||||
|
||||
|
||||
def test_calibrate_bloccata_senza_copia_409(finta):
|
||||
assert _anteprima(finta, FINTA).status_code == 200 # il grezzo c'e', ma non e' bloccato
|
||||
r = _calcola(finta, bloccata="true")
|
||||
assert r.status_code == 409 and r.json()["detail"] == "nessuna immagine bloccata su questa sorgente"
|
||||
|
||||
|
||||
def test_sblocca_204_idempotente_e_toglie_la_copia(finta):
|
||||
assert finta.delete("/camera/blocca", params={"source": FINTA}).status_code == 204 # niente da togliere
|
||||
assert _anteprima(finta, FINTA).status_code == 200
|
||||
assert _blocca(finta).status_code == 200
|
||||
assert finta.delete("/camera/blocca", params={"source": FINTA}).status_code == 204
|
||||
assert _calcola(finta, bloccata="true").status_code == 409
|
||||
assert finta.delete("/camera/blocca", params={"source": FINTA}).status_code == 204
|
||||
|
||||
|
||||
def test_rimuovere_l_immagine_toglie_la_copia_bloccata(client):
|
||||
assert _carica(client, "foto.png", una_scacchiera()).status_code == 201
|
||||
assert _anteprima(client, "foto.png").status_code == 200
|
||||
assert _blocca(client, "foto.png").status_code == 200
|
||||
assert client.delete("/camera/immagini", params={"nome": "foto.png"}).status_code == 204
|
||||
assert _carica(client, "foto.png", una_scacchiera()).status_code == 201
|
||||
r = client.post("/camera/calibrate", data={"source": "foto.png", "bloccata": "true", **CAMPI})
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
# ---- pagina (contenuto statico)
|
||||
|
||||
def test_pagina_riquadro_e_casella_nella_tab_calibrazione():
|
||||
ids = set(re.findall(r'id="([^"]+)"', PAGINA))
|
||||
assert {"immagineCalTab", "vuotoCalTab", "scattaCal", "bloccaImg", "msgBlocco"} <= ids
|
||||
assert "Nessuna immagine: «Scatta» o avviare l'anteprima nella tab Acquisizione" in PAGINA
|
||||
assert "Blocca immagine" in PAGINA
|
||||
tab = PAGINA.split('id="pannelloCal"')[1].split('id="sezioneSalvate"')[0]
|
||||
assert "immagineCalTab" in tab and "bloccaImg" in tab # nella sezione Calibrazione, non altrove
|
||||
|
||||
|
||||
def test_pagina_scegli_tab_non_ferma_piu_il_ciclo():
|
||||
assert "A.riprendi" not in PAGINA
|
||||
scegli = PAGINA.split("function scegliTab")[1].split("$(\"tabAcq\").onclick")[0]
|
||||
assert "fermaAnteprima" not in scegli and "avviaAnteprima" not in scegli
|
||||
|
||||
|
||||
def test_pagina_calcola_manda_bloccata_e_non_ferma_il_ciclo():
|
||||
calcola = PAGINA.split("async function calcola")[1].split("async function salvaNome")[0]
|
||||
assert 'f.append("bloccata", "true")' in calcola
|
||||
assert "if (bloccata)" in calcola and "else fermaAnteprima();" in calcola
|
||||
assert "immagine bloccata" in calcola
|
||||
assert "l'anteprima resta sospesa" not in PAGINA and "si può bloccare l'immagine" in PAGINA
|
||||
|
||||
|
||||
def test_pagina_blocco_rotte_e_cambio_sorgente():
|
||||
assert '"/camera/blocca"' in PAGINA and '"/camera/blocca?source="' in PAGINA
|
||||
assert "method: \"DELETE\"" in PAGINA
|
||||
cambio = PAGINA.split("function cambioSorgente")[1].split("// ---- Anteprima")[0]
|
||||
assert "rilasciaBlocco()" in cambio
|
||||
assert cambio.index("rilasciaBlocco()") < cambio.index("svuotaRiquadri()")
|
||||
|
||||
|
||||
def test_pagina_blocco_attende_l_anteprima_in_volo():
|
||||
# #94: il frame bloccato e' quello mostrato: flag A.bloccando, attesa di A.inVolo (20 ms, max 3 s), poi POST, flag tolto in finally
|
||||
blocco = PAGINA.split('$("bloccaImg").onchange')[1].split('$("avvia").onclick')[0]
|
||||
assert "A.bloccando++" in blocco and "A.bloccando--" in blocco and "finally" in blocco
|
||||
assert "k < 150" in blocco and "pausa(20)" in blocco
|
||||
assert blocco.index("A.inVolo") < blocco.index('fetch("/camera/blocca"')
|
||||
ciclo = PAGINA.split("async function ciclo")[1].split("function avviaAnteprima")[0]
|
||||
assert "while (A.bloccando)" in ciclo and ciclo.index("A.bloccando") < ciclo.index("unaAnteprima(1280)")
|
||||
scatta = PAGINA.split("async function scatta")[1].split("// #94: «Blocca immagine»")[0]
|
||||
assert "while (A.bloccando)" in scatta and scatta.index("A.bloccando") < scatta.index("unaAnteprima(4096)")
|
||||
@@ -82,12 +82,24 @@ def client(cartella, tmp_path):
|
||||
|
||||
|
||||
CAMPI = {"cols": "9", "rows": "6", "square_mm": "2"}
|
||||
CHIAVI_C1 = {"source", "px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows",
|
||||
CHIAVI_C1 = {"px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows",
|
||||
"square_mm", "image_width", "image_height", "scale_spread_pct", "calibrated_at"}
|
||||
# #84: campi in piu' per metodo planare (C2 con cols/rows) e per la densa
|
||||
CHIAVI_PLANARE = CHIAVI_C1 | {"metodo", "distorsione"}
|
||||
CHIAVI_PLANARE = CHIAVI_C1 | {"metodo", "distorsione", "origine"} # #91: + origine ("camera:X" | "file:Y")
|
||||
CHIAVI_DENSA = CHIAVI_PLANARE | {"residuo_rms_px", "residuo_max_px", "copertura_pct"}
|
||||
SORGENTE = "camera:folder"
|
||||
# #91: GET /camera/calibration = voce di libreria + "nome" + "source" (la sorgente che la usa)
|
||||
CHIAVI_ATTIVA = {"source", "nome"}
|
||||
|
||||
|
||||
def salva_con_nome(client, calcolo: dict, nome: str, **extra):
|
||||
"""#91: POST /camera/calibrazioni sul calcolo_id di una risposta di /camera/calibrate."""
|
||||
return client.post("/camera/calibrazioni", json={"calcolo_id": calcolo["calcolo_id"], "nome": nome, **extra})
|
||||
|
||||
|
||||
def carica_su(client, sorgente: str, nome: str):
|
||||
"""#91: PUT /camera/calibrazione-attiva."""
|
||||
return client.put("/camera/calibrazione-attiva", json={"source": sorgente, "nome": nome})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -157,34 +169,49 @@ def test_c2_immagine_caricata_senza_salvare(client):
|
||||
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
|
||||
assert r.status_code == 200, r.text
|
||||
corpo = r.json()
|
||||
assert CHIAVI_PLANARE | {"overlay_png", "saved"} == set(corpo)
|
||||
assert CHIAVI_PLANARE | {"overlay_png", "calcolo_id", "bloccata"} == set(corpo) # #94: + bloccata; #91: niente "saved", niente "source"
|
||||
assert corpo["metodo"] == "planare" and corpo["distorsione"] is False # #84: con cols/rows nulla cambia
|
||||
assert corpo["saved"] is False
|
||||
assert corpo["origine"] == "file:s.png"
|
||||
assert corpo["px_per_mm"] == pytest.approx(20.0, rel=0.005)
|
||||
# senza save non si scrive nulla
|
||||
# il calcolo non salva mai
|
||||
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
|
||||
assert not client.config.exists() or not list(client.config.glob("*.json"))
|
||||
assert not client.config.exists() or not list(client.config.iterdir())
|
||||
|
||||
|
||||
def test_c2_save_true_viene_ignorato(client):
|
||||
# #91: il vecchio campo save non salva piu' nulla
|
||||
r = client.post("/camera/calibrate", data={"source": "camera:folder", "save": "true", **CAMPI})
|
||||
assert r.status_code == 200, r.text
|
||||
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
|
||||
assert client.get("/camera/calibrazioni").json()["calibrazioni"] == []
|
||||
assert not client.config.exists() or not list(client.config.iterdir())
|
||||
|
||||
|
||||
def test_c2_c1_c3_ciclo_con_grab_dalla_sorgente(client):
|
||||
# nessun file: grab da camera:folder (la scacchiera in cartella), con save
|
||||
r = client.post("/camera/calibrate",
|
||||
data={"source": "camera:folder", "save": "true", **CAMPI})
|
||||
# nessun file: grab da camera:folder (la scacchiera in cartella)
|
||||
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["saved"] is True
|
||||
assert r.json()["origine"] == "camera:folder"
|
||||
assert salva_con_nome(client, r.json(), "Banco").status_code == 201 # #91: si salva con nome, dopo
|
||||
|
||||
# salvata ma non ancora caricata sulla sorgente
|
||||
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
|
||||
assert carica_su(client, "camera:folder", "Banco").status_code == 200
|
||||
|
||||
r1 = client.get("/camera/calibration", params={"source": "camera:folder"})
|
||||
assert r1.status_code == 200
|
||||
assert set(r1.json()) == CHIAVI_PLANARE # nessun overlay nel file salvato
|
||||
assert set(r1.json()) == CHIAVI_PLANARE | CHIAVI_ATTIVA # nessun overlay nel file salvato
|
||||
assert r1.json()["px_per_mm"] == pytest.approx(20.0, rel=0.005)
|
||||
assert r1.json()["source"] == "camera:folder"
|
||||
file_salvati = list(client.config.glob("*.json"))
|
||||
assert r1.json()["source"] == "camera:folder" and r1.json()["nome"] == "Banco"
|
||||
file_salvati = list((client.config / "calibrazioni").glob("*.json"))
|
||||
assert len(file_salvati) == 1
|
||||
assert json.loads(file_salvati[0].read_text(encoding="utf-8"))["cols"] == 9
|
||||
|
||||
# DELETE /camera/calibration scollega: la libreria resta
|
||||
assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204
|
||||
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
|
||||
# cancellare di nuovo e' idempotente
|
||||
assert len(list((client.config / "calibrazioni").glob("*.json"))) == 1
|
||||
# scollegare di nuovo e' idempotente
|
||||
assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204
|
||||
|
||||
|
||||
@@ -214,12 +241,14 @@ def test_c2_errori_camera_come_grab(client):
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_due_sorgenti_hanno_file_diversi(client):
|
||||
for nome in ("camera:a b", "camera:a_b"):
|
||||
r = client.post("/camera/calibrate", data={"source": nome, "save": "true", **CAMPI},
|
||||
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
|
||||
assert r.status_code == 200
|
||||
assert len(list(client.config.glob("*.json"))) == 2
|
||||
def test_due_nomi_simili_hanno_file_diversi(client):
|
||||
# #91: i nomi si ripuliscono come prima (hash sul nome originale): due nomi diversi non si pestano
|
||||
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI},
|
||||
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
|
||||
assert r.status_code == 200
|
||||
for nome in ("banco a b", "banco a_b"):
|
||||
assert salva_con_nome(client, r.json(), nome).status_code == 201
|
||||
assert len(list((client.config / "calibrazioni").glob("*.json"))) == 2
|
||||
|
||||
|
||||
def test_c4_pagina_html(client):
|
||||
@@ -287,14 +316,15 @@ def test_archivio_mappa_cache_e_invalidazione(tmp_path):
|
||||
_, mappa = calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), 2.0)
|
||||
arch = calibrazione.Archivio(tmp_path)
|
||||
assert arch.leggi_mappa("s") is None
|
||||
arch.salva("s", {"a": 1}, mappa)
|
||||
arch.salva_nome("cal", {"a": 1}, mappa) # #91: salvataggio con nome + attivazione sulla sorgente
|
||||
arch.attiva("s", "cal")
|
||||
prima = arch.leggi_mappa("s")
|
||||
assert prima is not None and arch.leggi_mappa("s") is prima # cache in memoria
|
||||
assert not list(tmp_path.glob("*.tmp*")) # niente temporanei rimasti
|
||||
arch.salva("s", {"a": 2}, mappa) # file riscritto -> cache invalidata
|
||||
assert not list(tmp_path.rglob("*.tmp*")) # niente temporanei rimasti
|
||||
arch.salva_nome("cal", {"a": 2}, mappa, sovrascrivi=True) # file riscritto -> cache invalidata
|
||||
assert arch.leggi_mappa("s") is not prima
|
||||
arch.salva("s", {"a": 3}) # salvataggio senza mappa (planare): la vecchia sparisce
|
||||
assert arch.leggi_mappa("s") is None and not list(tmp_path.glob("*.npz"))
|
||||
arch.salva_nome("cal", {"a": 3}, sovrascrivi=True) # salvataggio senza mappa (planare): la vecchia sparisce
|
||||
assert arch.leggi_mappa("s") is None and not list(tmp_path.rglob("*.npz"))
|
||||
|
||||
|
||||
# ---- #84: rotte con la calibrazione densa
|
||||
@@ -307,11 +337,10 @@ def test_c2_senza_cols_rows_e_densa(client_denso):
|
||||
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
|
||||
assert r.status_code == 200, r.text
|
||||
corpo = r.json()
|
||||
assert CHIAVI_DENSA | {"overlay_png", "saved"} == set(corpo)
|
||||
assert CHIAVI_DENSA | {"overlay_png", "calcolo_id", "bloccata"} == set(corpo)
|
||||
assert corpo["metodo"] == "densa" and corpo["distorsione"] is True
|
||||
assert (corpo["cols"], corpo["rows"]) == (15, 11)
|
||||
assert corpo["px_per_mm"] == pytest.approx(1.0 / corpo["mm_per_px"])
|
||||
assert corpo["saved"] is False
|
||||
assert not client_denso.config.exists() or not list(client_denso.config.iterdir())
|
||||
|
||||
|
||||
@@ -326,14 +355,17 @@ def test_grab_raddrizzato_se_calibrata_e_grezzo_a_richiesta(client_denso, cartel
|
||||
assert prima.headers["x-raddrizzato"] == "no" # non calibrata: frame com'e'
|
||||
assert np.array_equal(_grigio(prima), originale)
|
||||
|
||||
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
|
||||
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["saved"] is True
|
||||
json_salvati, mappe = list(client_denso.config.glob("*.json")), list(client_denso.config.glob("*.npz"))
|
||||
assert salva_con_nome(client_denso, r.json(), "Densa").status_code == 201 # #91: salvata con nome
|
||||
libreria = client_denso.config / "calibrazioni"
|
||||
json_salvati, mappe = list(libreria.glob("*.json")), list(libreria.glob("*.npz"))
|
||||
assert len(json_salvati) == 1 and len(mappe) == 1
|
||||
assert mappe[0].stem == json_salvati[0].stem # stesso nome del JSON
|
||||
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no" # salvata != attiva
|
||||
assert carica_su(client_denso, SORGENTE, "Densa").status_code == 200
|
||||
saved = client_denso.get("/camera/calibration", params={"source": SORGENTE}).json()
|
||||
assert set(saved) == CHIAVI_DENSA and saved["metodo"] == "densa"
|
||||
assert set(saved) == CHIAVI_DENSA | CHIAVI_ATTIVA and saved["metodo"] == "densa"
|
||||
|
||||
dopo = client_denso.post("/camera/grab", json={"source": SORGENTE})
|
||||
assert dopo.status_code == 200 and dopo.headers["x-raddrizzato"] == "si"
|
||||
@@ -358,19 +390,33 @@ def test_grab_grezzo_non_booleano_400(client_denso):
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_delete_toglie_anche_la_mappa(client_denso):
|
||||
client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
|
||||
assert list(client_denso.config.glob("*.npz"))
|
||||
def test_scollega_lascia_la_libreria_ed_elimina_toglie_la_mappa(client_denso):
|
||||
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
|
||||
salva_con_nome(client_denso, r.json(), "Densa")
|
||||
carica_su(client_denso, SORGENTE, "Densa")
|
||||
libreria = client_denso.config / "calibrazioni"
|
||||
assert list(libreria.glob("*.npz"))
|
||||
# scollegare: il frame torna com'e', ma la voce (e la mappa) restano in libreria
|
||||
assert client_denso.delete("/camera/calibration", params={"source": SORGENTE}).status_code == 204
|
||||
assert not list(client_denso.config.glob("*.npz")) and not list(client_denso.config.glob("*.json"))
|
||||
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no"
|
||||
assert list(libreria.glob("*.npz")) and list(libreria.glob("*.json"))
|
||||
# eliminare la voce toglie JSON e mappa
|
||||
carica_su(client_denso, SORGENTE, "Densa")
|
||||
assert client_denso.delete("/camera/calibrazioni", params={"nome": "Densa"}).status_code == 204
|
||||
assert not list(libreria.glob("*.npz")) and not list(libreria.glob("*.json"))
|
||||
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no"
|
||||
|
||||
|
||||
def test_ricalibrare_planare_toglie_la_mappa_densa(client_denso):
|
||||
client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
|
||||
assert list(client_denso.config.glob("*.npz"))
|
||||
planare = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "save": "true", **CAMPI},
|
||||
densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
|
||||
salva_con_nome(client_denso, densa.json(), "Cal")
|
||||
carica_su(client_denso, SORGENTE, "Cal")
|
||||
libreria = client_denso.config / "calibrazioni"
|
||||
assert list(libreria.glob("*.npz"))
|
||||
planare = client_denso.post("/camera/calibrate", data={"source": SORGENTE, **CAMPI},
|
||||
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
|
||||
assert planare.status_code == 200 and planare.json()["metodo"] == "planare"
|
||||
assert not list(client_denso.config.glob("*.npz")) # niente mappa vecchia che raddrizza a vuoto
|
||||
assert salva_con_nome(client_denso, planare.json(), "Cal").status_code == 409 # nome gia' preso
|
||||
assert salva_con_nome(client_denso, planare.json(), "Cal", sovrascrivi=True).status_code == 201
|
||||
assert not list(libreria.glob("*.npz")) # niente mappa vecchia che raddrizza a vuoto
|
||||
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no"
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
"""Test di #95: calibrazione manuale (mm/px impostato a mano) in scripts/camera_service.py e nella PAGINA.
|
||||
|
||||
Uso: uv run --python 3.13 --extra vision-worker --extra dev \
|
||||
--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q
|
||||
Sorgenti = immagini caricate in tmp_path: niente camera vera, niente ProgramData.
|
||||
"""
|
||||
import re
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from test_calibrazione import _png, cartella, client # fixture della calibrazione
|
||||
from test_immagini_affiancate import _anteprima, _calibrata, _carica
|
||||
from test_pagina_acquisizione import PAGINA
|
||||
|
||||
__all__ = ["cartella", "client"] # le fixture importate restano usate
|
||||
|
||||
SRC = "banco.png" # 64x48
|
||||
ALTRA = "grande.png" # 80x60, risoluzione diversa
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sorgenti(client):
|
||||
assert _carica(client, SRC, np.full((48, 64), 120, np.uint8)).status_code == 201
|
||||
assert _carica(client, ALTRA, np.full((60, 80), 120, np.uint8)).status_code == 201
|
||||
return client
|
||||
|
||||
|
||||
def _manuale(c, nome="Manuale", mm=0.05, source=SRC, **extra):
|
||||
return c.post("/camera/calibrazioni/manuale",
|
||||
json={"source": source, "nome": nome, "mm_per_px": mm, **extra})
|
||||
|
||||
|
||||
# ---- servizio
|
||||
|
||||
def test_manuale_201_con_scala_e_risoluzione_del_frame(sorgenti):
|
||||
r = _manuale(sorgenti, mm=0.05)
|
||||
assert r.status_code == 201, r.text
|
||||
m = r.json()
|
||||
assert m["nome"] == "Manuale" and m["metodo"] == "manuale"
|
||||
assert m["px_per_mm"] == pytest.approx(20.0) and m["mm_per_px"] == pytest.approx(0.05)
|
||||
assert (m["image_width"], m["image_height"]) == (64, 48)
|
||||
assert m["origine"] == SRC and m["square_mm"] is None and m["attiva_su"] == [] # non la attiva
|
||||
voci = sorgenti.get("/camera/calibrazioni").json()["calibrazioni"]
|
||||
assert [v["nome"] for v in voci] == ["Manuale"]
|
||||
|
||||
|
||||
def test_manuale_usa_il_frame_tenuto_se_c_e(sorgenti):
|
||||
assert _anteprima(sorgenti, SRC).status_code == 200 # tiene il grezzo 64x48
|
||||
r = _manuale(sorgenti)
|
||||
assert r.status_code == 201 and (r.json()["image_width"], r.json()["image_height"]) == (64, 48)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("valore", [0, -1, 0.000001, 100.5, "abc", None, True])
|
||||
def test_manuale_mm_non_valido_400(sorgenti, valore):
|
||||
r = _manuale(sorgenti, mm=valore)
|
||||
assert r.status_code == 400
|
||||
assert "mm/px deve essere un numero positivo" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_manuale_senza_source_o_nome_400_e_sorgente_ignota_404(sorgenti):
|
||||
assert sorgenti.post("/camera/calibrazioni/manuale", json={"nome": "x", "mm_per_px": 0.1}).status_code == 400
|
||||
assert _manuale(sorgenti, nome=" ").status_code == 400
|
||||
assert _manuale(sorgenti, source="nonesiste.png").status_code == 404
|
||||
|
||||
|
||||
def test_manuale_409_nome_doppio_e_sovrascrivi(sorgenti):
|
||||
assert _manuale(sorgenti, mm=0.05).status_code == 201
|
||||
r = _manuale(sorgenti, mm=0.1)
|
||||
assert r.status_code == 409 and "esiste già una calibrazione" in r.json()["detail"]
|
||||
r = _manuale(sorgenti, mm=0.1, sovrascrivi=True)
|
||||
assert r.status_code == 201 and r.json()["px_per_mm"] == pytest.approx(10.0)
|
||||
assert _manuale(sorgenti, mm=0.1, sovrascrivi="si").status_code == 400
|
||||
|
||||
|
||||
def test_attiva_manuale_ok_su_stessa_risoluzione_e_legge_px_per_mm(sorgenti):
|
||||
assert _manuale(sorgenti, mm=0.05).status_code == 201
|
||||
r = sorgenti.put("/camera/calibrazione-attiva", json={"source": SRC, "nome": "Manuale"})
|
||||
assert r.status_code == 200, r.text
|
||||
g = sorgenti.get("/camera/calibration", params={"source": SRC})
|
||||
assert g.status_code == 200 and g.json()["px_per_mm"] == pytest.approx(20.0)
|
||||
assert g.json()["nome"] == "Manuale" and g.json()["metodo"] == "manuale"
|
||||
|
||||
|
||||
def test_attiva_manuale_409_su_risoluzione_diversa(sorgenti):
|
||||
assert _manuale(sorgenti).status_code == 201
|
||||
r = sorgenti.put("/camera/calibrazione-attiva", json={"source": ALTRA, "nome": "Manuale"})
|
||||
assert r.status_code == 409 and "64×48" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_anteprima_calibrata_con_manuale_200_grezza(sorgenti):
|
||||
assert _manuale(sorgenti).status_code == 201
|
||||
assert sorgenti.put("/camera/calibrazione-attiva", json={"source": SRC, "nome": "Manuale"}).status_code == 200
|
||||
assert _anteprima(sorgenti, SRC).status_code == 200
|
||||
r = _calibrata(sorgenti, SRC)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.headers["X-Raddrizzato"] == "no" and r.headers["X-Calibrazione"] == "Manuale"
|
||||
assert (r.headers["X-Larghezza"], r.headers["X-Altezza"]) == ("64", "48")
|
||||
assert r.headers["content-type"] == "image/jpeg"
|
||||
|
||||
|
||||
def test_anteprima_calibrata_manuale_409_se_il_frame_ha_altra_risoluzione(sorgenti):
|
||||
assert _manuale(sorgenti).status_code == 201
|
||||
assert sorgenti.put("/camera/calibrazione-attiva", json={"source": SRC, "nome": "Manuale"}).status_code == 200
|
||||
assert _anteprima(sorgenti, ALTRA).status_code == 200
|
||||
# stessa sorgente attiva, ma il frame tenuto cambia (immagine sostituita con altra misura)
|
||||
assert _carica(sorgenti, SRC, np.full((60, 80), 9, np.uint8)).status_code == 201
|
||||
assert _anteprima(sorgenti, SRC).status_code == 200
|
||||
r = _calibrata(sorgenti, SRC)
|
||||
assert r.status_code == 409 and "64×48" in r.json()["detail"]
|
||||
|
||||
|
||||
# ---- pagina
|
||||
|
||||
def test_pagina_scheda_manuale_id_e_testi():
|
||||
ids = set(re.findall(r'id="([^"]+)"', PAGINA))
|
||||
assert {"mmPx", "pxMmCalc", "nomeManuale", "salvaManuale", "msgManuale"} <= ids
|
||||
tab = PAGINA.split('id="pannelloCal"')[1]
|
||||
assert tab.index('id="sezioneCalibrazione"') < tab.index('id="sezioneManuale"') < tab.index('id="sezioneSalvate"')
|
||||
assert "Calibrazione manuale" in PAGINA and "mm per pixel" in PAGINA
|
||||
assert "salvata: per usarla «Carica» o CONFERMA" in PAGINA
|
||||
assert "Sovrascrivi «" in PAGINA.split("async function salvaManuale")[1] # stessa conferma del Salva scacchiera
|
||||
assert "/camera/calibrazioni/manuale" in PAGINA and "alert(" not in PAGINA and "confirm(" not in PAGINA
|
||||
|
||||
|
||||
def test_pagina_manuale_senza_sorgente_pulsante_spento_con_nota():
|
||||
stato = PAGINA.split("function statoManuale")[1].split("function aggiornaManuale")[0]
|
||||
assert 'salvaManuale").disabled = !s' in stato and "Scegliere una sorgente" in stato
|
||||
assert "statoManuale()" in PAGINA.split("async function aggiornaStato")[1].split("async function caricaCalibrazioni")[0]
|
||||
|
||||
|
||||
def test_pagina_px_mm_calcolato_a_ogni_tasto():
|
||||
assert '$("mmPx").oninput = aggiornaManuale' in PAGINA
|
||||
assert '"= " + fmt(1 / v, 4) + " px/mm"' in PAGINA
|
||||
|
||||
|
||||
def test_pagina_metodo_manuale_e_campi_null():
|
||||
assert 'c.metodo === "densa" || c.metodo === "manuale" ? c.metodo : "planare"' in PAGINA
|
||||
righe = PAGINA.split("function righe")[1].split("function tabella")[0]
|
||||
assert 'd.metodo === "manuale"' in righe and "Lato dello scacco" in righe # la manuale ha il suo elenco, le altre il solito
|
||||
assert 'v === null || v === undefined' in PAGINA.split("function fmt")[1].split("function fmtCasella")[0]
|
||||
stato = PAGINA.split("async function aggiornaStato")[1].split("async function caricaCalibrazioni")[0]
|
||||
assert "Number(d.square_mm) > 0" in stato
|
||||
|
||||
|
||||
def test_pagina_calibrata_senza_raddrizzamento_dice_solo_scala():
|
||||
assert "solo scala, nessuna correzione ottica" in PAGINA
|
||||
assert 'r.headers.get("X-Raddrizzato") === "no"' in PAGINA
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Test dell'archivio unico di calibrazioni con nome (#91): scripts/calibrazione.py, classe Archivio.
|
||||
|
||||
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.
|
||||
"""
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from test_calibrazione import _carica, una_scacchiera_densa
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def mappa_densa():
|
||||
# #91: una mappa vera, calcolata una volta sola per tutti i test del modulo
|
||||
_carica()
|
||||
import calibrazione
|
||||
return calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), 2.0)[1]
|
||||
|
||||
|
||||
def _dati(**extra):
|
||||
base = {"px_per_mm": 20.0, "mm_per_px": 0.05, "image_width": 640, "image_height": 480,
|
||||
"metodo": "planare", "square_mm": 2.0, "calibrated_at": "2026-10-24T10:00:00+00:00",
|
||||
"origine": "file:s.png"}
|
||||
return {**base, **extra}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cal():
|
||||
_carica()
|
||||
import calibrazione
|
||||
return calibrazione
|
||||
|
||||
|
||||
# ---- nomi
|
||||
|
||||
@pytest.mark.parametrize("nome", ["", " ", "x" * 61, "a\x00b", "a\nb", "a\tb", None, 5])
|
||||
def test_nome_non_valido(cal, tmp_path, nome):
|
||||
with pytest.raises(cal.NomeNonValido):
|
||||
cal.Archivio(tmp_path).salva_nome(nome, _dati())
|
||||
|
||||
|
||||
def test_nome_valido_limiti_e_strip(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("x" * 60, _dati())
|
||||
meta = arch.salva_nome(" Banco 1 ", _dati())
|
||||
assert meta["nome"] == "Banco 1" # strip
|
||||
assert arch.leggi_nome("Banco 1")["nome"] == "Banco 1"
|
||||
|
||||
|
||||
def test_nome_unico_senza_maiuscole(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("Banco", _dati(px_per_mm=1.0))
|
||||
with pytest.raises(cal.NomeGiaUsato, match="Banco"):
|
||||
arch.salva_nome("BANCO", _dati(px_per_mm=2.0))
|
||||
assert arch.leggi_nome("banco")["px_per_mm"] == 1.0 # non toccata
|
||||
assert len(arch.elenco()) == 1
|
||||
|
||||
|
||||
def test_sovrascrivi_sostituisce_e_prende_il_nuovo_nome(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("Banco", _dati(px_per_mm=1.0))
|
||||
arch.salva_nome("BANCO", _dati(px_per_mm=2.0), sovrascrivi=True)
|
||||
voci = arch.elenco()
|
||||
assert len(voci) == 1 and voci[0]["nome"] == "BANCO" and voci[0]["px_per_mm"] == 2.0
|
||||
assert len(list((tmp_path / "calibrazioni").glob("*.json"))) == 1
|
||||
|
||||
|
||||
def test_salva_non_scrive_source_ne_overlay(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("A", _dati(source="camera:x", overlay_png="abc", saved=True))
|
||||
scritto = json.loads(next((tmp_path / "calibrazioni").glob("*.json")).read_text(encoding="utf-8"))
|
||||
assert "source" not in scritto and "overlay_png" not in scritto and "saved" not in scritto
|
||||
assert scritto["nome"] == "A" and scritto["origine"] == "file:s.png"
|
||||
|
||||
|
||||
def test_leggi_nome_inesistente_o_non_valido(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
assert arch.leggi_nome("nessuna") is None and arch.leggi_nome("") is None
|
||||
|
||||
|
||||
# ---- elenco e attive
|
||||
|
||||
def test_elenco_dalla_piu_recente_e_attiva_su(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("vecchia", _dati(calibrated_at="2026-01-01T00:00:00+00:00"))
|
||||
arch.salva_nome("nuova", _dati(calibrated_at="2026-10-01T00:00:00+00:00"))
|
||||
arch.attiva("camera:b", "nuova")
|
||||
arch.attiva("camera:a", "nuova")
|
||||
voci = arch.elenco()
|
||||
assert [v["nome"] for v in voci] == ["nuova", "vecchia"]
|
||||
assert voci[0]["attiva_su"] == ["camera:a", "camera:b"] and voci[1]["attiva_su"] == []
|
||||
assert set(voci[0]) == {"nome", "origine", "image_width", "image_height", "px_per_mm",
|
||||
"mm_per_px", "metodo", "square_mm", "calibrated_at", "attiva_su"}
|
||||
|
||||
|
||||
def test_attiva_leggi_scollega(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("Banco", _dati())
|
||||
assert arch.attiva_di("camera:a") is None and arch.leggi("camera:a") is None
|
||||
assert arch.attiva("camera:a", "banco") == "Banco" # nome come in libreria
|
||||
assert arch.attiva_di("camera:a") == "Banco"
|
||||
letta = arch.leggi("camera:a")
|
||||
assert letta["nome"] == "Banco" and letta["source"] == "camera:a" and letta["px_per_mm"] == 20.0
|
||||
assert json.loads((tmp_path / "attive.json").read_text(encoding="utf-8")) == {"camera:a": "Banco"}
|
||||
arch.scollega("camera:a")
|
||||
assert arch.attiva_di("camera:a") is None and arch.leggi("camera:a") is None
|
||||
assert arch.leggi_nome("Banco") is not None # la libreria resta
|
||||
arch.scollega("camera:a") # idempotente
|
||||
|
||||
|
||||
def test_attiva_nome_sconosciuto(cal, tmp_path):
|
||||
with pytest.raises(cal.NomeSconosciuto):
|
||||
cal.Archivio(tmp_path).attiva("camera:a", "nessuna")
|
||||
|
||||
|
||||
def test_due_sorgenti_stessa_calibrazione(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("Comune", _dati())
|
||||
arch.attiva("camera:a", "Comune")
|
||||
arch.attiva("camera:b", "Comune")
|
||||
assert arch.leggi("camera:a")["source"] == "camera:a" and arch.leggi("camera:b")["source"] == "camera:b"
|
||||
|
||||
|
||||
def test_elimina_nome_toglie_da_attive_e_mappa(cal, tmp_path, mappa_densa):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("Densa", _dati(metodo="densa"), mappa_densa)
|
||||
arch.salva_nome("Altra", _dati())
|
||||
arch.attiva("camera:a", "Densa")
|
||||
arch.attiva("camera:b", "Densa")
|
||||
arch.attiva("camera:c", "Altra")
|
||||
assert arch.leggi_mappa("camera:a") is not None
|
||||
assert arch.elimina_nome("DENSA") is True
|
||||
assert arch.leggi_nome("Densa") is None and not list((tmp_path / "calibrazioni").glob("*.npz"))
|
||||
assert arch.attiva_di("camera:a") is None and arch.attiva_di("camera:b") is None
|
||||
assert arch.leggi_mappa("camera:a") is None
|
||||
assert arch.attiva_di("camera:c") == "Altra"
|
||||
assert json.loads((tmp_path / "attive.json").read_text(encoding="utf-8")) == {"camera:c": "Altra"}
|
||||
assert arch.elimina_nome("Densa") is False # gia' tolta
|
||||
|
||||
|
||||
# ---- mappa
|
||||
|
||||
def test_leggi_mappa_risolve_l_attiva_con_cache(cal, tmp_path, mappa_densa):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
assert arch.leggi_mappa("camera:a") is None
|
||||
arch.salva_nome("Densa", _dati(metodo="densa"), mappa_densa)
|
||||
assert arch.leggi_mappa("camera:a") is None # non attiva su camera:a
|
||||
arch.attiva("camera:a", "Densa")
|
||||
prima = arch.leggi_mappa("camera:a")
|
||||
assert prima is not None and arch.leggi_mappa("camera:a") is prima # cache in memoria
|
||||
assert arch.leggi_mappa("camera:b") is None
|
||||
assert not list(tmp_path.rglob("*.tmp*")) # niente temporanei rimasti
|
||||
arch.salva_nome("Densa", _dati(metodo="densa"), mappa_densa, sovrascrivi=True) # file riscritto
|
||||
assert arch.leggi_mappa("camera:a") is not prima # cache invalidata
|
||||
arch.salva_nome("Densa", _dati(), sovrascrivi=True) # sovrascritta con una planare: la mappa sparisce
|
||||
assert arch.leggi_mappa("camera:a") is None and not list(tmp_path.rglob("*.npz"))
|
||||
|
||||
|
||||
# ---- migrazione dal formato vecchio
|
||||
|
||||
def _vecchia(cartella, cal, source, mappa=None, **extra):
|
||||
"""Scrive una calibrazione nel formato vecchio (un JSON per sorgente, + .npz) come faceva Archivio.salva."""
|
||||
dati = {"source": source, **_dati(**extra)}
|
||||
dati.pop("origine")
|
||||
nudo = cal.Archivio.__new__(cal.Archivio) # senza costruttore: niente migrazione mentre si prepara
|
||||
nudo.cartella = cartella
|
||||
percorso = nudo._file(source)
|
||||
cartella.mkdir(parents=True, exist_ok=True)
|
||||
percorso.write_text(json.dumps(dati), encoding="utf-8")
|
||||
if mappa is not None:
|
||||
mappa.save(str(percorso.with_suffix(".npz")))
|
||||
return percorso
|
||||
|
||||
|
||||
def test_migrazione_formato_vecchio(cal, tmp_path, mappa_densa):
|
||||
v1 = _vecchia(tmp_path, cal, "camera:601020", mappa_densa, metodo="densa", px_per_mm=7.0)
|
||||
v2 = _vecchia(tmp_path, cal, "camera:folder", px_per_mm=9.0)
|
||||
(tmp_path / "settaggi").mkdir()
|
||||
(tmp_path / "settaggi" / v1.name).write_text("{}", encoding="utf-8") # non si tocca
|
||||
arch = cal.Archivio(tmp_path)
|
||||
assert not v1.exists() and not v1.with_suffix(".npz").exists() and not v2.exists()
|
||||
assert {v["nome"] for v in arch.elenco()} == {"camera:601020", "camera:folder"}
|
||||
assert arch.attiva_di("camera:601020") == "camera:601020" and arch.attiva_di("camera:folder") == "camera:folder"
|
||||
letta = arch.leggi("camera:601020")
|
||||
assert letta["px_per_mm"] == 7.0 and letta["source"] == "camera:601020"
|
||||
assert arch.leggi_mappa("camera:601020") is not None # il .npz ha seguito il JSON
|
||||
assert len(list((tmp_path / "calibrazioni").glob("*.npz"))) == 1
|
||||
assert (tmp_path / "settaggi" / v1.name).read_text(encoding="utf-8") == "{}"
|
||||
|
||||
|
||||
def test_migrazione_idempotente(cal, tmp_path):
|
||||
_vecchia(tmp_path, cal, "camera:a")
|
||||
cal.Archivio(tmp_path)
|
||||
prima = sorted(p.name for p in tmp_path.rglob("*") if p.is_file())
|
||||
arch = cal.Archivio(tmp_path) # secondo avvio
|
||||
arch.migra()
|
||||
assert sorted(p.name for p in tmp_path.rglob("*") if p.is_file()) == prima
|
||||
assert [v["nome"] for v in arch.elenco()] == ["camera:a"]
|
||||
|
||||
|
||||
def test_migrazione_nome_gia_preso(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("camera:a", _dati(px_per_mm=1.0, calibrated_at="2020-01-01T00:00:00+00:00"))
|
||||
_vecchia(tmp_path, cal, "camera:a", px_per_mm=5.0)
|
||||
arch.migra()
|
||||
nomi = {v["nome"]: v["px_per_mm"] for v in arch.elenco()}
|
||||
assert nomi == {"camera:a": 1.0, "camera:a (2)": 5.0}
|
||||
assert arch.attiva_di("camera:a") == "camera:a (2)" # la sorgente usa la sua, non quella gia' presente
|
||||
# un altro file con lo stesso nome -> (3)
|
||||
_vecchia(tmp_path, cal, "camera:A", px_per_mm=6.0) # nome uguale senza maiuscole
|
||||
arch.migra()
|
||||
assert "camera:A (3)" in {v["nome"] for v in arch.elenco()}
|
||||
|
||||
|
||||
def test_migrazione_ignora_json_senza_source_e_attive(cal, tmp_path):
|
||||
tmp_path.mkdir(exist_ok=True)
|
||||
(tmp_path / "altro.json").write_text(json.dumps({"x": 1}), encoding="utf-8")
|
||||
(tmp_path / "rotto.json").write_text("{non json", encoding="utf-8")
|
||||
arch = cal.Archivio(tmp_path)
|
||||
arch.salva_nome("A", _dati())
|
||||
arch.attiva("source", "A") # una sorgente chiamata «source» non deve far scambiare attive.json per un vecchio file
|
||||
arch.migra()
|
||||
assert (tmp_path / "altro.json").exists() and (tmp_path / "rotto.json").exists()
|
||||
assert arch.attiva_di("source") == "A" and [v["nome"] for v in arch.elenco()] == ["A"]
|
||||
|
||||
|
||||
def test_cartella_assente_non_da_errore(cal, tmp_path):
|
||||
arch = cal.Archivio(tmp_path / "non" / "esiste")
|
||||
assert arch.elenco() == [] and arch.leggi("s") is None and arch.leggi_mappa("s") is None
|
||||
arch.scollega("s")
|
||||
assert arch.elimina_nome("x") is False
|
||||
@@ -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()
|
||||
@@ -0,0 +1,292 @@
|
||||
"""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"
|
||||
@@ -29,10 +29,10 @@ def test_offline_e_senza_finestre_del_browser():
|
||||
|
||||
def test_id_principali():
|
||||
ids = set(re.findall(r'id="([^"]+)"', PAGINA))
|
||||
attesi = {"sorgente", "avvia", "ferma", "scatta", "grezzo", "mirinoCasella", "immagine",
|
||||
attesi = {"sorgente", "avvia", "ferma", "scatta", "mirinoCasella", "immagine",
|
||||
"nitidezza", "nitidezzaMax", "saturati", "media", "dimensioni",
|
||||
"parametri", "salvaParametri", "quadrato", "acquisisci", "scegliFile", "file",
|
||||
"salva", "stato", "statoSorgente"} # #91: carica/elimina sono ora per riga, nell'elenco
|
||||
"parametri", "salvaParametri", "quadrato", "acquisisci", "caricaImmagine", "fileImmagine",
|
||||
"mostraCal", "salva", "stato", "statoSorgente"} # #91: carica/elimina sono ora per riga; #92: file e calibrata nell'Acquisizione
|
||||
assert attesi <= ids, attesi - ids
|
||||
# ogni id usato dal JS con $("...") esiste nel markup
|
||||
usati = set(re.findall(r'\$\("([^"]+)"\)', PAGINA))
|
||||
@@ -41,8 +41,8 @@ def test_id_principali():
|
||||
|
||||
def test_sezioni_e_testi_italiani():
|
||||
for testo in ("Sorgente", "Anteprima", "Parametri camera", "Calibrazione",
|
||||
"Salva come predefiniti", "Scegli immagine...", "Questa sorgente non ha parametri regolabili",
|
||||
"Immagine grezza (senza correzione)"):
|
||||
"Salva come predefiniti", "Carica immagine da file...", "Questa sorgente non ha parametri regolabili",
|
||||
"Attiva calibrazione"): # #92 #93
|
||||
assert testo in PAGINA, testo
|
||||
assert "it-IT" in PAGINA # numeri con la virgola
|
||||
assert "Choose File" not in PAGINA
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Test della pagina a due tab (#91): contenuto statico di PAGINA, senza servizio."""
|
||||
"""Test della pagina a due tab (#91, #92): contenuto statico di PAGINA, senza servizio."""
|
||||
|
||||
import importlib.util
|
||||
import re
|
||||
@@ -20,37 +20,58 @@ def test_due_tab_con_ruolo_e_selezione():
|
||||
assert '"tmf.acq.tab"' in PAGINA # tab scelta ricordata (via memoria(), protetta)
|
||||
|
||||
|
||||
def test_anteprima_fuori_dai_pannelli():
|
||||
# l'anteprima sta nella colonna sinistra, prima dei due pannelli: visibile in entrambe le tab
|
||||
i_ant = PAGINA.index('id="sezioneAnteprima"')
|
||||
def test_sorgente_in_cima_poi_riquadri():
|
||||
# #92: Sorgente, anteprima e parametri stanno nel pannello Acquisizione; Sorgente prima dei riquadri
|
||||
i_acq = PAGINA.index('id="pannelloAcq"')
|
||||
i_cal = PAGINA.index('id="pannelloCal"')
|
||||
assert i_ant < i_acq < i_cal
|
||||
# Sorgente e Parametri nel pannello Acquisizione, la calibrazione in quello Calibrazione
|
||||
assert i_acq < PAGINA.index('id="sezioneSorgente"') < PAGINA.index('id="sezioneParametri"') < i_cal
|
||||
i_sorg = PAGINA.index('id="sezioneSorgente"')
|
||||
i_ant = PAGINA.index('id="sezioneAnteprima"')
|
||||
i_riq = PAGINA.index('id="riquadri"')
|
||||
assert i_acq < i_sorg < i_ant < i_riq < PAGINA.index('id="sezioneParametri"') < i_cal
|
||||
assert PAGINA.index('id="sezioneCalibrazione"') > i_cal
|
||||
# barra: casella calibrata e mirino; la vecchia casella «Immagine grezza» non c'e' piu'
|
||||
assert 'id="mostraCal"' in PAGINA and "Attiva calibrazione" in PAGINA
|
||||
assert 'id="grezzo"' not in PAGINA and "Immagine grezza (senza correzione)" not in PAGINA
|
||||
# riquadri: «Immagine» sempre visibile, «Calibrata» nascosta finche' la casella e' spenta
|
||||
assert re.search(r'<div class="riquadro" id="riquadroCal" hidden>', PAGINA)
|
||||
assert "grezzo: true" in PAGINA
|
||||
|
||||
|
||||
def test_id_del_pannello_calibrazione():
|
||||
ids = set(re.findall(r'id="([^"]+)"', PAGINA))
|
||||
attesi = {"calSorgente", "origSorgente", "origFile", "scegliFile", "file", "quadrato", "acquisisci",
|
||||
"nomeCal", "salva", "msgSalva", "calibrazioni", "msgElenco", "stato", "risultato", "overlay"}
|
||||
attesi = {"calSorgente", "quadrato", "acquisisci", "nomeCal", "salva", "msgSalva", "calibrazioni",
|
||||
"msgElenco", "stato", "risultato", "overlay",
|
||||
"caricaImmagine", "fileImmagine", "rimuoviImmagine", "riquadroCal", "immagineCal", "mirinoCal"}
|
||||
assert attesi <= ids, attesi - ids
|
||||
usati = set(re.findall(r'\$\("([^"]+)"\)', PAGINA))
|
||||
assert usati <= ids, usati - ids
|
||||
assert len(re.findall(r'id="([^"]+)"', PAGINA)) == len(ids) # nessun id doppio
|
||||
|
||||
|
||||
def test_calibrazione_senza_file_ne_radio():
|
||||
# #92: l'immagine da file e' una sorgente di acquisizione, non un'opzione della calibrazione
|
||||
for sparito in ("origFile", "origSorgente", "scegliFile", "nomeFile", 'type="radio"', "Immagine da file", "Sorgente selezionata"):
|
||||
assert sparito not in PAGINA, sparito
|
||||
# l'unico input file e' nella sezione Sorgente, in cima alla tab Acquisizione
|
||||
assert PAGINA.count('type="file"') == 1
|
||||
assert PAGINA.index('type="file"') < PAGINA.index('id="pannelloCal"')
|
||||
assert "immagine grezza dalla tab Acquisizione" in PAGINA
|
||||
# il calcolo manda solo la sorgente (e il lato), mai il file
|
||||
assert 'f.append("source", sorgente)' in PAGINA and 'f.append("image"' not in PAGINA.split("async function calcola")[1].split("async function salvaNome")[0]
|
||||
|
||||
|
||||
def test_testi_del_pannello():
|
||||
for testo in ("Sorgente selezionata", "Immagine da file", "Scegli immagine...", "Calcola",
|
||||
"Calibrazioni salvate", "si cambia nella tab Acquisizione", "risoluzione diversa",
|
||||
"Carica", "Elimina", "Scollega", "Sovrascrivi «", "Clic ancora", "Nome"):
|
||||
for testo in ("Camera o immagine", "Carica immagine da file...", "Rimuovi immagine", "Calcola",
|
||||
"Calibrazioni salvate", "risoluzione diversa", "Calibrata",
|
||||
"Carica", "Elimina", "Scollega", "Sovrascrivi «", "Clic ancora", "Nome",
|
||||
"calibrata: ", "non calibrata"):
|
||||
assert testo in PAGINA, testo
|
||||
|
||||
|
||||
def test_rotte_del_contratto():
|
||||
for rotta in ("/camera/calibrazioni", "/camera/calibrazione-attiva", "/camera/calibrate",
|
||||
"/camera/calibration?source="):
|
||||
"/camera/calibration?source=", "/camera/immagini", "/camera/immagini?nome=",
|
||||
"/camera/anteprima/calibrata"): # #92
|
||||
assert '"' + rotta in PAGINA, rotta
|
||||
for parola in ("calcolo_id", "sovrascrivi", "method: \"PUT\"", "method: \"DELETE\"", "X-Larghezza", "X-Altezza"):
|
||||
assert parola in PAGINA, parola
|
||||
@@ -65,11 +86,58 @@ def test_nessuna_finestra_del_browser_e_offline():
|
||||
|
||||
def test_acquisizione_di_87_90_conservata():
|
||||
# anteprima, indicatori, mirino, grezzo, parametri camera e ciclo a 100 ms restano
|
||||
for parola in ("Parametri camera", "Immagine grezza (senza correzione)", "Mirino centrale",
|
||||
for parola in ("Parametri camera", "Mirino centrale",
|
||||
"Math.max(100 - (Date.now() - t0), 0)", "/camera/anteprima", "/camera/parametri"):
|
||||
assert parola in PAGINA, parola
|
||||
|
||||
|
||||
def test_pagina_stretta():
|
||||
assert "@media (max-width: 500px)" in PAGINA
|
||||
assert "overflow-x: auto" in PAGINA # l'elenco scorre di lato invece di rompere la pagina
|
||||
# #91: elenco a schede, niente tabella larga ne' scorrimento orizzontale
|
||||
assert 'id="calibrazioni" class="elenco"' in PAGINA and "overflow-x" not in PAGINA
|
||||
assert "Lettura dei parametri..." in PAGINA
|
||||
|
||||
|
||||
def test_ciclo_non_sospeso_nella_tab_calibrazione():
|
||||
# #94: il ciclo non si ferma piu' entrando in Calibrazione (prima #92: A.riprendi); Calcola senza blocco aspetta le richieste in volo
|
||||
assert "A.riprendi" not in PAGINA
|
||||
calcola = PAGINA.split("async function calcola")[1].split("async function salvaNome")[0]
|
||||
assert "fermaAnteprima();" in calcola and "A.inVolo) await pausa(50)" in calcola
|
||||
|
||||
|
||||
def test_calibrata_dopo_anteprima_e_dopo_carica():
|
||||
# #92: la calibrata segue ogni anteprima grezza riuscita e si aggiorna dopo Carica/Scollega; 404/409 al posto dell'immagine
|
||||
assert "await aggiornaCalibrata(lato);" in PAGINA
|
||||
assert PAGINA.count("aggiornaCalibrata();") >= 3
|
||||
assert "messaggioCalibrata(await dettaglio(r))" in PAGINA
|
||||
assert 'decodeURIComponent(r.headers.get("X-Calibrazione")' in PAGINA
|
||||
|
||||
|
||||
def test_elenco_sorgenti_senza_attese_e_calibrata_solo_se_serve():
|
||||
# #92: la voce dell'immagine caricata si aggiunge in pagina (niente rilettura dell'elenco); «Aggiorna» forza la ricerca camere
|
||||
assert "aggiungiSorgente(j);" in PAGINA and "caricaSorgenti(j.source)" not in PAGINA
|
||||
assert 'id="aggiornaSorgenti"' in PAGINA and ">Aggiorna</button>" in PAGINA
|
||||
assert '"?aggiorna=true"' in PAGINA
|
||||
# nessuna calibrazione attiva: il riquadro lo dice senza chiamare il servizio; dopo un 409 non si richiede piu'
|
||||
assert "if (A.calAttiva === false)" in PAGINA
|
||||
assert "nessuna calibrazione attiva su questa sorgente" in PAGINA
|
||||
assert "if (r.status === 409) A.bloccoCal = true;" in PAGINA and "if (A.bloccoCal) return;" in PAGINA
|
||||
assert PAGINA.count("A.bloccoCal = false;") >= 3 # cambio sorgente, Carica, Scollega
|
||||
|
||||
|
||||
def test_attiva_calibrazione_con_scelta_e_conferma():
|
||||
# #93: sotto la calibrata una combo e CONFERMA; la casella cambia nome
|
||||
for i in ('id="sceltaCal"', 'id="confermaCal"', 'id="msgSceltaCal"'):
|
||||
assert i in PAGINA, i
|
||||
assert "Attiva calibrazione" in PAGINA and ">CONFERMA</button>" in PAGINA
|
||||
assert "Mostra immagine calibrata" not in PAGINA
|
||||
assert PAGINA.index('id="riquadroCal"') < PAGINA.index('id="sceltaCal"') < PAGINA.index('id="confermaCal"') < PAGINA.index('id="msgSceltaCal"')
|
||||
# il gestore di CONFERMA usa PUT calibrazione-attiva (via caricaNome) o DELETE calibration (via scollega), messaggio sotto la calibrata
|
||||
gestore = PAGINA[PAGINA.index('$("confermaCal").onclick'):]
|
||||
gestore = gestore[:gestore.index("};")]
|
||||
assert 'caricaNome(nome, "msgSceltaCal")' in gestore and 'scollega("msgSceltaCal")' in gestore
|
||||
assert 'method: "PUT"' in PAGINA and "/camera/calibrazione-attiva" in PAGINA
|
||||
assert 'method: "DELETE"' in PAGINA and '"/camera/calibration?source="' in PAGINA
|
||||
# voce vuota «nessuna», suffisso di risoluzione, combo ridisegnata con l'elenco e spenta senza sorgente
|
||||
assert "— nessuna (non calibrata) —" in PAGINA and "— risoluzione diversa" in PAGINA
|
||||
assert "disegnaSceltaCal();" in PAGINA and "= !$(\"sorgente\").value" in PAGINA
|
||||
|
||||
@@ -541,6 +541,55 @@ def api_list_stations_for_maker():
|
||||
return jsonify(out), 200
|
||||
|
||||
|
||||
def _camera_services() -> list[dict]:
|
||||
"""#97: servizi di acquisizione distinti delle stazioni che hanno una camera_url."""
|
||||
services: dict[str, list[str]] = {}
|
||||
for st in _stations_for_maker():
|
||||
url = (st.get("camera_url") or "").strip().rstrip("/")
|
||||
if not url:
|
||||
continue
|
||||
label = f"{st.get('code') or ''} - {st.get('name') or ''}"
|
||||
services.setdefault(url, []).append(label)
|
||||
return [{"camera_url": u, "stazioni": s} for u, s in services.items()]
|
||||
|
||||
|
||||
@maker_bp.route("/api/camera/servizi", methods=["GET"])
|
||||
@login_required
|
||||
@maker_or_admin_required
|
||||
def api_camera_services():
|
||||
"""#97: servizi di acquisizione fra cui scegliere la camera di un task."""
|
||||
return jsonify({"servizi": _camera_services()}), 200
|
||||
|
||||
|
||||
@maker_bp.route("/api/camera/sorgenti", methods=["GET"])
|
||||
@login_required
|
||||
@maker_or_admin_required
|
||||
def api_camera_sources():
|
||||
"""#97: proxy di GET {url}/camera/sources; solo per i servizi delle stazioni."""
|
||||
url = (request.args.get("url") or "").strip().rstrip("/")
|
||||
# Niente richieste verso indirizzi arbitrari: url ammesso solo se e' un servizio noto
|
||||
if not url or url not in {s["camera_url"] for s in _camera_services()}:
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": _("Servizio di acquisizione non ammesso"),
|
||||
}), 400
|
||||
try:
|
||||
resp = http_requests.get(f"{url}/camera/sources", timeout=15)
|
||||
body = resp.json() if resp.status_code == 200 else None
|
||||
except (http_requests.RequestException, ValueError) as exc:
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": _("Servizio camera non raggiungibile: %(detail)s", detail=str(exc)),
|
||||
}), 502
|
||||
sources = body.get("sources") if isinstance(body, dict) else None
|
||||
if not isinstance(sources, list):
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": _("Risposta del servizio camera non valida (%(status)s)", status=resp.status_code),
|
||||
}), 502
|
||||
return jsonify({"sources": sources}), 200
|
||||
|
||||
|
||||
# Copia minima della logica di measure.api_vision_grab, ma per una stazione scelta per id
|
||||
@maker_bp.route("/api/stations/<int:station_id>/camera/grab", methods=["GET", "POST"])
|
||||
@login_required
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
// «Acquisizione» dell'editor task (#97): camera (servizio + sorgente) e modo di analisi
|
||||
// («impulsiva» / «in linea») dei task camera_measure e dxf_compare.
|
||||
// Legge /maker/api/camera/servizi e /maker/api/camera/sorgenti?url=. Lo stato scelto sta
|
||||
// nel componente padre (editTaskData.acquisition): questo componente carica solo gli elenchi.
|
||||
// Uso: x-data="taskAcquisition(() => editTaskData.acquisition, msgs)". Dipende da Alpine.js.
|
||||
function taskAcquisition(getAcq, msgs) {
|
||||
return {
|
||||
msgs: msgs || {},
|
||||
servizi: [],
|
||||
sources: [],
|
||||
loading: false,
|
||||
error: '',
|
||||
|
||||
async init() {
|
||||
await this.loadServizi();
|
||||
// Task con camera gia' scelta: elenco sorgenti del suo servizio
|
||||
const acq = getAcq();
|
||||
if (acq && acq.camera_url) await this.loadSources(acq.camera_url);
|
||||
},
|
||||
|
||||
async loadServizi() {
|
||||
try {
|
||||
const resp = await fetch('/maker/api/camera/servizi');
|
||||
if (resp.ok) this.servizi = (await resp.json()).servizi || [];
|
||||
} catch (e) { console.error('taskAcquisition servizi', e); }
|
||||
},
|
||||
|
||||
// Servizio salvato nel task ma non piu' fra quelli delle stazioni: resta selezionabile
|
||||
serviziConCorrente() {
|
||||
const acq = getAcq();
|
||||
const out = this.servizi.slice();
|
||||
if (acq && acq.camera_url && !out.some(s => s.camera_url === acq.camera_url)) {
|
||||
out.push({ camera_url: acq.camera_url, stazioni: [] });
|
||||
}
|
||||
return out;
|
||||
},
|
||||
|
||||
// Cambio servizio: la sorgente scelta non vale piu'
|
||||
async onServizio() {
|
||||
const acq = getAcq();
|
||||
acq.camera_source = '';
|
||||
this.sources = [];
|
||||
this.error = '';
|
||||
if (acq.camera_url) await this.loadSources(acq.camera_url);
|
||||
},
|
||||
|
||||
async loadSources(url) {
|
||||
this.loading = true;
|
||||
this.error = '';
|
||||
try {
|
||||
const resp = await fetch('/maker/api/camera/sorgenti?url=' + encodeURIComponent(url));
|
||||
const data = await resp.json().catch(() => ({}));
|
||||
if (resp.ok) this.sources = data.sources || [];
|
||||
else this.error = data.detail || this.msgs.genericError || '';
|
||||
} catch (e) {
|
||||
this.error = this.msgs.genericError || String(e);
|
||||
} finally { this.loading = false; }
|
||||
},
|
||||
|
||||
// Sorgente salvata ma non piu' elencata dal servizio: resta visibile
|
||||
sourcesConCorrente() {
|
||||
const acq = getAcq();
|
||||
const out = this.sources.slice();
|
||||
if (acq && acq.camera_source && !out.some(s => s.name === acq.camera_source)) {
|
||||
out.push({ name: acq.camera_source, label: acq.camera_source });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -547,6 +547,71 @@
|
||||
|
||||
<div class="tmf-card-body space-y-4">
|
||||
|
||||
<!-- ========== Acquisizione (#97): camera e modo di analisi, solo camera_measure / dxf_compare ========== -->
|
||||
<template x-if="task.task_type === 'camera_measure' || task.task_type === 'dxf_compare'">
|
||||
<div class="p-3 rounded-lg bg-[var(--bg-secondary)] border border-[var(--border-color)]">
|
||||
<p class="text-xs font-semibold text-[var(--text-secondary)] uppercase tracking-wide mb-2">
|
||||
{{ _('Acquisizione') }}
|
||||
</p>
|
||||
<!-- Lettura: camera e modo salvati nel task -->
|
||||
<template x-if="editingTask !== task.id">
|
||||
<div class="text-xs space-y-1">
|
||||
<template x-if="task.acquisition_json && task.acquisition_json.camera_url">
|
||||
<div class="space-y-1">
|
||||
<p><span class="text-[var(--text-secondary)]">{{ _('Servizio di acquisizione') }}:</span>
|
||||
<span class="font-mono" x-text="task.acquisition_json.camera_url"></span></p>
|
||||
<p><span class="text-[var(--text-secondary)]">{{ _('Sorgente') }}:</span>
|
||||
<span class="font-mono" x-text="task.acquisition_json.camera_source"></span></p>
|
||||
<p><span class="text-[var(--text-secondary)]">{{ _('Modo di analisi') }}:</span>
|
||||
<span x-text="task.acquisition_json.modo === 'in_linea' ? {{ _('Analisi in linea')|tojson_attr }} : {{ _('Analisi impulsiva')|tojson_attr }}"></span></p>
|
||||
</div>
|
||||
</template>
|
||||
<p x-show="!(task.acquisition_json && task.acquisition_json.camera_url)" class="text-[var(--text-muted)]">
|
||||
{{ _("Camera della stazione dell'operatore") }}
|
||||
</p>
|
||||
</div>
|
||||
</template>
|
||||
<!-- Modifica: servizio, sorgente, modo (salvati con «Salva» del task) -->
|
||||
<template x-if="editingTask === task.id">
|
||||
<div class="space-y-3"
|
||||
x-data="taskAcquisition(() => editTaskData.acquisition, {{ {'genericError': _('Errore durante il caricamento delle sorgenti')}|tojson_attr }})">
|
||||
<div class="grid grid-cols-1 sm:grid-cols-2 gap-3">
|
||||
<div>
|
||||
<label class="tmf-label text-xs">{{ _('Servizio di acquisizione') }}</label>
|
||||
<select class="tmf-input text-xs" x-model="editTaskData.acquisition.camera_url" @change="onServizio()">
|
||||
<option value="">{{ _("Camera della stazione dell'operatore") }}</option>
|
||||
<template x-for="sv in serviziConCorrente()" :key="sv.camera_url">
|
||||
<option :value="sv.camera_url"
|
||||
x-text="sv.camera_url + (sv.stazioni.length ? ' (' + sv.stazioni.join(', ') + ')' : '')"></option>
|
||||
</template>
|
||||
</select>
|
||||
</div>
|
||||
<div x-show="editTaskData.acquisition.camera_url">
|
||||
<label class="tmf-label text-xs">{{ _('Sorgente') }}</label>
|
||||
<select class="tmf-input text-xs" x-model="editTaskData.acquisition.camera_source" :disabled="loading">
|
||||
<option value="">{{ _('Scegli la sorgente') }}</option>
|
||||
<template x-for="src in sourcesConCorrente()" :key="src.name">
|
||||
<option :value="src.name" x-text="src.label || src.name"></option>
|
||||
</template>
|
||||
</select>
|
||||
</div>
|
||||
</div>
|
||||
<p x-show="error" class="text-xs text-red-500" x-text="error"></p>
|
||||
<div x-show="editTaskData.acquisition.camera_url" class="flex flex-wrap items-center gap-4 text-xs">
|
||||
<label class="flex items-center gap-2">
|
||||
<input type="radio" :name="'acq-modo-' + task.id" value="impulsiva" x-model="editTaskData.acquisition.modo">
|
||||
{{ _('Analisi impulsiva') }}
|
||||
</label>
|
||||
<label class="flex items-center gap-2">
|
||||
<input type="radio" :name="'acq-modo-' + task.id" value="in_linea" x-model="editTaskData.acquisition.modo">
|
||||
{{ _('Analisi in linea') }}
|
||||
</label>
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
</div>
|
||||
</template>
|
||||
|
||||
<!-- ========== Ricetta visione (#46): solo camera_measure / dxf_compare ========== -->
|
||||
<template x-if="task.task_type === 'camera_measure' || task.task_type === 'dxf_compare'">
|
||||
<div class="p-3 rounded-lg bg-[var(--bg-secondary)] border border-[var(--border-color)]"
|
||||
@@ -1335,6 +1400,7 @@
|
||||
<script src="{{ url_for('static', filename='js/vision_recipe.js') }}"></script>
|
||||
<script src="{{ url_for('static', filename='js/station_calibration.js') }}"></script>
|
||||
<script src="{{ url_for('static', filename='js/dxf_recipe_wizard.js') }}"></script>
|
||||
<script src="{{ url_for('static', filename='js/task_acquisition.js') }}"></script>{# #97 #}
|
||||
<script src="{{ url_for('static', filename='vendor/pdf-3.11.174.min.js') }}"></script>
|
||||
<script>if(typeof pdfjsLib!=='undefined')pdfjsLib.GlobalWorkerOptions.workerSrc='{{ url_for('static', filename='vendor/pdf.worker-3.11.174.min.js') }}';</script>
|
||||
<script src="{{ url_for('static', filename='js/annotation-viewer.js') }}?v=6"></script>
|
||||
@@ -1389,7 +1455,7 @@ function taskEditor() {
|
||||
|
||||
// ---- Task CRUD state ----
|
||||
editingTask: null,
|
||||
editTaskData: { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] },
|
||||
editTaskData: { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [], acquisition: { camera_url: '', camera_source: '', modo: 'impulsiva' } }, // #97
|
||||
newTask: { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] },
|
||||
MAX_NOTE_FIELDS: 20, // #55: limite campi per nota (stesso del backend)
|
||||
_fieldUid: 0,
|
||||
@@ -1575,6 +1641,18 @@ function taskEditor() {
|
||||
this.newTask = { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] };
|
||||
},
|
||||
|
||||
// #97: acquisition_json del task; null = camera della stazione (voce vuota o tipo senza camera)
|
||||
_acquisitionPayload(obj) {
|
||||
const a = obj.acquisition || {};
|
||||
const usesCamera = obj.task_type === 'camera_measure' || obj.task_type === 'dxf_compare';
|
||||
if (!usesCamera || !a.camera_url) return { ok: true, value: null };
|
||||
if (!a.camera_source) {
|
||||
this.errorMessage = {{ _("Scegli la sorgente della camera")|tojson }};
|
||||
return { ok: false };
|
||||
}
|
||||
return { ok: true, value: { camera_url: a.camera_url, camera_source: a.camera_source, modo: a.modo === 'in_linea' ? 'in_linea' : 'impulsiva' } };
|
||||
},
|
||||
|
||||
// ---- #53 #55: ciclo di misura e campi della nota ----
|
||||
|
||||
// #53: default di "nel ciclo" per tipo (stessa regola del backend)
|
||||
@@ -1785,7 +1863,13 @@ function taskEditor() {
|
||||
in_loop: task.in_loop != null ? !!task.in_loop : this.loopDefault(task.task_type || 'note'),
|
||||
in_loop_touched: false,
|
||||
at_close: !!task.at_close,
|
||||
fields: this._fieldsToState(task.fields)
|
||||
fields: this._fieldsToState(task.fields),
|
||||
// #97: camera e modo di analisi del task (acquisition_json), vuoti = camera della stazione
|
||||
acquisition: {
|
||||
camera_url: (task.acquisition_json && task.acquisition_json.camera_url) || '',
|
||||
camera_source: (task.acquisition_json && task.acquisition_json.camera_source) || '',
|
||||
modo: (task.acquisition_json && task.acquisition_json.modo) || 'impulsiva'
|
||||
}
|
||||
};
|
||||
// Ensure expanded
|
||||
this.expandedTask = task.id;
|
||||
@@ -1793,7 +1877,7 @@ function taskEditor() {
|
||||
|
||||
cancelEditTask() {
|
||||
this.editingTask = null;
|
||||
this.editTaskData = { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [] };
|
||||
this.editTaskData = { title: '', directive: '', description: '', task_type: 'note', in_loop: false, in_loop_touched: false, at_close: false, fields: [], acquisition: { camera_url: '', camera_source: '', modo: 'impulsiva' } }; // #97
|
||||
},
|
||||
|
||||
async updateTask(taskId) {
|
||||
@@ -1801,6 +1885,8 @@ function taskEditor() {
|
||||
this.errorMessage = '';
|
||||
const fp = this._fieldsPayload(this.editTaskData); // #55
|
||||
if (!fp.ok) return;
|
||||
const acq = this._acquisitionPayload(this.editTaskData); // #97
|
||||
if (!acq.ok) return;
|
||||
this.saving = true;
|
||||
|
||||
try {
|
||||
@@ -1818,7 +1904,8 @@ function taskEditor() {
|
||||
// #53 #55: ciclo di misura, compilazione a fine produzione, campi della nota
|
||||
in_loop: !!this.editTaskData.in_loop,
|
||||
at_close: this.editTaskData.task_type === 'note' && !!this.editTaskData.at_close,
|
||||
fields: fp.fields
|
||||
fields: fp.fields,
|
||||
acquisition_json: acq.value // #97
|
||||
})
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
"""gitea:Adriano/TieMeasureFlow#97 - camera e modo di analisi nel task visione (lato Maker).
|
||||
|
||||
api_client e requests sono finti: nessun servizio camera vero.
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import requests
|
||||
|
||||
from tests.conftest import api_get_router
|
||||
|
||||
ST1 = {"id": 4, "code": "ST-4", "name": "Banco", "camera_url": "http://acq:8765/", "camera_source": "camera:1"}
|
||||
ST2 = {"id": 5, "code": "ST-5", "name": "Linea", "camera_url": "http://acq:8765", "camera_source": "camera:2"}
|
||||
ST3 = {"id": 6, "code": "ST-6", "name": "Altra", "camera_url": "http://altro:9000", "camera_source": "x"}
|
||||
NO_CAMERA = {"id": 7, "code": "ST-7", "name": "Senza", "camera_url": None, "camera_source": None}
|
||||
SOURCES = {"sources": [
|
||||
{"name": "camera:601020", "kind": "camera", "label": "Basler 601020"},
|
||||
{"name": "pezzo.png", "kind": "image", "label": "immagine - pezzo.png"},
|
||||
]}
|
||||
|
||||
|
||||
def _stations(mock_api_client, stations=(ST1, ST2, ST3, NO_CAMERA)):
|
||||
mock_api_client.get.side_effect = api_get_router({"/api/stations": list(stations)})
|
||||
|
||||
|
||||
def _fake(status=200, body=None):
|
||||
resp = MagicMock(status_code=status, text="")
|
||||
if body is None:
|
||||
resp.json.side_effect = ValueError("no json")
|
||||
else:
|
||||
resp.json.return_value = body
|
||||
return resp
|
||||
|
||||
|
||||
class TestCameraServizi:
|
||||
def test_distinct_urls_with_stations(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
resp = logged_in_client.get("/maker/api/camera/servizi")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json() == {"servizi": [
|
||||
{"camera_url": "http://acq:8765", "stazioni": ["ST-4 - Banco", "ST-5 - Linea"]},
|
||||
{"camera_url": "http://altro:9000", "stazioni": ["ST-6 - Altra"]},
|
||||
]}
|
||||
|
||||
def test_no_stations_with_camera(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client, [NO_CAMERA])
|
||||
assert logged_in_client.get("/maker/api/camera/servizi").get_json() == {"servizi": []}
|
||||
|
||||
|
||||
class TestCameraSorgenti:
|
||||
def test_lists_sources(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
with patch("blueprints.maker.http_requests.get", return_value=_fake(200, SOURCES)) as get:
|
||||
resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765/")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json() == SOURCES
|
||||
args, kwargs = get.call_args
|
||||
assert args[0] == "http://acq:8765/camera/sources"
|
||||
assert kwargs["timeout"] == 15
|
||||
|
||||
def test_url_not_a_station_service_is_400(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
with patch("blueprints.maker.http_requests.get") as get:
|
||||
resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://evil:1")
|
||||
assert resp.status_code == 400
|
||||
get.assert_not_called()
|
||||
|
||||
def test_missing_url_is_400(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
with patch("blueprints.maker.http_requests.get") as get:
|
||||
assert logged_in_client.get("/maker/api/camera/sorgenti").status_code == 400
|
||||
get.assert_not_called()
|
||||
|
||||
def test_service_down_is_502(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
with patch("blueprints.maker.http_requests.get",
|
||||
side_effect=requests.ConnectionError("refused")):
|
||||
resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765")
|
||||
assert resp.status_code == 502
|
||||
assert "refused" in resp.get_json()["detail"]
|
||||
|
||||
def test_service_error_status_is_502(self, logged_in_client, mock_api_client):
|
||||
_stations(mock_api_client)
|
||||
with patch("blueprints.maker.http_requests.get", return_value=_fake(500, {"detail": "x"})):
|
||||
resp = logged_in_client.get("/maker/api/camera/sorgenti?url=http://acq:8765")
|
||||
assert resp.status_code == 502
|
||||
|
||||
|
||||
class TestTaskProxyForwardsAcquisition:
|
||||
def test_put_forwards_acquisition_json(self, logged_in_client, mock_api_client):
|
||||
acq = {"camera_url": "http://acq:8765", "camera_source": "camera:1", "modo": "in_linea"}
|
||||
mock_api_client.put.return_value = {"id": 11, "title": "T", "acquisition_json": acq}
|
||||
resp = logged_in_client.put("/maker/api/tasks/11", json={"title": "T", "acquisition_json": acq})
|
||||
assert resp.status_code == 200
|
||||
assert mock_api_client.put.call_args.kwargs["data"]["acquisition_json"] == acq
|
||||
|
||||
def test_put_forwards_null(self, logged_in_client, mock_api_client):
|
||||
mock_api_client.put.return_value = {"id": 11, "title": "T", "acquisition_json": None}
|
||||
logged_in_client.put("/maker/api/tasks/11", json={"title": "T", "acquisition_json": None})
|
||||
assert mock_api_client.put.call_args.kwargs["data"]["acquisition_json"] is None
|
||||
|
||||
|
||||
class TestTaskEditorAcquisitionUi:
|
||||
def _html(self, client, mock_api_client, task_type="camera_measure"):
|
||||
mock_api_client.get.side_effect = api_get_router({
|
||||
"/api/recipes/3": {"id": 3, "code": "R", "name": "R"},
|
||||
"/api/recipes/3/tasks": [{
|
||||
"id": 11, "title": "T", "task_type": task_type, "subtasks": [],
|
||||
"file_path": None, "order_index": 1,
|
||||
"acquisition_json": {"camera_url": "http://acq:8765", "camera_source": "camera:1",
|
||||
"modo": "in_linea"},
|
||||
}],
|
||||
})
|
||||
return client.get("/maker/recipes/3/tasks").get_data(as_text=True)
|
||||
|
||||
def test_section_and_fields(self, logged_in_client, mock_api_client):
|
||||
html = self._html(logged_in_client, mock_api_client)
|
||||
assert "taskAcquisition(" in html
|
||||
assert "js/task_acquisition.js" in html
|
||||
assert "Servizio di acquisizione" in html
|
||||
assert "Sorgente" in html
|
||||
assert "Analisi impulsiva" in html and "Analisi in linea" in html
|
||||
assert 'value="impulsiva"' in html and 'value="in_linea"' in html
|
||||
assert "Camera della stazione dell'operatore" in html or \
|
||||
"Camera della stazione dell'operatore" in html
|
||||
|
||||
def test_payload_carries_acquisition_json(self, logged_in_client, mock_api_client):
|
||||
html = self._html(logged_in_client, mock_api_client)
|
||||
assert "acquisition_json: acq.value" in html
|
||||
assert "_acquisitionPayload(this.editTaskData)" in html
|
||||
# il task salvato riempie lo stato di modifica
|
||||
assert "task.acquisition_json.camera_url" in html
|
||||
|
||||
def test_static_js_endpoints(self):
|
||||
import os
|
||||
path = os.path.join(os.path.dirname(__file__), "..", "static", "js", "task_acquisition.js")
|
||||
src = open(path, encoding="utf-8").read()
|
||||
assert "/maker/api/camera/servizi" in src
|
||||
assert "/maker/api/camera/sorgenti?url=" in src
|
||||
Reference in New Issue
Block a user