tngo
bf9c04bce4
The text lane now requires shared bigrams between the read and the winning row's name before it will answer, and the local lane passes the catalog signatures through so the decision can weigh text evidence. On the labeled shelf this moves the local lane from 66% to 80% end-to-end at full resolution (91% precision) and 66% at live resolution (95% precision, 2 false accepts), against 57%/35% for the matcher alone. Live fixes: - `_plan_classification` counts `in_flight` again. The increment had been dropped while removing a `label_frame_id` write, so the guard never blocked, every frame re-planned the same tracks, and the executor queue grew without bound until the server stopped answering. - Label memory boxes carry the shelf origin they were measured against and are shifted before the IoU test, so a panning camera no longer slides a stale box onto its neighbour. - The text lane is resolved before the session lock is taken, removing a lock-ordering hazard against the lane's own build lock. - A native crop always reads; the frame crop skips OCR only when it already has a reading. Stream (20 s pan, 4 runs): 71-79% coverage at 76-84% precision, against 62-71% at 73-75% before.
1270 lines
56 KiB
Python
1270 lines
56 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 asyncio
|
|
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 MAX_CROP_BYTES, 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 FakeGallery:
|
|
"""The gallery's own accept line, which the live aggregate must clear."""
|
|
|
|
head_threshold = 0.40
|
|
threshold = 0.55
|
|
|
|
def __init__(self, head=True):
|
|
self._head = head
|
|
|
|
def head_loaded(self):
|
|
return self._head
|
|
|
|
|
|
class CosineEngine(RecognitionEngine):
|
|
"""A gallery-scale engine: score_crop only returns scores the gallery accepted."""
|
|
|
|
def __init__(self, *args, head=True, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._fake_gallery = FakeGallery(head=head)
|
|
|
|
@property
|
|
def identity_scale(self):
|
|
return "cosine"
|
|
|
|
def identity_gallery(self):
|
|
return self._fake_gallery
|
|
|
|
def score_crop(self, crop, method=None):
|
|
scores, _ = self.matcher.score_crop(crop, method or "hybrid")
|
|
return scores
|
|
|
|
|
|
class LocalScaleEngine(RecognitionEngine):
|
|
"""The shipped default scale, with the lanes' evidence injected.
|
|
|
|
`local` is the branch that was wrong twice - it re-sorted the lane's own
|
|
order by value, and its early return on an empty local observation made the
|
|
text lane's "answer on its own" job unreachable - so it is tested here
|
|
rather than only through the real lane, which the fake matcher cannot drive.
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.lane_answers = []
|
|
self.ocr_calls = 0
|
|
|
|
@property
|
|
def identity_scale(self):
|
|
return "local"
|
|
|
|
def score_crop_lanes(self, crop, method=None, skip_ocr=False):
|
|
answer = self.lane_answers.pop(0) if self.lane_answers else ({}, [], "")
|
|
if skip_ocr:
|
|
# The real engine re-runs only the local lane and returns no text;
|
|
# the live path reuses the reading it cached on the track.
|
|
return answer[0], [], ""
|
|
self.ocr_calls += 1
|
|
return answer
|
|
|
|
|
|
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": "コクうま カロリーカット65%", "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": "アマニ油マヨネーズ", "price": 150.0,
|
|
"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, engine_cls=RecognitionEngine):
|
|
detector = FakeDetector()
|
|
matcher = matcher or FakeMatcher()
|
|
# These fixtures inject a fake matcher, so they exercise the `matcher`
|
|
# backend explicitly: the shipped default is the `local` lane, which reads
|
|
# real reference images and would ignore the fake.
|
|
engine = engine_cls(detector, matcher, FakeInventory(),
|
|
settings=__import__("engine").EngineSettings(
|
|
classify_workers=2, identity_backend="matcher"))
|
|
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_payload(session, frame_id, payload):
|
|
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 process(session, frame_id=1):
|
|
return process_payload(session, frame_id, frame_bytes())
|
|
|
|
|
|
def blurred_frame_bytes(width=640, height=480, value=120):
|
|
"""A flat grey capture: the whole frame is out of focus, not just a facing."""
|
|
ok, buffer = cv2.imencode(".jpg", np.full((height, width, 3), value, dtype="uint8"))
|
|
return buffer.tobytes()
|
|
|
|
|
|
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()
|
|
|
|
def test_a_pan_does_not_let_a_neighbour_inherit(self):
|
|
"""A label must not slide onto whatever moved into its screen position.
|
|
|
|
The memory box is stored in the coordinates of the frame it was
|
|
measured on. Under a pan the shelf slides beneath that box, so the
|
|
product now occupying the old screen position is a different one.
|
|
Comparing the raw box hands it the label at IoU 1.0; compensating by
|
|
the shelf movement drops the overlap to nothing.
|
|
"""
|
|
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"))
|
|
# The product leaves, so its track ages out and the next detection at
|
|
# that screen position is a new object rather than the same one.
|
|
detector.boxes = []
|
|
for frame_id in range(6, 10):
|
|
process(session, frame_id)
|
|
# The shelf slides 90 px: the labelled product is now 90 px to the
|
|
# left, and the screen position it used to occupy holds its neighbour.
|
|
session._frame_origin = [90.0, 0.0]
|
|
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, 10)
|
|
row = result["tracks"][0]
|
|
self.assertFalse(row["inherited"],
|
|
"the product that slid into the old box is not the labelled one")
|
|
engine.shutdown()
|
|
|
|
|
|
class CosineScaleTests(unittest.TestCase):
|
|
"""A gallery-scale score is gated by the gallery, not re-gated by the live path."""
|
|
|
|
def test_a_head_scale_match_is_labelled(self):
|
|
"""0.45 clears the head's own 0.40 line but not COSINE_ACCEPT (0.60).
|
|
|
|
score_crop only returns scores the gallery already accepted, so the live
|
|
aggregate must not re-apply the cosine accept line: that refused every
|
|
head match the gallery had accepted, which is the live regression.
|
|
"""
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
|
|
for frame_id in range(1, 5):
|
|
process(session, frame_id)
|
|
self.assertTrue(wait_for(lambda: any(
|
|
track["identity"]["state"] == "recognized" for track in process(session, 5)["tracks"]
|
|
)), "a gallery-accepted head match must be labelled")
|
|
engine.shutdown()
|
|
|
|
def test_a_close_cosine_pair_stays_ambiguous(self):
|
|
"""The margin gate still applies: two near-equal scores must not name a SKU."""
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
|
|
# The matcher contract is (scores, details); a bare dict unpacks into two
|
|
# strings and raises inside the future callback, where the failure is
|
|
# swallowed and the test passes for the wrong reason.
|
|
engine.matcher.score_crop = lambda crop, method=None: ({"sku_a": 0.45, "other": 0.44}, {})
|
|
for frame_id in range(1, 8):
|
|
process(session, frame_id)
|
|
result = process(session, 9)
|
|
for track in result["tracks"]:
|
|
self.assertIsNone(track["identity"]["sku_id"],
|
|
"a cosine pair inside the margin must not be named")
|
|
engine.shutdown()
|
|
|
|
def test_a_rejected_crop_cannot_dilute_an_accepted_one(self):
|
|
"""aggregate_votes divides by the whole window, so a miss must not lower the line.
|
|
|
|
A rejected crop first, then an accepted 0.45: the mean is 0.225. Gating the
|
|
aggregate at zero would label that, at a score the gallery calls unknown.
|
|
"""
|
|
session, engine, _, _ = make_session(FakeMatcher(score=0.45), engine_cls=CosineEngine)
|
|
answers = iter([{}, {"sku_a": 0.45}])
|
|
engine.matcher.score_crop = lambda crop, method=None: (next(answers, {}), {})
|
|
# The retry needs retry_gap frames before the second attempt is eligible.
|
|
for frame_id in range(1, 12):
|
|
process(session, frame_id)
|
|
result = process(session, 13)
|
|
for track in result["tracks"]:
|
|
self.assertIsNone(track["identity"]["sku_id"],
|
|
"a diluted mean must not be labelled below the gallery's line")
|
|
engine.shutdown()
|
|
|
|
|
|
class LocalScaleTests(unittest.TestCase):
|
|
"""The shipped default scale, which the fake matcher cannot reach.
|
|
|
|
`make_session` builds a `matcher`-backend engine, so `scale == "local"` in
|
|
`_on_classified` never runs in the rest of this file - and that branch is
|
|
the one that was wrong twice. These tests drive it with injected lane
|
|
evidence, so a regression in the local decision path fails here.
|
|
"""
|
|
|
|
def test_a_text_only_read_still_labels_the_track(self):
|
|
"""The text lane answers on its own when the local lane has nothing.
|
|
|
|
The early return on an empty local observation made this unreachable:
|
|
a SKU with no enrolled image can only ever be named by its text, so
|
|
dropping that path silently removed the lane's fill-in job.
|
|
"""
|
|
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
|
|
engine.lane_answers = [({}, [("sku_a", 0.9)], "コクうま 65%")]
|
|
result = process(session, 1)
|
|
self.assertTrue(
|
|
wait_for(lambda: any(
|
|
track["identity"]["sku_id"] == "sku_a" for track in process(session, 2)["tracks"]
|
|
)),
|
|
"a text-only read must label the track",
|
|
)
|
|
engine.shutdown()
|
|
|
|
def test_a_text_veto_blocks_a_local_accept(self):
|
|
"""A local match whose package text names another row must not be shown.
|
|
|
|
The sibling shares the artwork, so the geometry is real and the local
|
|
score is high; only the words differ. The text here is below its own
|
|
accept line, so it vetoes without filling in: the local answer must not
|
|
reach the wire at all.
|
|
|
|
`scores` is a dict keyed by SKU, which is what `score_crop` returns and
|
|
what `_on_classified` iterates; a list here raises inside the future
|
|
callback, where the failure is swallowed and the test passes for the
|
|
wrong reason.
|
|
"""
|
|
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
|
|
engine.lane_answers = [({"sku_a": 5.0}, [("other", 0.3)], "other マヨネーズ")] * 4
|
|
for frame_id in range(1, 14):
|
|
result = process(session, frame_id)
|
|
for track in result["tracks"]:
|
|
self.assertNotEqual(
|
|
track["identity"]["sku_id"], "sku_a",
|
|
"a vetoed local answer must not be labelled",
|
|
)
|
|
engine.shutdown()
|
|
|
|
def test_a_size_only_read_cannot_name_a_product(self):
|
|
"""A crop reading "400g" matches every 400g product's size line.
|
|
|
|
Size and number tokens are not identity text: without this gate the
|
|
text lane answered a distractor crop reading only "400g" as a product.
|
|
"""
|
|
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
|
|
engine.lane_answers = [({}, [("sku_a", 0.9)], "400g")] * 4
|
|
for frame_id in range(1, 14):
|
|
for track in process(session, frame_id)["tracks"]:
|
|
self.assertIsNone(
|
|
track["identity"]["sku_id"],
|
|
"a size-only read must not name a product",
|
|
)
|
|
engine.shutdown()
|
|
|
|
def test_ocr_runs_once_per_track(self):
|
|
"""Vision on every attempt is what made coverage pacing-dependent.
|
|
|
|
A crop costs 5 ms without OCR and 25 ms with it, so re-reading the same
|
|
package on every attempt cut throughput from 789 to 60 crops/s at 8
|
|
workers. The reading is cached on the track instead, and the local lane
|
|
keeps running on every attempt.
|
|
"""
|
|
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
|
|
# Below the text accept line, so the track stays undecided and retries:
|
|
# the retry is what would re-run OCR without the cache.
|
|
engine.lane_answers = [({}, [("sku_a", 0.3)], "コクうま 65%")] * 6
|
|
for frame_id in range(1, 14):
|
|
process(session, frame_id)
|
|
self.assertEqual(engine.ocr_calls, 1, "OCR must run once per track")
|
|
engine.shutdown()
|
|
|
|
def test_a_size_only_read_is_not_cached(self):
|
|
"""An unreadable crop must not veto the product for the track's life.
|
|
|
|
The cache used to keep any non-empty read, so a first crop that read
|
|
only "400g" was replayed on every later attempt: the track could never
|
|
be named even after a clean crop arrived, because the cached read
|
|
replaced it. Only a read that carries identity text is cached now, so
|
|
the next attempt gets a fresh OCR.
|
|
"""
|
|
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
|
|
engine.lane_answers = [
|
|
({}, [("sku_a", 0.9)], "400g"),
|
|
({}, [("sku_a", 0.9)], "コクうま 65%"),
|
|
] + [({}, [("sku_a", 0.9)], "コクうま 65%")] * 6
|
|
labelled = False
|
|
for frame_id in range(1, 14):
|
|
for track in process(session, frame_id)["tracks"]:
|
|
if track["identity"]["sku_id"] == "sku_a":
|
|
labelled = True
|
|
self.assertTrue(labelled, "a later readable crop must still name the track")
|
|
self.assertGreater(engine.ocr_calls, 1, "the size-only read must not be cached")
|
|
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()
|
|
|
|
def test_a_panned_shelf_reusing_the_screen_row_is_planned(self):
|
|
"""The same screen row on a shelf that has moved is a new rail."""
|
|
session, engine, _ = self.make_tagged_session()
|
|
session._tags.settings.pass_interval_frames = 0
|
|
session._tags.settings.settle_observations = 1
|
|
session._tags.settings.settle_passes = 1
|
|
|
|
class Panned:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def register(self, image):
|
|
self.calls += 1
|
|
# Content moved up 400 px, so the frame origin moves down the shelf.
|
|
return (0.0, -400.0) if self.calls == 2 else (0.0, 0.0)
|
|
|
|
session._shift_map = Panned()
|
|
process(session, 1)
|
|
time.sleep(0.05)
|
|
before = session._tags.stats()["passes_run"]
|
|
self.assertGreaterEqual(before, 1)
|
|
process(session, 2)
|
|
time.sleep(0.1)
|
|
self.assertGreaterEqual(session._frame_origin[1], 400.0)
|
|
self.assertGreater(
|
|
session._tags.stats()["passes_run"], before,
|
|
"a shelf 400 px away reuses screen row 0 and must be read again",
|
|
)
|
|
engine.shutdown()
|
|
|
|
def test_overlapping_frames_keep_their_own_origin(self):
|
|
"""A later worker must not become the previous-frame baseline or steal origin."""
|
|
session, engine, _ = self.make_tagged_session()
|
|
|
|
class Ordered:
|
|
def __init__(self):
|
|
self.inside = 0
|
|
self.max_inside = 0
|
|
self.calls = 0
|
|
self._inner = threading.Lock()
|
|
|
|
def register(self, image):
|
|
with self._inner:
|
|
self.inside += 1
|
|
self.max_inside = max(self.max_inside, self.inside)
|
|
self.calls += 1
|
|
call = self.calls
|
|
time.sleep(0.03)
|
|
with self._inner:
|
|
self.inside -= 1
|
|
return (0.0, -400.0)
|
|
|
|
session._shift_map = Ordered()
|
|
errors = []
|
|
|
|
def run(frame_id):
|
|
try:
|
|
session.process_frame(frame_id, frame_bytes())
|
|
except Exception as exc:
|
|
errors.append(exc)
|
|
|
|
early = threading.Thread(target=run, args=(1,))
|
|
late = threading.Thread(target=run, args=(2,))
|
|
early.start()
|
|
late.start()
|
|
early.join(2.0)
|
|
late.join(2.0)
|
|
self.assertEqual(errors, [])
|
|
self.assertFalse(early.is_alive() or late.is_alive())
|
|
self.assertEqual(session._shift_map.calls, 2)
|
|
self.assertEqual(session._shift_map.max_inside, 1)
|
|
self.assertEqual(session._frame_origin[1], 800.0)
|
|
engine.shutdown()
|
|
|
|
def test_cached_tag_projects_at_this_frame_origin(self):
|
|
"""A panned frame must not attach a tag that only overlaps the previous origin."""
|
|
from tags import TagRecord
|
|
|
|
session, engine, _ = self.make_tagged_session()
|
|
record = TagRecord(
|
|
tag_id="cached", box=[150, 700, 190, 760], band_index=0, name="cached", price=150,
|
|
flags=["tax_consistent"],
|
|
)
|
|
record.settled = True
|
|
session._tags._tags[record.tag_id] = record
|
|
|
|
class Panned:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def register(self, image):
|
|
self.calls += 1
|
|
return (0.0, -400.0) if self.calls == 2 else (0.0, 0.0)
|
|
|
|
session._shift_map = Panned()
|
|
first = session.process_frame(1, frame_bytes())
|
|
self.assertFalse(any(track.get("price_tag", {}).get("id") for track in first["tracks"]))
|
|
second = session.process_frame(2, frame_bytes())
|
|
bound = [track["price_tag"] for track in second["tracks"] if track.get("price_tag", {}).get("id")]
|
|
self.assertTrue(bound)
|
|
self.assertGreaterEqual(bound[0]["box"][1], 200)
|
|
self.assertLess(bound[0]["box"][1], 480)
|
|
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)
|
|
|
|
|
|
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))
|
|
|
|
def test_async_take_keeps_latest_and_wakes_on_close(self):
|
|
"""The server waiter must not miss a frame that beat registration, and close must wake it."""
|
|
session, engine, _, _ = make_session()
|
|
|
|
async def scenario():
|
|
session.submit_frame(1, frame_bytes())
|
|
session.submit_frame(2, frame_bytes())
|
|
first = await asyncio.wait_for(session.take_pending_async(), timeout=1.0)
|
|
self.assertEqual(first[0], 2)
|
|
|
|
waiting = asyncio.create_task(session.take_pending_async())
|
|
await asyncio.sleep(0)
|
|
self.assertFalse(waiting.done(), "an empty queue must not complete the waiter")
|
|
session.reset()
|
|
session.submit_frame(3, frame_bytes())
|
|
self.assertEqual((await asyncio.wait_for(waiting, timeout=1.0))[0], 3)
|
|
|
|
cancelled = asyncio.create_task(session.take_pending_async())
|
|
await asyncio.sleep(0)
|
|
cancelled.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await cancelled
|
|
session.submit_frame(4, frame_bytes())
|
|
self.assertEqual((await asyncio.wait_for(session.take_pending_async(), timeout=1.0))[0], 4)
|
|
|
|
closing = asyncio.create_task(session.take_pending_async())
|
|
await asyncio.sleep(0)
|
|
session.close()
|
|
self.assertIsNone(await asyncio.wait_for(closing, timeout=1.0))
|
|
self.assertFalse(session.submit_frame(5, frame_bytes()))
|
|
|
|
asyncio.run(scenario())
|
|
engine.shutdown()
|
|
|
|
|
|
|
|
class SmallBoxDetector(FakeDetector):
|
|
"""A facing too small for the frame crop to decide, so only a crop can."""
|
|
|
|
def __init__(self, box=None):
|
|
super().__init__()
|
|
self.boxes = [{
|
|
"id": 0, "box": list(box or [100, 100, 130, 200]), "confidence": 0.9,
|
|
"class_id": 0, "width": 30, "height": 100, "center": [115, 150],
|
|
}]
|
|
|
|
|
|
def make_crop_session(matcher=None, settings=None, detector=None):
|
|
detector = detector or SmallBoxDetector()
|
|
engine = RecognitionEngine(
|
|
detector, matcher or FakeMatcher(), FakeInventory(),
|
|
settings=__import__("engine").EngineSettings(
|
|
classify_workers=2, identity_backend="matcher"),
|
|
)
|
|
session = LiveSession("crop", engine, settings=settings or LiveSettings(method="hybrid"))
|
|
session.refresh_catalog()
|
|
session.enable_crops()
|
|
return session, engine, detector
|
|
|
|
|
|
def requested_crop(result):
|
|
for track in result["tracks"]:
|
|
if "crop_request" in track:
|
|
return track["crop_request"]
|
|
return None
|
|
|
|
|
|
def panned(detector, x):
|
|
"""Put the fake facing at x, the way a camera panning a shelf does.
|
|
|
|
The facing stays too narrow for the frame path, so only the crop lane can
|
|
decide it, and it stays under crop_below_px so a crop is asked about.
|
|
"""
|
|
detector.boxes[0].update({
|
|
"box": [x, 100, x + 36, 220], "width": 36, "height": 120, "center": [x + 18, 160],
|
|
})
|
|
|
|
|
|
class GatedMatcher(FakeMatcher):
|
|
"""Holds every classification open until the test lets it finish."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.gate = threading.Event()
|
|
|
|
def score_crop(self, crop, method="hybrid"):
|
|
self.gate.wait(2.0)
|
|
return super().score_crop(crop, method)
|
|
|
|
|
|
class SecondAnswerWaits(FakeMatcher):
|
|
"""Answers the first crop at once, holds the second one, and disagrees.
|
|
|
|
Two facings that overlap this much are the only way a track can be handed
|
|
a settled label while its own answer is still in the classifier, so the
|
|
test answers the first crop, lets the track settle, and only then hands
|
|
over a second crop whose answer names a different product.
|
|
"""
|
|
|
|
def __init__(self, sku="sku_a", other="other"):
|
|
super().__init__(sku=sku)
|
|
self.gate = threading.Event()
|
|
self.first_answered = threading.Event()
|
|
self.disagreeing = other
|
|
self._seen = 0
|
|
|
|
def score_crop(self, crop, method="hybrid"):
|
|
with self.lock:
|
|
self._seen += 1
|
|
turn = self._seen
|
|
self.calls += 1
|
|
if turn == 1:
|
|
self.first_answered.set()
|
|
return {self.sku: 40.0, "other": 0.1}, {}
|
|
self.gate.wait(2.0)
|
|
return {self.disagreeing: 40.0, self.sku: 0.1}, {}
|
|
|
|
|
|
class CropLaneTests(unittest.TestCase):
|
|
"""A full-resolution crop is extra evidence for one track, nothing else."""
|
|
|
|
def test_a_protocol_v1_client_is_never_asked(self):
|
|
session, engine, _ = make_crop_session()
|
|
session.enable_crops # noqa: B018 - documents that crops start disabled
|
|
session.settings.crops = False
|
|
process(session, 1)
|
|
result = process(session, 2)
|
|
self.assertIsNone(requested_crop(result))
|
|
self.assertEqual(result["stats"]["crops"]["requested"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_accepted_crop_labels_the_track_it_was_asked_about(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(request, "a box the frame crop cannot use must be asked about")
|
|
self.assertEqual(request["frame_id"], 2)
|
|
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["type"], "crop_ack")
|
|
self.assertTrue(wait_for(lambda: process(session, 3)["stats"]["crops"]["accepted"] == 1))
|
|
engine.shutdown()
|
|
|
|
def test_the_label_says_it_was_measured_on_a_crop(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertTrue(wait_for(lambda: "crop_request" not in process(session, 3)["tracks"][0]))
|
|
labelled = [t for t in process(session, 4)["tracks"] if t["identity"]["state"] == "recognized"]
|
|
self.assertTrue(labelled, "the crop should settle the track the frame could not")
|
|
self.assertEqual(labelled[0]["identity"]["sku_id"], "sku_a")
|
|
self.assertEqual(labelled[0]["identity"]["source"], "crop")
|
|
self.assertEqual(labelled[0]["identity"]["attempts"], 1)
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_can_arrive_after_the_next_frame(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
process(session, 3)
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["type"], "crop_ack")
|
|
self.assertTrue(wait_for(lambda: session._crops_accepted == 1))
|
|
engine.shutdown()
|
|
|
|
def test_a_delayed_crop_is_accepted_while_the_track_stays_on_its_product(self):
|
|
"""Movement alone is not staleness: the crop still shows the same product."""
|
|
session, engine, detector = make_crop_session()
|
|
panned(detector, 100)
|
|
process(session, 1)
|
|
panned(detector, 106)
|
|
request = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(request, "a facing the frame cannot use is asked about")
|
|
# The phone answers a frame late, on a track that has panned a little.
|
|
panned(detector, 112)
|
|
moved = process(session, 3)
|
|
self.assertEqual(moved["tracks"][0]["track_id"], request["track_id"])
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["type"], "crop_ack")
|
|
self.assertTrue(wait_for(lambda: process(session, 4)["stats"]["crops"]["accepted"] == 1))
|
|
labelled = [t for t in process(session, 5)["tracks"] if t["identity"]["state"] == "recognized"]
|
|
self.assertEqual(labelled[0]["identity"]["sku_id"], "sku_a")
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_for_a_track_that_slid_onto_its_neighbour_is_refused(self):
|
|
"""The crop photographs one product; a track that walked off it answers nothing."""
|
|
session, engine, detector = make_crop_session()
|
|
panned(detector, 100)
|
|
process(session, 1)
|
|
panned(detector, 118)
|
|
request = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(request)
|
|
# The pan carries the track onto the next product before the phone answers.
|
|
panned(detector, 136)
|
|
moved = process(session, 3)
|
|
self.assertEqual(moved["tracks"][0]["track_id"], request["track_id"],
|
|
"the id must survive the pan, or this refusal proves nothing")
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["code"], "CROP_STALE")
|
|
self.assertEqual(session._tracks[request["track_id"]].in_flight, 0,
|
|
"a refused crop gives the track's busy slot back")
|
|
result = process(session, 4)
|
|
self.assertEqual(result["stats"]["crops"]["accepted"], 0)
|
|
self.assertEqual(result["stats"]["crops"]["rejected"], 1)
|
|
self.assertNotEqual(result["tracks"][0]["identity"]["state"], "recognized")
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_answered_after_the_track_left_its_product_does_not_label_it(self):
|
|
"""A classification in flight cannot settle a track that has since moved on."""
|
|
matcher = GatedMatcher()
|
|
session, engine, detector = make_crop_session(matcher=matcher)
|
|
panned(detector, 100)
|
|
process(session, 1)
|
|
panned(detector, 118)
|
|
request = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(request)
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["type"], "crop_ack", "the track is still on the product here")
|
|
# The answer is still in the classifier when the pan takes the track off it.
|
|
panned(detector, 136)
|
|
process(session, 3)
|
|
matcher.gate.set()
|
|
track = session._tracks[request["track_id"]]
|
|
self.assertTrue(wait_for(lambda: session._tracks[request["track_id"]].in_flight == 0))
|
|
self.assertEqual(track.attempts, 0, "a crop of a product left behind is not an attempt")
|
|
self.assertIsNone(track.identity, "the neighbour must not wear the old product's label")
|
|
engine.shutdown()
|
|
|
|
def test_a_track_is_asked_once(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
process(session, 2)
|
|
result = process(session, 3)
|
|
self.assertIsNone(requested_crop(result), "one crop per track object")
|
|
self.assertEqual(result["stats"]["crops"]["requested"], 1)
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_for_another_frame_is_refused(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
reply = session.submit_crop(request["request_id"], request["track_id"], 999, frame_bytes())
|
|
self.assertEqual(reply["code"], "CROP_MISMATCH")
|
|
result = process(session, 3)
|
|
self.assertNotEqual(result["tracks"][0]["identity"]["state"], "recognized")
|
|
self.assertEqual(result["stats"]["crops"]["rejected"], 1)
|
|
self.assertEqual(result["stats"]["crops"]["accepted"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_for_another_track_is_refused(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
reply = session.submit_crop(request["request_id"], request["track_id"] + 41,
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(reply["code"], "CROP_MISMATCH")
|
|
self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_an_oversized_crop_is_refused(self):
|
|
session, engine, _ = make_crop_session(settings=LiveSettings(max_crop_bytes=2048))
|
|
session.enable_crops(2048)
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], b"\xff\xd8" + b"\x00" * 4096)
|
|
self.assertEqual(reply["code"], "CROP_TOO_LARGE")
|
|
self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_an_undecodable_crop_is_refused(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
reply = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], b"not an image")
|
|
self.assertEqual(reply["code"], "CROP_UNDECODABLE")
|
|
self.assertEqual(process(session, 3)["stats"]["crops"]["accepted"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_a_crop_cannot_be_replayed(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
first = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(first["type"], "crop_ack")
|
|
again = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(again["code"], "CROP_UNKNOWN_REQUEST")
|
|
engine.shutdown()
|
|
|
|
def test_an_unknown_request_is_refused(self):
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
reply = session.submit_crop("c-nope", 1, 1, frame_bytes())
|
|
self.assertEqual(reply["code"], "CROP_UNKNOWN_REQUEST")
|
|
engine.shutdown()
|
|
|
|
def test_an_unanswered_request_expires_and_frees_the_track(self):
|
|
session, engine, _ = make_crop_session(settings=LiveSettings(crop_ttl_frames=1))
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
process(session, 3)
|
|
# The request was made on frame 2 with a one-frame budget, so by frame 4
|
|
# it is out of date and the track it held is free again.
|
|
result = process(session, 4)
|
|
self.assertEqual(result["stats"]["crops"]["rejected"], 1, "an unanswered request is counted")
|
|
late = session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertEqual(late["code"], "CROP_UNKNOWN_REQUEST")
|
|
engine.shutdown()
|
|
|
|
def test_the_client_states_the_smaller_crop_limit(self):
|
|
session, engine, _ = make_crop_session()
|
|
self.assertEqual(session.enable_crops(4096)["max_crop_bytes"], 4096)
|
|
self.assertEqual(session.capabilities()["crop"], True)
|
|
engine.shutdown()
|
|
|
|
def test_a_refused_limit_never_turns_the_lane_on(self):
|
|
session, engine, _ = make_crop_session()
|
|
session.settings.crops = False # a lane no client has advertised yet
|
|
for bad in (True, 0, -1, 4096.0, "4096", MAX_CROP_BYTES + 1):
|
|
with self.subTest(limit=bad):
|
|
with self.assertRaises(ValueError):
|
|
session.enable_crops(bad)
|
|
self.assertFalse(session.capabilities()["crop"],
|
|
"a refused limit must leave the lane off")
|
|
self.assertEqual(session.enable_crops(4096)["max_crop_bytes"], 4096)
|
|
engine.shutdown()
|
|
|
|
def test_a_reset_while_the_crop_decodes_is_refused(self):
|
|
"""An ack promises this crop decided that track; a reset says it cannot."""
|
|
matcher = FakeMatcher()
|
|
session, engine, _ = make_crop_session(matcher=matcher)
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(request)
|
|
|
|
decoding = threading.Event()
|
|
resumed = threading.Event()
|
|
decode = engine.decode_image
|
|
|
|
def gated_decode(payload):
|
|
decoding.set()
|
|
resumed.wait(2.0)
|
|
return decode(payload)
|
|
|
|
engine.decode_image = gated_decode
|
|
replies = {}
|
|
|
|
def answer():
|
|
replies["reply"] = session.submit_crop(
|
|
request["request_id"], request["track_id"], request["frame_id"], frame_bytes()
|
|
)
|
|
|
|
worker = threading.Thread(target=answer)
|
|
worker.start()
|
|
self.assertTrue(decoding.wait(2.0), "the crop must be decoding before the reset")
|
|
session.reset()
|
|
resumed.set()
|
|
worker.join(5.0)
|
|
self.assertIn("reply", replies, "the answer never came back")
|
|
self.assertEqual(replies["reply"]["code"], "CROP_STALE")
|
|
self.assertEqual(session._crops_accepted, 0)
|
|
self.assertEqual(matcher.calls, 0, "a refused crop never reaches the classifier")
|
|
engine.shutdown()
|
|
|
|
def test_a_blurred_frame_is_not_asked_about(self):
|
|
"""More pixels of an out-of-focus scene are not more evidence."""
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
result = process_payload(session, 2, blurred_frame_bytes())
|
|
self.assertEqual(result["quality"]["hint"], "blur",
|
|
"the frame under test must be the blurry one")
|
|
self.assertIsNone(requested_crop(result))
|
|
self.assertEqual(result["stats"]["crops"]["requested"], 0)
|
|
engine.shutdown()
|
|
|
|
def test_a_distant_shelf_is_still_asked_about(self):
|
|
"""Too far to read at the frame's own resolution is the case a crop exists for."""
|
|
session, engine, detector = make_crop_session(
|
|
settings=LiveSettings(method="hybrid", min_crop_px=60)
|
|
)
|
|
detector.boxes[0].update({
|
|
"box": [300, 380, 336, 430], "width": 36, "height": 50, "center": [318, 405],
|
|
})
|
|
process(session, 1)
|
|
result = process(session, 2)
|
|
self.assertEqual(result["quality"]["hint"], "too_far",
|
|
"the frame under test must be the distant one")
|
|
self.assertIsNotNone(requested_crop(result))
|
|
engine.shutdown()
|
|
|
|
def test_a_facing_the_frame_clips_is_not_asked_about(self):
|
|
session, engine, detector = make_crop_session()
|
|
clipped = {"box": [0, 100, 36, 220], "width": 36, "height": 120, "center": [18, 160]}
|
|
detector.boxes[0].update(clipped)
|
|
process(session, 1)
|
|
result = process(session, 2)
|
|
self.assertIsNone(requested_crop(result), "half a facing is not worth the phone's pixels")
|
|
self.assertEqual(result["stats"]["crops"]["requested"], 0)
|
|
# The same facing, walked clear of the edge, is asked about: only the
|
|
# clip kept it out of the lane.
|
|
asked = None
|
|
for frame_id, x in enumerate([12, 24, 36], start=3):
|
|
detector.boxes[0].update({"box": [x, 100, x + 36, 220], "center": [x + 18, 160]})
|
|
result = process(session, frame_id)
|
|
asked = asked or requested_crop(result)
|
|
if asked is not None:
|
|
break
|
|
self.assertIsNotNone(asked, "a whole facing is worth the phone's pixels")
|
|
engine.shutdown()
|
|
|
|
def test_a_box_this_frame_did_not_measure_is_not_asked_about(self):
|
|
"""The phone crops by the box it is sent, so a stale box crops the wrong thing."""
|
|
session, engine, detector = make_crop_session()
|
|
process(session, 1)
|
|
detector.boxes[0]["confidence"] = 0.15 # too low for the tracker to trust
|
|
result = process(session, 2)
|
|
self.assertIsNone(result["quality"]["hint"], "only the detection is at fault here")
|
|
self.assertIsNone(requested_crop(result))
|
|
detector.boxes[0]["confidence"] = 0.9
|
|
self.assertIsNotNone(requested_crop(process(session, 3)),
|
|
"the same facing is asked about once a frame measures it again")
|
|
engine.shutdown()
|
|
|
|
def test_a_method_change_lets_the_small_facing_ask_again(self):
|
|
"""A crop minted under the old method is not an attempt under the new one."""
|
|
session, engine, _ = make_crop_session()
|
|
process(session, 1)
|
|
first = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(first)
|
|
session.configure(method="orb")
|
|
second = requested_crop(process(session, 3))
|
|
self.assertIsNotNone(second, "the new method gets a budget of its own")
|
|
self.assertNotEqual(second["request_id"], first["request_id"])
|
|
self.assertEqual(process(session, 4)["stats"]["crops"]["requested"], 2)
|
|
late = session.submit_crop(first["request_id"], first["track_id"],
|
|
first["frame_id"], frame_bytes())
|
|
self.assertEqual(late["code"], "CROP_UNKNOWN_REQUEST",
|
|
"the outstanding request was minted under settings the session left")
|
|
engine.shutdown()
|
|
|
|
def test_an_expired_request_leaves_exactly_one_more_ask(self):
|
|
session, engine, _ = make_crop_session(settings=LiveSettings(crop_ttl_frames=1))
|
|
process(session, 1)
|
|
first = requested_crop(process(session, 2))
|
|
self.assertIsNotNone(first)
|
|
process(session, 3)
|
|
# The one-frame budget ran out on frame 4, so the facing gets one more
|
|
# ask rather than being left with a request nobody will ever answer.
|
|
second = requested_crop(process(session, 4))
|
|
self.assertIsNotNone(second, "an expiry must not strand a facing only the crop can decide")
|
|
self.assertNotEqual(second["request_id"], first["request_id"])
|
|
process(session, 5)
|
|
process(session, 6)
|
|
result = process(session, 7)
|
|
self.assertIsNone(requested_crop(result), "the budget is bounded, not a loop")
|
|
self.assertEqual(result["stats"]["crops"]["requested"], 2)
|
|
engine.shutdown()
|
|
|
|
def test_a_drifted_label_lets_the_next_product_be_asked_about(self):
|
|
"""The facing in the frame is a different product, so it gets its own ask."""
|
|
session, engine, detector = make_crop_session()
|
|
panned(detector, 100)
|
|
process(session, 1)
|
|
request = requested_crop(process(session, 2))
|
|
session.submit_crop(request["request_id"], request["track_id"],
|
|
request["frame_id"], frame_bytes())
|
|
self.assertTrue(wait_for(lambda: session._tracks[request["track_id"]].state == "recognized"))
|
|
track_id = request["track_id"]
|
|
asked_again = None
|
|
for frame_id, x in enumerate([118, 136, 154, 172, 190, 208], start=3):
|
|
panned(detector, x)
|
|
result = process(session, frame_id)
|
|
self.assertIn(track_id, session._tracks, "the id must survive the pan")
|
|
again = requested_crop(result)
|
|
if again is not None and again["request_id"] != request["request_id"]:
|
|
asked_again = again
|
|
break
|
|
self.assertIsNotNone(asked_again,
|
|
"a drifted label must hand the new product its own crop ask")
|
|
self.assertGreater(asked_again["box"][0], request["box"][0],
|
|
"the new ask must be about where the product is now")
|
|
engine.shutdown()
|
|
|
|
def test_a_late_crop_cannot_repaint_a_settled_label(self):
|
|
"""A track that took a neighbour's label keeps it; a later crop does not overrule it."""
|
|
matcher = SecondAnswerWaits()
|
|
session, engine, detector = make_crop_session(
|
|
matcher=matcher, settings=LiveSettings(method="hybrid", min_crop_px=60)
|
|
)
|
|
# Two narrow facings that overlap by more than a label can be trusted
|
|
# across: that is the only shape in which one inherits the other's.
|
|
detector.boxes = [
|
|
{"id": 0, "box": [100, 100, 150, 340], "confidence": 0.9, "class_id": 0,
|
|
"width": 50, "height": 240, "center": [125, 220]},
|
|
{"id": 1, "box": [110, 100, 160, 340], "confidence": 0.9, "class_id": 0,
|
|
"width": 50, "height": 240, "center": [135, 220]},
|
|
]
|
|
process(session, 1)
|
|
requests = {row["track_id"]: row["crop_request"]
|
|
for row in process(session, 2)["tracks"] if "crop_request" in row}
|
|
self.assertEqual(len(requests), 2, "both facings are too small for the frame crop")
|
|
settled_id, waiting_id = list(requests)
|
|
|
|
session.submit_crop(requests[settled_id]["request_id"], settled_id,
|
|
requests[settled_id]["frame_id"], frame_bytes())
|
|
self.assertTrue(matcher.first_answered.wait(2.0), "the first crop must be decided")
|
|
session.submit_crop(requests[waiting_id]["request_id"], waiting_id,
|
|
requests[waiting_id]["frame_id"], frame_bytes())
|
|
|
|
inherited = process(session, 3)
|
|
row = next(t for t in inherited["tracks"] if t["track_id"] == waiting_id)
|
|
self.assertTrue(row["inherited"], "the facing takes the label its neighbour settled")
|
|
self.assertEqual(row["identity"]["sku_id"], "sku_a")
|
|
|
|
matcher.gate.set()
|
|
self.assertTrue(wait_for(lambda: session._tracks[waiting_id].in_flight == 0))
|
|
settled = next(t for t in process(session, 4)["tracks"] if t["track_id"] == waiting_id)
|
|
self.assertEqual(settled["identity"]["sku_id"], "sku_a",
|
|
"a crop that arrives late must not overrule a settled label")
|
|
self.assertTrue(settled["inherited"])
|
|
self.assertEqual(settled["identity"]["source"], "inherited")
|
|
self.assertEqual(session._tracks[waiting_id].attempts, 0,
|
|
"an answer that could not decide is not an attempt")
|
|
engine.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|