""" Per-pair embedding computation + on-disk cache. For every bi-temporal pair ``(T1, T2)`` of a ``TemporalDataset`` this module runs the frozen image encoder on both timesteps and stores the resulting L2-normalised vectors ``f_T1, f_T2`` (shape ``[N, D]`` each, aligned with an ordered pair list). This is the artefact every retrieval/benchmark/training step consumes — it decouples the (slow, GPU) encoding pass from the (fast, CPU) scoring passes and makes runs reproducible. Cache file: ``/__[__]__pair_embeddings.npz`` where `` = [_][_lora]`` (see ``cache_tag_for``); the rgb default adds no colour suffix. CLI: python -m src.embeddings --dataset dynamic_earthnet \ --root data/DynamicEarthNet --encoder clip_vitl14 """ from __future__ import annotations import argparse from dataclasses import dataclass from pathlib import Path from typing import List import numpy as np import torch from src.datasets.base import PairKey, TemporalDataset from src.encoders import get_encoder from src.features import compute_change_feature def cache_path(cache_dir: str | Path, dataset_name: str, encoder_name: str, tag: str = "") -> Path: suffix = f"__{tag}" if tag else "" return Path(cache_dir) / f"{dataset_name}__{encoder_name}{suffix}__pair_embeddings.npz" def patch_cache_path(cache_dir: str | Path, dataset_name: str, encoder_name: str, tag: str = "") -> Path: """Path of the per-patch embedding cache (sibling of :func:`cache_path`), keyed by the same ``[_][_lora]`` tag so it never aliases another split/colour combination.""" suffix = f"__{tag}" if tag else "" return Path(cache_dir) / f"{dataset_name}__{encoder_name}{suffix}__patch_embeddings.npz" _KNOWN_COLORS = ("rgb", "nrg", "ndvi") def color_tag(color_mode: str = "rgb") -> str: """Canonical colour suffix for cache/artefact names: ``_`` (empty for the rgb default). Single source of truth shared by ``cache_tag_for`` and the figure/export scripts so the suffix can never drift between them.""" return f"_{color_mode}" if color_mode != "rgb" else "" def cache_tag_for(split: str, color_mode: str = "rgb", lora: bool = False) -> str: """Canonical embedding-cache tag: ``[_][_lora]``. Single source of truth for the cache-tag string, matching the layout the committed caches were written with by ``scripts.run_pipeline`` (rgb adds no colour suffix; lora appends ``_lora``). Importers: ``run_pipeline``, ``train``, ``cv_eval``, ``eval_rerank``, ``make_comparison_figure``, the ``embeddings`` CLI, ``app``, ``export_results`` (grep ``cache_tag_for`` for the authoritative list). Keeping this in one place avoids the historical ``test``+``rgb``->empty-tag drift. The tag is positional (no delimiter between fields), so a ``split`` whose name ended in ``_`` or ``_lora`` could alias another (split, color, lora) combination — e.g. ``("test_nrg", "rgb")`` and ``("test", "nrg")`` both → ``"test_nrg"``. Splits are a closed set ({train,val,test,all}) so this never happens in practice; we assert it to keep the collision impossible rather than merely improbable. """ if color_mode not in _KNOWN_COLORS: raise ValueError(f"unknown color_mode {color_mode!r}; expected one of {_KNOWN_COLORS}") if any(split.endswith(f"_{c}") for c in _KNOWN_COLORS) or split.endswith("_lora"): raise ValueError( f"split {split!r} ends with a colour/lora suffix and would alias another " "cache tag; rename the split." ) lora_tag = "_lora" if lora else "" return f"{split}{color_tag(color_mode)}{lora_tag}" @dataclass class PairEmbeddingStore: """Ordered pair list + aligned ``f_T1`` / ``f_T2`` matrices.""" dataset_name: str encoder_name: str embed_dim: int pairs: List[PairKey] f_t1: np.ndarray # [N, D] float32, L2-normalised f_t2: np.ndarray # [N, D] float32, L2-normalised def __len__(self) -> int: return len(self.pairs) def change_features(self, mode: str = "difference") -> np.ndarray: """Δf for every pair via :func:`src.features.compute_change_feature`.""" t1 = torch.from_numpy(self.f_t1) t2 = torch.from_numpy(self.f_t2) return compute_change_feature(t1, t2, mode=mode).numpy().astype(np.float32) def save(self, path: str | Path) -> None: path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) np.savez( path, f_t1=self.f_t1.astype(np.float32), f_t2=self.f_t2.astype(np.float32), loc=np.array([p.location_id for p in self.pairs]), t1=np.array([p.t1_key for p in self.pairs]), t2=np.array([p.t2_key for p in self.pairs]), dataset_name=np.array(self.dataset_name), encoder_name=np.array(self.encoder_name), embed_dim=np.array(self.embed_dim), ) @classmethod def load(cls, path: str | Path) -> "PairEmbeddingStore": d = np.load(path, allow_pickle=False) pairs = [ PairKey(str(l), str(a), str(b)) for l, a, b in zip(d["loc"], d["t1"], d["t2"]) ] return cls( dataset_name=str(d["dataset_name"]), encoder_name=str(d["encoder_name"]), embed_dim=int(d["embed_dim"]), pairs=pairs, f_t1=d["f_t1"].astype(np.float32), f_t2=d["f_t2"].astype(np.float32), ) def compute_pair_embeddings( dataset: TemporalDataset, encoder, batch_size: int = 32, ) -> PairEmbeddingStore: """Encode both timesteps of every pair. Pairs whose tiles fail to load are skipped (real DEN occasionally misses a monthly tile).""" pairs: List[PairKey] = [] imgs_t1, imgs_t2 = [], [] for pair in dataset.list_pairs(): try: a, b = dataset.load_pair_images(pair) except FileNotFoundError as exc: print(f" skip {pair}: {exc}") continue pairs.append(pair) imgs_t1.append(a) imgs_t2.append(b) if not pairs: raise RuntimeError("No loadable pairs in dataset.") print(f"Encoding {len(pairs)} pairs ({2 * len(pairs)} images) with " f"'{encoder.name}' on {encoder.device} ...") f_t1 = encoder.encode_image(imgs_t1, batch_size=batch_size).astype(np.float32) f_t2 = encoder.encode_image(imgs_t2, batch_size=batch_size).astype(np.float32) return PairEmbeddingStore( dataset_name=dataset.name, encoder_name=encoder.name, embed_dim=int(f_t1.shape[1]), pairs=pairs, f_t1=f_t1, f_t2=f_t2, ) def load_or_compute( dataset: TemporalDataset, encoder, cache_dir: str | Path = "data/cache", force: bool = False, batch_size: int = 32, cache_tag: str = "", ) -> PairEmbeddingStore: path = cache_path(cache_dir, dataset.name, encoder.name, tag=cache_tag) if path.exists() and not force: store = PairEmbeddingStore.load(path) expected = [tuple(p) for p in dataset.list_pairs()] # Order-sensitive list equality is intentional: f_t1/f_t2 rows are aligned # positionally to this pair list, so a reordered-but-equal pair set is NOT # safely reusable and must be recomputed (the worst case here is a redundant # recompute, never silent row/label misalignment). if [tuple(p) for p in store.pairs] == expected: print(f"Loaded {len(store)} pair embeddings (mode=reused) from cache: {path}") return store print(f"Cache {path} stale (pair set changed: " f"{len(store.pairs)} cached vs {len(expected)} expected) " "-- recomputing.") store = compute_pair_embeddings(dataset, encoder, batch_size=batch_size) store.save(path) print(f"Saved {len(store)} pair embeddings (mode=recomputed) -> {path}") return store # --------------------------------------------------------------------------- # Per-patch embedding cache (localised / patch-level retrieval, REPORT B.10) # --------------------------------------------------------------------------- # Patch-level Δ-similarity needs per-patch embeddings for the *whole* corpus. # Encoding them at first ``approach="patch"`` query stalls the UI for the length # of a full GPU pass over every pair. This store decouples that encode (slow, # GPU) from the scoring pass (fast, CPU) exactly as the pair store does — a warm # cache makes the first patch query instant (the instant-search precompute). # Rows are aligned positionally to an ordered pair list; the pair list is the # pair store's (so engine code can index ``patch_t1[i]`` with the same ``i`` it # uses on ``store.pairs[i]``). @dataclass class PatchEmbeddingStore: """Ordered pair list + aligned per-patch ``patch_t1`` / ``patch_t2`` tensors (shape ``[N, P, D]`` each).""" dataset_name: str encoder_name: str pairs: List[PairKey] patch_t1: np.ndarray # [N, P, D] float32 patch_t2: np.ndarray # [N, P, D] float32 def __len__(self) -> int: return len(self.pairs) def save(self, path: str | Path) -> None: path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) np.savez( path, patch_t1=self.patch_t1.astype(np.float32), patch_t2=self.patch_t2.astype(np.float32), loc=np.array([p.location_id for p in self.pairs]), t1=np.array([p.t1_key for p in self.pairs]), t2=np.array([p.t2_key for p in self.pairs]), dataset_name=np.array(self.dataset_name), encoder_name=np.array(self.encoder_name), ) @classmethod def load(cls, path: str | Path) -> "PatchEmbeddingStore": d = np.load(path, allow_pickle=False) pairs = [ PairKey(str(l), str(a), str(b)) for l, a, b in zip(d["loc"], d["t1"], d["t2"]) ] return cls( dataset_name=str(d["dataset_name"]), encoder_name=str(d["encoder_name"]), pairs=pairs, patch_t1=d["patch_t1"].astype(np.float32), patch_t2=d["patch_t2"].astype(np.float32), ) def compute_patch_embeddings( dataset: TemporalDataset, encoder, pairs: List[PairKey], batch_size: int = 32, progress: bool = False, ) -> PatchEmbeddingStore: """Encode per-patch embeddings for every pair in *pairs* (kept in the given order so the result aligns with the pair store). Chunks the GPU passes to bound host memory. Raises if the encoder exposes no patch tokens.""" pairs = list(pairs) if not pairs: raise RuntimeError("No pairs to encode patches for.") p1, p2 = [], [] chunks = range(0, len(pairs), batch_size) if progress: try: from tqdm import tqdm chunks = tqdm(chunks, desc=f"patch-encode {encoder.name}", unit="batch") except ImportError: pass for s in chunks: imgs1, imgs2 = [], [] for pk in pairs[s:s + batch_size]: try: im1, im2 = dataset.load_pair_images(pk) except FileNotFoundError as exc: # Do NOT skip (rows must stay positionally aligned to the pair store); fail with a # clear, actionable message instead of an opaque crash deep in the encoder loop. raise RuntimeError( f"patch encode: tile for pair {pk} is missing ({exc}); the pair set changed since " "the pair-embedding cache was built — delete the matching *__patch_embeddings.npz " "(and *__pair_embeddings.npz) and recompute.") from exc imgs1.append(im1) imgs2.append(im2) a = encoder.encode_image_patches(imgs1) b = encoder.encode_image_patches(imgs2) if a is None or b is None: raise RuntimeError( f"{encoder.name} exposes no patch tokens; pick zero_shot/naive/peft.") p1.append(np.asarray(a)) p2.append(np.asarray(b)) return PatchEmbeddingStore( dataset_name=dataset.name, encoder_name=encoder.name, pairs=pairs, patch_t1=np.concatenate(p1, axis=0).astype(np.float32), patch_t2=np.concatenate(p2, axis=0).astype(np.float32), ) def load_or_compute_patches( dataset: TemporalDataset, encoder, pairs: List[PairKey], cache_dir: str | Path = "data/cache", force: bool = False, batch_size: int = 32, cache_tag: str = "", progress: bool = False, ) -> PatchEmbeddingStore: """Reuse the on-disk patch cache when its pair list matches *pairs*, else recompute + save. Mirrors :func:`load_or_compute`'s order-sensitive guard: a reordered-but-equal pair set is recomputed rather than risk row/label misalignment.""" path = patch_cache_path(cache_dir, dataset.name, encoder.name, tag=cache_tag) expected = [tuple(p) for p in pairs] if path.exists() and not force: store = PatchEmbeddingStore.load(path) if [tuple(p) for p in store.pairs] == expected: print(f"Loaded {len(store)} patch embeddings (mode=reused) from cache: {path}") return store print(f"Patch cache {path} stale (pair set changed: " f"{len(store.pairs)} cached vs {len(expected)} expected) -- recomputing.") store = compute_patch_embeddings(dataset, encoder, pairs, batch_size=batch_size, progress=progress) store.save(path) print(f"Saved {len(store)} patch embeddings (mode=recomputed) -> {path}") return store def main() -> None: ap = argparse.ArgumentParser(description="Precompute per-pair embeddings cache") ap.add_argument("--dataset", default="dynamic_earthnet") ap.add_argument("--root", default="data/DynamicEarthNet", help="Dataset root (DEN) or ignored for cache-only datasets") ap.add_argument("--pairing", default="bimonthly", choices=["bimonthly", "monthly", "seasonal-quartet"]) ap.add_argument("--encoder", default="clip_vitl14") ap.add_argument("--cache-dir", default="data/cache") ap.add_argument("--batch-size", type=int, default=32) ap.add_argument("--split", default="test", help="DEN preprocessed split: train|val|test|all") ap.add_argument("--color-mode", default="rgb", choices=["rgb", "nrg", "ndvi"]) ap.add_argument("--force", action="store_true") args = ap.parse_args() from src.datasets.registry import build_dataset color_mode = args.color_mode ds = build_dataset( args.dataset, root=args.root, pairing=args.pairing, split=None if args.split == "all" else args.split, color_mode=color_mode, ) enc = get_encoder(args.encoder) cache_tag = cache_tag_for(args.split, color_mode) store = load_or_compute( ds, enc, cache_dir=args.cache_dir, force=args.force, batch_size=args.batch_size, cache_tag=cache_tag, ) print(f"dataset={store.dataset_name} encoder={store.encoder_name} " f"N={len(store)} D={store.embed_dim}") if __name__ == "__main__": main()