diff --git a/recognization/code/engine.py b/recognization/code/engine.py index f655b7b..78f43ca 100644 --- a/recognization/code/engine.py +++ b/recognization/code/engine.py @@ -527,9 +527,12 @@ class RecognitionEngine: """Both lanes' evidence for one crop: local scores and text ranking. The live path accumulates evidence across looks, so it needs the lanes - separately rather than one fused decision: it averages the local scores - over the track's window and keeps the newest text reading, then decides - once. The still path calls `decide_crop`, which fuses a single crop. + separately rather than one fused decision. How it combines them depends + on the scale: the local lane decides on the newest crop (its value is a + margin over the runner-up, and averaging in an empty observation halves + it below the accept line), while the cosine and inlier scales average + the scores over the track's window. The text reading is always the + newest. The still path calls `decide_crop`, which fuses a single crop. """ scores = self.score_crop(crop, method) text_lane = self.identity_text_lane() diff --git a/recognization/code/eval_identity.py b/recognization/code/eval_identity.py index 7ecf476..2158fe7 100644 --- a/recognization/code/eval_identity.py +++ b/recognization/code/eval_identity.py @@ -35,8 +35,10 @@ from __future__ import annotations import argparse import csv +import hashlib import json import os +import subprocess import sys import time from collections import Counter @@ -86,6 +88,55 @@ def load_labels(path: str) -> Dict[str, Any]: return payload +def file_hash(path: str) -> Optional[str]: + """SHA-256 of a file's bytes, or None when it cannot be read.""" + try: + with open(path, "rb") as handle: + return hashlib.sha256(handle.read()).hexdigest() + except OSError: + return None + + +def catalog_hash(stock_dir: str) -> Optional[str]: + """SHA-256 over the catalog's CSV bytes and its image names. + + Image bytes are not hashed: a re-encoded view of the same product is the + same catalog, and hashing every packshot would make the number move for + reasons that do not change a single decision. A changed row or a different + set of views does change it. + """ + digest = hashlib.sha256() + csv_path = os.path.join(stock_dir, "stock.csv") + try: + with open(csv_path, "rb") as handle: + digest.update(handle.read()) + except OSError: + digest.update(b"missing-csv") + try: + names = sorted( + name for name in os.listdir(stock_dir) + if os.path.splitext(name)[1].lower() in (".jpg", ".jpeg", ".png") + ) + except OSError: + names = [] + digest.update(b"\n") + digest.update("\n".join(names).encode("utf-8")) + return digest.hexdigest() + + +def code_revision() -> Optional[str]: + """The checked-out revision, or None outside a git worktree.""" + try: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=os.path.dirname(os.path.abspath(__file__)), + capture_output=True, text=True, timeout=10, + ) + except (OSError, subprocess.SubprocessError): + return None + return result.stdout.strip() or None if result.returncode == 0 else None + + 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] @@ -539,6 +590,15 @@ def main() -> int: "image": image_path, "stock_dir": args.stock_dir, "labels": args.labels, + # The enrollment protocol's versioning rule: every report records the + # catalog hash, the label-file hash and the code revision, so a number + # can be traced to the exact inputs that produced it. + "provenance": { + "catalog_hash": catalog_hash(args.stock_dir), + "labels_hash": file_hash(args.labels), + "image_hash": file_hash(image_path), + "code_revision": code_revision(), + }, "facings": len(facings), "leakage": {"excluded": len(refused), "labels": refused}, "hub_share_limit": HUB_SHARE, diff --git a/recognization/code/rebuild_catalog.py b/recognization/code/rebuild_catalog.py index f8c9787..32392e3 100644 --- a/recognization/code/rebuild_catalog.py +++ b/recognization/code/rebuild_catalog.py @@ -21,6 +21,12 @@ images by default), whose image entries use ``file`` plus ``capture_session_id`` or ``session_id``. Pixel equality is not a leak test: re-cut crops pass it, so leakage is refused by capture session instead. +Two checks report rather than refuse, because both are properties of the +catalog as a whole: cross-SKU near-duplicates are flagged as +text-discriminated siblings (the near-unknown case only package text +separates), and ``--shelf-tags`` turns the completeness rule into a live-ready +verdict. Without that list the verdict is ``unknown``, not a guess. + python code/rebuild_catalog.py --source /tmp/lopia-clean \ --output stock-lopia-clean --csv stock-lopia/stock.csv python code/rebuild_catalog.py --source /tmp/lopia-clean \ @@ -278,6 +284,95 @@ def audit( return kept, refused +def text_discriminated_siblings( + kept: Dict[str, List[str]], source: str, names: Optional[Dict[str, str]] = None +) -> List[Dict[str, Any]]: + """Cross-SKU near-duplicates: the same artwork on two different products. + + These are the near-unknown case no threshold rejects. The geometry is real + because the sibling shares the brand's artwork, so only the package text + separates them. They are kept - they are legitimately different products - + but flagged, because a catalog that enrolls one and not the other will + accept the missing one as the enrolled one. + """ + views: List[Tuple[str, str, np.ndarray]] = [] + for sku, filenames in sorted(kept.items()): + for filename in filenames: + image = cv2.imread(os.path.join(source, filename)) + if image is not None: + views.append((sku, filename, image)) + flagged: List[Dict[str, Any]] = [] + for index, (sku, filename, image) in enumerate(views): + for other_sku, other_name, other in views[index + 1:]: + if other_sku == sku or image.shape != other.shape: + continue + similarity = cosine(image, other) + if similarity < DUPLICATE_COSINE: + continue + flagged.append({ + "sku": sku, + "view": filename, + "sibling_sku": other_sku, + "sibling_view": other_name, + "cosine": round(similarity, 4), + "name_similarity": round( + name_ocr_similarity( + (names or {}).get(sku, ""), (names or {}).get(other_sku, "") + ), 3, + ), + }) + return flagged + + +def live_ready_verdict( + kept: Dict[str, List[str]], names: Optional[Dict[str, str]], shelf_skus: Optional[Set[str]] +) -> Dict[str, Any]: + """The enrollment protocol's completeness rule, as a verdict. + + A bay is live-ready when every SKU on its shelf tags is enrolled or + explicitly listed as text-only. Without a shelf-tag list the rule cannot be + evaluated, and the verdict says so rather than guessing. + """ + catalog = set(names or {}) + enrolled = set(kept) + text_only = catalog - enrolled + if shelf_skus is None: + return { + "live_ready": None, + "reason": "no --shelf-tags given; the completeness rule needs the shelf's SKU list", + "enrolled": len(enrolled), + "text_only": sorted(text_only), + } + missing = sorted(shelf_skus - enrolled - text_only) + return { + "live_ready": not missing, + "reason": ( + "every shelf SKU is enrolled or text-only" + if not missing + else "on the shelf but absent from the catalog: " + ", ".join(missing) + ), + "enrolled": len(enrolled), + "text_only": sorted(text_only), + "shelf_skus": len(shelf_skus), + "missing_from_catalog": missing, + } + + +def load_shelf_skus(path: Optional[str]) -> Optional[Set[str]]: + """SKUs on the shelf's tags: one per line, or a JSON list.""" + if not path: + return None + with open(path, encoding="utf-8") as handle: + text = handle.read() + try: + payload = json.loads(text) + except json.JSONDecodeError: + return {line.strip() for line in text.splitlines() if line.strip()} + if isinstance(payload, list): + return {str(item).strip() for item in payload if str(item).strip()} + raise RebuildError(f"{path}: expected a JSON list or one SKU per line") + + def write_catalog(kept: Dict[str, List[str]], source: str, output: str, csv_path: Optional[str]) -> None: """Copy the kept views and the whole CSV into a staging dir, then swap it in.""" staging = tempfile.mkdtemp(prefix=".catalog-", dir=os.path.dirname(os.path.abspath(output)) or ".") @@ -329,6 +424,9 @@ def main(argv: Optional[Sequence[str]] = None) -> int: help="image capture metadata (default: SOURCE/manifest.json when present)") parser.add_argument("--eval-manifest", help="manifest of evaluation sessions; images captured in them are refused") + parser.add_argument("--shelf-tags", + help="the shelf's SKU list (JSON list or one per line) for the " + "completeness rule; without it the verdict is 'unknown'") parser.add_argument("--json", help="write the audit here") args = parser.parse_args(argv) @@ -378,6 +476,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int: # A row with no kept image is a text-only entry: the text lane still names # and vetoes it from the CSV alone, so it is a coverage gap, not a refusal. text_only = sorted(set(names) - set(kept)) + siblings = text_discriminated_siblings(kept, args.source, names) + verdict = live_ready_verdict(kept, names, load_shelf_skus(args.shelf_tags)) report = { "source": args.source, @@ -389,6 +489,8 @@ def main(argv: Optional[Sequence[str]] = None) -> int: "min_name_ocr": MIN_NAME_OCR, "ocr": engine.name if ocr_engine is not None else "skipped", "text_only_skus": text_only, + "text_discriminated_siblings": siblings, + "completeness": verdict, "skus": {sku: views for sku, views in sorted(kept.items())}, "refused": refused, } @@ -401,6 +503,17 @@ def main(argv: Optional[Sequence[str]] = None) -> int: print(f" {sku:<10} {len(views)} view(s): {', '.join(views)}") if text_only: print(f"\ntext-only CSV row(s), no enrolled image ({len(text_only)}): {', '.join(text_only)}") + if siblings: + print(f"\ntext-discriminated sibling(s) ({len(siblings)}):") + for row in siblings: + print(f" {row['sku']:<10} {row['view']:<20} ~ {row['sibling_sku']} " + f"({row['sibling_view']}) cosine {row['cosine']}") + if verdict["live_ready"] is None: + print(f"\ncompleteness: unknown - {verdict['reason']}") + elif verdict["live_ready"]: + print(f"\ncompleteness: live-ready ({verdict['reason']})") + else: + print(f"\ncompleteness: NOT live-ready - {verdict['reason']}") if refused: print(f"\nrefused {len(refused)} view(s):") for row in refused: diff --git a/recognization/code/test_rebuild_catalog.py b/recognization/code/test_rebuild_catalog.py new file mode 100644 index 0000000..12a5d94 --- /dev/null +++ b/recognization/code/test_rebuild_catalog.py @@ -0,0 +1,122 @@ +""" +Contract tests for the enrollment tool's reporting checks. + +The refusal rules (size floors, session leakage, name/OCR, duplicates) are what +the tool exists for; these tests cover the two checks that report rather than +refuse, because both are properties of the catalog as a whole and neither is +visible from a single view: cross-SKU near-duplicates, and the completeness +rule that decides whether a bay is live-ready. +""" + +from __future__ import annotations + +import os +import sys +import tempfile +import unittest + +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 ( # noqa: E402 + RebuildError, + live_ready_verdict, + load_shelf_skus, + text_discriminated_siblings, +) + + +def artwork(mark=None): + """A package face; `mark` adds the small text that separates siblings.""" + image = np.full((400, 300, 3), 200, dtype="uint8") + cv2.rectangle(image, (20, 20), (280, 380), (40, 40, 180), -1) + cv2.putText(image, "MAYO", (60, 200), cv2.FONT_HERSHEY_SIMPLEX, 2.0, (255, 255, 255), 4) + if mark: + cv2.putText(image, mark, (60, 300), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (255, 255, 255), 3) + return image + + +class SiblingFlagTests(unittest.TestCase): + """The near-unknown case: same artwork, different product.""" + + def make_source(self, images): + root = tempfile.mkdtemp(prefix="siblings-") + for name, image in images.items(): + cv2.imwrite(os.path.join(root, name), image) + return root + + def test_shared_artwork_across_skus_is_flagged(self): + root = self.make_source({"sku_x1.jpg": artwork(), "sku_x2.jpg": artwork("HALF")}) + flagged = text_discriminated_siblings({"sku_x1": ["sku_x1.jpg"], "sku_x2": ["sku_x2.jpg"]}, root) + self.assertEqual(len(flagged), 1, flagged) + self.assertEqual(flagged[0]["sku"], "sku_x1") + self.assertEqual(flagged[0]["sibling_sku"], "sku_x2") + self.assertGreater(flagged[0]["cosine"], 0.98) + + def test_distinct_artwork_is_not_flagged(self): + other = np.full((400, 300, 3), 30, dtype="uint8") + cv2.circle(other, (150, 200), 90, (200, 200, 40), -1) + root = self.make_source({"sku_x1.jpg": artwork(), "sku_x2.jpg": other}) + self.assertEqual( + text_discriminated_siblings({"sku_x1": ["sku_x1.jpg"], "sku_x2": ["sku_x2.jpg"]}, root), [] + ) + + def test_two_views_of_one_sku_are_not_siblings(self): + """The same-SKU duplicate rule already covers these; this flag is cross-SKU.""" + root = self.make_source({"sku_x1.jpg": artwork(), "sku_x1_2.jpg": artwork()}) + self.assertEqual( + text_discriminated_siblings({"sku_x1": ["sku_x1.jpg", "sku_x1_2.jpg"]}, root), [] + ) + + +class CompletenessVerdictTests(unittest.TestCase): + """The enrollment protocol's completeness rule, as a verdict.""" + + def test_without_a_shelf_list_the_verdict_is_unknown(self): + """Not a guess: the rule needs the shelf's SKU list to mean anything.""" + verdict = live_ready_verdict({"sku_a": ["a.jpg"]}, {"sku_a": "A"}, None) + self.assertIsNone(verdict["live_ready"]) + self.assertIn("shelf", verdict["reason"]) + + def test_every_shelf_sku_enrolled_is_live_ready(self): + verdict = live_ready_verdict({"sku_a": ["a.jpg"]}, {"sku_a": "A"}, {"sku_a"}) + self.assertTrue(verdict["live_ready"]) + + def test_a_text_only_row_satisfies_the_rule(self): + """A CSV row with no image still names and vetoes from its text.""" + verdict = live_ready_verdict({"sku_a": ["a.jpg"]}, {"sku_a": "A", "sku_b": "B"}, {"sku_a", "sku_b"}) + self.assertTrue(verdict["live_ready"]) + + def test_a_shelf_sku_absent_from_the_catalog_fails(self): + verdict = live_ready_verdict({"sku_a": ["a.jpg"]}, {"sku_a": "A"}, {"sku_a", "sku_z"}) + self.assertFalse(verdict["live_ready"]) + self.assertEqual(verdict["missing_from_catalog"], ["sku_z"]) + + +class ShelfTagListTests(unittest.TestCase): + def write(self, text): + path = os.path.join(tempfile.mkdtemp(prefix="tags-"), "tags.json") + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return path + + def test_a_json_list_is_read(self): + self.assertEqual(load_shelf_skus(self.write('["sku_a", "sku_b"]')), {"sku_a", "sku_b"}) + + def test_one_sku_per_line_is_read(self): + self.assertEqual(load_shelf_skus(self.write("sku_a\nsku_b\n")), {"sku_a", "sku_b"}) + + def test_no_path_means_no_list(self): + self.assertIsNone(load_shelf_skus(None)) + + def test_a_json_object_is_refused(self): + with self.assertRaises(RebuildError): + load_shelf_skus(self.write('{"sku_a": 1}')) + + +if __name__ == "__main__": + unittest.main()