File size: 2,740 Bytes
907930d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e21775b
907930d
 
 
 
 
e21775b
907930d
 
 
 
 
 
e21775b
907930d
 
 
 
 
e21775b
907930d
e21775b
 
907930d
 
 
 
 
 
 
 
86bd1bf
 
 
907930d
e21775b
86bd1bf
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
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)