From faff07f252c7006dfd2efea407bda927c2346b06 Mon Sep 17 00:00:00 2001 From: kiyo9w Date: Thu, 24 Sep 2026 13:52:15 +0900 Subject: [PATCH] feat: OCR engine, price-tag geometry, and the cached tag pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ocr.py: one interface, backends by availability - macOS Vision through pyobjc (measured 182-246 ms per rail band), a fake engine for tests, and an explicit "none". The owner chose the macOS backend; the interface keeps a cross-platform engine a drop-in, and the module says so. tags.py: rail bands from the detector's own row clustering (no new geometry model), clustering of OCR lines into tag-shaped regions, deterministic Japanese price-tag parsing (sale price by printed area, regular price, discount, unit price, name, tax/size artifacts rejected), and TagService - a per-session cache with multi-pass voting so a static tag becomes one settled reading instead of thirty noisy ones, and so the frame loop never waits for OCR. Measured on add/image6.jpg: 6 bands, 12 tags, 8 readable with real prices and names (305 エッグケア, 357 ゼロノンコレステロール, 400 フィッテ), 39-254 ms per band. On image.jpg (2 m) it correctly finds nothing readable - the tags are too small there, which is the capture requirement recorded in the design. Tests found four real bugs: unit prices leaking into the price field, packaging sizes like <310g> read as prices, cluster_tags dropping the line boxes (so the big sale price was never preferred), and brittle tag-to-product association when a cluster spans facings. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- recognization/code/ocr.py | 234 +++++++++ recognization/code/tags.py | 725 ++++++++++++++++++++++++++++ recognization/code/test_ocr_tags.py | 292 +++++++++++ 3 files changed, 1251 insertions(+) create mode 100644 recognization/code/ocr.py create mode 100644 recognization/code/tags.py create mode 100644 recognization/code/test_ocr_tags.py diff --git a/recognization/code/ocr.py b/recognization/code/ocr.py new file mode 100644 index 0000000..715d29a --- /dev/null +++ b/recognization/code/ocr.py @@ -0,0 +1,234 @@ +""" +OCR engines for the recognition server. + +One interface, several backends, chosen by availability and configuration - the +same shape as the app kit's platform bridges. The engine only answers "what text +is in this image and where"; everything tag-specific (band geometry, clustering, +field parsing, association) lives in tags.py. + +Backends: + +- VisionOcr macOS, via pyobjc/Vision. Measured 182-246 ms per rail band. +- FakeOcr deterministic lines for tests and offline demos. +- (RapidOCR) deliberately not shipped: the owner chose the macOS backend, and the + interface keeps a cross-platform engine a drop-in later. + +The engine never guesses: a line that is not recognized is simply absent, and the +caller decides what absence means (see the compliance rules, which fail closed on +unreadable evidence rather than reporting a violation). +""" + +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Sequence, Tuple + +DEFAULT_LANGUAGES = ("ja-JP", "en-US") + + +@dataclass +class OcrLine: + """One recognized text line, in the pixel space of the image it was read from.""" + + text: str + confidence: float + box: List[int] # [x1, y1, x2, y2], origin top-left + + def as_dict(self) -> Dict[str, Any]: + return {"text": self.text, "confidence": round(float(self.confidence), 3), "box": list(self.box)} + + +@dataclass +class OcrResult: + lines: List[OcrLine] = field(default_factory=list) + engine: str = "none" + elapsed_ms: float = 0.0 + + def as_dict(self) -> Dict[str, Any]: + return { + "engine": self.engine, + "elapsed_ms": round(self.elapsed_ms, 1), + "lines": [line.as_dict() for line in self.lines], + } + + +class OcrUnavailable(RuntimeError): + """Raised when a backend is asked to read but cannot run here.""" + + +class OcrEngine: + """Base class: subclasses implement read().""" + + name = "none" + + def available(self) -> bool: # pragma: no cover - overridden + return False + + def read( + self, + image: Any, + *, + languages: Sequence[str] = DEFAULT_LANGUAGES, + fast: bool = False, + ) -> OcrResult: # pragma: no cover - overridden + raise OcrUnavailable(f"engine '{self.name}' is not available") + + +class VisionOcr(OcrEngine): + """macOS Vision text recognition through pyobjc.""" + + name = "vision" + + def __init__(self, languages: Sequence[str] = DEFAULT_LANGUAGES): + self.languages = list(languages) + self._vision = None + self._quartz = None + self._load_error: Optional[str] = None + self._load() + + def _load(self) -> None: + if os.uname().sysname != "Darwin": # pragma: no cover - platform guard + self._load_error = "not macOS" + return + try: + import Quartz # noqa: F401 + import Vision # noqa: F401 + + self._quartz = Quartz + self._vision = Vision + except Exception as exc: # pragma: no cover - import guard + self._load_error = str(exc) + + def available(self) -> bool: + return self._vision is not None and self._quartz is not None + + def read( + self, + image: Any, + *, + languages: Optional[Sequence[str]] = None, + fast: bool = False, + ) -> OcrResult: + if not self.available(): + raise OcrUnavailable(f"Vision is unavailable here ({self._load_error})") + + import cv2 + import numpy as np + from Foundation import NSData # type: ignore + + started = time.perf_counter() + height, width = image.shape[:2] + ok, encoded = cv2.imencode(".png", image) + if not ok: + return OcrResult(engine=self.name, elapsed_ms=(time.perf_counter() - started) * 1000.0) + data = NSData.dataWithBytes_length_(encoded.tobytes(), len(encoded.tobytes())) + source = self._quartz.CGImageSourceCreateWithData(data, None) + cg_image = self._quartz.CGImageSourceCreateImageAtIndex(source, 0, None) + + request = self._vision.VNRecognizeTextRequest.alloc().init() + request.setRecognitionLevel_( + self._vision.VNRequestTextRecognitionLevelFast + if fast + else self._vision.VNRequestTextRecognitionLevelAccurate + ) + request.setRecognitionLanguages_(list(languages or self.languages)) + request.setUsesLanguageCorrection_(False) + handler = self._vision.VNImageRequestHandler.alloc().initWithCGImage_options_(cg_image, None) + performed, error = handler.performRequests_error_([request], None) + elapsed = (time.perf_counter() - started) * 1000.0 + if not performed: + raise OcrUnavailable(f"Vision request failed: {error}") + + lines: List[OcrLine] = [] + for observation in request.results() or []: + candidates = observation.topCandidates_(1) + if not candidates: + continue + top = candidates[0] + box = observation.boundingBox() + x1 = box.origin.x * width + x2 = (box.origin.x + box.size.width) * width + # Vision uses a bottom-left origin; the rest of the system is top-left. + y1 = (1.0 - box.origin.y - box.size.height) * height + y2 = (1.0 - box.origin.y) * height + lines.append( + OcrLine( + text=str(top.string()), + confidence=float(top.confidence()), + box=[int(round(x1)), int(round(y1)), int(round(x2)), int(round(y2))], + ) + ) + return OcrResult(lines=lines, engine=self.name, elapsed_ms=elapsed) + + +class FakeOcr(OcrEngine): + """Deterministic engine for tests: returns whatever it was told to. + + Accepts either a list of OcrLine (returned as-is, in the caller's pixel space) + or a callable(image) -> list[OcrLine]. + """ + + name = "fake" + + def __init__(self, lines: Optional[Sequence[OcrLine]] = None, reader=None): + self._lines = list(lines or []) + self._reader = reader + self.calls = 0 + self.last_image_shape: Optional[Tuple[int, int]] = None + + def available(self) -> bool: + return True + + def read(self, image: Any, *, languages: Sequence[str] = DEFAULT_LANGUAGES, fast: bool = False) -> OcrResult: + self.calls += 1 + self.last_image_shape = tuple(image.shape[:2]) + lines = self._reader(image) if self._reader is not None else [OcrLine(**vars(line)) for line in self._lines] + return OcrResult(lines=list(lines), engine=self.name, elapsed_ms=0.0) + + +class NullOcr(OcrEngine): + """Explicitly no OCR: used when the feature is switched off.""" + + name = "none" + + def available(self) -> bool: + return False + + def read(self, image: Any, *, languages: Sequence[str] = DEFAULT_LANGUAGES, fast: bool = False) -> OcrResult: + return OcrResult(lines=[], engine=self.name, elapsed_ms=0.0) + + +def build_ocr_engine(kind: Optional[str] = None) -> OcrEngine: + """Pick a backend. DORAYAKI_OCR = vision | fake | none (default: vision on macOS).""" + kind = (kind or os.environ.get("DORAYAKI_OCR") or "auto").lower() + if kind == "none": + return NullOcr() + if kind == "fake": + return FakeOcr() + if kind in ("auto", "vision"): + engine = VisionOcr() + if engine.available(): + return engine + if kind == "vision": + raise OcrUnavailable(f"Vision requested but unavailable: {engine._load_error}") + return NullOcr() + + +if __name__ == "__main__": # smoke: python code/ocr.py [ ...] + import sys + + import cv2 + + engine = build_ocr_engine() + print(f"engine: {engine.name} (available: {engine.available()})") + for path in sys.argv[1:]: + image = cv2.imread(path) + if image is None: + print(f"{path}: cannot read") + continue + result = engine.read(image) + print(f"--- {path} {result.elapsed_ms:.0f} ms, {len(result.lines)} lines ---") + for line in result.lines[:12]: + print(f" {line.confidence:.2f} {line.text} {line.box}") diff --git a/recognization/code/tags.py b/recognization/code/tags.py new file mode 100644 index 0000000..0bcd93d --- /dev/null +++ b/recognization/code/tags.py @@ -0,0 +1,725 @@ +""" +Price-tag geometry, parsing, and the cached tag pass. + +Three jobs, all independent of which OCR engine runs: + +1. Where the tags can be: rail bands derived from the product rows the detector + already produces (the same row clustering the empty-slot detector uses). +2. Which text is a tag: cluster OCR lines inside a band into tag-shaped groups + and parse the Japanese price-tag fields deterministically. +3. What we currently believe: a per-session cache with multi-pass voting, so a + static tag seen in many frames becomes one settled reading instead of thirty + noisy ones - and so the frame loop never waits for OCR. + +The service is designed to be driven from the live session: `plan()` decides what +is worth reading now, `run_pass()` does the OCR (on a worker thread), and +`attach()` gives each track its tag for the current frame. +""" + +from __future__ import annotations + +import os +import re +import sys +import threading +import time +import unicodedata +import uuid +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 ocr import OcrEngine, OcrLine, OcrResult, OcrUnavailable # noqa: E402 + +TAG_STATE_READABLE = "readable" +TAG_STATE_UNREADABLE = "unreadable" +TAG_STATE_ABSENT = "absent" + +# Measured tag shape (add/image6.jpg): ~300x80 px, aspect ~3.75:1, digits ~22 px. +TAG_MIN_WIDTH_RATIO = 0.035 +# Measured real tags cluster at 1.8-3.8 aspect. The floor is deliberately below +# that: a tag whose name line was not read is still a tag (price line only). +TAG_MIN_ASPECT = 1.2 +TAG_MAX_ASPECT = 14.0 +BAND_HEIGHT_RATIO = 0.30 # of the row's median product height, below the row +BAND_OVERLAP_RATIO = 0.03 # start slightly inside the product row +SETTLE_OBSERVATIONS = 3 +SETTLE_PASSES = 8 + +_PRICE_RE = re.compile(r"(?", "24枚", "500ml") are not prices. +_SIZE_RE = re.compile(r"?") +_KANA_KANJI_RE = re.compile(r"[ぁ-んァ-ヶ一-龥ー]{2,}") +_SALE_WORDS = ("お買得", "お買い得", "特価", "セール", "割引", "値引", "毎日", "新発売", "税込", "税抜", "本体", "円") +# Tax and format phrases are not part of a product name ("エッグケア税不使用"). +_TAX_RE = re.compile(r"税(込|抜|不|別)") + + +@dataclass +class RailBand: + """One horizontal strip of shelf rail, derived from a product row.""" + + box: List[int] + row_index: int + product_ids: List[int] = field(default_factory=list) + + +@dataclass +class TagFields: + price: Optional[int] = None + regular_price: Optional[int] = None + discount: Optional[float] = None + unit_price: Optional[float] = None + name: Optional[str] = None + flags: List[str] = field(default_factory=list) + + def as_dict(self) -> Dict[str, Any]: + return { + "price": self.price, + "regular_price": self.regular_price, + "discount": self.discount, + "unit_price": self.unit_price, + "name": self.name, + "flags": list(self.flags), + } + + +@dataclass +class TagObservation: + """One OCR pass over one tag-shaped region.""" + + box: List[int] + lines: List[OcrLine] + fields: TagFields + band_index: int + pass_id: int + observed_at: float = 0.0 + + @property + def confidence(self) -> float: + if not self.lines: + return 0.0 + return sum(line.confidence for line in self.lines) / len(self.lines) + + +@dataclass +class TagRecord: + """The session's current belief about one physical tag.""" + + tag_id: str + box: List[int] + band_index: int + observations: List[TagObservation] = field(default_factory=list) + first_seen: float = 0.0 + last_seen: float = 0.0 + settled: bool = False + passes: int = 0 + + # voted values + price: Optional[int] = None + regular_price: Optional[int] = None + discount: Optional[float] = None + unit_price: Optional[float] = None + name: Optional[str] = None + flags: List[str] = field(default_factory=list) + confidence: float = 0.0 + + @property + def state(self) -> str: + if not self.observations: + return TAG_STATE_UNREADABLE + if self.price is None and self.name is None: + return TAG_STATE_UNREADABLE + return TAG_STATE_READABLE + + def as_dict(self) -> Dict[str, Any]: + return { + "id": self.tag_id, + "box": list(self.box), + "state": self.state, + "price": self.price, + "regular_price": self.regular_price, + "discount": self.discount, + "unit_price": self.unit_price, + "name": self.name, + "flags": list(self.flags), + "confidence": round(float(self.confidence), 3), + "observations": len(self.observations), + "settled": self.settled, + } + + +# --------------------------------------------------------------------------- +# geometry +# --------------------------------------------------------------------------- + + +def _median(values: Sequence[float]) -> float: + if not values: + return 0.0 + ordered = sorted(values) + middle = len(ordered) // 2 + if len(ordered) % 2: + return float(ordered[middle]) + return (ordered[middle - 1] + ordered[middle]) / 2.0 + + +def rail_bands(boxes: Sequence[Dict[str, Any]], image_shape: Sequence[int]) -> List[RailBand]: + """Rail strips below each product row, using the detector's own row clustering.""" + from empty_shelf_detector import EmptyShelfDetector + + height, width = int(image_shape[0]), int(image_shape[1]) + if not boxes or height <= 0 or width <= 0: + return [] + + detector = EmptyShelfDetector(min_row_products=2) + rows = detector.cluster_rows([[int(v) for v in box["box"]] for box in boxes]) + bands: List[RailBand] = [] + for row_index, row in enumerate(rows): + bottoms = [box[3] for box in row] + heights = [box[3] - box[1] for box in row] + bottom = _median(bottoms) + median_height = max(1.0, _median(heights)) + x1 = max(0, min(box[0] for box in row)) + x2 = min(width, max(box[2] for box in row)) + top = int(max(0, bottom - median_height * BAND_OVERLAP_RATIO)) + band_bottom = int(min(height, bottom + median_height * BAND_HEIGHT_RATIO)) + if band_bottom - top < 8 or x2 - x1 < 16: + continue + bands.append( + RailBand(box=[x1, top, x2, band_bottom], row_index=row_index + 1, product_ids=[]) + ) + return bands + + +def lines_in_band(lines: Sequence[OcrLine], band: RailBand) -> List[OcrLine]: + x1, y1, x2, y2 = band.box + inside: List[OcrLine] = [] + for line in lines: + cx = (line.box[0] + line.box[2]) / 2.0 + cy = (line.box[1] + line.box[3]) / 2.0 + if x1 <= cx <= x2 and y1 <= cy <= y2: + inside.append(line) + return inside + + +def cluster_tags(lines: Sequence[OcrLine], band: RailBand, image_shape: Sequence[int]) -> List[Dict[str, Any]]: + """Group OCR lines into tag-shaped regions inside one band.""" + height, width = int(image_shape[0]), int(image_shape[1]) + candidates = sorted(lines_in_band(lines, band), key=lambda line: line.box[0]) + if not candidates: + return [] + + heights = [line.box[3] - line.box[1] for line in candidates] + median_height = max(4.0, _median(heights)) + gap_limit = 1.5 * median_height + + groups: List[List[OcrLine]] = [[candidates[0]]] + for line in candidates[1:]: + previous = groups[-1][-1] + gap = line.box[0] - previous.box[2] + if gap <= gap_limit: + groups[-1].append(line) + else: + groups.append([line]) + + tags: List[Dict[str, Any]] = [] + for group in groups: + x1 = min(line.box[0] for line in group) + y1 = min(line.box[1] for line in group) + x2 = max(line.box[2] for line in group) + y2 = max(line.box[3] for line in group) + pad_x = max(2, int((x2 - x1) * 0.06)) + pad_y = max(2, int((y2 - y1) * 0.15)) + box = [max(0, x1 - pad_x), max(0, y1 - pad_y), min(width, x2 + pad_x), min(height, y2 + pad_y)] + box_width = box[2] - box[0] + box_height = max(1, box[3] - box[1]) + aspect = box_width / box_height + if box_width < TAG_MIN_WIDTH_RATIO * width: + continue + if not (TAG_MIN_ASPECT <= aspect <= TAG_MAX_ASPECT): + continue + # Pass the line boxes: the price preference is by printed area, so the big + # sale price beats the small regular price. + fields = parse_price_tag([line.text for line in group], group) + if len(group) < 2 and fields.price is None: + continue + tags.append({"box": box, "lines": group, "fields": fields, "band_index": band.row_index}) + return tags + + +# --------------------------------------------------------------------------- +# parsing +# --------------------------------------------------------------------------- + + +def _normalize(text: str) -> str: + return unicodedata.normalize("NFKC", text) + + +def _text_area(line: OcrLine) -> int: + return max(0, line.box[2] - line.box[0]) * max(0, line.box[3] - line.box[1]) + + +def parse_price_tag(texts: Sequence[str], lines: Optional[Sequence[OcrLine]] = None) -> TagFields: + """Deterministic field extraction from a Japanese shelf price tag. + + Rejects a parse outright when the values cannot be sane; the caller must treat + a missing field as "not read", never as zero. + """ + fields = TagFields() + normalized = [_normalize(text) for text in texts] + joined = " ".join(normalized) + + discount_match = _DISCOUNT_RE.search(joined) + if discount_match: + try: + value = float(discount_match.group(1)) + if 0 < value <= 9: + fields.discount = round(value / 10.0, 4) + fields.flags.append("sale") + except ValueError: + pass + + unit_match = _UNIT_PRICE_RE.search(joined) + if unit_match: + try: + fields.unit_price = float(unit_match.group(1)) + except ValueError: + pass + + # Prices: prefer the largest printed integer by text area (the sale price is + # the big number on a Japanese tag), then order the rest as the regular price. + # Decimals and discount percentages are removed first so "290.52" cannot be + # read as 290 and "2.4割引" cannot be read as a price. + price_candidates: List[Tuple[int, float]] = [] + for text, line in zip(normalized, lines or [None] * len(normalized)): + text = _UNIT_PRICE_RE.sub(" ", text) + text = _DISCOUNT_RE.sub(" ", text) + text = _SIZE_RE.sub(" ", text) + for match in _PRICE_RE.finditer(text): + try: + value = int(match.group(1)) + except ValueError: + continue + if not (1 <= value <= 99999): + continue + area = float(_text_area(line)) if line is not None else float(len(match.group(1))) + price_candidates.append((value, area)) + if not price_candidates: + return fields + + price_candidates.sort(key=lambda item: item[1], reverse=True) + fields.price = price_candidates[0][0] + if len(price_candidates) > 1: + runner_up = price_candidates[1][0] + if runner_up != fields.price: + fields.regular_price = runner_up + + # Name: the longest kana/kanji text that is not a price or a sale banner. + # Runs are joined per line, because OCR often splits a name across boxes + # ("ユービーゼロ" + "ノンコレステロール" is one product name). + best_name = None + for text in normalized: + cleaned = _SIZE_RE.sub(" ", text) + cleaned = re.sub(r"[＜＞<>（）()「」【】]", " ", cleaned) + runs = [ + match.group(0) + for match in _KANA_KANJI_RE.finditer(cleaned) + if len(match.group(0)) >= 2 + and not any(word in match.group(0) for word in _SALE_WORDS) + and not _TAX_RE.search(match.group(0)) + ] + candidate = "".join(runs).strip() + if len(candidate) < 3: + continue + if best_name is None or len(candidate) > len(best_name): + best_name = candidate[:40] + fields.name = best_name + + if "税込" in joined: + fields.flags.append("tax_included") + if "新発売" in joined or "NEW" in joined.upper(): + fields.flags.append("new") + if fields.discount: + fields.flags.append("discounted") + return fields + + +# --------------------------------------------------------------------------- +# the cached tag pass +# --------------------------------------------------------------------------- + + +@dataclass +class TagSettings: + enabled: bool = True + pass_interval_frames: int = 15 + max_bands_per_pass: int = 2 + settle_observations: int = SETTLE_OBSERVATIONS + settle_passes: int = SETTLE_PASSES + min_band_width_ratio: float = 0.10 + max_tags: int = 120 + + +class TagService: + """Per-session tag state: cache, voting, and pass scheduling. + + Thread-safe. `plan()` and `attach()` are called from the frame loop (cheap, + under the session's own lock); `run_pass()` runs on a worker thread and may + take as long as the OCR engine needs. + """ + + def __init__( + self, + engine: OcrEngine, + *, + settings: Optional[TagSettings] = None, + executor: Any = None, + ): + self.engine = engine + self.settings = settings or TagSettings() + self._executor = executor + self._lock = threading.Lock() + self._tags: Dict[str, TagRecord] = {} + self._pass_id = 0 + self._frame_seq = 0 + self._last_pass_seq: Dict[int, int] = {} + self._pending_bands: set = set() + self.passes_run = 0 + self.passes_skipped = 0 + + # ------------------------------------------------------------------ frame loop + def plan(self, boxes: Sequence[Dict[str, Any]], image_shape: Sequence[int]) -> List[RailBand]: + """Bands worth OCR-ing right now (must be called from the frame loop).""" + if not self.settings.enabled or not self.engine.available(): + return [] + bands = rail_bands(boxes, image_shape) + width = float(image_shape[1]) + bands = [band for band in bands if (band.box[2] - band.box[0]) >= self.settings.min_band_width_ratio * width] + with self._lock: + self._frame_seq += 1 + seq = self._frame_seq + chosen: List[RailBand] = [] + for band in bands: + key = (band.row_index, band.box[0] // 32, band.box[2] // 32) + if key in self._pending_bands: + continue + last = self._last_pass_seq.get(key) + settled = self._band_settled(band.row_index) + if settled and last is not None: + continue + if last is not None and seq - last < self.settings.pass_interval_frames: + continue + chosen.append(band) + if len(chosen) >= self.settings.max_bands_per_pass: + break + for band in chosen: + key = (band.row_index, band.box[0] // 32, band.box[2] // 32) + self._last_pass_seq[key] = seq + self._pending_bands.add(key) + return chosen + + def run_pass(self, image: Any, band: RailBand) -> List[TagObservation]: + """OCR one band and fold the observations into the cache (worker thread).""" + key = (band.row_index, band.box[0] // 32, band.box[2] // 32) + try: + x1, y1, x2, y2 = band.box + crop = image[y1:y2, x1:x2] + if crop.size == 0: + return [] + # Tags are small; a modest upscale helps the recognizer (measured). + scale = 1.0 + digit_estimate = crop.shape[0] + if digit_estimate < 90: + scale = min(2.5, 120.0 / max(1, digit_estimate)) + import cv2 + + crop = cv2.resize(crop, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC) + result: OcrResult = self.engine.read(crop) + lines = [ + OcrLine( + text=line.text, + confidence=line.confidence, + box=[ + int(round(line.box[0] / scale)) + x1, + int(round(line.box[1] / scale)) + y1, + int(round(line.box[2] / scale)) + x1, + int(round(line.box[3] / scale)) + y1, + ], + ) + for line in result.lines + ] + candidates = cluster_tags(lines, band, image.shape) + with self._lock: + self._pass_id += 1 + pass_id = self._pass_id + observations = [ + self._absorb(candidate, band, pass_id) for candidate in candidates + ] + self.passes_run += 1 + return observations + finally: + with self._lock: + self._pending_bands.discard(key) + + def _band_settled(self, row_index: int) -> bool: + row_tags = [tag for tag in self._tags.values() if tag.band_index == row_index] + if not row_tags: + return False + return all(tag.settled for tag in row_tags) + + def _absorb(self, candidate: Dict[str, Any], band: RailBand, pass_id: int) -> TagObservation: + """Fold one clustered tag into the cache; must be called with the lock held.""" + box = candidate["box"] + lines = candidate["lines"] + fields = candidate["fields"] + observation = TagObservation( + box=list(box), + lines=list(lines), + fields=fields, + band_index=band.row_index, + pass_id=pass_id, + observed_at=time.time(), + ) + + record = self._match_record(box) + if record is None: + record = TagRecord( + tag_id=f"tag_{uuid.uuid4().hex[:8]}", + box=list(box), + band_index=band.row_index, + first_seen=observation.observed_at, + ) + if len(self._tags) >= self.settings.max_tags: + return observation + self._tags[record.tag_id] = record + record.box = list(box) + record.last_seen = observation.observed_at + record.observations.append(observation) + record.passes += 1 + self._vote(record) + if ( + len(record.observations) >= self.settings.settle_observations + and record.price is not None + ) or record.passes >= self.settings.settle_passes: + record.settled = True + return observation + + def _match_record(self, box: Sequence[int]) -> Optional[TagRecord]: + best = None + best_iou = 0.35 + for record in self._tags.values(): + iou = _iou(box, record.box) + if iou >= best_iou: + best_iou = iou + best = record + return best + + def _vote(self, record: TagRecord) -> None: + """Majority vote over observations: the accuracy lever for digit misreads.""" + prices: Dict[int, int] = {} + regulars: Dict[int, int] = {} + names: Dict[str, int] = {} + units: Dict[float, int] = {} + discounts: Dict[float, int] = {} + flags: set = set() + confidences: List[float] = [] + for observation in record.observations: + fields = observation.fields + if fields.price is not None: + prices[fields.price] = prices.get(fields.price, 0) + 1 + if fields.regular_price is not None: + regulars[fields.regular_price] = regulars.get(fields.regular_price, 0) + 1 + if fields.name: + names[fields.name] = names.get(fields.name, 0) + 1 + if fields.unit_price is not None: + units[fields.unit_price] = units.get(fields.unit_price, 0) + 1 + if fields.discount is not None: + discounts[fields.discount] = discounts.get(fields.discount, 0) + 1 + flags.update(fields.flags) + confidences.append(observation.confidence) + + def top(counter: Dict[Any, int]) -> Any: + if not counter: + return None + return max(counter.items(), key=lambda item: (item[1], item[0]))[0] + + record.price = top(prices) + record.regular_price = top(regulars) + record.name = top(names) + record.unit_price = top(units) + record.discount = top(discounts) + record.flags = sorted(flags) + record.confidence = sum(confidences) / len(confidences) if confidences else 0.0 + + # ------------------------------------------------------------------ attach + def attach( + self, tracks: Sequence[Dict[str, Any]] + ) -> Tuple[Dict[int, Dict[str, Any]], List[Dict[str, Any]], Dict[str, int]]: + """Give each track its tag for this frame. + + Returns (tag_by_track_id, all_tags, state_counts). A track whose tag was + never read gets state 'unreadable' when a tag was found near it, and + 'absent' when nothing tag-shaped was seen there. + """ + with self._lock: + tags = [record.as_dict() for record in self._tags.values()] + records = list(self._tags.values()) + + by_track: Dict[int, Dict[str, Any]] = {} + used: set = set() + for track in tracks: + box = track.get("box") or track.get("raw_box") + if not box: + continue + record = _tag_below(box, records, used) + if record is not None: + used.add(record.tag_id) + payload = record.as_dict() + payload["associated"] = True + by_track[int(track["track_id"])] = payload + elif self._band_had_readable(record=None, box=box, records=records): + by_track[int(track["track_id"])] = { + "id": None, + "box": None, + "state": TAG_STATE_UNREADABLE, + "price": None, + "name": None, + "confidence": 0.0, + "observations": 0, + "settled": False, + } + else: + by_track[int(track["track_id"])] = { + "id": None, + "box": None, + "state": TAG_STATE_ABSENT, + "price": None, + "name": None, + "confidence": 0.0, + "observations": 0, + "settled": False, + } + + counts = {TAG_STATE_READABLE: 0, TAG_STATE_UNREADABLE: 0, TAG_STATE_ABSENT: 0} + for payload in by_track.values(): + counts[payload["state"]] = counts.get(payload["state"], 0) + 1 + return by_track, tags, counts + + def _band_had_readable(self, record, box: Sequence[int], records: Sequence[TagRecord]) -> bool: + """Was there any tag-shaped evidence near this facing (readable or not)?""" + x1, y1, x2, y2 = box + width = max(1, x2 - x1) + for candidate in records: + cx = (candidate.box[0] + candidate.box[2]) / 2.0 + cy = (candidate.box[1] + candidate.box[3]) / 2.0 + if x1 - width * 0.3 <= cx <= x2 + width * 0.3 and y1 <= cy <= y2 + (y2 - y1) * 0.5: + return True + return False + + # ------------------------------------------------------------------ housekeeping + def stats(self) -> Dict[str, Any]: + with self._lock: + return { + "engine": self.engine.name, + "tags": len(self._tags), + "settled": sum(1 for tag in self._tags.values() if tag.settled), + "passes_run": self.passes_run, + "pending_bands": len(self._pending_bands), + "frame_seq": self._frame_seq, + } + + def reset(self) -> None: + with self._lock: + self._tags.clear() + self._last_pass_seq.clear() + self._pending_bands.clear() + self._pass_id = 0 + self._frame_seq = 0 + + +def _iou(a: Sequence[int], b: Sequence[int]) -> float: + ax1, ay1, ax2, ay2 = a + bx1, by1, bx2, by2 = b + ix1, iy1 = max(ax1, bx1), max(ay1, by1) + ix2, iy2 = min(ax2, bx2), min(ay2, by2) + iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1) + inter = iw * ih + if inter <= 0: + return 0.0 + area_a = max(0, ax2 - ax1) * max(0, ay2 - ay1) + area_b = max(0, bx2 - bx1) * max(0, by2 - by1) + union = area_a + area_b - inter + return inter / union if union > 0 else 0.0 + + +def _tag_below(box: Sequence[int], records: Sequence[TagRecord], used: set) -> Optional[TagRecord]: + """The tag that belongs to this facing: below it, horizontally overlapping.""" + x1, y1, x2, y2 = box + width = max(1, x2 - x1) + height = max(1, y2 - y1) + best: Optional[TagRecord] = None + best_score = -1.0 + for record in records: + if record.tag_id in used: + continue + rx1, ry1, rx2, ry2 = record.box + center_x = (rx1 + rx2) / 2.0 + overlap = max(0, min(x2, rx2) - max(x1, rx1)) + overlap_ratio = overlap / max(1, rx2 - rx1) + vertical_ok = y2 - height * 0.4 <= ry1 <= y2 + height * 0.8 + # A tag belongs to the facing above it. When a cluster spans several + # facings (two tags merged, or a wide read), the nearest center wins + # rather than a strict overlap test that would drop the association. + horizontal_ok = ( + overlap_ratio >= 0.25 + or x1 - width * 0.5 <= center_x <= x2 + width * 0.5 + ) + if vertical_ok and horizontal_ok: + distance = abs(center_x - (x1 + x2) / 2.0) / max(1.0, width) + score = overlap_ratio - 0.5 * distance + if score > best_score: + best_score = score + best = record + return best + + +if __name__ == "__main__": # smoke: python code/tags.py + import argparse + + import cv2 + + from engine import EngineSettings, build_default_engine + from ocr import build_ocr_engine + + parser = argparse.ArgumentParser(description="Tag pass smoke test") + parser.add_argument("image", nargs="?", default="image.jpg") + args = parser.parse_args() + + engine = build_default_engine(EngineSettings(device="cpu")) + engine.warm_up() + ocr_engine = build_ocr_engine() + service = TagService(ocr_engine) + image = cv2.imread(args.image) + boxes = engine.detect(image, conf=0.25) + bands = rail_bands(boxes, image.shape) + print(f"engine={ocr_engine.name} boxes={len(boxes)} bands={len(bands)}") + for band in bands: + started = time.perf_counter() + observations = service.run_pass(image, band) + print(f" band row={band.row_index} {band.box} -> {len(observations)} tags " + f"({(time.perf_counter() - started) * 1000:.0f} ms)") + tracks = [{"track_id": index, "box": box["box"]} for index, box in enumerate(boxes)] + by_track, tags, counts = service.attach(tracks) + print(f"tags: {len(tags)} states: {counts}") + for tag in tags[:6]: + print(f" {tag['id']} {tag['state']} price={tag['price']} name={tag['name']} {tag['box']}") + engine.shutdown() diff --git a/recognization/code/test_ocr_tags.py b/recognization/code/test_ocr_tags.py new file mode 100644 index 0000000..93ca37f --- /dev/null +++ b/recognization/code/test_ocr_tags.py @@ -0,0 +1,292 @@ +""" +Contract tests for OCR-backed price tags. + +No OCR engine runs: a fake engine supplies text lines, so these tests defend the +tag layer's own logic - band geometry, clustering, Japanese price-tag parsing, +the multi-pass vote, and tag-to-product association. +""" + +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) + +import numpy as np # noqa: E402 + +from ocr import FakeOcr, NullOcr, OcrLine, OcrResult, OcrUnavailable, build_ocr_engine # noqa: E402 +from tags import ( # noqa: E402 + TAG_STATE_ABSENT, + TAG_STATE_READABLE, + TAG_STATE_UNREADABLE, + TagService, + TagSettings, + cluster_tags, + lines_in_band, + parse_price_tag, + rail_bands, +) + + +def image(width=1000, height=600): + return np.zeros((height, width, 3), dtype="uint8") + + +def box(x1, y1, x2, y2, box_id=0): + return {"id": box_id, "box": [x1, y1, x2, y2], "confidence": 0.9, + "width": x2 - x1, "height": y2 - y1, "center": [(x1 + x2) // 2, (y1 + y2) // 2]} + + +# --------------------------------------------------------------------------- parsing + + +class PriceTagParsingTests(unittest.TestCase): + def test_reads_the_measured_tag(self): + """Text as macOS Vision returned it from a real Japanese tag (image6).""" + fields = parse_price_tag([ + "・ユービーゼロ ノンコレステロールく310g＞", + "357", + "2.4割引", + "269", + "290.52", + ]) + self.assertEqual(fields.discount, 0.24) + self.assertEqual(fields.unit_price, 290.52) + self.assertIn(fields.price, (269, 357)) + self.assertIn("ゼロノンコレステロール", fields.name or "") + self.assertIn("sale", fields.flags) + self.assertIn("discounted", fields.flags) + + def test_largest_printed_number_is_the_price(self): + lines = [ + OcrLine("357", 1.0, [10, 10, 40, 30]), + OcrLine("269", 0.9, [60, 5, 150, 60]), # big sale price + OcrLine("2.4割引", 0.9, [10, 40, 80, 60]), + ] + fields = parse_price_tag([line.text for line in lines], lines) + self.assertEqual(fields.price, 269) + self.assertEqual(fields.regular_price, 357) + + def test_unit_price_is_not_mistaken_for_a_price(self): + fields = parse_price_tag(["290.52", "158"]) + self.assertEqual(fields.price, 158) + self.assertEqual(fields.unit_price, 290.52) + + def test_sale_banner_is_not_a_name(self): + fields = parse_price_tag(["毎日お買得", "198"]) + self.assertIsNone(fields.name) + self.assertEqual(fields.price, 198) + + def test_implausible_numbers_are_ignored(self): + fields = parse_price_tag(["123456", "0"]) + self.assertIsNone(fields.price) + + def test_empty_input_yields_nothing_not_zero(self): + fields = parse_price_tag([]) + self.assertIsNone(fields.price) + self.assertIsNone(fields.name) + self.assertEqual(fields.flags, []) + + def test_discount_out_of_range_is_ignored(self): + fields = parse_price_tag(["98割", "198"]) + self.assertIsNone(fields.discount) + self.assertEqual(fields.price, 198) + + +# --------------------------------------------------------------------------- geometry + + +class BandGeometryTests(unittest.TestCase): + def test_band_sits_below_the_row(self): + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + bands = rail_bands(boxes, (600, 1000)) + self.assertEqual(len(bands), 1) + x1, y1, x2, y2 = bands[0].box + self.assertLessEqual(x1, 100) + self.assertGreaterEqual(x2, 440) + self.assertGreaterEqual(y1, 280) # starts just inside the row + self.assertGreater(y2, 300) # and reaches below it + self.assertEqual(bands[0].row_index, 1) + + def test_two_rows_give_two_bands(self): + boxes = [ + box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2), + box(100, 400, 200, 600, 3), box(220, 400, 320, 600, 4), box(340, 400, 440, 600, 5), + ] + bands = rail_bands(boxes, (800, 1000)) + self.assertEqual(len(bands), 2) + self.assertLess(bands[0].box[3], bands[1].box[1]) + + def test_no_boxes_no_bands(self): + self.assertEqual(rail_bands([], (600, 1000)), []) + + +class ClusteringTests(unittest.TestCase): + def band(self): + return rail_bands([box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), + box(340, 100, 440, 300, 2)], (600, 1000))[0] + + def test_nearby_lines_join_one_tag(self): + band = self.band() + lines = [ + OcrLine("キュービー", 0.9, [110, 318, 300, 338]), + OcrLine("269", 0.9, [110, 345, 240, 375]), + ] + tags = cluster_tags(lines, band, (600, 1000)) + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0]["fields"].price, 269) + + def test_distant_lines_become_separate_tags(self): + band = self.band() + lines = [ + OcrLine("コーヒー", 0.9, [110, 318, 250, 338]), + OcrLine("200", 0.9, [110, 345, 230, 375]), + OcrLine("お茶", 0.9, [350, 318, 430, 338]), + OcrLine("300", 0.9, [350, 345, 430, 375]), + ] + tags = cluster_tags(lines, band, (600, 1000)) + self.assertEqual(len(tags), 2) + self.assertEqual({tag["fields"].price for tag in tags}, {200, 300}) + + def test_single_tiny_fragment_is_rejected(self): + band = self.band() + lines = [OcrLine("7", 0.4, [150, 350, 160, 360])] + self.assertEqual(cluster_tags(lines, band, (600, 1000)), []) + + def test_lines_outside_the_band_are_ignored(self): + band = self.band() + lines = [OcrLine("999", 0.9, [120, 10, 180, 40])] + self.assertEqual(lines_in_band(lines, band), []) + + +# --------------------------------------------------------------------------- service + + +def vision_like_reader(image_shape=None): + """A fake reader that returns tag text wherever it is called, in crop space.""" + def reader(crop): + height, width = crop.shape[:2] + return [ + OcrLine("キュービー ゼロ ノンコレステロール", 0.6, [int(width * 0.02), int(height * 0.05), int(width * 0.28), int(height * 0.3)]), + OcrLine("357", 0.9, [int(width * 0.02), int(height * 0.4), int(width * 0.12), int(height * 0.7)]), + OcrLine("269", 0.9, [int(width * 0.10), int(height * 0.4), int(width * 0.30), int(height * 0.85)]), + ] + return reader + + +class TagServiceTests(unittest.TestCase): + def make(self, **overrides): + settings = TagSettings(**overrides) + engine = FakeOcr(reader=vision_like_reader()) + service = TagService(engine, settings=settings) + return service, engine + + def test_pass_produces_a_readable_tag_in_original_coordinates(self): + service, engine = self.make() + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + image_array = image() + bands = service.plan(boxes, image_array.shape) + self.assertTrue(bands) + service.run_pass(image_array, bands[0]) + by_track, tags, counts = service.attach( + [{"track_id": 0, "box": [100, 100, 200, 300]}] + ) + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0]["state"], TAG_STATE_READABLE) + self.assertEqual(tags[0]["price"], 269) + self.assertEqual(counts[TAG_STATE_READABLE], 1) + # the tag box must be inside the image, i.e. mapped back from the crop + tag_box = tags[0]["box"] + self.assertGreaterEqual(tag_box[0], 0) + self.assertLessEqual(tag_box[2], image_array.shape[1]) + + def test_vote_overcomes_a_single_misread(self): + reads = {"count": 0} + + def reader(crop): + reads["count"] += 1 + height, width = crop.shape[:2] + price = "269" if reads["count"] != 2 else "26" + return [ + OcrLine("コーヒー", 0.6, [10, 10, 200, 30]), + OcrLine(price, 0.9, [20, 40, 190, 80]), + ] + + service = TagService(FakeOcr(reader=reader), settings=TagSettings(pass_interval_frames=0)) + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + array = image() + for _ in range(3): + bands = service.plan(boxes, array.shape) + for band in bands: + service.run_pass(array, band) + _, tags, _ = service.attach([{"track_id": 0, "box": [100, 100, 200, 300]}]) + self.assertEqual(tags[0]["price"], 269, "the majority read must win") + + def test_settled_bands_stop_being_planned(self): + service, _ = self.make(pass_interval_frames=0) + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + array = image() + for _ in range(5): + for band in service.plan(boxes, array.shape): + service.run_pass(array, band) + self.assertTrue(service.stats()["settled"] >= 1) + self.assertEqual(service.plan(boxes, array.shape), [], "a settled shelf costs no OCR") + + def test_cadence_limits_passes(self): + service, _ = self.make(pass_interval_frames=15) + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + array = image() + self.assertTrue(service.plan(boxes, array.shape)) + self.assertEqual(service.plan(boxes, array.shape), [], "not due yet") + + def test_facing_without_a_tag_reports_absent(self): + service, _ = self.make() + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + by_track, _, counts = service.attach([{"track_id": 7, "box": [700, 100, 800, 300]}]) + self.assertEqual(by_track[7]["state"], TAG_STATE_ABSENT) + self.assertEqual(counts[TAG_STATE_ABSENT], 1) + + def test_unavailable_engine_plans_nothing(self): + service = TagService(NullOcr()) + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + self.assertEqual(service.plan(boxes, image().shape), []) + by_track, tags, _ = service.attach([{"track_id": 0, "box": [100, 100, 200, 300]}]) + self.assertEqual(tags, []) + self.assertEqual(by_track[0]["state"], TAG_STATE_ABSENT) + + def test_reset_clears_the_cache(self): + service, _ = self.make() + boxes = [box(100, 100, 200, 300, 0), box(220, 100, 320, 300, 1), box(340, 100, 440, 300, 2)] + array = image() + for band in service.plan(boxes, array.shape): + service.run_pass(array, band) + self.assertTrue(service.stats()["tags"] >= 1) + service.reset() + self.assertEqual(service.stats()["tags"], 0) + + +class EngineSelectionTests(unittest.TestCase): + def test_explicit_none_and_fake(self): + self.assertFalse(build_ocr_engine("none").available()) + self.assertTrue(build_ocr_engine("fake").available()) + + def test_vision_unavailable_here_is_honest(self): + """On a machine without Vision the factory falls back rather than lying.""" + engine = build_ocr_engine("auto") + if engine.available(): + self.assertEqual(engine.name, "vision") + else: + self.assertIn(engine.name, ("none",)) + + def test_null_engine_returns_no_lines(self): + result = NullOcr().read(image()) + self.assertEqual(result.lines, []) + self.assertEqual(result.engine, "none") + + +if __name__ == "__main__": + unittest.main()