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

"""Open-world false-accept probe: unrelated shelf photos through the shipped lane.

data/external-shelves/ holds Wikimedia Commons supermarket photos of shelves
that share nothing with the LOPIA catalog. Every accept here is a false accept
by construction, which is what makes the set useful: the image6 shelf can only
measure false accepts on the products it happens to contain.

Measured the way the server measures: a 1280-long-side frame detected at
`LiveSettings.conf`. The sources are 4032-4512 px photos, and detecting them at
full resolution and at the default `detect_conf` instead measures a
configuration the phone never runs, which is how a threshold change reads as a
pass on evidence the product does not produce.

    .venv/bin/python code/eval_distractors.py --stock stock-lopia-clean \
        data/external-shelves/wikimedia-commons
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import sys
import time
from typing import Any, Dict, List

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

from eval_identity import (  # noqa: E402
    LIVE_LONG_SIDE,
    LOCAL_ACCEPT,
    LocalLaneAdapter,
    at_live_resolution,
)
from identity_lanes import TEXT_MIN_SHARED  # noqa: E402


def main(argv: List[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("directories", nargs="+")
    parser.add_argument("--stock", default="stock-lopia-clean")
    parser.add_argument("--write", action="store_true")
    parser.add_argument("--limit", type=int, default=0, help="images per set (0 = all)")
    args = parser.parse_args(argv)

    from engine import build_default_engine, EngineSettings
    from live import LiveSettings

    engine = build_default_engine(EngineSettings(stock_dir=args.stock))
    lane = LocalLaneAdapter(args.stock)
    conf = LiveSettings().conf
    for directory in args.directories:
        manifest_path = os.path.join(directory, "manifest.json")
        try:
            with open(manifest_path) as handle:
                manifest = json.load(handle)
        except (OSError, ValueError):
            manifest = {"images": []}
        meta = {row["file"]: row for row in manifest.get("images", [])}
        rows: List[Dict[str, Any]] = []

        paths = sorted(
            path for path in glob.glob(os.path.join(directory, "*"))
            if os.path.splitext(path)[1].lower() in (".jpg", ".jpeg", ".png")
        )
        if args.limit:
            paths = paths[: args.limit]
        for path in paths:
            name = os.path.basename(path)
            image = cv2.imread(path)
            if image is None:
                continue
            started = time.perf_counter()
            frame = at_live_resolution(image)
            detections = engine.detect(frame, conf=conf)
            accepted = 0
            for detection in detections:
                box = detection.get("box") or detection.get("bbox")
                if not box:
                    continue
                x1, y1, x2, y2 = [int(v) for v in box]
                crop = frame[max(0, y1):y2, max(0, x1):x2]
                if crop.size == 0:
                    continue
                decision = lane.decide(crop, lane.rank(crop), accept=LOCAL_ACCEPT,
                                       margin=0.0, scale="full")
                if decision.get("best_sku"):
                    accepted += 1
            rows.append({
                "file": name,
                "title": meta.get(name, {}).get("title", ""),
                "detections": len(detections),
                "false_accepts": accepted,
                "ms": round((time.perf_counter() - started) * 1000),
            })

        detections = sum(r["detections"] for r in rows)
        accepts = sum(r["false_accepts"] for r in rows)
        report = {
            "dataset": os.path.basename(directory.rstrip("/")),
            "catalog": args.stock,
            "note": "Unrelated shelves: every accept is a false accept by construction.",
            # The configuration the numbers above were measured under, so the
            # stored evidence is checkable against the constants it claims.
            "config": {
                "frame_long_side": LIVE_LONG_SIDE,
                "detect_conf": conf,
                "local_accept": LOCAL_ACCEPT,
                "text_min_shared": TEXT_MIN_SHARED,
            },
            "images": len(rows),
            "detections": detections,
            "false_accepts": accepts,
            "false_accept_rate": round(accepts / detections, 4) if detections else None,
            "rows": rows,
        }
        print(f"{report['dataset']}: {len(rows)} images, {detections} detections, "
              f"{accepts} false accepts ({report['false_accept_rate'] or 0:.2%})")
        for row in sorted(rows, key=lambda r: -r["false_accepts"])[:5]:
            if row["false_accepts"]:
                print(f"    {row['false_accepts']:>3}  {row['file'][:55]}")
        if args.write:
            out = os.path.join(directory, "eval_result.json")
            with open(out, "w") as handle:
                json.dump(report, handle, ensure_ascii=False, indent=1)
            print(f"    wrote {out}")
    return 0


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