File size: 2,992 Bytes
9bd3ee0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """
OCR provider — uses RapidOCR (ONNX Runtime) under the hood.
RapidOCR runs PaddleOCR's models via ONNX Runtime instead of
PaddlePaddle, making it ~500MB lighter. Multilingual (80+ languages),
good accuracy, CPU-friendly.
Design:
- `rapidocr_onnxruntime` is an OPTIONAL dependency.
- The RapidOCR instance is loaded LAZILY on first use and cached
via cores.embedding.EmbeddingCache.
- If rapidocr_onnxruntime is not installed, is_available() returns False.
Install with: pip install rapidocr-onnxruntime
"""
from __future__ import annotations
import numpy as np
from config.settings import Settings, settings as _default_settings
from cores.embedding import EmbeddingCache
from pipeline.feature_extraction import PipelineOutput
from providers.base import BaseProvider, ProviderCapability
# Module-level cache for the RapidOCR instance
_rapidocr_cache = EmbeddingCache()
class RapidOCRProvider(BaseProvider):
name = "ocr"
capability = ProviderCapability.OCR
def __init__(self, settings: Settings | None = None) -> None:
super().__init__(settings=settings or _default_settings)
def is_available(self) -> bool:
try:
import rapidocr_onnxruntime # noqa: F401
return True
except ImportError:
return False
def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]:
img: np.ndarray = pipeline_output.image
ocr = self._get_ocr()
# RapidOCR accepts numpy arrays
result, _ = ocr(img)
if result is None:
raw = {"total_lines": 0, "text": ""}
normalized = {"text_blocks": [], "full_text": "", "language": None}
return raw, normalized
text_blocks: list[dict] = []
full_text_parts: list[str] = []
for line in result:
# line = [box_points, (text, confidence)]
box, (text, conf) = line
x_coords = [p[0] for p in box]
y_coords = [p[1] for p in box]
x1, y1 = int(min(x_coords)), int(min(y_coords))
x2, y2 = int(max(x_coords)), int(max(y_coords))
text_blocks.append({
"text": text,
"box": {"x": x1, "y": y1, "w": x2 - x1, "h": y2 - y1},
"confidence": round(float(conf), 4),
})
full_text_parts.append(text)
raw = {
"total_lines": len(text_blocks),
"engine": "rapidocr_onnxruntime",
}
normalized = {
"text_blocks": text_blocks,
"full_text": " ".join(full_text_parts),
"language": None, # RapidOCR auto-detects
}
return raw, normalized
def _get_ocr(self):
"""Lazy-load RapidOCR instance (cached)."""
return _rapidocr_cache.get_or_load(
"rapidocr",
lambda: self._create_ocr(),
)
def _create_ocr(self):
from rapidocr_onnxruntime import RapidOCR
return RapidOCR()
|