"""Metric contracts for eval_live. No model, no real walk, no fabricated accuracy.""" from __future__ import annotations import json import os import shutil import subprocess 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 eval_live import ( # noqa: E402 EvalInputError, identity_level, labelled_top1, leaked_labels, load_index, score_recording, score_tracks, track_churn, ) def track(level, sku=None, track_id=1, box=None, relabels=0, restarts=0, tag=None): identity = {"identity_level": level, "state": "recognized" if level == "accepted" else "unrecognized"} if level == "accepted": identity["sku_id"] = sku row = { "track_id": track_id, "box": box or [0, 0, 10, 10], "identity": identity, "relabels": relabels, "restarts": restarts, } if tag is not None: row["price_tag"] = tag return row def write_jpeg(path, image): ok, encoded = cv2.imencode(".jpg", image) assert ok with open(path, "wb") as handle: handle.write(encoded.tobytes()) class DecisionDenominatorTests(unittest.TestCase): def test_settled_excludes_unsettled_and_rates_use_that_denominator(self): scored = score_tracks( [ track("accepted", "sku_a"), track("ambiguous"), track("rejected"), track("unknown"), {"identity": {"state": "unconfirmed", "identity_level": None}}, ] ) self.assertEqual(scored["settled"], 4) self.assertEqual(scored["unsettled"], 1) self.assertEqual(scored["accepted_rate"], 0.25) self.assertIsNone(score_tracks([{"identity": {"state": "unconfirmed"}}])["accepted_rate"]) def test_wire_state_fills_a_missing_level(self): self.assertEqual( identity_level({"identity": {"state": "needs_confirmation"}}), "ambiguous" ) self.assertEqual(identity_level({"identity": {"state": "unconfirmed"}}), "unsettled") def test_readable_tags_are_counted_apart_from_detected(self): scored = score_tracks( [ track("accepted", "sku_a", tag={"state": "readable"}), track("rejected", tag={"state": "unreadable"}), track("rejected", tag={"state": "absent"}), track("rejected"), ] ) self.assertEqual(scored["tags_readable"], 1) self.assertEqual(scored["tags_detected"], 2) self.assertEqual(scored["tags_absent"], 2) class Top1AndChurnTests(unittest.TestCase): def test_top1_counts_only_accepted_matches_over_overlapping_labels(self): frames = [ { "frame": 1, "tracks": [ track("accepted", "sku_a", box=[0, 0, 10, 10]), track("accepted", "sku_b", track_id=2, box=[100, 100, 110, 110]), ], }, {"frame": 2, "tracks": [track("rejected", box=[0, 0, 10, 10])]}, {"frame": 3, "tracks": []}, ] labels = [ {"frame": 1, "box": [0, 0, 10, 10], "sku_id": "sku_a"}, {"frame": 1, "box": [100, 100, 110, 110], "sku_id": "sku_a"}, {"frame": 2, "box": [0, 0, 10, 10], "sku_id": "sku_a"}, {"frame": 3, "box": [0, 0, 10, 10], "sku_id": "sku_a"}, ] scored = labelled_top1(frames, labels, iou_min=0.5) self.assertEqual((scored["correct"], scored["wrong"], scored["rejected"], scored["unmatched"]), (1, 1, 1, 1)) self.assertEqual(scored["top1"], 0.3333) def test_sku_churn_counts_a_renamed_track_not_the_first_name(self): frames = [ {"tracks": [track("accepted", "sku_a", track_id=7)]}, {"tracks": [track("accepted", "sku_a", track_id=7)]}, {"tracks": [track("accepted", "sku_b", track_id=7, relabels=1)]}, {"tracks": [track("rejected", track_id=8, restarts=2)]}, ] churn = track_churn(frames) self.assertEqual(churn["sku_changes"], 1) self.assertEqual(churn["tracks_with_sku_change"], 1) self.assertEqual(churn["relabels"], 1) self.assertEqual(churn["restarts"], 2) self.assertEqual(churn["per_track"][0]["track_id"], 7) class LeakageTests(unittest.TestCase): def setUp(self): self.root = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) def test_a_pasted_catalog_crop_is_excluded_and_a_real_facing_is_not(self): stock = os.path.join(self.root, "stock") os.makedirs(stock) template = np.full((20, 16, 3), 40, dtype=np.uint8) template[2:8, 2:8] = (255, 0, 0) write_jpeg(os.path.join(stock, "sku_a.jpeg"), template) frame = np.zeros((40, 40, 3), dtype=np.uint8) frame[5:25, 10:26] = template recording = os.path.join(self.root, "walk") os.makedirs(recording) write_jpeg(os.path.join(recording, "000000.jpg"), frame) with open(os.path.join(recording, "index.jsonl"), "w", encoding="utf-8") as handle: handle.write(json.dumps({"frame": 0, "path": "000000.jpg", "tracks": []}) + "\n") labels = [ {"frame": 0, "box": [10, 5, 26, 25], "sku_id": "sku_a"}, {"frame": 0, "box": [0, 0, 8, 8], "sku_id": "sku_b"}, ] held, leaked = leaked_labels(recording, load_index(recording), labels, stock) self.assertEqual(len(leaked), 1) self.assertEqual(leaked[0]["leak"], "crop_is_template") self.assertEqual([row["sku_id"] for row in held], ["sku_b"]) def test_score_recording_names_the_leak_and_refuses_it_as_top1(self): stock = os.path.join(self.root, "stock") os.makedirs(stock) image = np.full((12, 12, 3), 9, dtype=np.uint8) write_jpeg(os.path.join(stock, "sku_a.jpg"), image) with open(os.path.join(stock, "stock.csv"), "w", encoding="utf-8") as handle: handle.write("SKU_ID,product_name\nsku_a,A\n") recording = os.path.join(self.root, "session") os.makedirs(recording) shutil.copy(os.path.join(stock, "sku_a.jpg"), os.path.join(recording, "000000.jpg")) with open(os.path.join(recording, "index.jsonl"), "w", encoding="utf-8") as handle: handle.write( json.dumps( { "frame": 1, "path": "000000.jpg", "tracks": [track("accepted", "sku_a", box=[0, 0, 12, 12])], } ) + "\n" ) with open(os.path.join(recording, "labels.json"), "w", encoding="utf-8") as handle: json.dump({"labels": [{"frame": 1, "box": [0, 0, 12, 12], "sku_id": "sku_a"}]}, handle) report = score_recording(recording, stock_dir=stock) self.assertEqual(report["leakage"]["excluded"], 1) self.assertEqual(report["labelled_top1"]["labels"], 0) self.assertEqual(report["labelled_top1"]["top1"], None) self.assertEqual(report["decisions"]["accepted"], 1) self.assertTrue(report["fingerprints"]["sample"]["sha256"]) self.assertEqual(len(report["fingerprints"]["catalog"]["version"]), 12) self.assertTrue(any("template-derived" in gap for gap in report["evidence"])) class CliTests(unittest.TestCase): def test_missing_recording_is_an_explicit_error(self): completed = subprocess.run( [sys.executable, os.path.join(current_dir, "eval_live.py"), "/no/such/recording"], capture_output=True, text=True, check=False, ) self.assertEqual(completed.returncode, 2) self.assertIn("error", json.loads(completed.stdout)) def test_the_same_pinned_recording_scores_twice_the_same(self): root = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, root, ignore_errors=True) stock = os.path.join(root, "stock") os.makedirs(stock) with open(os.path.join(stock, "stock.csv"), "w", encoding="utf-8") as handle: handle.write("SKU_ID\nsku_a\n") recording = os.path.join(root, "session") os.makedirs(recording) payload = np.zeros((8, 8, 3), dtype=np.uint8) write_jpeg(os.path.join(recording, "000000.jpg"), payload) line = { "frame": 4, "path": "000000.jpg", "tracks": [ track("accepted", "sku_a", tag={"state": "readable"}), track("ambiguous", track_id=2), ], } with open(os.path.join(recording, "index.jsonl"), "w", encoding="utf-8") as handle: handle.write(json.dumps(line) + "\n") with open(os.path.join(recording, "labels.json"), "w", encoding="utf-8") as handle: json.dump({"labels": [{"frame": 4, "box": [0, 0, 8, 8], "sku_id": "sku_a"}]}, handle) first = score_recording(recording, stock_dir=stock) second = score_recording(recording, stock_dir=stock) self.assertEqual(first, second) self.assertEqual(first["decisions"]["accepted"], 1) self.assertEqual(first["decisions"]["ambiguous"], 1) self.assertEqual(first["labelled_top1"]["correct"], 1) self.assertIsNone(first["fingerprints"]["model"]["weights_sha256"]) def test_a_broken_index_line_is_an_input_error(self): root = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, root, ignore_errors=True) with open(os.path.join(root, "index.jsonl"), "w", encoding="utf-8") as handle: handle.write("{not json}\n") with self.assertRaises(EvalInputError): load_index(root) class IdentityContractTests(unittest.TestCase): def test_missing_gallery_weights_exit_without_a_matcher_score(self): root = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, root, ignore_errors=True) recording = os.path.join(root, "session") os.makedirs(recording) write_jpeg(os.path.join(recording, "000000.jpg"), np.zeros((8, 8, 3), dtype=np.uint8)) with open(os.path.join(recording, "index.jsonl"), "w", encoding="utf-8") as handle: handle.write(json.dumps({"frame": 1, "path": "000000.jpg", "tracks": []}) + "\n") missing = os.path.join(root, "no-such-dinov2.safetensors") report_path = os.path.join(recording, "eval_live.json") env = os.environ.copy() env["DORAYAKI_DINOV2_WEIGHTS"] = missing completed = subprocess.run( [ sys.executable, os.path.join(current_dir, "eval_live.py"), recording, "--replay", "--identity", "gallery", "--json", report_path, ], capture_output=True, text=True, check=False, env=env, ) self.assertEqual(completed.returncode, 2) payload = json.loads(completed.stdout) self.assertIn("DINOv2", payload["error"]) self.assertIn("matcher", payload["error"]) self.assertFalse(os.path.exists(report_path)) self.assertNotIn("accepted_rate", completed.stdout) if __name__ == "__main__": unittest.main()