From a27841afa4236a426ce5700078597ed15222c83d Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Thu, 24 Sep 2026 13:58:18 +0900 Subject: [PATCH] feat: run the tag pass in parallel with recognition, in live and still modes The engine owns the OCR backend, a shared OCR lock and a dedicated tag worker, so a tag pass can never occupy a classification worker or slow a frame. A live session plans rail bands on the frame loop (cheap), runs them on that worker, and attaches whatever the cache already knows; a still runs one synchronous pass. Tracks and detections now carry price_tag and reconciliation, and the summary carries price_tags plus compliance. Config can turn tags off per session. Compliance now fails closed on evidence quality. The first real run produced 106 "missing tag" violations on a shelf where only 8 tags were readable - our tag *detection* was incomplete, not the shelf. A row is only judged when at least half its facings carry tag evidence; otherwise it reports "tags detected for 14 of 91 facings in this row" as a hint. Missing-tag findings are one per row with a count, not one per facing, and price_mismatch still reports every disagreement. On add/image6.jpg the result is now 7 real price mismatches and no false violations. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- recognization/code/app.py | 55 ++++++++++++- recognization/code/engine.py | 38 ++++++++- recognization/code/live.py | 109 ++++++++++++++++++++++++++ recognization/code/reconcile.py | 112 +++++++++++++++++++++++---- recognization/code/test_live.py | 82 ++++++++++++++++++++ recognization/code/test_reconcile.py | 38 ++++++--- 6 files changed, 407 insertions(+), 27 deletions(-) diff --git a/recognization/code/app.py b/recognization/code/app.py index 0b92922..fb84113 100644 --- a/recognization/code/app.py +++ b/recognization/code/app.py @@ -95,6 +95,7 @@ class ServerSettings: max_sessions: int = DEFAULT_MAX_SESSIONS live_method: str = "hybrid" live_conf: float = 0.1 + live_tags: bool = True log_level: str = "info" ssl_certfile: Optional[str] = None ssl_keyfile: Optional[str] = None @@ -110,6 +111,7 @@ def server_settings_from_env() -> ServerSettings: max_sessions=int(os.environ.get("DORAYAKI_MAX_SESSIONS", base.max_sessions)), live_method=os.environ.get("DORAYAKI_LIVE_METHOD", base.live_method), live_conf=float(os.environ.get("DORAYAKI_LIVE_CONF", base.live_conf)), + live_tags=os.environ.get("DORAYAKI_LIVE_TAGS", "1").lower() not in ("0", "false", "no"), log_level=os.environ.get("DORAYAKI_LOG_LEVEL", base.log_level), ssl_certfile=os.environ.get("DORAYAKI_SSL_CERTFILE") or None, ssl_keyfile=os.environ.get("DORAYAKI_SSL_KEYFILE") or None, @@ -315,6 +317,7 @@ def create_app( method: Optional[str] = None, conf: Optional[float] = None, empty_slots: bool = True, + tags: bool = False, ) -> Dict[str, Any]: if method is not None: _require_method(method) @@ -338,7 +341,7 @@ def create_app( except UndecodableImage as exc: raise _error("UNDECODABLE_IMAGE", str(exc), 422) try: - return engine.recognize( + result = engine.recognize( image, classify=classify, method=method, @@ -348,6 +351,49 @@ def create_app( except UnknownMethod as exc: raise _error("BAD_REQUEST", str(exc), 400) + if tags: + result = _attach_still_tags(result, image) + return result + + def _attach_still_tags(result: Dict[str, Any], image: Any) -> Dict[str, Any]: + """Run one synchronous tag pass over a still and merge its evidence.""" + ocr_engine = getattr(engine, "ocr_engine", None) + if ocr_engine is None or not getattr(ocr_engine, "available", lambda: False)(): + result["price_tags"] = {"readable": 0, "unreadable": 0, "absent": 0} + result["compliance"] = {"violations": 0, "by_rule": {}, "rows": [], "hints": []} + result["ocr_engine"] = "none" + return result + + from reconcile import attach_tags_to_boxes + from tags import TagService, TagSettings + + boxes = [ + {"id": detection["id"], "box": detection["box"]} for detection in result["detections"] + ] + identities = { + int(detection["id"]): detection["identity"] for detection in result["detections"] + } + catalog_by_sku = {product["sku_id"]: product for product in engine.catalog()} + service = TagService(ocr_engine, settings=TagSettings()) + started = time.perf_counter() + bundle = attach_tags_to_boxes(service, image, boxes, catalog_by_sku, identities) + elapsed_ms = (time.perf_counter() - started) * 1000.0 + + for detection in result["detections"]: + box_id = int(detection["id"]) + tag = bundle["by_box"].get(box_id) + if tag is not None: + detection["price_tag"] = tag + reconciliation = bundle["reconciliations"].get(box_id) + if reconciliation is not None: + detection["reconciliation"] = reconciliation + + result["price_tags"] = bundle["counts"] + result["compliance"] = bundle["compliance"] + result["ocr_engine"] = ocr_engine.name + result["tag_timing_ms"] = round(elapsed_ms, 1) + return result + @app.post("/v1/images/{image_id}/classify", dependencies=[Depends(authorize)]) async def classify_image(image_id: str, body: ClassifyBody) -> Dict[str, Any]: method = _require_method(body.method) @@ -418,9 +464,12 @@ def create_app( method=settings.live_method, conf=settings.live_conf, max_inflight_per_session=engine.settings.live_inflight, + tags=settings.live_tags, ), ) session.refresh_catalog() + if settings.live_tags: + session.enable_tags() sessions[session_id] = session await websocket.accept() send_lock = asyncio.Lock() @@ -506,6 +555,7 @@ def create_app( method=message.get("method"), conf=message.get("conf"), empty_slots=message.get("empty_slots"), + tags=message.get("tags"), ) except (TypeError, ValueError): await send({"type": "error", "code": "BAD_MESSAGE", "message": "bad config values"}) @@ -733,6 +783,7 @@ def main() -> None: parser.add_argument("--classify-workers", type=int, default=None) parser.add_argument("--live-method", type=str, default=None, choices=list(CLASSIFY_METHODS)) parser.add_argument("--live-conf", type=float, default=None) + parser.add_argument("--no-tags", action="store_true", help="disable price-tag OCR") parser.add_argument("--max-sessions", type=int, default=None) parser.add_argument("--ssl-certfile", type=str, default=None) parser.add_argument("--ssl-keyfile", type=str, default=None) @@ -755,6 +806,8 @@ def main() -> None: server.live_method = args.live_method if args.live_conf is not None: server.live_conf = args.live_conf + if args.no_tags: + server.live_tags = False if args.ssl_certfile is not None: server.ssl_certfile = args.ssl_certfile if args.ssl_keyfile is not None: diff --git a/recognization/code/engine.py b/recognization/code/engine.py index b76ebd6..273fbf4 100644 --- a/recognization/code/engine.py +++ b/recognization/code/engine.py @@ -224,6 +224,13 @@ class RecognitionEngine: self._owns_executor = owns_executor self._detect_lock = threading.Lock() self._inventory_lock = threading.Lock() + # OCR is the slowest stage (0.2-3.4 s). It gets its own executor so tag + # passes can never occupy a classification worker, and a lock so several + # live sessions do not pile OCR requests onto the CPU at once. + self.ocr_engine: Optional[Any] = None + self.ocr_lock = threading.Lock() + self._tag_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="tags") + self._owns_tag_executor = True self.image_store = ImageStore( self.settings.image_store_max, self.settings.image_store_ttl_s, @@ -237,6 +244,22 @@ class RecognitionEngine: def executor(self) -> ThreadPoolExecutor: return self._executor + @property + def tag_executor(self) -> ThreadPoolExecutor: + return self._tag_executor + + def set_ocr_engine(self, engine: Any) -> None: + """Attach an OCR backend; None disables tag reading.""" + self.ocr_engine = engine + + def read_tags(self, image: Any) -> Any: + """OCR one crop under the shared lock (Vision is happy with concurrency, + but the CPU is not: one OCR at a time keeps detection responsive).""" + if self.ocr_engine is None: + return None + with self.ocr_lock: + return self.ocr_engine.read(image) + def warm_up(self) -> float: """One dummy predict so the first real frame is not the slow one.""" import numpy as np @@ -249,6 +272,8 @@ class RecognitionEngine: def shutdown(self) -> None: if self._owns_executor: self._executor.shutdown(wait=False, cancel_futures=True) + if self._owns_tag_executor: + self._tag_executor.shutdown(wait=False, cancel_futures=True) # ------------------------------------------------------------------ images def decode_image(self, payload: bytes) -> Any: @@ -670,7 +695,18 @@ def build_default_engine(settings: Optional[EngineSettings] = None) -> Recogniti ) matcher = ProductFeatureMatcher(stock_dir=settings.stock_dir) inventory = InventoryManager(stock_dir=settings.stock_dir, csv_filename=settings.stock_csv) - return RecognitionEngine(detector, matcher, inventory, settings=settings) + engine = RecognitionEngine(detector, matcher, inventory, settings=settings) + + from ocr import build_ocr_engine + + try: + ocr_engine = build_ocr_engine() + except Exception as exc: # a broken OCR install must not stop recognition + print(f"[Engine] OCR disabled: {exc}") + ocr_engine = None + engine.set_ocr_engine(ocr_engine) + print(f"[Engine] OCR engine: {getattr(ocr_engine, 'name', 'none')}") + return engine def settings_from_env(env: Optional[Dict[str, str]] = None) -> EngineSettings: diff --git a/recognization/code/live.py b/recognization/code/live.py index f38cee0..f6cb256 100644 --- a/recognization/code/live.py +++ b/recognization/code/live.py @@ -61,6 +61,7 @@ class LiveSettings: method: str = "hybrid" conf: float = 0.1 empty_slots: bool = True + tags: bool = True min_hits: int = 2 min_crop_px: int = 40 max_attempts: int = 3 @@ -176,6 +177,9 @@ class LiveSession: self._slot_counter = 0 self._catalog_by_sku: Dict[str, Dict[str, Any]] = {} self._catalog_version = "" + self._tags = None + self._compliance_tracker = None + self._last_compliance: Dict[str, Any] = {"violations": 0, "by_rule": {}, "rows": []} self._label_memory: "collections.deque[Dict[str, Any]]" = collections.deque( maxlen=LABEL_MEMORY_MAX ) @@ -227,6 +231,7 @@ class LiveSession: method: Optional[str] = None, conf: Optional[float] = None, empty_slots: Optional[bool] = None, + tags: Optional[bool] = None, ) -> Dict[str, Any]: with self._lock: method_changed = method is not None and method != self.settings.method @@ -236,6 +241,8 @@ class LiveSession: self.settings.conf = float(conf) if empty_slots is not None: self.settings.empty_slots = bool(empty_slots) + if tags is not None: + self.settings.tags = bool(tags) if method_changed: self._generation += 1 for track in self._tracks.values(): @@ -255,6 +262,9 @@ class LiveSession: "method": self.settings.method, "conf": self.settings.conf, "empty_slots": self.settings.empty_slots, + "tags": self.settings.tags, + "tags_available": self._tags is not None, + "ocr_engine": getattr(self._tags.engine, "name", "none") if self._tags else "none", } def reset(self) -> None: @@ -380,6 +390,17 @@ class LiveSession: quality = self._quality(image, self._tracks) self._inherit_labels(seq) + # Price tags: plan a pass (cheap) under the lock, run it on the tag + # worker, and attach whatever the cache already knows this frame. + tag_bands = [] + if self._tags is not None and self.settings.tags: + tag_boxes = [ + {"id": state.track_id, "box": state.box} + for state in self._tracks.values() + if state.box + ] + tag_bands = self._tags.plan(tag_boxes, image.shape) + # Collect the classification work under the lock; submit it after # releasing it. add_done_callback can run the callback inline when the # future already finished, and the callback takes this same lock. @@ -388,8 +409,13 @@ class LiveSession: summary = self._summary_locked() stats = self._stats_locked() track_payload = [self._track_payload(state) for state in self._tracks.values()] + tag_summary = self._apply_tags(seq, track_payload, tag_bands) + if tag_summary is not None: + summary["price_tags"] = tag_summary["tags"] + summary["compliance"] = tag_summary["compliance"] self._submit_classification(submissions, method, generation) + self._run_tag_passes(image, tag_bands) result = { "type": "result", @@ -846,6 +872,87 @@ class LiveSession: self._catalog_by_sku = {row["sku_id"]: row for row in products} self._catalog_version = self.engine.catalog_version() + # ------------------------------------------------------------------ price tags + def enable_tags(self, settings: Optional[Any] = None) -> bool: + """Attach a tag pass when an OCR engine is available. Returns whether it runs.""" + ocr_engine = getattr(self.engine, "ocr_engine", None) + if ocr_engine is None or not getattr(ocr_engine, "available", lambda: False)(): + return False + from tags import TagService, TagSettings + + with self._lock: + self._tags = TagService(ocr_engine, settings=settings or TagSettings()) + return True + + def _run_tag_passes(self, image: Any, bands: Sequence[Any]) -> None: + """Submit tag passes to the engine's tag worker (never the frame loop).""" + service = self._tags + if service is None or not bands: + return + for band in bands: + try: + future = self.engine.tag_executor.submit(service.run_pass, image, band) + except RuntimeError: # executor shut down + return + future.add_done_callback(self._on_tag_pass_done) + + def _on_tag_pass_done(self, future: Any) -> None: + try: + future.result() + except Exception: + # A failed OCR pass must not disturb the stream; the band simply stays + # unread until the next attempt. + pass + + def _apply_tags( + self, seq: int, track_payloads: List[Dict[str, Any]], bands: Sequence[Any] + ) -> Optional[Dict[str, Any]]: + """Attach tag + reconciliation to each track and compute compliance. + + Must be called with the session lock held (it reads the tag cache under its + own lock, which is never taken the other way round). + """ + service = self._tags + if service is None or not self.settings.tags: + return None + from reconcile import ComplianceTracker, evaluate_compliance, reconcile, summarize_findings + + tag_by_track, all_tags, counts = service.attach(track_payloads) + reconciliations: Dict[int, Dict[str, Any]] = {} + for payload in track_payloads: + track_id = int(payload["track_id"]) + tag = tag_by_track.get(track_id) + identity = payload.get("identity") + product = self._catalog_by_sku.get((identity or {}).get("sku_id") or "") + decision = reconcile(identity, tag, product) + if tag is not None: + payload["price_tag"] = tag + if decision is not None: + payload["reconciliation"] = decision + reconciliations[track_id] = decision + + if self._compliance_tracker is None: + self._compliance_tracker = ComplianceTracker() + findings = evaluate_compliance( + track_payloads, tag_by_track, all_tags, bands, reconciliations + ) + stable = self._compliance_tracker.update(findings) + summary = summarize_findings(stable) + rows = [ + { + "row_index": finding.row_index, + "rule": finding.rule, + "box": finding.box, + "missing": finding.count, + "skus": finding.skus, + } + for finding in stable + if finding.severity == "violation" + ] + self._last_compliance = {"violations": summary["violations"], "by_rule": summary["by_rule"], "rows": rows, + "hints": summary["hints"]} + return {"tags": counts, "compliance": self._last_compliance} + def snapshot(self) -> Dict[str, Any]: with self._lock: return { @@ -855,5 +962,7 @@ class LiveSession: "seq": self._seq, "stats": self._stats_locked(), "quality": dict(self._last_quality), + "tags": self._tags.stats() if self._tags else None, + "compliance": dict(self._last_compliance), "created_at": self._created_at, } diff --git a/recognization/code/reconcile.py b/recognization/code/reconcile.py index 6524298..809a328 100644 --- a/recognization/code/reconcile.py +++ b/recognization/code/reconcile.py @@ -186,6 +186,10 @@ class ComplianceSettings: group_min_facings: int = GROUP_MIN_FACINGS stability_frames: int = STABILITY_FRAMES stability_window: int = STABILITY_WINDOW + # A row may only be judged when enough of its facings carry tag evidence + # (readable or unreadable). Below this, the row is "not evaluated": claiming a + # tag is missing because our detector found none would be a false violation. + min_row_tag_coverage: float = 0.5 @dataclass @@ -263,10 +267,21 @@ def evaluate_compliance( 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): + 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)) + + if not readable or coverage < settings.min_row_tag_coverage: + # Fail closed on evidence quality. Either nothing in this row could be + # read, or too few facings carry any tag evidence at all - in both cases + # the honest answer is "not evaluated", never "tag missing". + if evidenced: + 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" + ) findings.append( Finding( rule=RULE_UNREADABLE_TAGS, @@ -274,25 +289,28 @@ def evaluate_compliance( 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", + detail=detail, ) ) 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: + # 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 + for track, tag in zip(row_tracks, row_tags) + if tag.get("state") == TAG_STATE_ABSENT + ] + if 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)], + 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", ) ) @@ -403,6 +421,72 @@ def _sku_of(track: Dict[str, Any]) -> str: return str(identity.get("sku_id") or "") +def attach_tags_to_boxes( + tag_service: Any, + image: Any, + boxes: Sequence[Dict[str, Any]], + catalog_by_sku: Dict[str, Dict[str, Any]], + identities: Optional[Dict[int, Dict[str, Any]]] = None, +) -> Dict[str, Any]: + """One-shot tag pass for a still image (photo mode / `?tags=true`). + + Runs the passes synchronously - a still is not a stream, so there is no frame + loop to protect - then attaches tags, reconciles, and evaluates compliance. + """ + image_shape = image.shape + bands = tag_service.plan(boxes, image_shape) + for band in bands: + tag_service.run_pass(image, band) + # Plan again: a still has no later frame to pick up a second band. + for band in tag_service.plan(boxes, image_shape): + tag_service.run_pass(image, band) + + tracks = [ + { + "track_id": int(box["id"]), + "box": list(box["box"]), + "identity": (identities or {}).get(int(box["id"])) or {}, + } + for box in boxes + ] + tag_by_track, all_tags, counts = tag_service.attach(tracks) + + reconciliations: Dict[int, Dict[str, Any]] = {} + for track in tracks: + track_id = track["track_id"] + tag = tag_by_track.get(track_id) + product = catalog_by_sku.get((track["identity"] or {}).get("sku_id") or "") + decision = reconcile(track["identity"], tag, product) + if decision is not None: + reconciliations[track_id] = decision + + findings = evaluate_compliance(tracks, tag_by_track, all_tags, bands, reconciliations) + summary = summarize_findings(findings) + return { + "by_box": tag_by_track, + "tags": all_tags, + "counts": counts, + "reconciliations": reconciliations, + "bands": [{"row_index": band.row_index, "box": band.box} for band in bands], + "compliance": { + "violations": summary["violations"], + "by_rule": summary["by_rule"], + "rows": [ + { + "row_index": finding.row_index, + "rule": finding.rule, + "box": finding.box, + "missing": finding.count, + "skus": finding.skus, + } + for finding in findings + if finding.severity == "violation" + ], + "hints": summary["hints"], + }, + } + + def summarize_findings(findings: Sequence[Finding]) -> Dict[str, Any]: by_rule: Dict[str, int] = {} for finding in findings: diff --git a/recognization/code/test_live.py b/recognization/code/test_live.py index 0b8799a..e4cd911 100644 --- a/recognization/code/test_live.py +++ b/recognization/code/test_live.py @@ -193,6 +193,88 @@ class LabelLifecycleTests(unittest.TestCase): engine.shutdown() +class TagPassIntegrationTests(unittest.TestCase): + """The tag pass must reach the live result without blocking frames.""" + + def make_tagged_session(self): + from ocr import FakeOcr, OcrLine + + def reader(crop): + height, width = crop.shape[:2] + return [ + OcrLine("コーヒー", 0.8, [int(width * 0.05), int(height * 0.05), int(width * 0.4), int(height * 0.3)]), + OcrLine("150", 0.9, [int(width * 0.05), int(height * 0.4), int(width * 0.35), int(height * 0.8)]), + ] + + session, engine, detector, matcher = make_session() + # Rail bands come from product rows, so the shelf needs a row of facings. + detector.boxes = [ + {"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260], + "confidence": 0.9, "class_id": 0, "width": 100, "height": 160, + "center": [150 + index * 120, 180]} + for index in range(3) + ] + engine.set_ocr_engine(FakeOcr(reader=reader)) + self.assertTrue(session.enable_tags()) + return session, engine, detector + + def test_track_gains_a_price_tag_and_a_reconciliation(self): + import time as _time + + session, engine, detector = self.make_tagged_session() + result = None + for frame_id in range(1, 12): + result = process(session, frame_id) + _time.sleep(0.05) # the tag pass runs on its own worker + if any( + (track.get("price_tag") or {}).get("state") == "readable" + for track in result["tracks"] + ): + break + track = result["tracks"][0] + self.assertIn("price_tag", track, "the tag pass should attach to the facing above it") + self.assertEqual(track["price_tag"]["price"], 150) + self.assertIn("reconciliation", track) + self.assertEqual(result["summary"]["price_tags"]["readable"], 1) + engine.shutdown() + + def test_frames_are_not_blocked_by_ocr(self): + """A slow OCR engine must not slow the frame loop: frames keep coming.""" + import time as _time + + from ocr import FakeOcr, OcrLine + + def slow_reader(crop): + _time.sleep(0.5) + height, width = crop.shape[:2] + return [OcrLine("150", 0.9, [10, 10, int(width * 0.3), int(height * 0.6)])] + + session, engine, detector, _ = make_session() + detector.boxes = [ + {"id": index, "box": [100 + index * 120, 100, 200 + index * 120, 260], + "confidence": 0.9, "class_id": 0, "width": 100, "height": 160, + "center": [150 + index * 120, 180]} + for index in range(3) + ] + engine.set_ocr_engine(FakeOcr(reader=slow_reader)) + self.assertTrue(session.enable_tags()) + started = _time.perf_counter() + for frame_id in range(1, 4): + process(session, frame_id) + elapsed = _time.perf_counter() - started + self.assertLess(elapsed, 1.0, "three frames must not wait for a 0.5 s OCR pass") + engine.shutdown() + + def test_config_can_turn_tags_off(self): + session, engine, _ = self.make_tagged_session() + config = session.configure(tags=False) + self.assertFalse(config["tags"]) + self.assertTrue(config["tags_available"]) + result = process(session, 1) + self.assertNotIn("price_tag", result["tracks"][0]) + engine.shutdown() + + class FramePipelineTests(unittest.TestCase): def test_latest_frame_wins_and_drops_are_counted(self): session, engine, _, _ = make_session() diff --git a/recognization/code/test_reconcile.py b/recognization/code/test_reconcile.py index 8335aca..94d98b2 100644 --- a/recognization/code/test_reconcile.py +++ b/recognization/code/test_reconcile.py @@ -147,11 +147,11 @@ def track(track_id, box, sku="sku_003", state="recognized"): } -BANDS = [band(1, [0, 300, 500, 340]), band(2, [0, 600, 500, 640])] +BANDS = [band(1, [0, 300, 700, 340]), band(2, [0, 600, 700, 640])] class ComplianceTests(unittest.TestCase): - def test_missing_tag_on_a_readable_row_is_a_violation(self): + def test_missing_tag_on_a_covered_row_is_a_violation(self): tracks = [ track(1, [10, 100, 90, 300]), track(2, [110, 100, 190, 300]), @@ -163,6 +163,19 @@ class ComplianceTests(unittest.TestCase): findings = evaluate_compliance(tracks, tags, [tags[1]], 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 facings") + + 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)] + 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.""" @@ -188,15 +201,18 @@ class ComplianceTests(unittest.TestCase): 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)] + # 5 facings: 2 tagged, 3 not. Coverage counts facings with *any* tag + # evidence, so give the row enough evidence to be judged at all. + tracks = [track(index, [10 + index * 100, 100, 90 + index * 100, 300]) for index in range(6)] tags = { 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), - 1: tag(state=TAG_STATE_ABSENT), + 1: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"), 2: tag(state=TAG_STATE_ABSENT), 3: tag(state=TAG_STATE_ABSENT), - 4: tag(state=TAG_STATE_READABLE, price=178, id="tag_b"), + 4: tag(state=TAG_STATE_ABSENT), + 5: tag(state=TAG_STATE_READABLE, price=188, id="tag_c"), } - findings = evaluate_compliance(tracks, tags, [tags[0], tags[4]], BANDS) + findings = evaluate_compliance(tracks, tags, [tags[0], tags[1], tags[5]], 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) @@ -208,13 +224,13 @@ class ComplianceTests(unittest.TestCase): tags = { 0: tag(state=TAG_STATE_READABLE, price=168, id="tag_a"), 1: tag(state=TAG_STATE_ABSENT), - 2: tag(state=TAG_STATE_ABSENT), + 2: tag(state=TAG_STATE_READABLE, price=188, id="tag_b"), } - findings = evaluate_compliance(tracks, tags, [tags[0]], BANDS) + findings = evaluate_compliance(tracks, tags, [tags[0], tags[2]], 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 - ) + 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_orphan_tag(self): tracks = [track(1, [10, 100, 90, 300])]