241 lines
9.7 KiB
Python
241 lines
9.7 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
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|