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