product-recognition-demo/recognization/code/eval_identity.py

"""Score identity lanes on labelled shelf facings. Architecture-neutral.

The original matcher is a candidate final architecture, not just a baseline.
Every lane is scored the same way, on the same crops, at the same resolutions:

    local          RootSIFT pyramid + OCR text lane (the shipped default)
    matcher        full-catalog SIFT/ORB/BRISK + RANSAC (the original demo)
    gallery        DINOv2 embedding, raw or with the trained head
    hybrid-verify  DINO shortlist, then the matcher re-ranks inside it

Reported per lane, as four separate ceilings rather than one conflated number:

    catalog ceiling      in-catalog facings with an enrolled view / all in-catalog
    identity ceiling     ungated top-1 rank correct on enrolled / enrolled
    decision ceiling     accepted-and-correct on enrolled / enrolled
    end-to-end coverage  accepted-and-correct / all in-catalog facings

plus precision over every accept — including accepts of unenrolled in-catalog
facings, which are false accepts and the most dangerous kind — the prediction
histogram, and the largest single-SKU share of accepts (the L09-apocalypse
detector). The hybrid lane also reports shortlist recall, because a true SKU
outside the shortlist is a hard ceiling no verifier can lift.

    python code/eval_identity.py \
        --labels docs/research/2026-09-28-recognition-architecture/evidence/image6_v2.json \
        --image add/image6.jpg --stock-dir stock-lopia-clean --lanes matcher --live

Labels are {"image", "facings": [{"box", "label"}]} with "unknown" for facings
not in the catalog and "exclude" for rows two label passes disagreed on. A
facing whose crop is a catalog template is refused as leakage, not scored:
those crops are how the store was enrolled.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import subprocess
import sys
import time
from collections import Counter
from typing import Any, Collection, Dict, List, Optional, Sequence, Tuple

current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
    sys.path.insert(0, current_dir)

import cv2  # noqa: E402
import numpy as np  # noqa: E402

from rebuild_catalog import sku_of  # noqa: E402

from identity import LOCAL_ACCEPT  # noqa: E402

UNKNOWN = "unknown"
# Two independent label passes disagreed here, so the row is not ground truth.
# Scoring it would be scoring a coin flip.
EXCLUDE = "exclude"
LANES = ("local", "matcher", "van", "gallery", "hybrid-verify")
# A single SKU above this share of accepts is the failure the last audit found:
# every dark package collapsing onto one enrolled product.
HUB_SHARE = 0.30
# Live frames arrive at this long side; the original demo classified a 2048 px
# still. Scoring both is how resolution stops being an invisible variable.
LIVE_LONG_SIDE = 1280

# Pixel equality is not a leak test: a template re-cut from the same photo with
# a different box passes a same-shape comparison. A 64x64 grayscale signature
# catches most of those, but it is a heuristic, not a proof — measured on
# image6_v2 against stock-lopia, re-cuts in the realistic 0.94..1.06 box band
# land 7..37 MAD from their facing's signature while the closest pair of
# genuinely distinct same-SKU facings is 23.1, so the distributions overlap and
# no threshold is clean. 20.0 catches 81% of that band with zero false
# positives on 5,886 genuine facing pairs and 7 clean packshots. The real fix is
# the enrollment protocol's session manifest (DECISION §8); this is the cheap
# backstop that runs without one.
LEAK_SIGNATURE_MAD = 20.0


def load_labels(path: str) -> Dict[str, Any]:
    with open(path) as handle:
        payload = json.load(handle)
    if not isinstance(payload, dict) or "facings" not in payload:
        raise SystemExit(f"{path}: expected {{'image', 'facings': [...]}}")
    return payload


def file_hash(path: str) -> Optional[str]:
    """SHA-256 of a file's bytes, or None when it cannot be read."""
    try:
        with open(path, "rb") as handle:
            return hashlib.sha256(handle.read()).hexdigest()
    except OSError:
        return None


def catalog_hash(stock_dir: str) -> Optional[str]:
    """SHA-256 over the catalog's CSV bytes and its image names.

    Image bytes are not hashed: a re-encoded view of the same product is the
    same catalog, and hashing every packshot would make the number move for
    reasons that do not change a single decision. A changed row or a different
    set of views does change it.
    """
    digest = hashlib.sha256()
    csv_path = os.path.join(stock_dir, "stock.csv")
    try:
        with open(csv_path, "rb") as handle:
            digest.update(handle.read())
    except OSError:
        digest.update(b"missing-csv")
    try:
        names = sorted(
            name for name in os.listdir(stock_dir)
            if os.path.splitext(name)[1].lower() in (".jpg", ".jpeg", ".png")
        )
    except OSError:
        names = []
    digest.update(b"\n")
    digest.update("\n".join(names).encode("utf-8"))
    return digest.hexdigest()


def code_revision() -> Optional[str]:
    """The checked-out revision, or None outside a git worktree."""
    try:
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=os.path.dirname(os.path.abspath(__file__)),
            capture_output=True, text=True, timeout=10,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    return result.stdout.strip() or None if result.returncode == 0 else None


def crop_of(image: Any, box: Sequence[int]) -> Any:
    x1, y1, x2, y2 = [int(v) for v in box]
    return image[max(0, y1):y2, max(0, x1):x2]


def at_live_resolution(image: Any, long_side: int = LIVE_LONG_SIDE) -> Any:
    height, width = image.shape[:2]
    if max(height, width) <= long_side:
        return image
    scale = long_side / float(max(height, width))
    return cv2.resize(image, (int(width * scale), int(height * scale)), interpolation=cv2.INTER_AREA)


def _signature(image: Any) -> Any:
    """64x64 grayscale signature. Cheap, and tolerant of a re-cut."""
    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) if image.ndim == 3 else image
    return cv2.resize(gray, (64, 64), interpolation=cv2.INTER_AREA).astype(np.float32)


def _signatures(image: Any) -> List[Any]:
    """The crop's signature and a centre 90% crop's.

    A re-cut box shifts the content inside the frame. Comparing a second,
    tighter crop against the template absorbs that shift: measured on the
    realistic re-cut band it catches 81% of re-cuts against 68% for the single
    signature, at no cost in false positives.
    """
    height, width = image.shape[:2]
    dy, dx = int(height * 0.05), int(width * 0.05)
    centre = image[dy:height - dy, dx:width - dx]
    return [_signature(image), _signature(centre if centre.size else image)]


_TEMPLATES: Dict[str, List[Tuple[str, str, Tuple[int, int], List[Any]]]] = {}


def _catalog_templates(stock_dir: str) -> List[Tuple[str, str, Tuple[int, int], List[Any]]]:
    """(stem, name, shape, signatures) per catalog image, read once per stock dir.

    Cached because the signature is what catches a re-cut crop, and re-reading
    every template for every facing is what made the old check expensive.
    """
    cached = _TEMPLATES.get(stock_dir)
    if cached is None:
        cached = []
        for name in sorted(os.listdir(stock_dir)):
            stem, ext = os.path.splitext(name)
            if ext.lower() not in (".jpg", ".jpeg", ".png"):
                continue
            template = cv2.imread(os.path.join(stock_dir, name))
            if template is None:
                continue
            cached.append((stem, name, template.shape[:2], _signatures(template)))
        _TEMPLATES[stock_dir] = cached
    return cached


def leaked(image: Any, box: Sequence[int], stock_dir: str) -> Optional[str]:
    """Name of the catalog template this crop is, or None when it is held out.

    Byte-identical crops and near-identical pastes both count: enrolment crops
    scored as test data are training-set accuracy, not recognition.

    Pixel equality is not a leak test. A template re-cut from the same photo
    with a slightly different box is the same leak and passes a same-shape
    comparison, so grayscale signatures are compared as well. That check is a
    heuristic with overlapping distributions (see LEAK_SIGNATURE_MAD): it
    catches most re-cuts and no genuine facing in the measured set, but a
    determined re-cut can still pass. The enrollment protocol's session manifest
    is the real refusal; this is the backstop for catalogs that lack one.
    """
    crop = crop_of(image, box)
    if crop.size == 0:
        return None
    crop_signatures = _signatures(crop)
    for stem, name, shape, signatures in _catalog_templates(stock_dir):
        if min(float(np.abs(signature - crop_signature).mean())
               for signature in signatures for crop_signature in crop_signatures) < LEAK_SIGNATURE_MAD:
            return stem
        if shape != crop.shape[:2]:
            continue
        template = cv2.imread(os.path.join(stock_dir, name))
        if template is None:
            continue
        if np.array_equal(template, crop):
            return stem
        left = crop.astype(np.float32).ravel()
        right = template.astype(np.float32).ravel()
        denom = float(np.linalg.norm(left) * np.linalg.norm(right))
        if denom and float((left * right).sum() / denom) >= 0.98:
            return stem
    return None


def catalog_sku_ids(stock_dir: str) -> List[str]:
    """SKU ids from the store's catalog CSV, falling back to enrolled images.

    Candidate sets must come from the catalog, never from the labels: a lane
    told which SKUs the shelf holds is being scored against the answer key, and
    a retrieval failure can never show up.
    """
    path = os.path.join(stock_dir, "stock.csv")
    if os.path.isfile(path):
        with open(path, newline="") as handle:
            ids = [row["SKU_ID"].strip() for row in csv.DictReader(handle)
                   if row.get("SKU_ID", "").strip()]
        if ids:
            return sorted(set(ids))
    return sorted({sku_of(name) for name in os.listdir(stock_dir)
                   if os.path.splitext(name)[1].lower() in (".jpg", ".jpeg", ".png")})


def group_by_sku(scores: Dict[str, float]) -> Dict[str, float]:
    """Collapse per-view template names onto their SKU.

    The matcher scores `sku_l01_1` and `sku_l01_2` separately; a SKU's score is
    its best view, which is also how the gallery ranks several views. `sku_of`
    is the canonical collapse, so a view outside the `sku_l` prefix groups with
    its SKU instead of ranking as a product of its own.
    """
    grouped: Dict[str, float] = {}
    for name, value in (scores or {}).items():
        sku = sku_of(name)
        grouped[sku] = max(grouped.get(sku, 0.0), float(value))
    return grouped


class Lane:
    """One identity architecture, scored identically to every other."""

    name = "lane"

    def rank(self, crop: Any) -> List[Tuple[str, float]]:
        """Catalog candidates, best first. Empty means nothing to answer."""
        raise NotImplementedError

    def shortlist(self, crop: Any, size: int) -> List[str]:
        """Candidate SKUs the verifier is allowed to consider."""
        return [sku for sku, _ in self.rank(crop)[:size]]

    def decide(self, crop: Any, ranked: List[Tuple[str, float]], *, accept: float,
               margin: float, scale: str) -> Dict[str, Any]:
        """The lane's own decision. Default: the shared gate over its ranking.

        A lane whose decision is not `decide_identity` over one ranking - the
        fused local+text lane is one - overrides this, so every lane is still
        scored through the same four ceilings.
        """
        from identity import decide_identity

        return decide_identity(dict(ranked), method=self.name, scale=scale,
                               min_score=accept, min_margin=margin)


class MatcherLane(Lane):
    """The original demo: local features and RANSAC over the whole catalog."""

    name = "matcher"

    def __init__(self, stock_dir: str, method: str = "hybrid") -> None:
        from feature_matcher import ProductFeatureMatcher

        self.matcher = ProductFeatureMatcher(stock_dir=stock_dir)
        self.method = method

    def rank(self, crop: Any) -> List[Tuple[str, float]]:
        scores, _ = self.matcher.score_crop(crop, method=self.method)
        return sorted(group_by_sku(scores).items(), key=lambda kv: kv[1], reverse=True)


class VanLane(MatcherLane):
    """Van's contract: the matcher's top SKU, always answered.

    The `matcher` lane applies `decide_identity`, which is not what the original
    demo did - it answered every crop with its best-scoring SKU and had no
    accept line at all. Comparing the new system against the gated matcher
    understates the baseline's coverage and overstates its precision, so the
    completion gate's "Van" row is this lane: same scores, no gate.
    """

    name = "van"

    def decide(self, crop: Any, ranked: List[Tuple[str, float]], *, accept: float,
               margin: float, scale: str) -> Dict[str, Any]:
        from identity import IDENTITY_ACCEPTED

        if not ranked:
            return {
                "status": IDENTITY_ACCEPTED, "method": self.name, "best_sku": None,
                "score": 0.0, "confidence": 0.0, "identity_level": IDENTITY_ACCEPTED,
                "ranked_candidates": [], "details": {}, "message": "no candidate",
                "source": None,
            }
        best, score = ranked[0]
        return {
            "status": IDENTITY_ACCEPTED, "method": self.name, "best_sku": best,
            "score": round(float(score), 2), "confidence": round(float(score), 4),
            "identity_level": IDENTITY_ACCEPTED,
            "ranked_candidates": [{"sku_id": sku, "score": round(float(v), 2)}
                                  for sku, v in ranked[:5]],
            "details": {}, "message": "Always-argmax baseline", "source": "van",
        }


class LocalLaneAdapter(Lane):
    """The shipped default: RootSIFT pyramid plus the OCR text lane.

    Wraps `identity_lanes` so the production decision is what gets scored,
    rather than a re-implementation of it. `rank` returns the lane's own order
    (its rank key, not its gate value), and `decide` runs the same
    `DecisionPolicy` the server runs.
    """

    name = "local"

    def __init__(self, stock_dir: str, ocr: Any = None) -> None:
        from identity_lanes import LocalLane, TextLane

        self.lane = LocalLane(stock_dir)
        self.text_lane = TextLane.from_catalog(
            _catalog_rows(stock_dir), stock_dir, ocr if ocr is not None else _ocr_engine()
        )

    def rank(self, crop: Any) -> List[Tuple[str, float]]:
        return self.lane.rank(crop)

    def decide(self, crop: Any, ranked: List[Tuple[str, float]], *, accept: float,
               margin: float, scale: str) -> Dict[str, Any]:
        from identity_lanes import DecisionPolicy

        read = self.text_lane.read(crop)
        text_rank = sorted(
            self.text_lane.scores_from_text(read).items(),
            key=lambda kv: kv[1], reverse=True,
        )
        return DecisionPolicy(local_accept=accept).decide(ranked, text_rank, text_read=read)


def _catalog_rows(stock_dir: str) -> List[Dict[str, Any]]:
    """The catalog's rows, in the shape TextLane.from_catalog expects."""
    from inventory import InventoryManager

    return InventoryManager(stock_dir).products.values()


def _ocr_engine() -> Any:
    from ocr import build_ocr_engine

    return build_ocr_engine()


class GalleryLane(Lane):
    """DINOv2 embedding, with the store's trained head when it has one."""

    name = "gallery"

    def __init__(self, stock_dir: str, sku_ids: Sequence[str], image_size: int = 518) -> None:
        from gallery import ProductGallery

        self.gallery = ProductGallery(stock_dir, list(sku_ids), image_size=image_size)
        if not self.gallery.load():
            raise SystemExit(f"gallery unavailable for {stock_dir}")

    def rank(self, crop: Any) -> List[Tuple[str, float]]:
        # Ungated: the lane's own accept floor is applied once, by score_lane.
        # Gating here too would reject before the floor is even consulted.
        return self.gallery.rank(crop)


class HybridLane(Lane):
    """DINO shortlist, then the original matcher re-ranks inside it."""

    name = "hybrid-verify"

    def __init__(self, shortlist_lane: Lane, verifier: MatcherLane, size: int = 5) -> None:
        self.shortlist_lane = shortlist_lane
        self.verifier = verifier
        self.size = int(size)
        self.recall_hits = 0
        self.recall_total = 0

    def rank(self, crop: Any) -> List[Tuple[str, float]]:
        allowed = self.shortlist_lane.shortlist(crop, self.size)
        if not allowed:
            return []
        scores, _ = self.verifier.matcher.score_crop(crop, method=self.verifier.method)
        grouped = group_by_sku(scores)
        ranked = [(sku, float(grouped.get(sku, 0.0))) for sku in allowed]
        return sorted(ranked, key=lambda kv: kv[1], reverse=True)


def score_lane(
    lane: Lane,
    facings: Sequence[Dict[str, Any]],
    image: Any,
    *,
    accept: float,
    margin: float,
    scale: str,
    enrolled_skus: Collection[str],
) -> Dict[str, Any]:
    """One lane over one shelf. Accepts only unique, above-floor answers.

    `top1` and `coverage` were both accepted-correct over enrolled facings, so
    ranking and deciding were never separated. They are now four quantities:

      catalog_ceiling      in-catalog facings with an enrolled view / all in-catalog
      identity_ceiling     ungated top-1 rank correct on enrolled / enrolled
      decision_ceiling     accepted-and-correct on enrolled / enrolled
      end_to_end_coverage  accepted-and-correct / all in-catalog facings

    Precision divides by every accept, unenrolled in-catalog facings included:
    the SKU is on the shelf but has no reference, so an accept there is wrong
    and is the failure the enrollment protocol exists to prevent.
    """
    from identity import decide_identity

    in_catalog = [f for f in facings if f["label"] != UNKNOWN]
    out_catalog = [f for f in facings if f["label"] == UNKNOWN]
    enrolled_count = sum(1 for f in in_catalog if f["label"] in enrolled_skus)
    unenrolled_count = len(in_catalog) - enrolled_count

    correct = wrong = rejected = top1_correct = 0
    unknown_false = unenrolled_false = 0
    accepts: Counter = Counter()
    per_sku: Dict[str, List[int]] = {}
    started = time.perf_counter()

    for facing in facings:
        crop = crop_of(image, facing["box"])
        if crop.size == 0:
            continue
        ranked = lane.rank(crop)
        decision = lane.decide(crop, ranked, accept=accept, margin=margin, scale=scale)
        # decide_identity reports an accept as status "success"; identity_level
        # is the stable field across both the inlier and cosine scales.
        answered = decision.get("best_sku") if decision.get("identity_level") == "accepted" else None
        truth = facing["label"]
        if truth != UNKNOWN:
            bucket = per_sku.setdefault(truth, [0, 0])
            bucket[1] += 1
            if truth in enrolled_skus and ranked and ranked[0][0] == truth:
                top1_correct += 1
            if answered == truth:
                correct += 1
                bucket[0] += 1
            elif answered:
                wrong += 1
                if truth not in enrolled_skus:
                    unenrolled_false += 1
        elif answered:
            wrong += 1
            unknown_false += 1
        if answered:
            accepts[answered] += 1
        else:
            rejected += 1

    elapsed = time.perf_counter() - started
    total_accepts = sum(accepts.values())
    hub = max(accepts.values()) / total_accepts if total_accepts else 0.0
    return {
        "lane": lane.name,
        "accept": accept,
        "margin": margin,
        "in_catalog": len(in_catalog),
        "out_catalog": len(out_catalog),
        "enrolled": enrolled_count,
        "correct": correct,
        "wrong": wrong,
        "rejected": rejected,
        "catalog_ceiling": round(enrolled_count / len(in_catalog), 4) if in_catalog else None,
        "identity_ceiling": round(top1_correct / enrolled_count, 4) if enrolled_count else None,
        "decision_ceiling": round(correct / enrolled_count, 4) if enrolled_count else None,
        "end_to_end_coverage": round(correct / len(in_catalog), 4) if in_catalog else None,
        "false_accepts": wrong,
        "unknown_false_accepts": unknown_false,
        "unenrolled_false_accepts": unenrolled_false,
        "false_accept_rate": round(unknown_false / len(out_catalog), 4) if out_catalog else None,
        "near_unknown_false_accept_rate": round(unenrolled_false / unenrolled_count, 4) if unenrolled_count else None,
        "precision": round(correct / total_accepts, 4) if total_accepts else None,
        "max_sku_share": round(hub, 4),
        "hub_sku": accepts.most_common(1)[0][0] if accepts else None,
        "hub_alarm": hub > HUB_SHARE,
        "histogram": dict(accepts.most_common()),
        "per_sku": {sku: {"correct": hit, "total": seen} for sku, (hit, seen) in sorted(per_sku.items())},
        "ms_per_crop": round(elapsed * 1000 / max(1, len(facings)), 1),
    }


def shortlist_recall(
    lane: HybridLane,
    facings: Sequence[Dict[str, Any]],
    image: Any,
    enrolled_skus: Collection[str],
) -> Dict[str, Any]:
    """Share of enrolled facings whose true SKU survives the shortlist.

    This is the hybrid's hard ceiling: a verifier cannot pick a SKU the
    shortlist already dropped. The denominator is enrolled facings only — a
    facing whose SKU has no reference cannot be retrieved by any shortlist, so
    counting it here reports a retrieval failure that is really a catalog gap.
    Measured on image6_v2, that distinction is 67% vs 98% at top-5.
    """
    hits = total = 0
    misses: List[Dict[str, Any]] = []
    for facing in facings:
        if facing["label"] == UNKNOWN or facing["label"] not in enrolled_skus:
            continue
        crop = crop_of(image, facing["box"])
        if crop.size == 0:
            continue
        total += 1
        allowed = lane.shortlist_lane.shortlist(crop, lane.size)
        if facing["label"] in allowed:
            hits += 1
        else:
            misses.append({"index": facing.get("index"), "truth": facing["label"], "shortlist": allowed})
    return {
        "size": lane.size,
        "hits": hits,
        "total": total,
        "recall": round(hits / total, 4) if total else None,
        "misses": misses,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--labels", required=True, help="labelled facings JSON")
    parser.add_argument("--image", help="shelf photo; defaults to the labels' own image")
    parser.add_argument("--stock-dir", default="stock-lopia")
    parser.add_argument("--lanes", default="local,matcher,gallery,hybrid-verify")
    parser.add_argument("--shortlist", type=int, default=5)
    parser.add_argument("--accept", type=float, default=None, help="override the lane's own floor")
    parser.add_argument("--margin", type=float, default=None)
    parser.add_argument("--live", action="store_true", help="also score at live resolution")
    parser.add_argument("--json", help="write the report here")
    args = parser.parse_args()

    payload = load_labels(args.labels)
    image_path = args.image or payload["image"]
    image = cv2.imread(image_path)
    if image is None:
        raise SystemExit(f"cannot read {image_path}")

    facings = payload["facings"]
    held: List[Dict[str, Any]] = []
    refused: List[Dict[str, Any]] = []
    for facing in facings:
        if facing["label"] == EXCLUDE:
            continue
        template = leaked(image, facing["box"], args.stock_dir)
        if template:
            refused.append({**facing, "template": template})
        else:
            held.append(facing)

    wanted = [name.strip() for name in args.lanes.split(",") if name.strip()]
    # Candidate sets come from the catalog, never from the labels: a lane told
    # which SKUs the shelf holds is being scored against the answer key.
    catalog_skus = catalog_sku_ids(args.stock_dir)
    # A SKU with no enrolled view cannot be answered by any lane. Counting those
    # facings as lane failures hides the real ceiling: catalog coverage.
    enrolled_skus = {sku_of(name) for name in os.listdir(args.stock_dir)
                     if os.path.splitext(name)[1].lower() in (".jpg", ".jpeg", ".png")}
    unenrolled = Counter(f["label"] for f in held if f["label"] != UNKNOWN and f["label"] not in enrolled_skus)

    lanes: Dict[str, Lane] = {}
    if "matcher" in wanted or "hybrid-verify" in wanted or "van" in wanted:
        lanes["matcher"] = MatcherLane(args.stock_dir)
    if "van" in wanted:
        lanes["van"] = VanLane(args.stock_dir)
    if "local" in wanted:
        lanes["local"] = LocalLaneAdapter(args.stock_dir)
    if "gallery" in wanted or "hybrid-verify" in wanted:
        lanes["gallery"] = GalleryLane(args.stock_dir, catalog_skus)
    if "hybrid-verify" in wanted:
        lanes["hybrid-verify"] = HybridLane(lanes["gallery"], lanes["matcher"], size=args.shortlist)
    # Boxes are in the original image's coordinates. Scoring a downscaled copy
    # needs the boxes scaled with it, or every crop is taken from the wrong place.
    resolutions = [("full", image, 1.0)]
    if args.live:
        live = at_live_resolution(image)
        resolutions.append(("live", live, live.shape[1] / float(image.shape[1])))

    report: Dict[str, Any] = {
        "image": image_path,
        "stock_dir": args.stock_dir,
        "labels": args.labels,
        # The enrollment protocol's versioning rule: every report records the
        # catalog hash, the label-file hash and the code revision, so a number
        # can be traced to the exact inputs that produced it.
        "provenance": {
            "catalog_hash": catalog_hash(args.stock_dir),
            "labels_hash": file_hash(args.labels),
            "image_hash": file_hash(image_path),
            "code_revision": code_revision(),
        },
        "facings": len(facings),
        "leakage": {"excluded": len(refused), "labels": refused},
        "hub_share_limit": HUB_SHARE,
        "enrollment": {
            "catalog_skus": catalog_skus,
            "enrolled_skus": sorted(enrolled_skus),
            "unenrolled_skus": dict(unenrolled.most_common()),
            "facings_without_enrollment": sum(unenrolled.values()),
            "scored_facings": len(held),
        },
        "results": {},
    }

    for resolution, picture, factor in resolutions:
        scaled = held if factor == 1.0 else [
            {**f, "box": [int(round(v * factor)) for v in f["box"]]} for f in held
        ]
        rows = []
        for name in wanted:
            lane = lanes[name]
            scale = {"gallery": "cosine", "local": "local"}.get(name, "inlier")
            default_accept = {"cosine": 0.60, "local": LOCAL_ACCEPT}.get(scale, 1.0)
            default_margin = 0.04 if scale == "cosine" else 0.15
            accept = args.accept if args.accept is not None else default_accept
            margin = args.margin if args.margin is not None else default_margin
            row = score_lane(lane, scaled, picture, accept=accept, margin=margin, scale=scale,
                             enrolled_skus=enrolled_skus)
            if isinstance(lane, HybridLane):
                row["shortlist_recall"] = shortlist_recall(lane, scaled, picture, enrolled_skus)
            rows.append(row)
        report["results"][resolution] = rows

    if args.json:
        with open(args.json, "w") as handle:
            json.dump(report, handle, indent=1)

    enrollment = report["enrollment"]
    if enrollment["unenrolled_skus"]:
        print(f"\nenrollment ceiling: {enrollment['facings_without_enrollment']} facing(s) belong to "
              f"{len(enrollment['unenrolled_skus'])} SKU(s) with no enrolled view — "
              f"unanswerable by any lane: {enrollment['unenrolled_skus']}")
    for resolution, rows in report["results"].items():
        print(f"\n{resolution} resolution — {enrollment['scored_facings']} scored facings "
              f"({sum(1 for f in held if f['label'] != UNKNOWN)} in-catalog, "
              f"{sum(1 for f in held if f['label'] == UNKNOWN)} unknown)")
        print(f"  {'lane':<14} {'catalog':>8} {'identity':>9} {'decision':>9} {'end-to-end':>11} "
              f"{'precision':>10} {'false acc':>10} {'max SKU':>8} {'ms/crop':>8}")
        for row in rows:
            cells = [f"{row[key]:.0%}" if row[key] is not None else "-" for key in
                     ("catalog_ceiling", "identity_ceiling", "decision_ceiling",
                      "end_to_end_coverage", "precision")]
            flag = " HUB!" if row["hub_alarm"] else ""
            print(f"  {row['lane']:<14} {cells[0]:>8} {cells[1]:>9} {cells[2]:>9} {cells[3]:>11} "
                  f"{cells[4]:>10} {row['false_accepts']:>10} {row['max_sku_share']:>7.0%} "
                  f"{row['ms_per_crop']:>8.1f}{flag}")
            if "shortlist_recall" in row:
                recall = row["shortlist_recall"]
                share = f"{recall['recall']:.0%}" if recall["recall"] is not None else "-"
                print(f"    shortlist top-{recall['size']} recall: {share} "
                      f"({recall['hits']}/{recall['total']}) — hard ceiling for this lane")
    if refused:
        print(f"\nleakage: {len(refused)} facing(s) refused as catalog templates")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())