tngo
2b7dd0c193
Some checks failed
tests / model-free (pull_request) Has been cancelled
The record said '2 false accepts of 48 unknowns', which conflated two fields: the 2 are 0 unknown accepts plus 1 unenrolled in-catalog facing and 1 enrolled misidentification. Also restores the hybrid-verify line the lane vocabulary edit dropped, and reframes the l09 note: its hubness ratio is 1.76 full / 1.36 live against G7's limit of 2.0, so the gate passes and the 30% internal share heuristic is what raises the alarm.
608 lines
26 KiB
Python
608 lines
26 KiB
Python
"""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 json
|
|
import os
|
|
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", "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 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 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
|
|
|
|
text_rank = sorted(
|
|
self.text_lane.scores_from_text(self.text_lane.read(crop)).items(),
|
|
key=lambda kv: kv[1], reverse=True,
|
|
)
|
|
return DecisionPolicy(local_accept=accept).decide(ranked, text_rank)
|
|
|
|
|
|
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:
|
|
lanes["matcher"] = MatcherLane(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,
|
|
"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())
|