acquisitore: archivio unico di calibrazioni con nome, migrazione dal formato per sorgente (Refs #91)

Archivio: libreria calibrazioni/<nome>-<sha>.json (+npz), attive.json per sorgente,
salva_nome/leggi_nome/elimina_nome/attiva/scollega/attiva_di/elenco; leggi e
leggi_mappa risolvono la calibrazione attiva. Le rotte seguono nel commit successivo.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
This commit is contained in:
2026-10-24 16:06:05 +02:00
parent 828471d27a
commit e7d5489438
3 changed files with 488 additions and 39 deletions
+249 -34
View File
@@ -3,6 +3,9 @@
Cosa fa: da una foto di scacchiera piana calcola la scala (px/mm) con Cosa fa: da una foto di scacchiera piana calcola la scala (px/mm) con
TieVision-Core (visionsuite.calibration), e conserva un risultato per sorgente TieVision-Core (visionsuite.calibration), e conserva un risultato per sorgente
camera in un JSON (scrittura atomica). 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 #84: `calibra_scacchiera_densa(pixels, square_mm)` calibra da UNA foto di
scacchiera che riempie tutto il campo (ottica telecentrica): trova da sola la scacchiera che riempie tutto il campo (ottica telecentrica): trova da sola la
griglia, costruisce la mappa di distorsione (DistortionMap) e il mm/px griglia, costruisce la mappa di distorsione (DistortionMap) e il mm/px
@@ -24,6 +27,7 @@ import os
import re import re
import tempfile import tempfile
import threading import threading
import unicodedata # #91 caratteri di controllo nei nomi
from collections import deque from collections import deque
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
@@ -463,43 +467,87 @@ def ora_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds") return datetime.now(timezone.utc).isoformat(timespec="seconds")
class Archivio: class NomeNonValido(ValueError):
"""Un JSON per sorgente in `cartella`; nome file dalla sorgente ripulito. """#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): def __init__(self, cartella: Path | str):
self.cartella = Path(cartella) self.cartella = Path(cartella)
self._cache: dict = {} # percorso .npz -> (mtime_ns, MappaPronta) self._cache: dict = {} # percorso .npz -> (mtime_ns, MappaPronta)
self._lock = threading.Lock() 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: def _file(self, sorgente: str) -> Path:
# caratteri sicuri + hash dell'originale: due sorgenti diverse non si pestano # 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" pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", sorgente).strip("._") or "sorgente"
sigla = hashlib.sha1(sorgente.encode("utf-8")).hexdigest()[:8] sigla = hashlib.sha1(sorgente.encode("utf-8")).hexdigest()[:8]
return self.cartella / f"{pulito[:60]}-{sigla}.json" return self.cartella / f"{pulito[:60]}-{sigla}.json"
def _file_mappa(self, sorgente: str) -> Path: @property
return self._file(sorgente).with_suffix(".npz") # #84 stesso nome del JSON def _libreria(self) -> Path:
return self.cartella / "calibrazioni"
def leggi(self, sorgente: str) -> dict | None: def _file_nome(self, nome: str) -> Path:
try: # #91: stesso schema di _file, ma l'hash e' sul nome senza maiuscole/minuscole
return json.loads(self._file(sorgente).read_text(encoding="utf-8")) pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", nome).strip("._") or "calibrazione"
except (OSError, ValueError): sigla = hashlib.sha1(nome.casefold().encode("utf-8")).hexdigest()[:8]
return None # assente o illeggibile = non calibrata return self._libreria / f"{pulito[:60]}-{sigla}.json"
def salva(self, sorgente: str, dati: dict, mappa: DistortionMap | None = None) -> None: def _file_mappa_nome(self, nome: str) -> Path:
destinazione = self._file(sorgente) return self._file_nome(nome).with_suffix(".npz") # #84 stesso nome del JSON
self.cartella.mkdir(parents=True, exist_ok=True)
# #84: prima la mappa (o la rimozione di una vecchia), poi il JSON che la dichiara @property
if mappa is not None: def _file_attive(self) -> Path:
self._salva_mappa(sorgente, mappa) return self.cartella / "attive.json"
else:
self._togli_mappa(sorgente) # ricalibrazione planare: la vecchia mappa non vale piu' # ---- 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 # 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: try:
with os.fdopen(fd, "w", encoding="utf-8") as f: with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(dati, f, ensure_ascii=False, indent=2) json.dump(dati, f, ensure_ascii=False, indent=2)
@@ -511,9 +559,128 @@ class Archivio:
pass pass
raise raise
def _salva_mappa(self, sorgente: str, mappa: DistortionMap) -> None: def _leggi_json(self, percorso: Path) -> dict | None:
destinazione = self._file_mappa(sorgente) try:
fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp.npz") # .npz: numpy non aggiunge nulla 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) os.close(fd)
try: try:
mappa.save(tmp) mappa.save(tmp)
@@ -527,8 +694,8 @@ class Archivio:
with self._lock: with self._lock:
self._cache.pop(str(destinazione), None) self._cache.pop(str(destinazione), None)
def _togli_mappa(self, sorgente: str) -> None: def _togli_mappa(self, nome: str) -> None:
destinazione = self._file_mappa(sorgente) destinazione = self._file_mappa_nome(nome)
try: try:
destinazione.unlink() destinazione.unlink()
except FileNotFoundError: except FileNotFoundError:
@@ -536,9 +703,64 @@ class Archivio:
with self._lock: with self._lock:
self._cache.pop(str(destinazione), None) 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: def leggi_mappa(self, sorgente: str) -> MappaPronta | None:
"""La mappa salvata, pronta per `raddrizza`; None se la sorgente non ne ha una.""" """La mappa della calibrazione attiva, pronta per `raddrizza`; None se non ne ha una."""
percorso = self._file_mappa(sorgente) 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) chiave = str(percorso)
try: try:
mtime = percorso.stat().st_mtime_ns mtime = percorso.stat().st_mtime_ns
@@ -557,10 +779,3 @@ class Archivio:
with self._lock: with self._lock:
self._cache[chiave] = (mtime, pronta) self._cache[chiave] = (mtime, pronta)
return 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
+6 -5
View File
@@ -287,14 +287,15 @@ def test_archivio_mappa_cache_e_invalidazione(tmp_path):
_, mappa = calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), 2.0) _, mappa = calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), 2.0)
arch = calibrazione.Archivio(tmp_path) arch = calibrazione.Archivio(tmp_path)
assert arch.leggi_mappa("s") is None 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") prima = arch.leggi_mappa("s")
assert prima is not None and arch.leggi_mappa("s") is prima # cache in memoria 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 assert not list(tmp_path.rglob("*.tmp*")) # niente temporanei rimasti
arch.salva("s", {"a": 2}, mappa) # file riscritto -> cache invalidata arch.salva_nome("cal", {"a": 2}, mappa, sovrascrivi=True) # file riscritto -> cache invalidata
assert arch.leggi_mappa("s") is not prima assert arch.leggi_mappa("s") is not prima
arch.salva("s", {"a": 3}) # salvataggio senza mappa (planare): la vecchia sparisce 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.glob("*.npz")) assert arch.leggi_mappa("s") is None and not list(tmp_path.rglob("*.npz"))
# ---- #84: rotte con la calibrazione densa # ---- #84: rotte con la calibrazione densa
+233
View File
@@ -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