"""Sample gallery: ordered manifest + lazy thumbnail cache. Backs the ``GET /api/samples`` and ``GET /api/samples/{id}/thumbnail`` endpoints. The manifest lives at ``samples/manifest.json``; the actual ``.tif`` images sit beside it. Thumbnails are generated on first request and cached in memory for the process lifetime - no per-request disk I/O. """ from __future__ import annotations import json from pathlib import Path import cv2 THUMBNAIL_SIZE: int = 128 THUMBNAIL_JPEG_QUALITY: int = 85 class SampleLibrary: """In-memory sample registry built from ``manifest.json``.""" def __init__(self, base_dir: Path, entries: list[dict]) -> None: self._base_dir = base_dir self._entries: list[dict] = entries self._by_id: dict[str, dict] = {e["id"]: e for e in entries} self._thumb_cache: dict[str, bytes] = {} @classmethod def from_directory(cls, dir_path: str) -> "SampleLibrary": """Read ``/manifest.json`` and validate every referenced file. Raises: FileNotFoundError: If the manifest or any referenced ``.tif`` is missing. ValueError: If the manifest is malformed. """ base = Path(dir_path) manifest_path = base / "manifest.json" if not manifest_path.exists(): raise FileNotFoundError(f"Sample manifest missing: {manifest_path}") with manifest_path.open("r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict) or "samples" not in data: raise ValueError(f"Malformed manifest at {manifest_path}: missing 'samples' key.") entries: list[dict] = data["samples"] if not isinstance(entries, list) or not entries: raise ValueError(f"Malformed manifest at {manifest_path}: empty sample list.") seen_ids: set[str] = set() for e in entries: for k in ("id", "label", "organ", "filename"): if k not in e: raise ValueError(f"Sample entry missing '{k}': {e}") if e["id"] in seen_ids: raise ValueError(f"Duplicate sample id: {e['id']!r}") seen_ids.add(e["id"]) tif_path = base / e["filename"] if not tif_path.exists(): raise FileNotFoundError( f"Sample file missing for id={e['id']!r}: {tif_path}" ) return cls(base_dir=base, entries=entries) def list(self) -> list[dict]: """JSON-serializable list of sample metadata (without file paths).""" return [ {"id": e["id"], "label": e["label"], "organ": e["organ"]} for e in self._entries ] def load_bytes(self, sample_id: str) -> bytes: """Return raw file bytes for the given sample id. Raises: KeyError: If ``sample_id`` is not in the manifest. """ entry = self._by_id.get(sample_id) if entry is None: raise KeyError(sample_id) path = self._base_dir / entry["filename"] return path.read_bytes() def thumbnail(self, sample_id: str) -> bytes: """128x128 JPEG thumbnail bytes; built and cached on first request. Raises: KeyError: If ``sample_id`` is not in the manifest. """ cached = self._thumb_cache.get(sample_id) if cached is not None: return cached entry = self._by_id.get(sample_id) if entry is None: raise KeyError(sample_id) path = self._base_dir / entry["filename"] bgr = cv2.imread(str(path), cv2.IMREAD_COLOR) if bgr is None: raise ValueError(f"Could not decode sample image: {path}") h, w = bgr.shape[:2] side = min(h, w) y0 = (h - side) // 2 x0 = (w - side) // 2 crop = bgr[y0:y0 + side, x0:x0 + side] interp = cv2.INTER_AREA if side > THUMBNAIL_SIZE else cv2.INTER_CUBIC thumb = cv2.resize(crop, (THUMBNAIL_SIZE, THUMBNAIL_SIZE), interpolation=interp) ok, buf = cv2.imencode( ".jpg", thumb, [cv2.IMWRITE_JPEG_QUALITY, THUMBNAIL_JPEG_QUALITY] ) if not ok: raise RuntimeError(f"Could not JPEG-encode thumbnail for {sample_id}.") out = buf.tobytes() self._thumb_cache[sample_id] = out return out def has(self, sample_id: str) -> bool: """True if a sample with this id exists in the manifest.""" return sample_id in self._by_id def __len__(self) -> int: # pragma: no cover - convenience for logs return len(self._entries)