Spaces:
Sleeping
Sleeping
| from typing import Optional, List | |
| import logging | |
| import easyocr | |
| import numpy as np | |
| from PIL import Image | |
| import io | |
| import anyio | |
| logger = logging.getLogger(__name__) | |
| class OCREngine: | |
| """Wrapper for EasyOCR library.""" | |
| def __init__(self, languages: Optional[List[str]] = None, gpu: bool = False): | |
| """ | |
| Initializes the EasyOCR reader. | |
| Args: | |
| languages: List of language codes (e.g., ['en'] for English). | |
| Defaults to ['en'] if not provided. | |
| gpu: Whether to use GPU acceleration. Defaults to False (CPU only). | |
| """ | |
| self.languages = languages or ['en'] | |
| self.gpu = gpu | |
| self.reader = None | |
| self._init_reader() | |
| def _init_reader(self): | |
| """Lazy initialization of the EasyOCR reader.""" | |
| try: | |
| logger.info(f"Initializing EasyOCR with languages={self.languages}, gpu={self.gpu}...") | |
| self.reader = easyocr.Reader(self.languages, gpu=self.gpu) | |
| logger.info("EasyOCR reader initialized successfully") | |
| except Exception as e: | |
| raise RuntimeError(f"Failed to initialize EasyOCR: {e}") | |
| async def detect_text(self, content: bytes, req_id: str = "Unknown") -> str: | |
| """ | |
| Performs text detection on image bytes asynchronously. | |
| Args: | |
| content: Image bytes (PNG, JPG, etc.) | |
| req_id: Request ID for trace logging. | |
| Returns: | |
| Extracted text as a string. | |
| """ | |
| try: | |
| # Wrap the CPU-bound OCR logic in a thread to keep the event loop free | |
| return await anyio.to_thread.run_sync(self._detect_text_sync, content, req_id) | |
| except Exception as e: | |
| logger.error(f"EasyOCR async text detection failed: {e}", exc_info=True) | |
| raise RuntimeError(f"EasyOCR async text detection failed: {e}") | |
| def _detect_text_sync(self, content: bytes, req_id: str) -> str: | |
| """Synchronous implementation of text detection.""" | |
| logger.info(f"[{req_id}] ⚙️ EasyOCR starting processing thread. Thread CPU bound task starting.") | |
| # Convert bytes to PIL Image | |
| pil_image = Image.open(io.BytesIO(content)) | |
| # EasyOCR works with numpy arrays | |
| # Convert PIL to numpy array | |
| np_image = np.array(pil_image) | |
| # Perform OCR | |
| # detail=0 returns only text, skipping bounding box and confidence math (faster) | |
| # paragraph=True groups text together which is slightly faster for joining | |
| result = self.reader.readtext(np_image, detail=0, paragraph=True) | |
| logger.info(f"[{req_id}] 🏁 EasyOCR thread finished. Extracted {len(result)} paragraphs.") | |
| return "\n".join(result) | |