diff --git a/recognization/code/reconcile.py b/recognization/code/reconcile.py new file mode 100644 index 0000000..6524298 --- /dev/null +++ b/recognization/code/reconcile.py @@ -0,0 +1,416 @@ +""" +Reconciliation (item vs price tag) and the first compliance rule family. + +Two pure layers, no I/O: + +- `reconcile()` compares one accepted identity with its tag and the catalog entry, + and publishes a state plus the individual signals. It never changes the product + identity: a disagreement is a second opinion, displayed next to the first. +- `evaluate_compliance()` turns settled evidence into findings. Every rule fails + closed on evidence quality: a row whose tags are merely unreadable produces a + capture hint, never a violation. +""" + +from __future__ import annotations + +import os +import re +import sys +import unicodedata +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +current_dir = os.path.dirname(os.path.abspath(__file__)) +if current_dir not in sys.path: + sys.path.insert(0, current_dir) + +from tags import TAG_STATE_ABSENT, TAG_STATE_READABLE, TAG_STATE_UNREADABLE # noqa: E402 + +RECONCILED_CONFIRMED = "confirmed" +RECONCILED_MISMATCH = "mismatch" +RECONCILED_UNVERIFIED = "unverified" +RECONCILED_NO_TAG = "no_tag" +RECONCILED_TAG_ONLY = "tag_only" + +RULE_MISSING_TAG = "missing_price_tag" +RULE_MISSING_TAG_GROUP = "missing_tag_group" +RULE_ORPHAN_TAG = "orphan_tag" +RULE_PRICE_MISMATCH = "price_mismatch" +RULE_DUPLICATE_TAG = "duplicate_tag" +RULE_UNREADABLE_TAGS = "unreadable_tags" + +NAME_MATCH_THRESHOLD = 0.4 +GROUP_MIN_FACINGS = 3 +STABILITY_FRAMES = 3 +STABILITY_WINDOW = 5 + + +# --------------------------------------------------------------------------- +# text matching +# --------------------------------------------------------------------------- + + +def normalize_name(text: Optional[str]) -> str: + if not text: + return "" + text = unicodedata.normalize("NFKC", text) + text = re.sub(r"<[^>]*>", "", text) + text = re.sub(r"[\s・、。,.\-–—_/]+", "", text) + return text.lower() + + +def longest_common_substring_ratio(a: str, b: str) -> float: + """Length of the longest common substring over the shorter string.""" + if not a or not b: + return 0.0 + shorter, longer = (a, b) if len(a) <= len(b) else (b, a) + best = 0 + for start in range(len(shorter)): + for end in range(start + best + 1, len(shorter) + 1): + if shorter[start:end] in longer: + best = end - start + else: + break + return best / len(shorter) + + +# --------------------------------------------------------------------------- +# reconciliation +# --------------------------------------------------------------------------- + + +@dataclass +class Signal: + kind: str # "jan" | "name" | "price" + agrees: bool + item: Any + tag: Any + note: Optional[str] = None + + def as_dict(self) -> Dict[str, Any]: + payload = {"kind": self.kind, "agrees": self.agrees, "item": self.item, "tag": self.tag} + if self.note: + payload["note"] = self.note + return payload + + +def reconcile( + identity: Optional[Dict[str, Any]], + tag: Optional[Dict[str, Any]], + product: Optional[Dict[str, Any]], +) -> Optional[Dict[str, Any]]: + """Compare one facing with its tag. + + Returns None when there is nothing to reconcile (no accepted identity and no + readable tag). A tag with no product above it is `tag_only`. + """ + identity_state = (identity or {}).get("state") + tag_state = (tag or {}).get("state", TAG_STATE_ABSENT) + accepted = identity_state == "recognized" + readable_tag = tag_state == TAG_STATE_READABLE + + if not accepted and not readable_tag: + return None + + if accepted and tag_state == TAG_STATE_ABSENT: + return {"state": RECONCILED_NO_TAG, "signals": [], "message": None} + if accepted and tag_state == TAG_STATE_UNREADABLE: + return { + "state": RECONCILED_UNVERIFIED, + "signals": [], + "message": "tag found but its text was not read", + } + if not accepted and readable_tag: + return {"state": RECONCILED_TAG_ONLY, "signals": [], "message": None} + + signals: List[Signal] = [] + + # JAN is the only exact signal; a checksummed barcode is not a guess. + item_jan = str((product or {}).get("jan") or "").strip() + tag_jan = str((tag or {}).get("jan") or "").strip() + if item_jan and tag_jan: + signals.append(Signal("jan", item_jan == tag_jan, item_jan, tag_jan)) + + item_name = normalize_name((product or {}).get("product_name")) + tag_name = normalize_name((tag or {}).get("name")) + if item_name and tag_name: + ratio = longest_common_substring_ratio(item_name, tag_name) + signals.append( + Signal("name", ratio >= NAME_MATCH_THRESHOLD, item_name[:32], tag_name[:32], + note=f"match {ratio:.2f}") + ) + + item_price = (product or {}).get("price") + tag_price = (tag or {}).get("price") + tag_regular = (tag or {}).get("regular_price") + if item_price is not None and (tag_price is not None or tag_regular is not None): + candidates = [value for value in (tag_price, tag_regular) if value is not None] + agrees = any(abs(float(item_price) - float(value)) < 0.5 for value in candidates) + signals.append(Signal("price", agrees, item_price, tag_price if tag_price is not None else tag_regular)) + + if not signals: + return { + "state": RECONCILED_UNVERIFIED, + "signals": [], + "message": "tag read, but nothing comparable in the catalog", + } + + if any(signal.kind == "jan" for signal in signals): + decided = [signal for signal in signals if signal.kind == "jan"][0] + state = RECONCILED_CONFIRMED if decided.agrees else RECONCILED_MISMATCH + elif any(signal.agrees for signal in signals): + state = RECONCILED_CONFIRMED + elif all(not signal.agrees for signal in signals): + state = RECONCILED_MISMATCH + else: + state = RECONCILED_UNVERIFIED + + message = None + if state == RECONCILED_MISMATCH: + price_signal = next((signal for signal in signals if signal.kind == "price"), None) + name_signal = next((signal for signal in signals if signal.kind == "name"), None) + if price_signal is not None: + message = f"tag says {price_signal.tag}, catalog says {price_signal.item}" + elif name_signal is not None: + message = "tag name does not match the catalog entry" + return {"state": state, "signals": [signal.as_dict() for signal in signals], "message": message} + + +# --------------------------------------------------------------------------- +# compliance +# --------------------------------------------------------------------------- + + +@dataclass +class ComplianceSettings: + group_min_facings: int = GROUP_MIN_FACINGS + stability_frames: int = STABILITY_FRAMES + stability_window: int = STABILITY_WINDOW + + +@dataclass +class Finding: + rule: str + severity: str # "violation" | "hint" + row_index: Optional[int] + box: List[int] + count: int = 1 + skus: List[str] = field(default_factory=list) + detail: Optional[str] = None + + def key(self) -> Tuple[Any, ...]: + return (self.rule, self.row_index, self.box[0] // 32, self.box[1] // 32, self.count) + + def as_dict(self) -> Dict[str, Any]: + payload = { + "rule": self.rule, + "severity": self.severity, + "row_index": self.row_index, + "box": list(self.box), + "count": self.count, + "skus": list(self.skus), + } + if self.detail: + payload["detail"] = self.detail + return payload + + +class ComplianceTracker: + """Keeps findings only while they persist, so the UI never flickers.""" + + def __init__(self, settings: Optional[ComplianceSettings] = None): + self.settings = settings or ComplianceSettings() + self._history: List[set] = [] + + def update(self, findings: Sequence[Finding]) -> List[Finding]: + current = {finding.key(): finding for finding in findings} + self._history.append(set(current)) + if len(self._history) > self.settings.stability_window: + self._history.pop(0) + stable: List[Finding] = [] + for key, finding in current.items(): + seen = sum(1 for frame in self._history if key in frame) + if seen >= self.settings.stability_frames: + stable.append(finding) + return stable + + def reset(self) -> None: + self._history.clear() + + +def evaluate_compliance( + tracks: Sequence[Dict[str, Any]], + tag_by_track: Dict[int, Dict[str, Any]], + all_tags: Sequence[Dict[str, Any]], + bands: Sequence[Any], + reconciliations: Optional[Dict[int, Dict[str, Any]]] = None, + settings: Optional[ComplianceSettings] = None, +) -> List[Finding]: + """Findings from settled evidence. Rules with unreadable evidence stay silent. + + `tag_by_track` is the per-facing attachment; `all_tags` is every tag the pass + found, so a tag nobody claimed can be reported as an orphan. + """ + settings = settings or ComplianceSettings() + findings: List[Finding] = [] + + by_row: Dict[int, List[Dict[str, Any]]] = {} + for track in tracks: + row_index = _row_for(track.get("box") or track.get("raw_box"), bands) + by_row.setdefault(row_index, []).append(track) + + for row_index, row_tracks in by_row.items(): + row_tracks.sort(key=lambda track: (track.get("box") or [0])[0]) + row_tags = [tag_by_track.get(int(track["track_id"]), {}) for track in row_tracks] + readable = [tag for tag in row_tags if tag.get("state") == TAG_STATE_READABLE] + if not readable: + # Fail closed: nothing in this row could be read, so nothing can be + # claimed about missing tags here. Report the capture problem instead. + if any(tag.get("state") == TAG_STATE_UNREADABLE for tag in row_tags): + findings.append( + Finding( + rule=RULE_UNREADABLE_TAGS, + severity="hint", + row_index=row_index, + box=_row_box(row_tracks), + count=len(row_tracks), + detail="move closer: the price tags in this row could not be read", + ) + ) + continue + + # missing_price_tag, per facing + missing: List[Dict[str, Any]] = [] + for track, tag in zip(row_tracks, row_tags): + if tag.get("state") == TAG_STATE_ABSENT: + missing.append(track) + for track in missing: + findings.append( + Finding( + rule=RULE_MISSING_TAG, + severity="violation", + row_index=row_index, + box=list(track.get("box") or track.get("raw_box") or [0, 0, 0, 0]), + count=1, + skus=[_sku_of(track)], + ) + ) + + # 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) + continue + if len(run) >= settings.group_min_facings: + findings.append(_group_finding(row_index, run)) + run = [] + if len(run) >= settings.group_min_facings: + findings.append(_group_finding(row_index, run)) + + # price_mismatch from reconciliation + for track in tracks: + track_id = int(track["track_id"]) + 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", [])): + continue + findings.append( + Finding( + rule=RULE_PRICE_MISMATCH, + severity="violation", + 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=reconciliation.get("message"), + ) + ) + + # orphan_tag: a readable tag that no facing claims (a tag with no product above + # it is either an empty slot with a leftover tag or a mis-clustered read). + attached_ids = { + tag.get("id") for tag in tag_by_track.values() if tag.get("id") + } + for tag in all_tags: + if tag.get("state") != TAG_STATE_READABLE or not tag.get("id"): + continue + if tag["id"] in attached_ids: + continue + findings.append( + Finding( + rule=RULE_ORPHAN_TAG, + severity="violation", + row_index=None, + box=list(tag.get("box") or [0, 0, 0, 0]), + count=1, + detail="price tag with no product above it", + ) + ) + + return findings + + +def _group_finding(row_index: int, run: Sequence[Dict[str, Any]]) -> 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) + x2 = max(box[2] for box in boxes) + y2 = max(box[3] for box in boxes) + return Finding( + rule=RULE_MISSING_TAG_GROUP, + severity="violation", + row_index=row_index, + 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", + ) + + +def _row_box(tracks: Sequence[Dict[str, Any]]) -> List[int]: + boxes = [track.get("box") or track.get("raw_box") or [0, 0, 0, 0] for track in tracks] + 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 _row_for(box: Optional[Sequence[int]], bands: Sequence[Any]) -> int: + """Which rail band (i.e. row) a facing sits above.""" + if not box or not bands: + return 0 + x_center = (box[0] + box[2]) / 2.0 + y_bottom = box[3] + best_index = 0 + best_distance = None + for band in bands: + bx1, by1, bx2, by2 = band.box + if not (bx1 - 40 <= x_center <= bx2 + 40): + continue + distance = abs(by1 - y_bottom) + if best_distance is None or distance < best_distance: + best_distance = distance + best_index = band.row_index + return best_index + + +def _sku_of(track: Dict[str, Any]) -> str: + identity = track.get("identity") or {} + return str(identity.get("sku_id") or "") + + +def summarize_findings(findings: Sequence[Finding]) -> Dict[str, Any]: + by_rule: Dict[str, int] = {} + for finding in findings: + if finding.severity != "violation": + continue + by_rule[finding.rule] = by_rule.get(finding.rule, 0) + 1 + return { + "violations": sum(by_rule.values()), + "by_rule": by_rule, + "hints": [finding.as_dict() for finding in findings if finding.severity != "violation"], + } diff --git a/recognization/code/test_reconcile.py b/recognization/code/test_reconcile.py new file mode 100644 index 0000000..8335aca --- /dev/null +++ b/recognization/code/test_reconcile.py @@ -0,0 +1,288 @@ +""" +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}, + } + + +BANDS = [band(1, [0, 300, 500, 340]), band(2, [0, 600, 500, 640])] + + +class ComplianceTests(unittest.TestCase): + def test_missing_tag_on_a_readable_row_is_a_violation(self): + tracks = [ + track(1, [10, 100, 90, 300]), + track(2, [110, 100, 190, 300]), + ] + tags = { + 1: tag(state=TAG_STATE_READABLE, price=168), + 2: tag(state=TAG_STATE_ABSENT), + } + findings = evaluate_compliance(tracks, tags, [tags[1]], BANDS) + rules = [finding.rule for finding in findings] + self.assertIn(RULE_MISSING_TAG, rules) + + 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): + tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(5)] + tags = { + 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), + 1: tag(state=TAG_STATE_ABSENT), + 2: tag(state=TAG_STATE_ABSENT), + 3: tag(state=TAG_STATE_ABSENT), + 4: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"), + } + findings = evaluate_compliance(tracks, tags, [tags[0], tags[4]], 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 facings", groups[0].detail) + + def test_two_missing_facings_are_not_a_group(self): + tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(3)] + tags = { + 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), + 1: tag(state=TAG_STATE_ABSENT), + 2: tag(state=TAG_STATE_ABSENT), + } + findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS) + self.assertNotIn(RULE_MISSING_TAG_GROUP, [finding.rule for finding in findings]) + self.assertEqual( + len([finding for finding in findings if finding.rule == RULE_MISSING_TAG]), 2 + ) + + 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): + tracks = [track(1, [10, 100, 90, 300])] + tags = {1: tag(state=TAG_STATE_READABLE, price=269)} + 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_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)], + 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() diff --git a/recognization/requirements.txt b/recognization/requirements.txt index 809bd19..76ecafb 100644 --- a/recognization/requirements.txt +++ b/recognization/requirements.txt @@ -12,3 +12,7 @@ fastapi>=0.100.0 uvicorn>=0.20.0 python-multipart>=0.0.6 requests>=2.28.0 +# macOS OCR backend (price tags); other platforms run without OCR until a +# cross-platform backend is added - see docs/design-price-tags-and-compliance.md +pyobjc-framework-Vision>=10; sys_platform == "darwin" +pyobjc-framework-Quartz>=10; sys_platform == "darwin"