vnguyen/product-recognition-demo
1
0
Fork 1
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat: OCR engine, price-tag geometry, and the cached tag pass

Browse source 
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>
This commit is contained in:
kiyo9w 2026-09-24 13:52:15 +09:00
parent 1bafb74058
commit faff07f252
3 changed files with 1251 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

234
recognization/code/ocr.py Normal file
View file

@ -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 <image> [<image> ...]
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}")

725
recognization/code/tags.py Normal file
View file

@ -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"(?<!\d)(\d{2,5})(?!\d)")
# A unit price is a decimal (290.52). The 割 lookahead keeps "2.4割引" - a discount -
# from being read as a unit price.
_UNIT_PRICE_RE = re.compile(r"(?<![\d.])(\d{1,5}\.\d{1,2})(?![\d.])(?!\s*割)")
_DISCOUNT_RE = re.compile(r"(\d{1,2}(?:\.\d)?)\s*割")
# Packaging sizes ("<310g>", "24枚", "500ml") are not prices.
_SIZE_RE = re.compile(r"<?\d{1,5}(?:\.\d)?\s*(?:g|kg|ml|mL|L|枚|個|袋|入|本|包|缶|ℓ)>?")
_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 <image>
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()

292
recognization/code/test_ocr_tags.py Normal file
View file

@ -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()