Enrolling a packshot from an external set is normal - it is how F-12 was closed - and it silently turns one of those rows into training data. The external sets were scored by an ad-hoc script that never checked, so sku_l06_1.jpg was being reported as a held-out correct accept while being byte-identical to the enrolled view. code/eval_external.py now scores those sets with the same leaked() check the shelf evaluator runs, and refuses a row that has since been enrolled. That check needed fixing first. Its 64x64 signature fires on two different photographs of the same product: the l09 rear label from a retailer page sits 19.5 from the enrolled front packshot, under the 20.0 line, while being a different photograph (resized MAD 26.3). A signature hit is now confirmed by correlation - a genuine re-cut, re-encode or rescale correlates at 0.99+, that pair at 0.28 - so the two signals together are decisive where either alone is not. Verified against both directions: identical, q55 re-encode and 700px rescale are caught; the different-photo pair is not. Also adds the van backend (always-argmax, no gate) so the stream comparison in DECISION.md is reproducible rather than asserted, with two tests pinning that it answers where the shipped gate abstains and still refuses an empty score set. External result on the corrected basis: 19 scored, 12 correct, 7 abstain, 0 false accepts, 1 leak excluded. Suite 307 -> 309.
116 lines
4.7 KiB
Python
116 lines
4.7 KiB
Python
"""Score external-proxy image sets through the shipped identity lane.
|
|
|
|
These sets are not Tier 2 (see data/external-eval/README.md). They exist to test
|
|
the pipeline on data it had no part in building, so the one thing this script
|
|
must never do is score an image that has since been enrolled - that is
|
|
training-set accuracy wearing a held-out label.
|
|
|
|
`leaked()` is the same check the shelf evaluator runs, and it is applied here
|
|
for that reason: enrolling a packshot from an external set is a normal thing to
|
|
do (it is how F-12 was closed), and it silently converts one of these rows into
|
|
a leak unless something refuses it.
|
|
|
|
.venv/bin/python code/eval_external.py --stock stock-lopia-clean \
|
|
data/external-eval/kewpie data/external-eval/other-mayo
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
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 LOCAL_ACCEPT, LocalLaneAdapter, leaked # noqa: E402
|
|
|
|
|
|
def score_set(lane: Any, directory: str, stock_dir: str) -> Dict[str, Any]:
|
|
rows: List[Dict[str, Any]] = []
|
|
for path in sorted(glob.glob(os.path.join(directory, "*.jpg"))):
|
|
name = os.path.basename(path)
|
|
truth = "sku_" + name.split("_")[1]
|
|
image = cv2.imread(path)
|
|
if image is None:
|
|
rows.append({"file": name, "truth": truth, "verdict": "unreadable"})
|
|
continue
|
|
# The whole image is the crop: an external packshot is not a shelf facing.
|
|
box = (0, 0, image.shape[1], image.shape[0])
|
|
leak = leaked(image, box, stock_dir)
|
|
if leak:
|
|
rows.append({
|
|
"file": name, "truth": truth, "verdict": "leaked",
|
|
"leaked_into": leak,
|
|
"note": "enrolled in the catalog; scoring it would be training accuracy",
|
|
})
|
|
continue
|
|
decision = lane.decide(
|
|
image, lane.rank(image), accept=LOCAL_ACCEPT, margin=0.0, scale="full"
|
|
)
|
|
best = decision.get("best_sku")
|
|
verdict = "correct" if best == truth else ("abstain" if not best else "wrong")
|
|
rows.append({
|
|
"file": name, "truth": truth, "best": best,
|
|
"confidence": decision.get("confidence"),
|
|
"status": decision.get("status"),
|
|
"source": decision.get("source"),
|
|
"verdict": verdict,
|
|
})
|
|
|
|
scored = [r for r in rows if r["verdict"] in ("correct", "abstain", "wrong")]
|
|
counts = {v: sum(1 for r in scored if r["verdict"] == v)
|
|
for v in ("correct", "abstain", "wrong")}
|
|
return {
|
|
"dataset": os.path.basename(directory.rstrip("/")),
|
|
"catalog": stock_dir,
|
|
"note": "External proxy images, not Tier 2. Scored through the shipped "
|
|
"LocalLaneAdapter + DecisionPolicy.",
|
|
"rows": rows,
|
|
"scored": len(scored),
|
|
"leaked": sum(1 for r in rows if r["verdict"] == "leaked"),
|
|
**counts,
|
|
}
|
|
|
|
|
|
def main(argv: List[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("directories", nargs="+", help="external set directories")
|
|
parser.add_argument("--stock", default="stock-lopia-clean")
|
|
parser.add_argument("--write", action="store_true",
|
|
help="write eval_result.json into each set directory")
|
|
args = parser.parse_args(argv)
|
|
|
|
lane = LocalLaneAdapter(args.stock)
|
|
totals = {"correct": 0, "abstain": 0, "wrong": 0, "leaked": 0, "scored": 0}
|
|
for directory in args.directories:
|
|
report = score_set(lane, directory, args.stock)
|
|
for key in totals:
|
|
totals[key] += report[key]
|
|
print(f"{report['dataset']}: correct {report['correct']} | "
|
|
f"abstain {report['abstain']} | wrong {report['wrong']} | "
|
|
f"leaked {report['leaked']} of {report['scored']} scored")
|
|
for row in report["rows"]:
|
|
if row["verdict"] in ("wrong", "leaked"):
|
|
print(f" {row['verdict'].upper():<7} {row['file']} "
|
|
f"truth={row['truth']} best={row.get('best')} "
|
|
f"{row.get('leaked_into', '')}")
|
|
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}")
|
|
print(f"\nTOTAL: {totals['scored']} scored, correct {totals['correct']}, "
|
|
f"abstain {totals['abstain']}, FALSE ACCEPTS {totals['wrong']}, "
|
|
f"leaked {totals['leaked']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|