providers/ocr/rapidocr_provider.py

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