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.
740 lines
31 KiB
Python
740 lines
31 KiB
Python
"""
|
|
Contract tests for the recognition engine.
|
|
|
|
No YOLO weights and no network: a fake detector and a fake matcher stand in for
|
|
the model, so these tests defend the engine's own contracts - normalization,
|
|
locking, isolation between clients, empty-slot geometry, and the still-image
|
|
store.
|
|
"""
|
|
|
|
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)
|
|
|
|
from engine import ( # noqa: E402
|
|
EngineSettings,
|
|
ImageExpired,
|
|
ImageStore,
|
|
RecognitionEngine,
|
|
UndecodableImage,
|
|
UnknownMethod,
|
|
identity_payload,
|
|
)
|
|
|
|
|
|
class TextLaneSignatureTests(unittest.TestCase):
|
|
"""The lane must score a crop against the row's name, not its packshot.
|
|
|
|
l09's packshot prints "マヨネーズ", which is l10's actual name. Scoring the
|
|
packshot's OCR as part of l09's signature let l09 explain an l10 crop's
|
|
text better than l10's own name could, so l09 won l10's facings. Measured
|
|
on image6: l10 correct went 2 -> 5 of 11, wrong 3 -> 1, precision 95% ->
|
|
97%, and the l09 hub alarm cleared.
|
|
"""
|
|
|
|
ROWS = [
|
|
{"sku_id": "sku_l09", "product_name": "ピュアセレクト コクうま カロリーカット65% 360g"},
|
|
{"sku_id": "sku_l10", "product_name": "ピュアセレクト マヨネーズ 450g"},
|
|
{"sku_id": "sku_l03", "product_name": "キューピー からしマヨネーズ 200g"},
|
|
]
|
|
|
|
def lane(self):
|
|
from identity_lanes import TextLane
|
|
|
|
return TextLane.from_catalog(self.ROWS, "no-such-dir", ocr_engine=None)
|
|
|
|
def test_a_read_of_the_plain_mayonnaise_names_the_plain_mayonnaise(self):
|
|
scores = self.lane().scores_from_text(
|
|
"AJINOMOTO とれて3日以内の 国産新鮮たまご ピュア セレクト マヨネーズ"
|
|
)
|
|
self.assertEqual(max(scores, key=scores.get), "sku_l10")
|
|
|
|
def test_a_read_of_the_variant_names_the_variant(self):
|
|
scores = self.lane().scores_from_text("ピュアセレクト コクうま カロリーカット65%")
|
|
self.assertEqual(max(scores, key=scores.get), "sku_l09")
|
|
|
|
def test_shared_brand_copy_alone_does_not_name_a_row(self):
|
|
"""Every row prints マヨネーズ, so it must not decide between them.
|
|
|
|
The lane scores l10 and l03 equally on that word alone; the policy's
|
|
margin gate is what turns the tie into no answer.
|
|
"""
|
|
from identity_lanes import DecisionPolicy
|
|
|
|
scores = self.lane().scores_from_text("マヨネーズ")
|
|
self.assertEqual(scores["sku_l10"], scores["sku_l03"])
|
|
decision = DecisionPolicy().decide([], scores, text_read="マヨネーズ")
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_a_signature_never_pairs_across_two_words(self):
|
|
"""A catalog name must not claim text it does not contain.
|
|
|
|
"キューピー アマニ油" normalizes to "キユピ アマニ油"; with the space
|
|
dropped it was "キユピアマニ油", which contains "ピア" - a pair neither
|
|
word has. That pair is rare across the catalog, so IDF weighted it as
|
|
distinctive and the garbage read "キユピアク5C" was accepted as l02.
|
|
"""
|
|
from rebuild_catalog import signature_bigrams
|
|
|
|
name = "キューピー アマニ油マヨネーズ 200g"
|
|
self.assertNotIn("ピア", signature_bigrams(name))
|
|
self.assertIn("アマ", signature_bigrams(name))
|
|
|
|
def test_a_read_keeps_matching_across_the_spaces_ocr_inserts(self):
|
|
"""The read is the noisy side, so its spaces must not break a match.
|
|
|
|
Vision prints "ピュア セレクト" where the catalog name has
|
|
"ピュアセレクト". Refusing to pair across the read's spaces dropped
|
|
two correct l10 labels below the accept line, so the read pairs
|
|
across them and the signature is the side that stays strict.
|
|
"""
|
|
from rebuild_catalog import bigrams
|
|
|
|
self.assertIn("アセ", bigrams("ピュア セレクト マヨネーズ"))
|
|
|
|
def test_a_garbage_read_does_not_name_a_row(self):
|
|
"""The read that the fake pair used to accept must not name a row.
|
|
|
|
It shares only "キユ" and "ユピ" with l02's name, both of which every
|
|
Kewpie row prints. Bounded to the catalog's own vocabulary the read
|
|
keeps just those two pairs, so it scores the same against every
|
|
Kewpie row and the margin gate refuses it - the same mechanism that
|
|
already handles a read of shared brand copy alone.
|
|
"""
|
|
from identity_lanes import DecisionPolicy, TextLane
|
|
|
|
rows = self.ROWS + [{"sku_id": "sku_l02", "product_name": "キューピー アマニ油マヨネーズ 200g"}]
|
|
read = "キユピアク5C"
|
|
scores = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None).scores_from_text(read)
|
|
ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
|
|
self.assertEqual(ranked[0][1], ranked[1][1], "the read cannot tell the Kewpie rows apart")
|
|
decision = DecisionPolicy().decide([], ranked, text_read=read)
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_a_read_of_real_text_still_names_its_row(self):
|
|
"""Bounding the denominator must not cost a correct read its answer.
|
|
|
|
This is the facing the strict-read variant broke: l10's own name read
|
|
with OCR's spacing, which has to clear the accept line by margin.
|
|
"""
|
|
from identity_lanes import TEXT_ACCEPT, TEXT_MARGIN, TextLane
|
|
|
|
rows = self.ROWS + [{"sku_id": "sku_l02", "product_name": "キューピー アマニ油マヨネーズ 200g"}]
|
|
scores = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None).scores_from_text(
|
|
"213日以月A ピュア セレクト マヨネーズ"
|
|
)
|
|
ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
|
|
self.assertEqual(ranked[0][0], "sku_l10")
|
|
self.assertGreaterEqual(ranked[0][1], TEXT_ACCEPT)
|
|
self.assertGreaterEqual(ranked[0][1] - ranked[1][1], TEXT_MARGIN)
|
|
|
|
def test_one_shared_pair_is_not_enough_to_name_a_row(self):
|
|
"""A ratio of 1.0 on a single pair is not evidence.
|
|
|
|
"XXイス" contains exactly one pair the catalog knows, and the row that
|
|
contains it scores 1.0 with a full margin over every other row. On
|
|
image6 that shape named sku_l08 from the read "サソース". The floor
|
|
asks how much of the read the row actually explains, so one pair is
|
|
refused while the same read with two is accepted.
|
|
"""
|
|
from identity_lanes import DecisionPolicy, TextLane
|
|
|
|
rows = [
|
|
{"sku_id": "sku_l08", "product_name": "オイスターソース"},
|
|
{"sku_id": "sku_l01", "product_name": "キューピー マヨネーズ"},
|
|
]
|
|
lane = TextLane.from_catalog(rows, "no-such-dir", ocr_engine=None)
|
|
|
|
thin = sorted(lane.scores_from_text("XXイス").items(), key=lambda kv: kv[1], reverse=True)
|
|
self.assertEqual(thin[0][1], 1.0, "the ratio alone cannot see the difference")
|
|
self.assertIsNone(
|
|
DecisionPolicy(signatures=lane.signatures).decide([], thin, text_read="XXイス")["best_sku"]
|
|
)
|
|
|
|
enough = sorted(lane.scores_from_text("ZZソース").items(), key=lambda kv: kv[1], reverse=True)
|
|
self.assertEqual(
|
|
DecisionPolicy(signatures=lane.signatures).decide(
|
|
[], enough, text_read="ZZソース"
|
|
)["best_sku"],
|
|
"sku_l08",
|
|
)
|
|
|
|
|
|
class FakeDetector:
|
|
"""Records concurrency and returns fixed boxes."""
|
|
|
|
weights_path = "fake.pt"
|
|
|
|
def __init__(self, boxes=None):
|
|
self.boxes = boxes if boxes is not None else [
|
|
{"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9, "class_id": 0,
|
|
"width": 50, "height": 80, "center": [35, 50]},
|
|
{"id": 1, "box": [70, 10, 120, 90], "confidence": 0.8, "class_id": 0,
|
|
"width": 50, "height": 80, "center": [95, 50]},
|
|
]
|
|
self.concurrent = 0
|
|
self.max_concurrent = 0
|
|
self.calls = 0
|
|
self.lock = threading.Lock()
|
|
|
|
def detect(self, image, conf_threshold=0.25, iou_threshold=0.45):
|
|
with self.lock:
|
|
self.calls += 1
|
|
self.concurrent += 1
|
|
self.max_concurrent = max(self.max_concurrent, self.concurrent)
|
|
time.sleep(0.01)
|
|
with self.lock:
|
|
self.concurrent -= 1
|
|
return {"status": "success", "boxes": [dict(b) for b in self.boxes]}
|
|
|
|
def detect_empty_slots(self, image_shape, boxes, min_gap_ratio=0.75):
|
|
# One gap between the two boxes, expressed in the frame the engine hands
|
|
# over. Scaling with that frame is what makes the normalization test
|
|
# meaningful: the engine must translate the answer back to the client's
|
|
# pixel space.
|
|
height, width = int(image_shape[0]), int(image_shape[1])
|
|
x1, x2 = int(width * 0.30), int(width * 0.35)
|
|
return [{
|
|
"id": "empty_1",
|
|
"box": [x1, int(height * 0.10), x2, int(height * 0.90)],
|
|
"width": x2 - x1,
|
|
"height": int(height * 0.80),
|
|
"missing_items_est": 1,
|
|
"center": [(x1 + x2) // 2, height // 2],
|
|
"row_index": 1,
|
|
}]
|
|
|
|
|
|
class FakeMatcher:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def score_crop(self, crop, method="hybrid"):
|
|
self.calls += 1
|
|
time.sleep(0.01)
|
|
return {"sku_a": 40.0, "sku_b": 2.0}, {"sku_a": {"score": 40.0}}
|
|
|
|
def score_crop_fast(self, crop, shortlist=5):
|
|
return self.score_crop(crop)
|
|
|
|
def classify_crop(self, crop, method="hybrid"):
|
|
from identity import decide_identity
|
|
|
|
scores, details = self.score_crop(crop, method=method)
|
|
return decide_identity(scores, details=details, method=method)
|
|
|
|
def generate_match_visualization(self, crop, sku, method="surf", max_matches=30):
|
|
return crop
|
|
|
|
|
|
class FakeInventory:
|
|
def __init__(self):
|
|
self.products = {
|
|
"sku_a": {"sku_id": "sku_a", "product_name": "Alpha", "price": 100.0,
|
|
"number_in": 10, "number_out": 2, "number_on": 8,
|
|
"receipt_date": "2026-01-01", "expiration_date": "2026-12-31",
|
|
"local_image_path": None},
|
|
"sku_b": {"sku_id": "sku_b", "product_name": "Beta", "price": 200.0,
|
|
"number_in": 5, "number_out": 1, "number_on": 4,
|
|
"receipt_date": "2026-01-02", "expiration_date": "2026-12-30",
|
|
"local_image_path": None},
|
|
}
|
|
self.lock = threading.Lock()
|
|
|
|
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):
|
|
with self.lock:
|
|
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_engine(**overrides):
|
|
# The fake matcher is what these tests exercise, so the backend is pinned to
|
|
# `matcher`; the shipped default is the `local` lane, which reads real
|
|
# reference images and would ignore the fake.
|
|
overrides.setdefault("identity_backend", "matcher")
|
|
settings = EngineSettings(device="cpu", classify_workers=4, **overrides)
|
|
detector = FakeDetector()
|
|
matcher = FakeMatcher()
|
|
inventory = FakeInventory()
|
|
engine = RecognitionEngine(detector, matcher, inventory, settings=settings)
|
|
return engine, detector, matcher, inventory
|
|
|
|
|
|
def make_image(width=200, height=120):
|
|
import numpy as np
|
|
|
|
return np.zeros((height, width, 3), dtype="uint8")
|
|
|
|
|
|
class EngineBasicsTests(unittest.TestCase):
|
|
def test_detect_normalizes_and_reindexes(self):
|
|
engine, _, _, _ = make_engine()
|
|
boxes = engine.detect(make_image())
|
|
self.assertEqual(len(boxes), 2)
|
|
self.assertEqual([b["id"] for b in boxes], [0, 1])
|
|
self.assertEqual(boxes[0]["width"], 50)
|
|
self.assertEqual(boxes[0]["center"], [35, 50])
|
|
|
|
def test_detector_calls_never_overlap(self):
|
|
engine, detector, _, _ = make_engine()
|
|
threads = [threading.Thread(target=engine.detect, args=(make_image(),)) for _ in range(8)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
self.assertEqual(detector.calls, 8)
|
|
self.assertEqual(detector.max_concurrent, 1, "the model must be used by one caller at a time")
|
|
|
|
def test_unknown_method_is_rejected(self):
|
|
engine, _, _, _ = make_engine()
|
|
with self.assertRaises(UnknownMethod):
|
|
engine.score_crop(make_image(), method="not-a-method")
|
|
|
|
def test_undecodable_payload_raises(self):
|
|
engine, _, _, _ = make_engine()
|
|
with self.assertRaises(UndecodableImage):
|
|
engine.decode_image(b"this is not an image")
|
|
|
|
def test_crop_is_a_copy(self):
|
|
engine, _, _, _ = make_engine()
|
|
image = make_image()
|
|
crop = engine.crop(image, [0, 0, 10, 10])
|
|
crop[0, 0, 0] = 255
|
|
self.assertEqual(int(image[0, 0, 0]), 0)
|
|
|
|
|
|
class EngineClassificationTests(unittest.TestCase):
|
|
def test_classify_boxes_accepts_detection_dicts_without_size_fields(self):
|
|
"""The API hands classify_boxes stored detections, which carry no width/height."""
|
|
engine, _, _, _ = make_engine()
|
|
detections = [
|
|
{"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9, "identity": identity_payload(None)},
|
|
{"id": 1, "box": [70, 10, 120, 90], "confidence": 0.8, "identity": identity_payload(None)},
|
|
]
|
|
rows = engine.classify_boxes(make_image(), detections, method="hybrid")
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual(rows[0]["identity"]["state"], "recognized")
|
|
self.assertEqual(rows[0]["identity"]["sku_id"], "sku_a")
|
|
|
|
def test_an_accepted_detection_carries_its_catalog_row(self):
|
|
"""The phone must not have to look up a name and price from a bare SKU.
|
|
|
|
`classify_boxes` returned only the identity, so every accepted
|
|
still-image answer reached the client as an id with no product row:
|
|
measured through the real server, 41 of 41 accepted detections had no
|
|
`product`. The live path attached it; the still path did not.
|
|
"""
|
|
engine, _, _, _ = make_engine()
|
|
detections = [
|
|
{"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9,
|
|
"identity": identity_payload(None)},
|
|
]
|
|
rows = engine.classify_boxes(make_image(), detections, method="hybrid")
|
|
self.assertEqual(rows[0]["identity"]["state"], "recognized")
|
|
self.assertEqual(rows[0]["product"]["product_name"], "Alpha")
|
|
self.assertEqual(rows[0]["product"]["price"], 100.0)
|
|
|
|
result = engine.recognize(make_image(), method="hybrid")
|
|
accepted = [
|
|
row for row in result["detections"]
|
|
if row["identity"]["identity_level"] == "accepted"
|
|
]
|
|
self.assertTrue(accepted, "the fake detector and matcher accept a box")
|
|
for row in accepted:
|
|
self.assertIn("product", row, "an accepted answer must carry its row")
|
|
|
|
def test_a_failing_crop_reports_unrecognized_not_pending(self):
|
|
"""One bad crop must not fail the batch, and must not read as pending.
|
|
|
|
`identity_payload(None)` is the "not yet classified" shape; an
|
|
operational failure has to stay distinguishable from absence.
|
|
"""
|
|
engine, _, matcher, _ = make_engine()
|
|
|
|
def explode(crop, method=None):
|
|
raise RuntimeError("classifier blew up")
|
|
|
|
matcher.score_crop = explode
|
|
rows = engine.classify_boxes(make_image(), [
|
|
{"id": 0, "box": [10, 10, 60, 90], "confidence": 0.9},
|
|
], method="hybrid")
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["identity"]["state"], "unrecognized")
|
|
self.assertIsNone(rows[0]["identity"]["sku_id"])
|
|
|
|
def test_parallel_classification_matches_sequential(self):
|
|
engine, _, matcher, _ = make_engine()
|
|
boxes = [
|
|
{"id": index, "box": [10 + index, 10, 60 + index, 90], "confidence": 0.9,
|
|
"width": 50, "height": 80, "center": [35, 50]}
|
|
for index in range(12)
|
|
]
|
|
parallel = engine.classify_boxes(make_image(), boxes, method="hybrid")
|
|
sequential = [engine.score_crop(engine.crop(make_image(), box["box"]), "hybrid") for box in boxes]
|
|
self.assertEqual(len(parallel), 12)
|
|
self.assertEqual(len(sequential), 12)
|
|
for row in parallel:
|
|
self.assertEqual(row["identity"]["sku_id"], "sku_a")
|
|
|
|
def test_summary_counts_states(self):
|
|
engine, _, _, _ = make_engine()
|
|
detections = [
|
|
{"identity": {"state": "recognized", "sku_id": "sku_a"}},
|
|
{"identity": {"state": "recognized", "sku_id": "sku_a"}},
|
|
{"identity": {"state": "needs_confirmation", "sku_id": None}},
|
|
{"identity": {"state": "unrecognized", "sku_id": None}},
|
|
]
|
|
summary = engine.summarize(detections, [])
|
|
self.assertEqual(summary["recognized"], 2)
|
|
self.assertEqual(summary["needs_confirmation"], 1)
|
|
self.assertEqual(summary["unrecognized"], 1)
|
|
self.assertEqual(summary["by_sku"], [{"sku_id": "sku_a", "facings": 2}])
|
|
|
|
|
|
class EmptySlotNormalizationTests(unittest.TestCase):
|
|
def test_slots_scale_with_the_frame(self):
|
|
"""The detector's thresholds are absolute pixels, so a half-size frame must
|
|
produce the same slots (scaled) as the reference-size frame."""
|
|
engine, _, _, _ = make_engine()
|
|
boxes_small = [
|
|
{"box": [5, 5, 30, 45]},
|
|
{"box": [35, 5, 60, 45]},
|
|
]
|
|
slots_small = engine.empty_slots((60, 100), boxes_small)
|
|
boxes_big = [
|
|
{"box": [10, 10, 60, 90]},
|
|
{"box": [70, 10, 120, 90]},
|
|
]
|
|
slots_big = engine.empty_slots((120, 200), boxes_big)
|
|
self.assertEqual(len(slots_small), 1)
|
|
self.assertEqual(len(slots_big), 1)
|
|
# the small frame is scaled up to the reference width, so the returned
|
|
# slot comes back in the small frame's own pixels
|
|
self.assertLess(slots_small[0]["box"][2], slots_big[0]["box"][2])
|
|
|
|
|
|
class ImageStoreTests(unittest.TestCase):
|
|
def test_roundtrip_and_expiry(self):
|
|
store = ImageStore(max_items=2, ttl_s=0.05, max_bytes=10_000_000)
|
|
stored = store.put(make_image())
|
|
self.assertEqual(store.get(stored.image_id).width, 200)
|
|
time.sleep(0.08)
|
|
with self.assertRaises(ImageExpired):
|
|
store.get(stored.image_id)
|
|
|
|
def test_bounded_by_count(self):
|
|
store = ImageStore(max_items=2, ttl_s=60, max_bytes=10_000_000)
|
|
ids = [store.put(make_image()).image_id for _ in range(4)]
|
|
with self.assertRaises(ImageExpired):
|
|
store.get(ids[0])
|
|
self.assertEqual(store.get(ids[-1]).width, 200)
|
|
|
|
|
|
class CatalogTests(unittest.TestCase):
|
|
def test_catalog_and_stock_update(self):
|
|
engine, _, _, _ = make_engine()
|
|
catalog = engine.catalog()
|
|
self.assertEqual(len(catalog), 2)
|
|
self.assertEqual(catalog[0]["image_url"], "/v1/catalog/sku_a/image")
|
|
product = engine.update_stock("sku_a", -1)
|
|
self.assertEqual(product["number_on"], 7)
|
|
self.assertEqual(product["number_out"], 3)
|
|
self.assertIsNone(engine.update_stock("nope", -1))
|
|
|
|
|
|
class GalleryLoadTests(unittest.TestCase):
|
|
def test_a_failed_gallery_load_returns_none_and_is_not_retried(self):
|
|
engine, _, _, _ = make_engine(identity_backend="gallery", stock_dir="does-not-exist")
|
|
|
|
class BrokenGallery:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def load(self):
|
|
raise FileNotFoundError("DINOv2 weights not found")
|
|
|
|
import gallery as gallery_module
|
|
|
|
original = gallery_module.ProductGallery
|
|
gallery_module.ProductGallery = BrokenGallery
|
|
try:
|
|
self.assertIsNone(engine.identity_gallery())
|
|
self.assertIsNone(engine.identity_gallery())
|
|
finally:
|
|
gallery_module.ProductGallery = original
|
|
engine.shutdown()
|
|
self.assertIs(engine._gallery, False)
|
|
|
|
|
|
class LocalLaneTests(unittest.TestCase):
|
|
"""The shipped default backend, which the fake matcher cannot exercise.
|
|
|
|
A two-SKU catalog of real images is built in a temp dir, so the lane reads
|
|
genuine reference files and the decision is the one production makes.
|
|
"""
|
|
|
|
def make_catalog(self):
|
|
import tempfile
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
root = tempfile.mkdtemp(prefix="lane-catalog-")
|
|
rng = np.random.default_rng(7)
|
|
for sku, base in (("sku_a", (200, 60, 60)), ("sku_b", (60, 60, 200))):
|
|
image = np.zeros((400, 300, 3), dtype="uint8")
|
|
image[:, :] = base
|
|
# Structured, high-contrast marks: SIFT needs corners, not noise.
|
|
for index in range(12):
|
|
x = int(rng.integers(10, 280))
|
|
y = int(rng.integers(10, 380))
|
|
cv2.rectangle(image, (x, y), (x + 40, y + 40), (255, 255, 255), 3)
|
|
cv2.putText(image, sku[-1].upper(), (x + 5, y + 30),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 0, 0), 2)
|
|
cv2.imwrite(os.path.join(root, f"{sku}.jpg"), image)
|
|
with open(os.path.join(root, "stock.csv"), "w", encoding="utf-8") as handle:
|
|
handle.write("SKU_ID,Product_Name,Price,Number_In,Number_Out,Number_On\n")
|
|
handle.write("sku_a,Alpha,100,1,0,1\n")
|
|
handle.write("sku_b,Beta,200,1,0,1\n")
|
|
return root
|
|
|
|
def test_decide_crop_names_the_matching_sku(self):
|
|
import cv2
|
|
|
|
root = self.make_catalog()
|
|
engine, _, _, _ = make_engine(identity_backend="local", stock_dir=root)
|
|
try:
|
|
self.assertEqual(engine.identity_scale, "local")
|
|
self.assertIsNotNone(engine.identity_lane())
|
|
# The reference itself is the easiest possible query for its own lane.
|
|
crop = cv2.imread(os.path.join(root, "sku_a.jpg"))
|
|
decision = engine.decide_crop(crop)
|
|
self.assertEqual(decision.get("best_sku"), "sku_a")
|
|
self.assertEqual(decision.get("identity_level"), "accepted")
|
|
finally:
|
|
engine.shutdown()
|
|
|
|
def test_score_crop_is_keyed_by_canonical_sku(self):
|
|
import cv2
|
|
|
|
root = self.make_catalog()
|
|
engine, _, _, _ = make_engine(identity_backend="local", stock_dir=root)
|
|
try:
|
|
scores = engine.score_crop(cv2.imread(os.path.join(root, "sku_b.jpg")))
|
|
self.assertTrue(scores)
|
|
# No view-suffixed ids: every key is a catalog SKU.
|
|
self.assertTrue(all("_" not in sku[4:] for sku in scores), scores)
|
|
finally:
|
|
engine.shutdown()
|
|
|
|
def test_a_featureless_crop_is_not_named(self):
|
|
import numpy as np
|
|
|
|
root = self.make_catalog()
|
|
engine, _, _, _ = make_engine(identity_backend="local", stock_dir=root)
|
|
try:
|
|
blank = np.zeros((200, 150, 3), dtype="uint8")
|
|
decision = engine.decide_crop(blank)
|
|
self.assertIsNone(decision.get("best_sku"))
|
|
finally:
|
|
engine.shutdown()
|
|
|
|
def test_warm_up_builds_both_lanes(self):
|
|
root = self.make_catalog()
|
|
engine, _, _, _ = make_engine(identity_backend="local", stock_dir=root)
|
|
try:
|
|
engine.warm_up()
|
|
# Built eagerly, so the first live frame never pays for them.
|
|
self.assertIsNotNone(engine._lane)
|
|
self.assertIsNotNone(engine._text_lane)
|
|
finally:
|
|
engine.shutdown()
|
|
|
|
|
|
class DecisionPolicyTests(unittest.TestCase):
|
|
"""The two jobs the text lane does, and the boundary between them."""
|
|
|
|
def policy(self):
|
|
from identity_lanes import DecisionPolicy
|
|
|
|
return DecisionPolicy()
|
|
|
|
def test_local_answer_stands_when_the_text_agrees(self):
|
|
decision = self.policy().decide([("sku_a", 5.0)], [("sku_a", 0.9)])
|
|
self.assertEqual(decision["best_sku"], "sku_a")
|
|
self.assertEqual(decision["source"], "local")
|
|
|
|
def test_text_vetoes_a_local_answer_naming_another_row(self):
|
|
"""The near-unknown case: real geometry, different words.
|
|
|
|
The veto blocks the local answer, and the text lane then names its own
|
|
row - the two jobs compose, so the result is the text's SKU, never the
|
|
vetoed local one.
|
|
"""
|
|
decision = self.policy().decide(
|
|
[("sku_a", 5.0)], [("sku_b", 0.9)], text_read="コクうま 65%"
|
|
)
|
|
self.assertEqual(decision["best_sku"], "sku_b")
|
|
self.assertEqual(decision["source"], "text")
|
|
|
|
def test_text_mentioning_the_local_candidate_is_not_a_veto(self):
|
|
"""A crop reading "マヨネーズ" scores every mayonnaise row.
|
|
|
|
The local candidate is usually one of them, so a veto needs the text to
|
|
*prefer* another row, not merely to mention it: here the text mentions
|
|
sku_b but explains sku_a better, so the local answer stands.
|
|
"""
|
|
decision = self.policy().decide([("sku_a", 5.0)], [("sku_b", 0.4), ("sku_a", 0.45)])
|
|
self.assertEqual(decision["best_sku"], "sku_a")
|
|
self.assertEqual(decision["source"], "local")
|
|
|
|
def test_the_lanes_own_order_wins_over_the_value_order(self):
|
|
"""Order and value are different quantities on the local lane.
|
|
|
|
The order comes from `rank_key` (inliers weighted by hue); the value is
|
|
the margin over the runner-up. Re-sorting by value names a different SKU
|
|
than the lane chose - here sku_b has the larger value but sku_a is the
|
|
lane's own first choice, and sku_a is what must be answered.
|
|
"""
|
|
decision = self.policy().decide([("sku_a", 3.0), ("sku_b", 5.0)], [])
|
|
self.assertEqual(decision["best_sku"], "sku_a")
|
|
|
|
def test_a_size_only_read_does_not_fill_in(self):
|
|
"""Size and number tokens are not identity text."""
|
|
decision = self.policy().decide([], [("sku_b", 0.9)], text_read="400g")
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_a_size_only_read_still_vetoes(self):
|
|
"""The veto keeps "65%" usable: the local lane supplied the evidence."""
|
|
decision = self.policy().decide(
|
|
[("sku_a", 5.0)], [("sku_b", 0.9)], text_read="65%"
|
|
)
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_a_named_read_with_a_size_still_fills_in(self):
|
|
decision = self.policy().decide([], [("sku_b", 0.9)], text_read="コクうま 65%")
|
|
self.assertEqual(decision["best_sku"], "sku_b")
|
|
|
|
def test_text_fills_in_when_local_has_nothing(self):
|
|
decision = self.policy().decide([], [("sku_b", 0.9)], text_read="コクうま 65%")
|
|
self.assertEqual(decision["best_sku"], "sku_b")
|
|
self.assertEqual(decision["source"], "text")
|
|
|
|
def test_a_ranking_without_a_read_does_not_fill_in(self):
|
|
"""Omitting the read is the safe default, not an ungated one.
|
|
|
|
A caller that has a ranking and no read has not shown the text can name
|
|
anything, so the fill-in job stays closed.
|
|
"""
|
|
decision = self.policy().decide([], [("sku_b", 0.9)])
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_quantity_strings_are_not_names(self):
|
|
"""Every pure quantity string, including the ones a digits-only rule let through.
|
|
|
|
"400g 50%" and "450g カロリー50%" survived the first version of this
|
|
gate: it counted non-digit bigrams, and the unit's own letters ("0G",
|
|
"0%") are non-digits. Cutting the whole quantity token is what closes
|
|
it. "450g カロリー50%" still passes, and correctly so - "カロリー" is
|
|
four characters of real text, and it is the text lane's job to find
|
|
that no catalog row is named by it.
|
|
"""
|
|
from identity_lanes import carries_identity
|
|
|
|
for read in ("400g", "400g 50%", "500ml", "65%", "3日", "中辛"):
|
|
with self.subTest(read=read):
|
|
self.assertFalse(carries_identity(read), f"{read!r} is not a name")
|
|
for read in ("コクうま 65%", "450g カロリー50%"):
|
|
with self.subTest(read=read):
|
|
self.assertTrue(carries_identity(read), f"{read!r} carries name text")
|
|
|
|
def test_text_below_its_own_line_does_not_fill_in(self):
|
|
decision = self.policy().decide([], [("sku_b", 0.3)])
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_a_close_text_pair_does_not_fill_in(self):
|
|
"""The margin gate: two rows the text explains equally must not name one."""
|
|
decision = self.policy().decide([], [("sku_b", 0.9), ("sku_c", 0.85)])
|
|
self.assertIsNone(decision["best_sku"])
|
|
|
|
def test_no_evidence_is_rejected(self):
|
|
decision = self.policy().decide([], [])
|
|
self.assertIsNone(decision["best_sku"])
|
|
self.assertEqual(decision["identity_level"], "rejected")
|
|
|
|
|
|
class LocalLaneConcurrencyTests(unittest.TestCase):
|
|
"""The lane is shared by the classify pool, so `rank` must be thread-safe.
|
|
|
|
One `cv2.BFMatcher` is reused across calls. The classify pool runs up to
|
|
eight workers, so a lane that mutated matcher state per call would return
|
|
another crop's answer under load - the failure would look like a
|
|
recognition error, not a crash.
|
|
"""
|
|
|
|
def test_concurrent_ranking_matches_solo_ranking(self):
|
|
import threading
|
|
|
|
import cv2
|
|
|
|
from identity_lanes import LocalLane
|
|
|
|
stock_dir = os.path.join(os.path.dirname(current_dir), "stock-lopia-clean")
|
|
if not os.path.isdir(stock_dir):
|
|
self.skipTest("no catalog checked out")
|
|
image_path = os.path.join(os.path.dirname(current_dir), "add", "image6.jpg")
|
|
if not os.path.isfile(image_path):
|
|
self.skipTest("no evaluation photo checked out")
|
|
lane = LocalLane(stock_dir)
|
|
image = cv2.imread(image_path)
|
|
crops = [image[700:1000, 600 + 40 * i:800 + 40 * i] for i in range(8)]
|
|
solo = lane.rank(crops[0])
|
|
results = {}
|
|
errors = []
|
|
|
|
def work(index: int) -> None:
|
|
try:
|
|
for _ in range(6):
|
|
results[index] = lane.rank(crops[index])
|
|
except Exception as error: # noqa: BLE001 - reported below
|
|
errors.append(repr(error))
|
|
|
|
threads = [threading.Thread(target=work, args=(i,)) for i in range(8)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(results[0], solo, "a shared matcher must not cross crops")
|
|
|
|
|
|
class CatalogImageTests(unittest.TestCase):
|
|
def test_catalog_image_path_is_resolved_through_inventory(self):
|
|
engine, _, _, _ = make_engine()
|
|
self.assertIsNone(engine.catalog_image_path("../../etc/passwd"))
|
|
self.assertIsNone(engine.catalog_image_path("sku_a"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|