src/benchmark.py

"""
Label-grounded retrieval benchmark + seasonal-vs-permanent error analysis.

Replaces the old identity-diagonal ``train.evaluate_retrieval`` hack with a
real information-retrieval evaluation: a fixed natural-language query set,
each query mapped to a *relevance rule* over the dataset's ``PairLabel``s
(derived from DEN's pixel-wise LULC labels). We then rank the whole pair
corpus with a :class:`~src.retrieval.ChangeRetriever` and compute Recall@K
and mAP, plus an error report on "semantic drift" (seasonal transitions —
e.g. snow-melt — wrongly retrieved for permanent-change queries).

CLI:
    python -m src.benchmark --root data/DynamicEarthNet --encoder clip_vitl14 \
        --approach zero_shot
"""
from __future__ import annotations

import argparse
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional

import numpy as np

from src.datasets.base import PairLabel, TemporalDataset
from src.embeddings import cache_tag_for, load_or_compute
from src.encoders import get_encoder
from src.retrieval import APPROACHES, ChangeRetriever
from src.stats import rank_order

# Used by ``_is_seasonal`` to flag snow-involving transitions in the
# seasonal-drift report. Datasets without snow labels are unaffected (no
# pair matches, mask is all-False).
SNOW = "snow_and_ice"


# Prompt templates for zero-shot text ensembling. Averaging a frozen encoder's
# text embedding over several phrasings is a standard, training-free way to
# stabilise CLIP retrieval (the query is short and prompt-sensitive). Stays
# within the brief: frozen backbone, cosine scoring, no fine-tuning.
PROMPT_TEMPLATES = (
    "{q}",
    "a satellite image showing {q}",
    "remote sensing imagery of {q}",
    "an aerial view of {q}",
    "land-cover change: {q}",
)


def encode_query(encoder, text: str, ensemble: bool = False,
                 templates=PROMPT_TEMPLATES) -> np.ndarray:
    """Return an L2-normalised text embedding for *text*. With ``ensemble``,
    average the encoder's embeddings over ``templates`` and renormalise."""
    if not ensemble:
        return encoder.encode_text(text)[0].astype(np.float32)
    embs = encoder.encode_text([t.format(q=text) for t in templates]).astype(np.float32)
    v = embs.mean(axis=0)
    n = float(np.linalg.norm(v))
    return v / n if n > 1e-8 else v


def _t1(lb: PairLabel) -> Optional[str]:
    return lb.dominant_t1_class


def _t2(lb: PairLabel) -> Optional[str]:
    return lb.dominant_t2_class


@dataclass
class Query:
    """A natural-language query + a predicate deciding pair relevance.

    ``category`` is ``"permanent"`` or ``"seasonal"`` — used by the error
    analysis to measure seasonal/permanent confusion.
    """

    text: str
    category: str
    predicate: Callable[[PairLabel], bool]


def _transition(src=None, dst=None) -> Callable[[PairLabel], bool]:
    def pred(lb: PairLabel) -> bool:
        if lb is None or lb.stable:
            return False
        if _t1(lb) == _t2(lb):
            return False
        if src is not None and _t1(lb) != src:
            return False
        if dst is not None and _t2(lb) != dst:
            return False
        return True
    return pred


# --- fraction-based relevance --------------------------------------------------
# The default ``_transition`` predicate only fires when the *dominant* class of
# the whole tile flips — discarding all localised change (a 10% wetland gain on
# an agriculture-majority tile reads as "stable"). Empirically only 8.6% of DEN
# pairs flip dominant class, almost all wetland<->agriculture, which starves
# every other query of positives. These predicates instead use the per-class
# pixel-change fractions ``derive_pair_label`` already computes, so a pair is
# relevant when the target class gains/loses >= ``thresh`` of valid pixels.

def _gained(cls: str, thresh: float = 0.05) -> Callable[[PairLabel], bool]:
    def pred(lb: PairLabel) -> bool:
        if lb is None:
            return False
        return lb.class_change_mask_fraction.get(cls, {}).get("gained_fraction", 0.0) >= thresh
    return pred


def _lost(cls: str, thresh: float = 0.05) -> Callable[[PairLabel], bool]:
    def pred(lb: PairLabel) -> bool:
        if lb is None:
            return False
        return lb.class_change_mask_fraction.get(cls, {}).get("lost_fraction", 0.0) >= thresh
    return pred


# Per-dataset query sets live in ``src/queries/`` (one module per dataset).
# They self-register into the query registry and are resolved by
# ``run_benchmark`` via ``dataset.name``.


# Bump when the serialized JSON layout (``to_dict``) changes incompatibly.
SCHEMA_VERSION = 1


@dataclass
class QueryResult:
    text: str
    category: str
    n_relevant: int
    recall_at_k: Dict[int, float]
    ap: float
    seasonal_drift_at_k: Dict[int, float]  # frac of non-relevant top-K that are seasonal

    def to_dict(self) -> Dict:
        # K stored as *string* keys for JSON stability.
        return {
            "text": self.text,
            "category": self.category,
            "n_relevant": int(self.n_relevant),
            "recall_at_k": {str(k): float(v) for k, v in self.recall_at_k.items()},
            "ap": float(self.ap),
            "seasonal_drift_at_k": {str(k): float(v)
                                    for k, v in self.seasonal_drift_at_k.items()},
        }

    @classmethod
    def from_dict(cls, d: Dict) -> "QueryResult":
        return cls(
            text=d["text"],
            category=d["category"],
            n_relevant=int(d["n_relevant"]),
            recall_at_k={int(k): float(v) for k, v in d["recall_at_k"].items()},
            ap=float(d["ap"]),
            seasonal_drift_at_k={int(k): float(v)
                                 for k, v in d["seasonal_drift_at_k"].items()},
        )


@dataclass
class BenchmarkReport:
    approach: str
    encoder: str
    dataset: str
    n_pairs: int
    per_query: List[QueryResult]

    @property
    def macro_recall(self) -> Dict[int, float]:
        ks = self.per_query[0].recall_at_k.keys() if self.per_query else []
        return {
            k: float(np.mean([q.recall_at_k[k] for q in self.per_query]))
            for k in ks
        }

    @property
    def mAP(self) -> float:
        return float(np.mean([q.ap for q in self.per_query])) if self.per_query else 0.0

    @property
    def _ks(self) -> List[int]:
        return sorted(self.per_query[0].recall_at_k) if self.per_query else []

    def macro_seasonal_drift(self) -> Dict[int, float]:
        """Mean seasonal-drift@K over *permanent* queries (mirrors ``to_table``).

        0.0 for every K when there are no permanent queries (or no snow class).
        """
        perm = [q for q in self.per_query if q.category == "permanent"]
        if not perm:
            return {k: 0.0 for k in self._ks}
        return {k: float(np.mean([q.seasonal_drift_at_k[k] for q in perm]))
                for k in self._ks}

    def to_dict(self, *, color_mode: str = "rgb", split: Optional[str] = None,
                lora: bool = False) -> Dict:
        ks = self._ks
        mr = self.macro_recall
        sd = self.macro_seasonal_drift()
        return {
            "schema_version": SCHEMA_VERSION,
            "dataset": self.dataset,
            "encoder": self.encoder,
            "approach": self.approach,
            "split": split,
            "color_mode": color_mode,
            "lora": bool(lora),
            "n_pairs": int(self.n_pairs),
            "k_values": [int(k) for k in ks],
            "macro": {
                "mAP": self.mAP,
                "recall_at_k": {str(k): mr[k] for k in ks},
                "seasonal_drift_at_k": {str(k): sd[k] for k in ks},
            },
            "per_query": [q.to_dict() for q in self.per_query],
        }

    @classmethod
    def from_dict(cls, d: Dict) -> "BenchmarkReport":
        """Reconstruct the core report. Top-level ``split``/``color_mode``/``lora``
        metadata stays in the source dict (read directly by ``results_io``)."""
        return cls(
            approach=d["approach"],
            encoder=d["encoder"],
            dataset=d["dataset"],
            n_pairs=int(d["n_pairs"]),
            per_query=[QueryResult.from_dict(q) for q in d["per_query"]],
        )

    def to_table(self) -> str:
        ks = sorted(self.per_query[0].recall_at_k) if self.per_query else []
        head = (f"\n=== Benchmark: {self.dataset} | {self.encoder} | "
                f"approach={self.approach} | N={self.n_pairs} ===")
        cols = "  ".join(f"R@{k}" for k in ks)
        lines = [head, f"{'query':52s} {'#rel':>4s}  {cols}  {'AP':>5s}"]
        for q in self.per_query:
            r = "  ".join(f"{q.recall_at_k[k]:.2f}" for k in ks)
            lines.append(f"{q.text[:52]:52s} {q.n_relevant:4d}  {r}  {q.ap:5.3f}")
        mr = self.macro_recall
        lines.append("-" * len(lines[1]))
        lines.append(f"{'MACRO':52s} {'':>4s}  "
                     + "  ".join(f"{mr[k]:.2f}" for k in ks)
                     + f"  {self.mAP:5.3f}  (mAP)")
        # seasonal drift summary over permanent queries
        perm = [q for q in self.per_query if q.category == "permanent"]
        if perm:
            sd = {k: float(np.mean([q.seasonal_drift_at_k[k] for q in perm]))
                  for k in ks}
            lines.append("seasonal drift @K (permanent queries, lower=better): "
                         + "  ".join(f"R@{k}={sd[k]:.2f}" for k in ks))
        return "\n".join(lines)


def _average_precision(rel: np.ndarray) -> float:
    """rel: boolean array in ranked order. Standard AP."""
    n_rel = int(rel.sum())
    if n_rel == 0:
        return 0.0
    hits = np.cumsum(rel)
    ranks = np.arange(1, len(rel) + 1)
    precision_at_hit = (hits / ranks)[rel]
    return float(precision_at_hit.sum() / n_rel)


def _is_seasonal(lb: Optional[PairLabel]) -> bool:
    if lb is None:
        return False
    return SNOW in (lb.dominant_t1_class, lb.dominant_t2_class)


def run_benchmark(
    dataset: TemporalDataset,
    retriever: ChangeRetriever,
    approach: str = "zero_shot",
    queries: Optional[List[Query]] = None,
    k_values=(1, 3, 5, 10),
) -> BenchmarkReport:
    if queries is None:
        from src.queries import get_queries
        queries = get_queries(dataset.name)
        if not queries:
            raise ValueError(
                f"No query set registered for dataset '{dataset.name}'. "
                "Add a module under src/queries/ that calls register_queries(...)"
            )
    pairs = retriever.store.pairs
    labels = [dataset.get_pair_label(p) for p in pairs]
    seasonal_mask = np.array([_is_seasonal(lb) for lb in labels])

    per_query: List[QueryResult] = []
    for q in queries:
        rel = np.array([bool(lb is not None and q.predicate(lb)) for lb in labels])
        if rel.sum() == 0:
            continue  # query has no positives in this corpus — not evaluable
        scores = retriever.score_all(q.text, approach=approach)
        order = rank_order(scores, rel)
        rel_ranked = rel[order]
        seas_ranked = seasonal_mask[order]

        recall, drift = {}, {}
        for k in k_values:
            topk = order[:k]
            recall[k] = float(rel[topk].sum() / rel.sum())
            non_rel = ~rel_ranked[:k]
            drift[k] = float(seas_ranked[:k][non_rel].mean()) if non_rel.any() else 0.0
        per_query.append(QueryResult(
            text=q.text, category=q.category, n_relevant=int(rel.sum()),
            recall_at_k=recall, ap=_average_precision(rel_ranked),
            seasonal_drift_at_k=drift,
        ))

    return BenchmarkReport(
        approach=approach, encoder=retriever.encoder.name,
        dataset=dataset.name, n_pairs=len(pairs), per_query=per_query,
    )


def main() -> None:
    ap = argparse.ArgumentParser(description="Label-grounded change-retrieval benchmark")
    ap.add_argument("--dataset", default="dynamic_earthnet")
    ap.add_argument("--root", default="data/DynamicEarthNet")
    ap.add_argument("--pairing", default="bimonthly",
                    choices=["bimonthly", "monthly", "seasonal-quartet"])
    ap.add_argument("--encoder", default="clip_vitl14")
    ap.add_argument("--approach", default="zero_shot",
                    choices=list(APPROACHES) + ["all"])
    ap.add_argument("--cache-dir", default="data/cache")
    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"],
                    help="Image colour mode (must match the cached embeddings).")
    args = ap.parse_args()

    from src.datasets.registry import build_dataset
    ds = build_dataset(args.dataset, root=args.root, pairing=args.pairing,
                       split=None if args.split == "all" else args.split,
                       color_mode=args.color_mode)
    enc = get_encoder(args.encoder)
    # Key the embedding cache by split + colour via the canonical helper, matching
    # train.py / run_pipeline.py. Without a tag this loaded the un-split-tagged
    # ("empty tag") cache regardless of --split, so a stale legacy cache could
    # silently feed the wrong split's pairs into the benchmark.
    store = load_or_compute(ds, enc, cache_dir=args.cache_dir,
                            cache_tag=cache_tag_for(args.split, args.color_mode))
    retriever = ChangeRetriever(store, enc)

    approaches = ["naive", "zero_shot"] if args.approach == "all" else [args.approach]
    for appr in approaches:
        if appr == "peft":
            print("PEFT requested but no adapter wired in CLI; skip (use src.train).")
            continue
        report = run_benchmark(ds, retriever, approach=appr)
        print(report.to_table())


if __name__ == "__main__":
    main()