fix: restore reliable authenticated live recognition
This commit is contained in:
kiyo9w
parent
0e23425ef0
commit
934615b593
40 changed files with 2556 additions and 820 deletions
241
recognization/code/test_eval_live.py
Normal file
241
recognization/code/test_eval_live.py
Normal file
|
|
@ -0,0 +1,241 @@
|
|||
"""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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue