tngo/product-recognition-demo
1
0
Fork 0
forked from vnguyen/product-recognition-demo
Code Pull requests Activity

feat(identity): local-feature matcher plus text lane as the default backend

Browse source 
Replaces the DINO gallery as the recognition path with the measured design
from docs/research/2026-09-28-recognition-architecture/DECISION.md: RootSIFT
over a per-SKU reference pyramid, similarity-constrained RANSAC and a hue
check, fused with an OCR text lane that vetoes a match whose package text
names a different catalog row and answers on its own when geometry finds
nothing.

- identity_lanes.py: LocalLane, TextLane, DecisionPolicy. The text lane
  builds each row's signature from the catalog name plus its packshot OCR, so
  a row with no enrolled image is still answerable.
- engine.py: every backend groups scores by canonical SKU, so no
  view-suffixed id reaches the wire and a name and price always resolve.
  decide_crop fuses the lanes; score_crop_lanes feeds the live path. Both
  lanes are built in warm_up (non-fatally) so the first frame is not the slow
  one, and the text lane shares the engine's OCR lock.
- live.py: decides per crop on the lane's own rank order. The lane's order
  comes from its rank key and its value is only the gate, so re-sorting by
  value named a different SKU than the lane chose; averaging in an empty
  observation also diluted a good confidence below the accept line.
- identity.py: a `local` score scale whose gate is one absolute line, because
  the value is already a margin over the runner-up.
- app.py, run-demo-server.sh: `--identity local` is the default; the demo
  server no longer starts on the leaked `stock-lopia` catalog.
- eval_live.py: the replay vocabulary accepts `local`, so a recording from
  the shipped server can be scored.
- README.md: documents the three backends and what each is for.

Measured on the corrected image6 labels at live resolution: 36/61 in-catalog
facings named at 95% precision, 0 of 48 unknowns accepted, 11 labels sourced
from the text lane alone. Through the real server: /v1/recognize 90%
precision, /v1/stream 100% precision, no view-suffixed ids on either.
This commit is contained in:
tngo 2026-09-28 22:27:21 +09:00
parent 25dbf132e7
commit aaed5d388f
44 changed files with 7847 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

544
recognization/code/eval_identity.py Normal file
View file

@ -0,0 +1,544 @@
"""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:
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
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 = ("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]]
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 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, verifier: Lane, size: int = 5) -> None:
self.shortlist_lane = shortlist
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 = decide_identity(dict(ranked), method=lane.name, scale=scale,
min_score=accept, min_margin=margin)
# 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="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 "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 = "cosine" if name == "gallery" else "inlier"
accept = args.accept if args.accept is not None else (0.60 if scale == "cosine" else 1.0)
margin = args.margin if args.margin is not None else (0.04 if scale == "cosine" else 0.15)
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())