diff --git a/scripts/calibrazione.py b/scripts/calibrazione.py index 9c21070..bf78ca5 100644 --- a/scripts/calibrazione.py +++ b/scripts/calibrazione.py @@ -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: /calibrazioni/-.json (+ .npz se + densa, stesso nome). Attive: /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 ) 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 /*.json con "source" (una per sorgente) entra in libreria. + + Nome = valore di "source" (gia' preso: " (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 diff --git a/scripts/tests/test_calibrazione.py b/scripts/tests/test_calibrazione.py index 5b31c26..b2a0d62 100644 --- a/scripts/tests/test_calibrazione.py +++ b/scripts/tests/test_calibrazione.py @@ -287,14 +287,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 diff --git a/scripts/tests/test_calibrazioni_archivio.py b/scripts/tests/test_calibrazioni_archivio.py new file mode 100644 index 0000000..48705bc --- /dev/null +++ b/scripts/tests/test_calibrazioni_archivio.py @@ -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