5094b5f6ca
Every threshold was fitted on image6, where each crop is a facing of a product the shelf holds, so the set could not price a false accept. Measured on 4580 crops from 64 unrelated Wikimedia shelf photos at the shipped configuration: LOCAL_ACCEPT 2.0 -> 4.0 false accepts 60 (1.31%) -> 3 (0.07%) TEXT_MIN_SHARED 2 -> 4 the two-bigram reads that named sku_l12 via おい/いし On image6 that is 44 shown / 42 correct at full resolution - 95.5% precision, 65.6% coverage - the highest-coverage cell of the joint sweep that passes both G2 (>= 95%) and G3 (<= 1%). At live resolution 56% end-to-end at 97%. The text lane's evidence floor gates its answer only: zeroing the row's score would reorder the ranking and make the margin gate compare a different pair than the one it was measured on. code/sweep_accept_lines.py sweeps both axes jointly through the shipped DecisionPolicy; the earlier table swept one axis while pinning the other, so its baseline rows were not the old line.
338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""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
|
|
IDENTITY_BACKENDS,
|
|
EvalInputError,
|
|
identity_level,
|
|
labelled_top1,
|
|
leaked_labels,
|
|
load_index,
|
|
recorded_identity,
|
|
resolve_identity,
|
|
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, identity="matcher")
|
|
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, identity="matcher")
|
|
second = score_recording(recording, stock_dir=stock, identity="matcher")
|
|
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)
|
|
|
|
def test_a_silent_recording_does_not_become_the_matcher(self):
|
|
with self.assertRaises(EvalInputError):
|
|
resolve_identity(None, None)
|
|
self.assertEqual(resolve_identity("matcher", None), "matcher")
|
|
self.assertEqual(resolve_identity(None, "gallery"), "gallery")
|
|
with self.assertRaises(EvalInputError):
|
|
resolve_identity("matcher", "gallery")
|
|
|
|
def test_recorded_backend_is_read_and_a_mix_is_refused(self):
|
|
self.assertEqual(recorded_identity([{"identity_backend": "Gallery"}]), "gallery")
|
|
self.assertIsNone(recorded_identity([{"frame": 1}]))
|
|
self.assertEqual(
|
|
recorded_identity([{"identity": {"backend": "matcher", "sku_id": "sku_a"}}]),
|
|
"matcher",
|
|
)
|
|
with self.assertRaises(EvalInputError):
|
|
recorded_identity([
|
|
{"identity_backend": "gallery"},
|
|
{"identity_backend": "matcher"},
|
|
])
|
|
|
|
def test_van_is_accepted_by_the_cli_and_by_a_recording_that_names_it(self):
|
|
self.assertIn("van", IDENTITY_BACKENDS)
|
|
self.assertEqual(resolve_identity("van", None), "van")
|
|
self.assertEqual(recorded_identity([{"identity_backend": "van"}]), "van")
|
|
root = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, root, ignore_errors=True)
|
|
stock = os.path.join(root, "stock")
|
|
os.makedirs(stock)
|
|
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))
|
|
line = {"frame": 1, "path": "000000.jpg", "tracks": [], "identity_backend": "van"}
|
|
with open(os.path.join(recording, "index.jsonl"), "w", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(line) + "\n")
|
|
for extra in ([], ["--identity", "van"]):
|
|
report_path = os.path.join(root, "eval_live.json")
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
os.path.join(current_dir, "eval_live.py"),
|
|
recording,
|
|
"--stock-dir",
|
|
stock,
|
|
"--json",
|
|
report_path,
|
|
*extra,
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stdout + completed.stderr)
|
|
with open(report_path, "r", encoding="utf-8") as handle:
|
|
report = json.load(handle)
|
|
self.assertEqual(report["identity"]["backend"], "van")
|
|
self.assertEqual(report["fingerprints"]["model"]["identity"], "van")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|