cf7e8a5a88
Two false positives came out of running the rules against a real shelf, and both had the same shape: the rules were arguing with our own detector instead of with the shelf. 1. One tag per product, not per facing. Measured: a row carries 16-18 facings but only 4-5 tags, because four facings of a mayonnaise share a single tag. Counting facings invents missing tags on a shelf that is perfectly labelled. The rules now group a row's facings into products - by identity when the facings were recognised, by size and spacing otherwise - and count products. 2. A product's tag may sit anywhere along its run. Measured: a run of ピュアセレクト facings was split by geometry because one facing was detected narrower than its neighbours (54 px beside 150 px), and the fragment was then reported as missing the tag that was sitting further along its own run. A product now counts as missing only when no tag exists for its SKU anywhere in the row, and a fragment far narrower than the row's typical product is not judged at all. 3. A mismatch needs one trustworthy anchor. Measured noise: a wrong identity plus a garbled tag read (name match 0.11, no JAN) produced a confident "tag says 90, catalog says 188" that was really two bad reads disagreeing. Without the tag's JAN or a name that resembles the product, that is now a hint, not a violation. Measured on add/image6.jpg through the real still path: 0 false violations without identities (was 1), and with identities the only finding left is a genuine tag-versus-catalog price difference. Coverage hints now read "tags detected for 2 of 4 products in this row". 143 tests pass. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
358 lines
16 KiB
Python
358 lines
16 KiB
Python
"""
|
||
Contract tests for reconciliation and the compliance rule family.
|
||
|
||
Pure functions over evidence, so these run without OCR, weights, or a camera.
|
||
The rules that matter most here are the fail-closed ones: unreadable evidence
|
||
must never become a violation.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
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 reconcile import ( # noqa: E402
|
||
ComplianceSettings,
|
||
ComplianceTracker,
|
||
RECONCILED_CONFIRMED,
|
||
RECONCILED_MISMATCH,
|
||
RECONCILED_NO_TAG,
|
||
RECONCILED_TAG_ONLY,
|
||
RECONCILED_UNVERIFIED,
|
||
RULE_MISSING_TAG,
|
||
RULE_MISSING_TAG_GROUP,
|
||
RULE_ORPHAN_TAG,
|
||
RULE_PRICE_MISMATCH,
|
||
RULE_UNREADABLE_TAGS,
|
||
evaluate_compliance,
|
||
longest_common_substring_ratio,
|
||
normalize_name,
|
||
reconcile,
|
||
summarize_findings,
|
||
)
|
||
from tags import TAG_STATE_ABSENT, TAG_STATE_READABLE, TAG_STATE_UNREADABLE # noqa: E402
|
||
|
||
|
||
def accepted(sku="sku_003"):
|
||
return {"state": "recognized", "sku_id": sku, "identity_level": "accepted", "confidence": 0.9}
|
||
|
||
|
||
def tag(state=TAG_STATE_READABLE, **overrides):
|
||
payload = {
|
||
"id": "tag_1",
|
||
"box": [100, 300, 200, 340],
|
||
"state": state,
|
||
"price": None,
|
||
"regular_price": None,
|
||
"name": None,
|
||
"confidence": 0.8,
|
||
"observations": 3,
|
||
"settled": True,
|
||
}
|
||
payload.update(overrides)
|
||
return payload
|
||
|
||
|
||
def product(price=168.0, name="Tropicana エッセンシャルズ マルチビタミン", jan=None):
|
||
payload = {"sku_id": "sku_003", "product_name": name, "price": price}
|
||
if jan:
|
||
payload["jan"] = jan
|
||
return payload
|
||
|
||
|
||
class NameMatchingTests(unittest.TestCase):
|
||
def test_normalization_strips_noise(self):
|
||
self.assertEqual(normalize_name("Ｔｒｏｐｉｃａｎａ エッセンシャルズ"), "tropicanaエッセンシャルズ")
|
||
self.assertEqual(normalize_name("キュービー<310g>"), "キュービー")
|
||
|
||
def test_lcs_ratio(self):
|
||
self.assertEqual(longest_common_substring_ratio("tropicana", "tropicana"), 1.0)
|
||
self.assertGreater(longest_common_substring_ratio("tropicanaエッセンシャルズ", "tropicana"), 0.5)
|
||
self.assertEqual(longest_common_substring_ratio("", "x"), 0.0)
|
||
|
||
|
||
class ReconcileTests(unittest.TestCase):
|
||
def test_price_agreement_confirms(self):
|
||
result = reconcile(accepted(), tag(price=168), product(price=168.0))
|
||
self.assertEqual(result["state"], RECONCILED_CONFIRMED)
|
||
self.assertTrue(any(signal["kind"] == "price" and signal["agrees"] for signal in result["signals"]))
|
||
|
||
def test_regular_price_agreement_confirms(self):
|
||
result = reconcile(accepted(), tag(price=140, regular_price=168), product(price=168.0))
|
||
self.assertEqual(result["state"], RECONCILED_CONFIRMED)
|
||
|
||
def test_price_disagreement_is_a_mismatch_with_both_values(self):
|
||
result = reconcile(accepted(), tag(price=269), product(price=168.0))
|
||
self.assertEqual(result["state"], RECONCILED_MISMATCH)
|
||
self.assertIn("269", result["message"])
|
||
self.assertIn("168", result["message"])
|
||
|
||
def test_name_agreement_confirms_even_without_a_price(self):
|
||
result = reconcile(accepted(), tag(name="Tropicana エッセンシャルズ"), product(price=None, name="Tropicana エッセンシャルズ マルチビタミン"))
|
||
self.assertEqual(result["state"], RECONCILED_CONFIRMED)
|
||
|
||
def test_jan_disagreement_is_a_mismatch(self):
|
||
result = reconcile(accepted(), tag(price=168, jan="4902105000001"), product(price=168.0, jan="4902105000002"))
|
||
self.assertEqual(result["state"], RECONCILED_MISMATCH)
|
||
|
||
def test_jan_agreement_confirms(self):
|
||
result = reconcile(accepted(), tag(price=None, jan="4902105000001"), product(price=None, jan="4902105000001"))
|
||
self.assertEqual(result["state"], RECONCILED_CONFIRMED)
|
||
|
||
def test_tag_read_but_nothing_comparable(self):
|
||
result = reconcile(accepted(), tag(name="コーヒー"), product(price=None, name=None))
|
||
self.assertEqual(result["state"], RECONCILED_UNVERIFIED)
|
||
|
||
def test_unreadable_tag_is_unverified_not_missing(self):
|
||
result = reconcile(accepted(), tag(state=TAG_STATE_UNREADABLE), product())
|
||
self.assertEqual(result["state"], RECONCILED_UNVERIFIED)
|
||
|
||
def test_absent_tag(self):
|
||
result = reconcile(accepted(), tag(state=TAG_STATE_ABSENT), product())
|
||
self.assertEqual(result["state"], RECONCILED_NO_TAG)
|
||
|
||
def test_readable_tag_without_a_product(self):
|
||
result = reconcile(None, tag(price=269), None)
|
||
self.assertEqual(result["state"], RECONCILED_TAG_ONLY)
|
||
|
||
def test_nothing_to_reconcile_returns_none(self):
|
||
self.assertIsNone(reconcile(None, None, None))
|
||
self.assertIsNone(reconcile({"state": "unconfirmed"}, tag(state=TAG_STATE_ABSENT), product()))
|
||
|
||
|
||
# --------------------------------------------------------------------------- compliance
|
||
|
||
|
||
def band(row_index, box):
|
||
class Band:
|
||
pass
|
||
|
||
instance = Band()
|
||
instance.row_index = row_index
|
||
instance.box = box
|
||
return instance
|
||
|
||
|
||
def track(track_id, box, sku="sku_003", state="recognized"):
|
||
return {
|
||
"track_id": track_id,
|
||
"box": box,
|
||
"raw_box": box,
|
||
"identity": {"state": state, "sku_id": sku if state == "recognized" else None},
|
||
}
|
||
|
||
|
||
# Wide enough that separate products can sit apart in a row.
|
||
BANDS = [band(1, [0, 300, 3000, 340]), band(2, [0, 600, 3000, 640])]
|
||
|
||
|
||
class ComplianceTests(unittest.TestCase):
|
||
def test_missing_tag_on_a_covered_row_is_a_violation(self):
|
||
# Four separate products (wide gaps), three with tag evidence - 75%, above
|
||
# the 70% bar - and one with none.
|
||
tracks = [
|
||
track(index, [10 + index * 300, 100, 140 + index * 300, 300], sku=f"sku_{index}")
|
||
for index in range(4)
|
||
]
|
||
tags = {
|
||
0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"),
|
||
1: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"),
|
||
2: tag(state=TAG_STATE_ABSENT),
|
||
3: tag(state=TAG_STATE_READABLE, price=188, id="tag_c"),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0], tags[1], tags[3]], BANDS)
|
||
rules = [finding.rule for finding in findings]
|
||
self.assertIn(RULE_MISSING_TAG, rules)
|
||
missing = [finding for finding in findings if finding.rule == RULE_MISSING_TAG][0]
|
||
self.assertEqual(missing.count, 1, "one finding per row, counting the products")
|
||
|
||
def test_a_row_with_poor_tag_coverage_is_not_evaluated(self):
|
||
"""The real failure mode: our detector found tags for only a few of many
|
||
facings. Claiming the rest are missing would be a false violation."""
|
||
tracks = [track(index, [10 + index * 300, 100, 90 + index * 300, 300]) for index in range(6)]
|
||
tags = {index: tag(state=TAG_STATE_ABSENT) for index in range(6)}
|
||
tags[0] = tag(state=TAG_STATE_READABLE, price=168, id="tag_a")
|
||
findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS)
|
||
self.assertEqual([finding.rule for finding in findings], [RULE_UNREADABLE_TAGS])
|
||
self.assertEqual(findings[0].severity, "hint")
|
||
self.assertEqual(summarize_findings(findings)["violations"], 0)
|
||
|
||
def test_unreadable_row_yields_a_hint_never_a_violation(self):
|
||
"""The demo image case: tags exist but cannot be read. Nothing may be claimed."""
|
||
tracks = [
|
||
track(1, [10, 100, 90, 300]),
|
||
track(2, [110, 100, 190, 300]),
|
||
track(3, [210, 100, 290, 300]),
|
||
]
|
||
tags = {
|
||
1: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_1"),
|
||
2: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_2"),
|
||
3: tag(state=TAG_STATE_UNREADABLE, price=None, id="tag_3"),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[1], tags[2], tags[3]], BANDS)
|
||
self.assertEqual([finding.rule for finding in findings], [RULE_UNREADABLE_TAGS])
|
||
self.assertEqual(findings[0].severity, "hint")
|
||
self.assertEqual(summarize_findings(findings)["violations"], 0)
|
||
|
||
def test_a_row_with_no_tag_evidence_at_all_is_silent(self):
|
||
tracks = [track(1, [10, 100, 90, 300])]
|
||
tags = {1: tag(state=TAG_STATE_ABSENT)}
|
||
findings = evaluate_compliance(tracks, tags, [], BANDS)
|
||
self.assertEqual(findings, [])
|
||
|
||
def test_group_of_consecutive_facings_without_tags(self):
|
||
# Coverage counts *products* with any tag evidence, so the row needs enough
|
||
# of it to be judged at all (bar is 70%: 7 of 10 products).
|
||
tracks = [
|
||
track(index, [10 + index * 300, 100, 70 + index * 300, 300], sku=f"sku_{index}")
|
||
for index in range(10)
|
||
]
|
||
tags = {index: tag(state=TAG_STATE_READABLE, price=168 + index, id=f"tag_{index}") for index in range(10)}
|
||
for absent in (2, 3, 4):
|
||
tags[absent] = tag(state=TAG_STATE_ABSENT, id=None)
|
||
findings = evaluate_compliance(tracks, tags, [t for t in tags.values() if t["id"]], BANDS)
|
||
groups = [finding for finding in findings if finding.rule == RULE_MISSING_TAG_GROUP]
|
||
self.assertEqual(len(groups), 1)
|
||
self.assertEqual(groups[0].count, 3)
|
||
self.assertEqual(groups[0].row_index, 1)
|
||
self.assertIn("3 products", groups[0].detail)
|
||
|
||
def test_a_product_whose_tag_sits_elsewhere_in_the_row_is_not_missing(self):
|
||
"""Measured false positive: a run of ピュアセレクト facings was split by
|
||
geometry, and the piece was reported as missing the tag that was sitting
|
||
further along its own run. Identity and the row's tags both guard it."""
|
||
tracks = [
|
||
track(0, [10, 100, 160, 300], sku="sku_a"),
|
||
track(1, [170, 100, 320, 300], sku="sku_a"),
|
||
track(2, [330, 100, 460, 300], sku="sku_a"),
|
||
]
|
||
tags = {
|
||
0: tag(state=TAG_STATE_READABLE, price=209, id="tag_a"),
|
||
1: tag(state=TAG_STATE_ABSENT),
|
||
2: tag(state=TAG_STATE_ABSENT),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS)
|
||
self.assertEqual(findings, [], "the product's tag is on its own run")
|
||
|
||
def test_two_missing_products_are_not_a_group(self):
|
||
tracks = [
|
||
track(index, [10 + index * 300, 100, 140 + index * 300, 300], sku=f"sku_{index}")
|
||
for index in range(4)
|
||
]
|
||
tags = {
|
||
0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"),
|
||
1: tag(state=TAG_STATE_ABSENT),
|
||
2: tag(state=TAG_STATE_READABLE, price=188, id="tag_b"),
|
||
3: tag(state=TAG_STATE_READABLE, price=198, id="tag_c"),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0], tags[2], tags[3]], BANDS)
|
||
self.assertNotIn(RULE_MISSING_TAG_GROUP, [finding.rule for finding in findings])
|
||
missing = [finding for finding in findings if finding.rule == RULE_MISSING_TAG]
|
||
self.assertEqual(len(missing), 1)
|
||
self.assertEqual(missing[0].count, 1)
|
||
|
||
def test_facings_of_one_product_share_a_tag(self):
|
||
"""The measured domain rule: this store prints one tag per product, not per
|
||
facing - a row carries ~16 facings but only ~4 tags. Counting facings would
|
||
accuse a perfectly labelled shelf."""
|
||
# Four facings of the same product (touching, same size) under one tag.
|
||
tracks = [track(index, [10 + index * 80, 100, 90 + index * 80, 300]) for index in range(4)]
|
||
tags = {
|
||
0: tag(state=TAG_STATE_READABLE, price=269, id="tag_a"),
|
||
1: tag(state=TAG_STATE_ABSENT),
|
||
2: tag(state=TAG_STATE_ABSENT),
|
||
3: tag(state=TAG_STATE_ABSENT),
|
||
}
|
||
findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS)
|
||
self.assertEqual(findings, [], "one tag covers the whole product: nothing is wrong")
|
||
|
||
def test_orphan_tag(self):
|
||
tracks = [track(1, [10, 100, 90, 300])]
|
||
tags = {1: tag(state=TAG_STATE_READABLE, price=168, id="tag_a")}
|
||
all_tags = [tags[1], tag(state=TAG_STATE_READABLE, price=999, id="tag_lonely", box=[300, 600, 400, 640])]
|
||
findings = evaluate_compliance(tracks, tags, all_tags, BANDS)
|
||
orphans = [finding for finding in findings if finding.rule == RULE_ORPHAN_TAG]
|
||
self.assertEqual(len(orphans), 1)
|
||
self.assertEqual(orphans[0].box, [300, 600, 400, 640])
|
||
|
||
def test_price_mismatch_becomes_a_finding(self):
|
||
# A reportable mismatch carries an anchor: the tag's JAN was read, so we
|
||
# know which tag we are comparing.
|
||
tracks = [track(1, [10, 100, 90, 300])]
|
||
tags = {1: tag(state=TAG_STATE_READABLE, price=269, jan="357")}
|
||
reconciliations = {1: reconcile(accepted(), tags[1], product(price=168.0))}
|
||
findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS, reconciliations)
|
||
mismatches = [finding for finding in findings if finding.rule == RULE_PRICE_MISMATCH]
|
||
self.assertEqual(len(mismatches), 1)
|
||
self.assertIn("269", mismatches[0].detail)
|
||
|
||
def test_a_mismatch_without_an_anchor_is_only_a_hint(self):
|
||
"""Measured noise: a wrong identity plus a garbled tag read (name match
|
||
0.11, no JAN) produced a confident "tag says 90, catalog says 188" that was
|
||
really two bad reads disagreeing. Without an anchor it is a hint."""
|
||
tracks = [track(1, [10, 100, 90, 300])]
|
||
tags = {1: tag(state=TAG_STATE_READABLE, price=90, name="ラスお要をおいしくだべおまにすくg)")}
|
||
reconciliations = {1: reconcile(accepted(), tags[1], product(price=188.0))}
|
||
findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS, reconciliations)
|
||
self.assertNotIn(RULE_PRICE_MISMATCH, [finding.rule for finding in findings])
|
||
hints = [finding for finding in findings if finding.severity == "hint"]
|
||
self.assertTrue(hints)
|
||
self.assertEqual(summarize_findings(findings)["violations"], 0)
|
||
|
||
def test_summary_counts_only_violations(self):
|
||
findings = [
|
||
evaluate_compliance(
|
||
[track(1, [10, 100, 90, 300])],
|
||
{1: tag(state=TAG_STATE_READABLE, price=269, jan="357")},
|
||
[tag(state=TAG_STATE_READABLE, price=269, jan="357")],
|
||
BANDS,
|
||
{1: reconcile(accepted(), tag(price=269), product(price=168.0))},
|
||
)
|
||
][0]
|
||
summary = summarize_findings(findings)
|
||
self.assertGreaterEqual(summary["violations"], 1)
|
||
self.assertIn(RULE_PRICE_MISMATCH, summary["by_rule"])
|
||
|
||
|
||
class StabilityTests(unittest.TestCase):
|
||
def test_findings_need_to_persist(self):
|
||
tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5))
|
||
from reconcile import Finding
|
||
|
||
finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10])
|
||
first = tracker.update([finding])
|
||
self.assertEqual(len(first), 0, "a single frame must not publish a violation")
|
||
tracker.update([finding])
|
||
stable = tracker.update([finding])
|
||
self.assertEqual(len(stable), 1)
|
||
|
||
def test_a_transient_finding_never_publishes(self):
|
||
"""The rule is windowed: a finding must appear in 3 of the last 5 frames.
|
||
Two sightings spread over six frames is a flicker, not a violation."""
|
||
tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5))
|
||
from reconcile import Finding
|
||
|
||
finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10])
|
||
published = 0
|
||
for index in range(6):
|
||
published += len(tracker.update([finding] if index in (0, 2) else []))
|
||
self.assertEqual(published, 0)
|
||
|
||
def test_a_finding_present_in_three_of_five_frames_publishes(self):
|
||
tracker = ComplianceTracker(ComplianceSettings(stability_frames=3, stability_window=5))
|
||
from reconcile import Finding
|
||
|
||
finding = Finding(rule=RULE_MISSING_TAG, severity="violation", row_index=1, box=[0, 0, 10, 10])
|
||
published = 0
|
||
for index in range(5):
|
||
published += len(tracker.update([finding] if index in (0, 2, 4) else []))
|
||
self.assertGreater(published, 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|