diff --git a/recognization/code/reconcile.py b/recognization/code/reconcile.py index 00376f3..271cee7 100644 --- a/recognization/code/reconcile.py +++ b/recognization/code/reconcile.py @@ -244,6 +244,72 @@ class ComplianceTracker: self._history.clear() +def _sku_of_track(track: Dict[str, Any]) -> Optional[str]: + identity = track.get("identity") or {} + if identity.get("state") == "recognized" and identity.get("sku_id"): + return str(identity["sku_id"]) + return None + + +def group_facings( + row_tracks: Sequence[Dict[str, Any]], + max_gap_ratio: float = 0.35, + size_tolerance: float = 0.45, +) -> List[List[Dict[str, Any]]]: + """Group a row's facings into products. + + A store prints one price tag per *product*, not per facing: measured on the + demo shelf, a row carries 16-18 facings but only 4-5 tags, because the four + facings of one mayonnaise share a single tag. Counting facings therefore + invents missing tags on a shelf that is perfectly labelled, so the rules count + *groups*. + + Identity wins over geometry when both facings were recognised: measured, a + run of ピュアセレクト facings was split by geometry alone because one facing + was detected narrower than its neighbours, and the product was then reported + as missing the tag that was in fact sitting further along its own run. + """ + ordered = sorted( + (track for track in row_tracks if track.get("box") or track.get("raw_box")), + key=lambda track: (track.get("box") or track.get("raw_box"))[0], + ) + groups: List[List[Dict[str, Any]]] = [] + for track in ordered: + box = track.get("box") or track.get("raw_box") + if groups: + previous = groups[-1][-1] + pbox = previous.get("box") or previous.get("raw_box") + gap = box[0] - pbox[2] + width = max(1, pbox[2] - pbox[0]) + height = max(1, pbox[3] - pbox[1]) + width2 = max(1, box[2] - box[0]) + height2 = max(1, box[3] - box[1]) + sku, previous_sku = _sku_of_track(track), _sku_of_track(previous) + close = gap <= max_gap_ratio * width + if sku and previous_sku: + same_product = sku == previous_sku and close + else: + same_product = close and ( + abs(height2 - height) / height <= size_tolerance + and abs(width2 - width) / width <= size_tolerance + ) + if same_product: + groups[-1].append(track) + continue + groups.append([track]) + return groups + + +def _group_box(group: Sequence[Dict[str, Any]]) -> List[int]: + boxes = [track.get("box") or track.get("raw_box") for track in group] + return [ + min(box[0] for box in boxes), + min(box[1] for box in boxes), + max(box[2] for box in boxes), + max(box[3] for box in boxes), + ] + + def evaluate_compliance( tracks: Sequence[Dict[str, Any]], tag_by_track: Dict[int, Dict[str, Any]], @@ -272,7 +338,21 @@ def evaluate_compliance( evidenced = [ tag for tag in row_tags if tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE) ] - coverage = len(evidenced) / max(1, len(row_tracks)) + # Count products, not facings: one tag serves a run of identical facings. + products = group_facings(row_tracks) + product_tags = [ + [ + tag_by_track.get(int(track["track_id"]), {}) + for track in group + ] + for group in products + ] + products_evidenced = [ + group + for group, tags in zip(products, product_tags) + if any(tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE) for tag in tags) + ] + coverage = len(products_evidenced) / max(1, len(products)) if not readable or coverage < settings.min_row_tag_coverage: # Fail closed on evidence quality. Either nothing in this row could be @@ -282,7 +362,10 @@ def evaluate_compliance( detail = ( "move closer: the price tags in this row could not be read" if not readable - else f"tags detected for {len(evidenced)} of {len(row_tracks)} facings in this row" + else ( + f"tags detected for {len(products_evidenced)} of {len(products)} products " + f"in this row" + ) ) findings.append( Finding( @@ -297,36 +380,60 @@ def evaluate_compliance( continue # missing_price_tag: one finding per row (the UI highlights the row), with - # the facings that lack a tag listed rather than one finding per facing. - missing = [ - track + # the products that lack a tag listed rather than one finding per facing. + # A product's tag may sit anywhere along its run of facings, so a group + # only counts as missing when no tag exists for its SKU elsewhere in the + # row either. This is the domain rule, applied as a guard. + skus_with_tags = { + _sku_of_track(track) for track, tag in zip(row_tracks, row_tags) - if tag.get("state") == TAG_STATE_ABSENT + if tag.get("state") in (TAG_STATE_READABLE, TAG_STATE_UNREADABLE) + } - {None} + # A fragment of a product is not a product. Measured without identities: a + # facing detected at 54 px wide beside 150 px neighbours was split off and + # reported as missing the tag that belonged to its own run. Anything much + # narrower than the row's typical product is not judged. + widths = sorted( + max(1, (_group_box(group)[2] - _group_box(group)[0])) for group in products + ) + typical_width = widths[len(widths) // 2] if widths else 0 + missing_groups = [ + group + for group, tags in zip(products, product_tags) + if all(tag.get("state") == TAG_STATE_ABSENT for tag in tags) + and not (_sku_of_track(group[0]) in skus_with_tags) + and (_group_box(group)[2] - _group_box(group)[0]) >= 0.6 * typical_width ] - if missing: + if missing_groups: findings.append( Finding( rule=RULE_MISSING_TAG, severity="violation", row_index=row_index, - box=_row_box(missing), - count=len(missing), - skus=[_sku_of(track) for track in missing], - detail=f"{len(missing)} of {len(row_tracks)} facings in this row have no price tag", + box=_row_box([track for group in missing_groups for track in group]), + count=len(missing_groups), + skus=[_sku_of(group[0]) for group in missing_groups], + detail=( + f"{len(missing_groups)} of {len(products)} products in this row " + f"have no price tag" + ), ) ) - # missing_tag_group: consecutive facings without a tag - run: List[Dict[str, Any]] = [] - for track, tag in zip(row_tracks, row_tags): - if tag.get("state") == TAG_STATE_ABSENT: - run.append(track) + # missing_tag_group: consecutive products without a tag + missing_ids = {id(group) for group in missing_groups} + run: List[List[Dict[str, Any]]] = [] + for group, tags in zip(products, product_tags): + if id(group) in missing_ids: + run.append(group) continue if len(run) >= settings.group_min_facings: - findings.append(_group_finding(row_index, run)) + findings.append( + _group_finding(row_index, [track for g in run for track in g], len(run)) + ) run = [] if len(run) >= settings.group_min_facings: - findings.append(_group_finding(row_index, run)) + findings.append(_group_finding(row_index, [track for g in run for track in g], len(run))) # price_mismatch from reconciliation for track in tracks: @@ -334,7 +441,33 @@ def evaluate_compliance( reconciliation = (reconciliations or {}).get(track_id) if not reconciliation or reconciliation.get("state") != RECONCILED_MISMATCH: continue - if not any(signal.get("kind") == "price" for signal in reconciliation.get("signals", [])): + signals = reconciliation.get("signals", []) + if not any(signal.get("kind") == "price" for signal in signals): + continue + # A mismatch needs one trustworthy anchor before it is a finding. Measured: + # a wrong identity plus a garbled tag read (name match 0.11, no JAN) + # produced a confident-looking "tag says 90, catalog says 188" that was + # really just noise. With an anchor - the tag's JAN, or a name that + # actually resembles the product - the disagreement is worth reporting. + tag_for_track = tag_by_track.get(track_id, {}) + name_signal = next( + (signal for signal in signals if signal.get("kind") == "name"), None + ) + has_anchor = bool(tag_for_track.get("jan")) or bool( + name_signal and float(name_signal.get("note", "0").split()[-1] or 0) >= 0.35 + ) + if not has_anchor: + findings.append( + Finding( + rule=RULE_UNREADABLE_TAGS, + severity="hint", + row_index=_row_for(track.get("box") or track.get("raw_box"), bands), + box=list(track.get("box") or track.get("raw_box") or [0, 0, 0, 0]), + count=1, + skus=[_sku_of(track)], + detail="the tag under this product could not be read well enough to compare", + ) + ) continue findings.append( Finding( @@ -372,7 +505,7 @@ def evaluate_compliance( return findings -def _group_finding(row_index: int, run: Sequence[Dict[str, Any]]) -> Finding: +def _group_finding(row_index: int, run: Sequence[Dict[str, Any]], products: int = 0) -> Finding: boxes = [track.get("box") or track.get("raw_box") or [0, 0, 0, 0] for track in run] x1 = min(box[0] for box in boxes) y1 = min(box[1] for box in boxes) @@ -385,7 +518,9 @@ def _group_finding(row_index: int, run: Sequence[Dict[str, Any]]) -> Finding: box=[x1, y1, x2, y2], count=len(run), skus=[_sku_of(track) for track in run], - detail=f"{len(run)} facings in a row without a price tag", + detail=( + f"{products or len(run)} products in a row without a price tag" + ), ) diff --git a/recognization/code/test_reconcile.py b/recognization/code/test_reconcile.py index 465628d..a53ea79 100644 --- a/recognization/code/test_reconcile.py +++ b/recognization/code/test_reconcile.py @@ -147,13 +147,18 @@ def track(track_id, box, sku="sku_003", state="recognized"): } -BANDS = [band(1, [0, 300, 700, 340]), band(2, [0, 600, 700, 640])] +# 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): - # 3 of 4 facings carry tag evidence (75% >= the 70% bar), one has none. - tracks = [track(index, [10 + index * 150, 100, 140 + index * 150, 300]) for index in range(4)] + # 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"), @@ -164,12 +169,12 @@ class ComplianceTests(unittest.TestCase): 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 facings") + 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 * 100, 100, 90 + index * 100, 300]) for index in range(6)] + 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) @@ -201,9 +206,12 @@ class ComplianceTests(unittest.TestCase): self.assertEqual(findings, []) def test_group_of_consecutive_facings_without_tags(self): - # Coverage counts facings with *any* tag evidence, so the row needs enough - # of it to be judged at all (bar is 70%). - tracks = [track(index, [10 + index * 70, 100, 70 + index * 70, 300]) for index in range(10)] + # 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) @@ -212,10 +220,30 @@ class ComplianceTests(unittest.TestCase): self.assertEqual(len(groups), 1) self.assertEqual(groups[0].count, 3) self.assertEqual(groups[0].row_index, 1) - self.assertIn("3 facings", groups[0].detail) + self.assertIn("3 products", groups[0].detail) - def test_two_missing_facings_are_not_a_group(self): - tracks = [track(index, [10 + index * 150, 100, 140 + index * 150, 300]) for index in range(4)] + 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), @@ -228,6 +256,21 @@ class ComplianceTests(unittest.TestCase): 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")} @@ -238,20 +281,35 @@ class ComplianceTests(unittest.TestCase): 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)} + 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)}, - [tag(state=TAG_STATE_READABLE, price=269)], + {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))}, )