research(051): catalog product retrieval from one photo + confidence gate

research/delighting/.gitignore

@@ -55,3 +55,11 @@ render047/node_modules/
 render047/pnpm-lock.yaml
 render047/package.json
 results/047/boards/
+
+# report 051 -- regenerable heavy artifacts (embeddings/caches gitignored, meta+metrics committed)
+results/051/*.npz
+results/051/delight_cache/
+catalog051/__pycache__/
+results/051/*_board.json
+results/051/*_perquery.json
+results/051/*_gatecurve.json

research/delighting/catalog051/build_clean_index.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+"""Report 051 — build the DINOv2 retrieval index over the canonical CLEAN CORPUS.
+
+The clean corpus (report 021/024, results/corpus/clean_manifest.json, 1,281
+images across Bullseye/Oceanside/Youghiogheny/Wissmach) is the registry the app
+actually ships. In the 051 benchmark it plays two roles:
+  (1) a realistic DISTRACTOR pool for realpairs retrieval (the query's true
+      product is a Delphi SKU not in this corpus; a good index must not rank a
+      Bullseye look-alike above the real target), and
+  (2) the substrate for the query-representation ablation and the out-of-catalog
+      confidence-gate negatives.
+
+One index entry per image; product key = registry_id (clean corpus is ~1
+canonical image per SKU). Embeddings are gitignored (regenerate with this
+script); the meta sidecar is small and committed.
+
+Raw corpus images are LOCAL-ONLY (gitignored on main); read read-only via the
+absolute main-repo path, never copied or committed (report 015/019/021 posture).
+"""
+import argparse
+import json
+import os
+import sys
+import numpy as np
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+from embed import Embedder  # noqa: E402
+
+# main-repo checkout holds the gitignored corpus (same convention as
+# catalog_texture_audit.resolve_default_registry)
+MAIN_REPO = "/Users/dominiquepiche-meunier/Documents/vitraux"
+CATALOG_DIR = os.path.join(MAIN_REPO, "frontend", "public", "assets", "catalog_images")
+CLEAN_MANIFEST = os.path.join(HERE, "..", "results", "corpus", "clean_manifest.json")
+OUT_DIR = os.path.join(HERE, "..", "results", "051")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--backbone", default="dinov2-small")
+    ap.add_argument("--manifest", default=CLEAN_MANIFEST)
+    ap.add_argument("--catalog-dir", default=CATALOG_DIR)
+    ap.add_argument("--out", default=os.path.join(OUT_DIR, "clean_index_dinov2.npz"))
+    ap.add_argument("--meta-out", default=os.path.join(OUT_DIR, "clean_index_meta.json"))
+    ap.add_argument("--limit", type=int, default=None)
+    args = ap.parse_args()
+
+    man = json.load(open(args.manifest))
+    imgs = man["images"]
+    if args.limit:
+        imgs = imgs[:args.limit]
+
+    entries, paths, missing = [], [], 0
+    for im in imgs:
+        p = os.path.join(args.catalog_dir, im["file"])
+        if not os.path.exists(p):
+            missing += 1
+            continue
+        entries.append({
+            "entry_id": f"clean::{im['registry_id']}",
+            "product_id": im["registry_id"],
+            "source": "clean_corpus",
+            "brand": im["manufacturer"],
+            "glass_class": im["extractor_class"],
+            "category": im.get("category"),
+            "name": im.get("name"),
+            "confidence": im.get("confidence"),
+            "file": im["file"],
+        })
+        paths.append(p)
+    print(f"{len(paths)} images to embed ({missing} missing on disk)")
+
+    emb = Embedder(backbone=args.backbone)
+    vecs = emb.embed(paths, normalize=True, progress=True)
+
+    os.makedirs(os.path.dirname(args.out), exist_ok=True)
+    np.savez_compressed(args.out, embeddings=vecs,
+                        entry_ids=np.array([e["entry_id"] for e in entries]))
+    json.dump({"backbone": args.backbone, "dim": int(vecs.shape[1]),
+               "n": len(entries), "entries": entries},
+              open(args.meta_out, "w"), indent=1)
+    print(f"wrote {args.out} ({vecs.shape}) and {args.meta_out}")
+
+
+if __name__ == "__main__":
+    main()

research/delighting/catalog051/embed.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+"""Report 051 — shared self-supervised visual embedder for catalog retrieval.
+
+DINOv2-small (facebook/dinov2-small, 384-d) via HF transformers on MPS.
+Verified runnable in this env before building around it (lab lesson, report 028):
+loads in ~10 s, forward pass on MPS, returns a 384-d CLS/pooler embedding.
+
+The embedder accepts image *paths*, PIL images, or HxWx3 uint8/float arrays, so
+the same code embeds catalog swatches, wild query photos, and derived query
+representations (delighted-T, luma-quotient) that live only in memory.
+
+CLIP (openai ViT-B/32) is provided as an alternative backbone behind the same
+interface for the backbone-choice ablation.
+"""
+import os
+import sys
+import numpy as np
+from PIL import Image
+
+import torch
+
+
+def _to_pil(x):
+    if isinstance(x, Image.Image):
+        return x.convert("RGB")
+    if isinstance(x, str):
+        return Image.open(x).convert("RGB")
+    if isinstance(x, np.ndarray):
+        a = x
+        if a.dtype != np.uint8:
+            a = np.clip(a, 0.0, 1.0) * 255.0 if a.max() <= 1.0 + 1e-6 else np.clip(a, 0, 255)
+            a = a.astype(np.uint8)
+        if a.ndim == 2:
+            a = np.stack([a] * 3, axis=-1)
+        return Image.fromarray(a).convert("RGB")
+    raise TypeError(f"cannot coerce {type(x)} to PIL image")
+
+
+class Embedder:
+    def __init__(self, backbone="dinov2-small", device=None, batch_size=16):
+        self.backbone = backbone
+        self.batch_size = batch_size
+        if device is None:
+            device = "mps" if torch.backends.mps.is_available() else "cpu"
+        self.device = device
+        if backbone.startswith("dinov2"):
+            from transformers import AutoImageProcessor, AutoModel
+            name = f"facebook/{backbone}"
+            self.proc = AutoImageProcessor.from_pretrained(name)
+            self.model = AutoModel.from_pretrained(name).eval().to(device)
+            self.dim = self.model.config.hidden_size
+            self._kind = "dinov2"
+        elif backbone.startswith("clip"):
+            from transformers import CLIPProcessor, CLIPModel
+            name = "openai/clip-vit-base-patch32"
+            self.proc = CLIPProcessor.from_pretrained(name)
+            self.model = CLIPModel.from_pretrained(name).eval().to(device)
+            self.dim = self.model.config.projection_dim
+            self._kind = "clip"
+        else:
+            raise ValueError(f"unknown backbone {backbone}")
+
+    @torch.no_grad()
+    def embed(self, items, normalize=True, progress=False):
+        """items: iterable of path|PIL|ndarray. Returns (N, dim) float32."""
+        items = list(items)
+        out = np.zeros((len(items), self.dim), dtype=np.float32)
+        for start in range(0, len(items), self.batch_size):
+            chunk = items[start:start + self.batch_size]
+            pil = [_to_pil(x) for x in chunk]
+            if self._kind == "dinov2":
+                inp = self.proc(images=pil, return_tensors="pt").to(self.device)
+                res = self.model(**inp)
+                # pooler_output = layernorm(CLS); robust global descriptor
+                feat = res.pooler_output if res.pooler_output is not None else res.last_hidden_state[:, 0]
+            else:  # clip
+                inp = self.proc(images=pil, return_tensors="pt").to(self.device)
+                feat = self.model.get_image_features(**inp)
+            feat = feat.float().cpu().numpy()
+            out[start:start + len(chunk)] = feat
+            if progress:
+                sys.stderr.write(f"\r  embedded {start + len(chunk)}/{len(items)}")
+                sys.stderr.flush()
+        if progress:
+            sys.stderr.write("\n")
+        if normalize:
+            n = np.linalg.norm(out, axis=1, keepdims=True)
+            n[n == 0] = 1.0
+            out = out / n
+        return out
+
+
+if __name__ == "__main__":
+    # self-test
+    emb = Embedder()
+    x = (np.random.rand(300, 300, 3) * 255).astype(np.uint8)
+    v = emb.embed([x, x])
+    print(f"backbone={emb.backbone} dim={emb.dim} device={emb.device} "
+          f"out={v.shape} norm={np.linalg.norm(v[0]):.3f} "
+          f"selfsim={float(v[0] @ v[1]):.4f}")

research/delighting/catalog051/embed_cache.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Report 051 — path-keyed embedding cache.
+
+Embeds a set of images once per (backbone, representation) and memoizes to an
+npz so index build, query eval, the representation ablation and the confidence
+gate all reuse the same vectors. Keyed by "repr::abspath" so raw / delighted-T /
+luma-quotient variants of the same file never collide.
+"""
+import os
+import numpy as np
+from embed import Embedder
+
+
+class EmbedCache:
+    def __init__(self, cache_path, backbone="dinov2-small"):
+        self.cache_path = cache_path
+        self.backbone = backbone
+        self.vecs = {}
+        if os.path.exists(cache_path):
+            z = np.load(cache_path, allow_pickle=True)
+            keys = z["keys"]
+            arr = z["vecs"]
+            self.vecs = {str(k): arr[i] for i, k in enumerate(keys)}
+        self._emb = None
+
+    def _embedder(self):
+        if self._emb is None:
+            self._emb = Embedder(backbone=self.backbone)
+        return self._emb
+
+    def get(self, paths, repr_name="raw", transform=None, progress=True):
+        """Return (N, d) for paths under repr_name. `transform(path)->image` is
+        applied for non-raw representations (image can be path/PIL/ndarray)."""
+        keys = [f"{repr_name}::{os.path.abspath(p)}" for p in paths]
+        missing = [(i, p, k) for i, (p, k) in enumerate(zip(paths, keys)) if k not in self.vecs]
+        if missing:
+            emb = self._embedder()
+            items = []
+            for _, p, _ in missing:
+                items.append(transform(p) if transform is not None else p)
+            new = emb.embed(items, normalize=True, progress=progress)
+            for (i, p, k), v in zip(missing, new):
+                self.vecs[k] = v.astype(np.float32)
+            self.save()
+        d = len(next(iter(self.vecs.values())))
+        out = np.zeros((len(paths), d), dtype=np.float32)
+        for i, k in enumerate(keys):
+            out[i] = self.vecs[k]
+        return out
+
+    def save(self):
+        os.makedirs(os.path.dirname(self.cache_path), exist_ok=True)
+        keys = list(self.vecs.keys())
+        arr = np.stack([self.vecs[k] for k in keys]) if keys else np.zeros((0, 384), np.float32)
+        np.savez_compressed(self.cache_path, keys=np.array(keys), vecs=arr)

research/delighting/catalog051/make_board.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Report 051, scope 7 — qualitative board: query photo | top-3 retrieved | correct?
+
+Rows are stratified (hits, near-misses, hard misses) and rendered as small
+DOWNSCALED crops only (raw catalog/realpairs photography is local-only and never
+redistributed; small board thumbnails are committed, consistent with prior
+reports). A green frame = top-1 correct product; the correct candidate (if in
+top-3) gets a green frame; wrong candidates are framed red.
+"""
+import argparse
+import json
+import os
+from PIL import Image, ImageDraw, ImageFont
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+OUT_DIR = os.path.join(HERE, "..", "results", "051")
+TH = 150      # thumbnail size
+PAD = 8
+LABEL_H = 34
+
+
+def thumb(path, size=TH):
+    try:
+        im = Image.open(path).convert("RGB")
+    except Exception:
+        im = Image.new("RGB", (size, size), (40, 40, 40))
+    im.thumbnail((size, size), Image.LANCZOS)
+    canvas = Image.new("RGB", (size, size), (25, 25, 25))
+    canvas.paste(im, ((size - im.width) // 2, (size - im.height) // 2))
+    return canvas
+
+
+def framed(im, color, w=4):
+    d = ImageDraw.Draw(im)
+    for i in range(w):
+        d.rectangle([i, i, im.width - 1 - i, im.height - 1 - i], outline=color)
+    return im
+
+
+def text_strip(width, lines, bg=(15, 15, 15), fg=(230, 230, 230)):
+    strip = Image.new("RGB", (width, LABEL_H), bg)
+    d = ImageDraw.Draw(strip)
+    try:
+        font = ImageFont.truetype("/System/Library/Fonts/Supplemental/Arial.ttf", 11)
+    except Exception:
+        font = ImageFont.load_default()
+    for i, ln in enumerate(lines[:2]):
+        d.text((3, 2 + i * 15), ln[:46], fill=fg, font=font)
+    return strip
+
+
+def make_row(entry, topk=3):
+    cells = []
+    # query cell
+    q = thumb(entry["query_path"])
+    framed(q, (90, 160, 255))
+    lbl = text_strip(TH, [f"QUERY {entry['query_capture']}",
+                          (entry["query_name"] or "")[:40]])
+    cells.append((q, lbl))
+    for c in entry["candidates"][:topk]:
+        im = thumb(c.get("path"))
+        framed(im, (60, 200, 90) if c["correct"] else (210, 70, 70))
+        src = "cat" if c.get("source") == "clean_corpus" else "rp"
+        cells.append((im, text_strip(TH, [f"#{'OK' if c['correct'] else 'x'} {c['score']:.3f} {src}",
+                                          (c["name"] or "")[:40]])))
+    row_w = len(cells) * TH + (len(cells) + 1) * PAD
+    row_h = TH + LABEL_H + 2 * PAD
+    row = Image.new("RGB", (row_w, row_h), (0, 0, 0))
+    x = PAD
+    for im, lbl in cells:
+        row.paste(im, (x, PAD))
+        row.paste(lbl, (x, PAD + TH))
+        x += TH + PAD
+    return row
+
+
+def stratify_board(board, per_q, seed=0):
+    import random
+    rng = random.Random(seed)
+    hits, near, miss = [], [], []
+    for b, pq in zip(board, per_q):
+        r = pq["rank"]
+        (hits if r == 1 else near if r in (2, 3) else miss).append(b)
+    for lst in (hits, near, miss):
+        rng.shuffle(lst)
+    return hits, near, miss
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--board", default=os.path.join(OUT_DIR, "realpairs_bench_board.json"))
+    ap.add_argument("--perquery", default=os.path.join(OUT_DIR, "realpairs_bench_perquery.json"))
+    ap.add_argument("--out", default=os.path.join(OUT_DIR, "board_qualitative.jpg"))
+    ap.add_argument("--rows", type=int, default=15)
+    args = ap.parse_args()
+
+    board = json.load(open(args.board))
+    per_q = json.load(open(args.perquery))
+    hits, near, miss = stratify_board(board, per_q)
+    n = args.rows
+    sel = ([("HIT", b) for b in hits[:n // 3]] +
+           [("NEAR", b) for b in near[:n // 3]] +
+           [("MISS", b) for b in miss[:n - 2 * (n // 3)]])
+
+    rows = [make_row(b) for _, b in sel]
+    if not rows:
+        print("no rows"); return
+    W = max(r.width for r in rows)
+    header_h = 24
+    H = header_h + sum(r.height for r in rows) + PAD
+    board_img = Image.new("RGB", (W, H), (0, 0, 0))
+    d = ImageDraw.Draw(board_img)
+    d.text((6, 6), "051 wild->clean retrieval  |  blue=query  green=correct product  red=wrong  (cat=corpus distractor, rp=realpairs)",
+           fill=(200, 200, 200))
+    y = header_h
+    for r in rows:
+        board_img.paste(r, (0, y)); y += r.height
+    board_img.save(args.out, quality=88)
+    print(f"wrote {args.out} ({W}x{H}, {len(rows)} rows: "
+          f"{len([s for s in sel if s[0]=='HIT'])} hit / "
+          f"{len([s for s in sel if s[0]=='NEAR'])} near / "
+          f"{len([s for s in sel if s[0]=='MISS'])} miss)")
+
+
+if __name__ == "__main__":
+    main()