tngo
771e9285d7
`classify_boxes` returned only the identity, so every accepted still-image answer reached the client as a bare SKU id with no name or price. Measured through the real server on add/image6.jpg: 41 of 41 accepted detections had no `product`. The live path attached the row; the still path did not, which is the F-02 finding the decision document recorded as still live in production. The row is attached where the decision is made, so both paths carry it, and `recognize` passes it through to the wire. Verified through the running server: 41 of 41 accepted answers now carry product_name, price, and stock counts. Also corrects the decision document's full-resolution claim, which was measured before the reference-upscaling and text-signature fixes: it said 77% precision with 11 false accepts and a hub alarm, where the current lane reads 88% with 5 and no alarm. The accept line does not need to differ by input path - correct facings sit at p25 2.07 (full) and 2.87 (live), wrong ones at p75 1.00 and 1.66, so 2.0 is the boundary in both - and the results table and the stream G1 numbers are refreshed to the current measurements.
647 lines
26 KiB
Python
647 lines
26 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"])
|
|
|
|
|
|
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()
|