235 lines
8 KiB
Python
235 lines
8 KiB
Python
"""
|
|
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)
|
|
raw = encoded.tobytes()
|
|
data = NSData.dataWithBytes_length_(raw, len(raw))
|
|
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}")
|