from __future__ import annotations import logging import threading from typing import TYPE_CHECKING, Optional import cv2 if TYPE_CHECKING: import easyocr logger = logging.getLogger("uvicorn") def _load_reader(languages: list[str]) -> easyocr.Reader: """Load the OCR model. easyocr and torch are imported here: they take seconds to import. Args: languages (list[str]): The languages the reader must recognize. Raises: OSError: When the model files can't be downloaded or read. RuntimeError: When torch can't load the model files. Returns: easyocr.Reader: The loaded reader. """ import easyocr import torch return easyocr.Reader(languages, gpu=torch.cuda.is_available()) class OCRReader: _instance: Optional["OCRReader"] = None _lock = threading.Lock() ocr_reader: easyocr.Reader def __new__(cls): """Return the one instance, loading the model on the first call. Raises: OSError: When the model files can't be downloaded or read. RuntimeError: When torch can't load the model files. Returns: OCRReader: The shared instance, with its reader loaded. """ # Double-checked: once loaded the fast path takes no lock, and while # the model is still loading every other caller waits here for it. if cls._instance is None: with cls._lock: if cls._instance is None: logger.info("Loading the OCR model into memory...") instance = super().__new__(cls) instance.ocr_reader = _load_reader(["en", "fr"]) # Published only once it can read: a caller racing the # warm-up must never get an instance without a reader. cls._instance = instance return cls._instance def read_text(self, img: cv2.typing.MatLike | None): res = self.ocr_reader.readtext(img, detail=0) if not isinstance(res, list): return None return " ".join([str(item) for item in res])