dc83e75a34
A tag is only readable when its own arithmetic corroborates the price, so the fixture needs the 162.00 that belongs with its 150 (150 x 1.08 = 162.00). Without it the fixture asserted the old behaviour. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""
|
|
Contract tests for the live session.
|
|
|
|
Everything runs against a fake engine, so these tests are fast and deterministic.
|
|
They defend the behaviours a live shelf view depends on: labels that appear
|
|
without blocking frames, labels that never silently change, restarted ids that
|
|
are re-verified rather than inherited, and backpressure that always keeps the
|
|
newest frame.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
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 engine import RecognitionEngine # noqa: E402
|
|
from live import LiveSession, LiveSettings # noqa: E402
|
|
|
|
|
|
class FakeDetector:
|
|
weights_path = "fake.pt"
|
|
|
|
def __init__(self):
|
|
self.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [150, 180]}]
|
|
|
|
def detect(self, image, conf_threshold=0.25, iou_threshold=0.45):
|
|
return {"status": "success", "boxes": [dict(b) for b in self.boxes]}
|
|
|
|
def detect_empty_slots(self, image_shape, boxes, min_gap_ratio=0.75):
|
|
return []
|
|
|
|
|
|
class FakeMatcher:
|
|
def __init__(self, sku="sku_a", score=40.0, delay=0.0):
|
|
self.sku = sku
|
|
self.score = score
|
|
self.delay = delay
|
|
self.calls = 0
|
|
self.lock = threading.Lock()
|
|
|
|
def score_crop(self, crop, method="hybrid"):
|
|
with self.lock:
|
|
self.calls += 1
|
|
if self.delay:
|
|
time.sleep(self.delay)
|
|
# The runner-up stays far below the winner so the decision is about the
|
|
# configured score, not about ambiguity.
|
|
return {self.sku: self.score, "other": 0.1}, {}
|
|
|
|
def score_crop_fast(self, crop, shortlist=5):
|
|
return self.score_crop(crop)
|
|
|
|
|
|
class FakeInventory:
|
|
"""Stateful stand-in for InventoryManager, so stock assertions are real."""
|
|
|
|
def __init__(self):
|
|
self.products = {
|
|
"sku_a": {"sku_id": "sku_a", "product_name": "Product sku_a", "price": 150.0,
|
|
"number_in": 10, "number_out": 1, "number_on": 9,
|
|
"receipt_date": "", "expiration_date": "", "local_image_path": None},
|
|
"other": {"sku_id": "other", "product_name": "Product other", "price": 150.0,
|
|
"number_in": 10, "number_out": 1, "number_on": 9,
|
|
"receipt_date": "", "expiration_date": "", "local_image_path": None},
|
|
}
|
|
|
|
def list_all_skus(self):
|
|
return list(self.products)
|
|
|
|
def get_product(self, sku_id):
|
|
return self.products.get(sku_id)
|
|
|
|
def update_stock(self, sku_id, delta):
|
|
product = self.products.get(sku_id)
|
|
if product is None:
|
|
return False
|
|
product["number_on"] = max(0, product["number_on"] + delta)
|
|
if delta < 0:
|
|
product["number_out"] += abs(delta)
|
|
else:
|
|
product["number_in"] += delta
|
|
return True
|
|
|
|
|
|
def make_session(matcher=None, settings=None):
|
|
detector = FakeDetector()
|
|
matcher = matcher or FakeMatcher()
|
|
engine = RecognitionEngine(detector, matcher, FakeInventory(),
|
|
settings=__import__("engine").EngineSettings(classify_workers=2))
|
|
session = LiveSession("test", engine, settings=settings or LiveSettings(method="hybrid"))
|
|
session.refresh_catalog()
|
|
return session, engine, detector, matcher
|
|
|
|
|
|
def frame_bytes(width=640, height=480):
|
|
ok, buffer = cv2.imencode(".jpg", np.zeros((height, width, 3), dtype="uint8"))
|
|
return buffer.tobytes()
|
|
|
|
|
|
def process(session, frame_id=1):
|
|
payload = frame_bytes()
|
|
session.submit_frame(frame_id, payload)
|
|
pending = session.take_pending(timeout=1.0)
|
|
assert pending is not None, "the session dropped a frame it should have kept"
|
|
return session.process_frame(*pending)
|
|
|
|
|
|
def wait_for(predicate, timeout=2.0):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.01)
|
|
return False
|
|
|
|
|
|
class LabelLifecycleTests(unittest.TestCase):
|
|
def test_first_frame_is_detected_then_labelled(self):
|
|
session, engine, _, matcher = make_session()
|
|
first = process(session, 1)
|
|
self.assertEqual(first["tracks"][0]["identity"]["state"], "unconfirmed")
|
|
self.assertIsNone(first["tracks"][0]["identity"]["sku_id"])
|
|
|
|
process(session, 2) # second frame: eligible for classification
|
|
self.assertTrue(wait_for(lambda: matcher.calls > 0), "classification should be scheduled")
|
|
self.assertTrue(wait_for(lambda: any(
|
|
track["identity"]["state"] == "recognized" for track in process(session, 3)["tracks"]
|
|
)), "the product should be labelled once classified")
|
|
engine.shutdown()
|
|
|
|
def test_label_is_never_shown_without_acceptance(self):
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.4)) # below the accept threshold
|
|
for frame_id in range(1, 8):
|
|
result = process(session, frame_id)
|
|
for track in result["tracks"]:
|
|
self.assertIsNone(track["identity"]["sku_id"],
|
|
"a weak match must never name a SKU")
|
|
engine.shutdown()
|
|
|
|
def test_weak_match_ends_as_needs_confirmation(self):
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.4))
|
|
for frame_id in range(1, 8):
|
|
process(session, frame_id)
|
|
result = process(session, 8)
|
|
self.assertTrue(wait_for(lambda: process(session, 9)["tracks"][0]["identity"]["state"]
|
|
== "needs_confirmation", timeout=3.0))
|
|
engine.shutdown()
|
|
|
|
def test_restarted_id_is_re_verified_not_inherited(self):
|
|
session, engine, detector, matcher = make_session()
|
|
for frame_id in range(1, 5):
|
|
process(session, frame_id)
|
|
self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
|
|
== "recognized"))
|
|
first_relabels = process(session, 6)["tracks"][0]["relabels"]
|
|
|
|
detector.boxes = [] # the product leaves the frame
|
|
for frame_id in range(7, 14):
|
|
process(session, frame_id)
|
|
detector.boxes = [{"id": 0, "box": [100, 100, 200, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [150, 180]}]
|
|
result = process(session, 14)
|
|
track = result["tracks"][0]
|
|
self.assertGreaterEqual(track["relabels"], first_relabels + 1,
|
|
"a restarted id must count as re-verified")
|
|
self.assertGreaterEqual(track["restarts"], 1)
|
|
engine.shutdown()
|
|
|
|
def test_inherited_label_is_flagged(self):
|
|
"""A fresh track covering a labelled product inherits the label, and says so."""
|
|
session, engine, detector, matcher = make_session()
|
|
for frame_id in range(1, 5):
|
|
process(session, frame_id)
|
|
self.assertTrue(wait_for(lambda: process(session, 5)["tracks"][0]["identity"]["state"]
|
|
== "recognized"))
|
|
# same place, new detection id => the tracker treats it as a new object
|
|
detector.boxes = [{"id": 0, "box": [102, 100, 202, 260], "confidence": 0.9, "class_id": 0,
|
|
"width": 100, "height": 160, "center": [152, 180]}]
|
|
result = process(session, 6)
|
|
self.assertTrue(result["tracks"][0]["inherited"] or result["tracks"][0]["identity"]["state"]
|
|
== "recognized")
|
|
engine.shutdown()
|
|
|
|
|
|
class TagPassIntegrationTests(unittest.TestCase):
|
|
"""The tag pass must reach the live result without blocking frames."""
|
|
|
|
def make_tagged_session(self):
|
|
from ocr import FakeOcr, OcrLine
|
|
|
|
def reader(crop):
|
|
height, width = crop.shape[:2]
|
|
return [
|
|
OcrLine("コーヒー", 0.8, [int(width * 0.05), int(height * 0.05), int(width * 0.4), int(height * 0.3)]),
|
|
OcrLine("150", 0.9, [int(width * 0.05), int(height * 0.4), int(width * 0.35), int(height * 0.8)]),
|
|
# 150 x 1.08 = 162.00, so the tag's arithmetic corroborates the price.
|
|
OcrLine("162.00", 0.6, [int(width * 0.55), int(height * 0.5), int(width * 0.8), int(height * 0.75)]),
|
|
]
|
|
|
|
session, engine, detector, matcher = make_session()
|
|
# Rail bands come from product rows, so the shelf needs a row of facings.
|
|
detector.boxes = [
|
|
{"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260],
|
|
"confidence": 0.9, "class_id": 0, "width": 100, "height": 160,
|
|
"center": [150 + index * 120, 180]}
|
|
for index in range(3)
|
|
]
|
|
engine.set_ocr_engine(FakeOcr(reader=reader))
|
|
self.assertTrue(session.enable_tags())
|
|
return session, engine, detector
|
|
|
|
def test_track_gains_a_price_tag_and_a_reconciliation(self):
|
|
import time as _time
|
|
|
|
session, engine, detector = self.make_tagged_session()
|
|
result = None
|
|
for frame_id in range(1, 12):
|
|
result = process(session, frame_id)
|
|
_time.sleep(0.05) # the tag pass runs on its own worker
|
|
if any(
|
|
(track.get("price_tag") or {}).get("state") == "readable"
|
|
for track in result["tracks"]
|
|
):
|
|
break
|
|
track = result["tracks"][0]
|
|
self.assertIn("price_tag", track, "the tag pass should attach to the facing above it")
|
|
self.assertEqual(track["price_tag"]["price"], 150)
|
|
self.assertIn("reconciliation", track)
|
|
self.assertEqual(result["summary"]["price_tags"]["readable"], 1)
|
|
engine.shutdown()
|
|
|
|
def test_frames_are_not_blocked_by_ocr(self):
|
|
"""A slow OCR engine must not slow the frame loop: frames keep coming."""
|
|
import time as _time
|
|
|
|
from ocr import FakeOcr, OcrLine
|
|
|
|
def slow_reader(crop):
|
|
_time.sleep(0.5)
|
|
height, width = crop.shape[:2]
|
|
return [OcrLine("150", 0.9, [10, 10, int(width * 0.3), int(height * 0.6)])]
|
|
|
|
session, engine, detector, _ = make_session()
|
|
detector.boxes = [
|
|
{"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260],
|
|
"confidence": 0.9, "class_id": 0, "width": 100, "height": 160,
|
|
"center": [150 + index * 120, 180]}
|
|
for index in range(3)
|
|
]
|
|
engine.set_ocr_engine(FakeOcr(reader=slow_reader))
|
|
self.assertTrue(session.enable_tags())
|
|
started = _time.perf_counter()
|
|
for frame_id in range(1, 4):
|
|
process(session, frame_id)
|
|
elapsed = _time.perf_counter() - started
|
|
self.assertLess(elapsed, 1.0, "three frames must not wait for a 0.5 s OCR pass")
|
|
engine.shutdown()
|
|
|
|
def test_config_can_turn_tags_off(self):
|
|
session, engine, _ = self.make_tagged_session()
|
|
config = session.configure(tags=False)
|
|
self.assertFalse(config["tags"])
|
|
self.assertTrue(config["tags_available"])
|
|
result = process(session, 1)
|
|
self.assertNotIn("price_tag", result["tracks"][0])
|
|
engine.shutdown()
|
|
|
|
|
|
class FramePipelineTests(unittest.TestCase):
|
|
def test_latest_frame_wins_and_drops_are_counted(self):
|
|
session, engine, _, _ = make_session()
|
|
session.submit_frame(1, frame_bytes())
|
|
session.submit_frame(2, frame_bytes())
|
|
session.submit_frame(3, frame_bytes())
|
|
pending = session.take_pending(timeout=1.0)
|
|
self.assertEqual(pending[0], 3, "the newest frame must be the one processed")
|
|
stats = session.snapshot()["stats"]
|
|
self.assertEqual(stats["frames_received"], 3)
|
|
self.assertEqual(stats["frames_dropped"], 2)
|
|
engine.shutdown()
|
|
|
|
def test_undecodable_frame_raises(self):
|
|
session, engine, _, _ = make_session()
|
|
with self.assertRaises(Exception):
|
|
session.process_frame(1, b"not an image")
|
|
engine.shutdown()
|
|
|
|
def test_result_payload_shape(self):
|
|
session, engine, _, _ = make_session()
|
|
result = process(session, 1)
|
|
for key in ("type", "frame_id", "seq", "width", "height", "tracks", "empty_slots",
|
|
"quality", "summary", "timing_ms", "stats"):
|
|
self.assertIn(key, result)
|
|
self.assertEqual(result["type"], "result")
|
|
self.assertIn("unconfirmed", result["summary"])
|
|
track = result["tracks"][0]
|
|
for key in ("track_id", "box", "raw_box", "confidence", "first_seen_seq", "relabels",
|
|
"restarts", "identity"):
|
|
self.assertIn(key, track)
|
|
engine.shutdown()
|
|
|
|
def test_quality_hint_reports_glare(self):
|
|
session, engine, _, _ = make_session()
|
|
ok, buffer = cv2.imencode(".jpg", np.full((480, 640, 3), 255, dtype="uint8"))
|
|
session.submit_frame(1, buffer.tobytes())
|
|
result = session.process_frame(*session.take_pending(timeout=1.0))
|
|
self.assertTrue(result["quality"]["glare"] > 0.5)
|
|
self.assertEqual(result["quality"]["hint"], "glare")
|
|
engine.shutdown()
|
|
|
|
|
|
class ClassificationBudgetTests(unittest.TestCase):
|
|
def test_per_session_inflight_is_capped(self):
|
|
matcher = FakeMatcher(delay=0.3)
|
|
session, engine, detector, _ = make_session(
|
|
matcher, LiveSettings(method="hybrid", max_inflight_per_session=2)
|
|
)
|
|
detector.boxes = [
|
|
{"id": index, "box": [10 + index * 120, 10, 110 + index * 120, 200],
|
|
"confidence": 0.9, "class_id": 0, "width": 100, "height": 190,
|
|
"center": [60 + index * 120, 105]}
|
|
for index in range(6)
|
|
]
|
|
process(session, 1)
|
|
process(session, 2)
|
|
self.assertLessEqual(matcher.calls, 2, "only the cap may be in flight at once")
|
|
engine.shutdown()
|
|
|
|
def test_method_change_clears_labels_and_ignores_late_results(self):
|
|
session, engine, _, matcher = make_session(FakeMatcher(delay=0.25))
|
|
process(session, 1)
|
|
process(session, 2)
|
|
session.configure(method="orb")
|
|
self.assertEqual(session.config()["method"], "orb")
|
|
result = process(session, 3)
|
|
for track in result["tracks"]:
|
|
self.assertNotEqual(track["identity"]["state"], "recognized",
|
|
"a late result from the old method must not label a track")
|
|
engine.shutdown()
|
|
|
|
def test_inspect_reports_track_not_in_view(self):
|
|
session, engine, _, _ = make_session()
|
|
process(session, 1)
|
|
with self.assertRaises(KeyError):
|
|
session.inspect(999)
|
|
engine.shutdown()
|
|
|
|
def test_close_stops_the_loop(self):
|
|
session, engine, _, _ = make_session()
|
|
session.close()
|
|
self.assertTrue(session.closed)
|
|
self.assertIsNone(session.take_pending(timeout=0.05))
|
|
engine.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|