Spaces:
Sleeping
Sleeping
File size: 2,975 Bytes
49cc409 b283975 49cc409 b283975 49cc409 | 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 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Interface abstraite commune à tous les adaptateurs moteurs OCR."""
from __future__ import annotations
import hashlib
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@dataclass
class EngineResult:
"""Résultat brut produit par un moteur OCR sur une image."""
engine_name: str
image_path: str
text: str
duration_seconds: float
error: Optional[str] = None
metadata: dict = field(default_factory=dict)
@property
def success(self) -> bool:
return self.error is None
@property
def image_sha256(self) -> str:
return hashlib.sha256(Path(self.image_path).read_bytes()).hexdigest()
class BaseOCREngine(ABC):
"""Classe de base dont héritent tous les adaptateurs OCR.
Chaque adaptateur doit implémenter :
- ``name`` : identifiant unique du moteur
- ``version()`` : retourne la version du moteur sous forme de chaîne
- ``_run_ocr(image_path)`` : logique d'exécution OCR, retourne le texte brut
Attribut de classe
------------------
execution_mode : ``"io"`` (défaut) ou ``"cpu"``
Indique au runner quel type d'exécuteur utiliser :
- ``"io"`` → ``ThreadPoolExecutor`` (moteurs API / réseau)
- ``"cpu"`` → ``ProcessPoolExecutor`` (moteurs CPU-intensifs : Tesseract, Pero, Kraken)
"""
execution_mode: str = "io"
"""``"io"`` pour ThreadPoolExecutor (défaut), ``"cpu"`` pour ProcessPoolExecutor."""
def __init__(self, config: Optional[dict] = None) -> None:
self.config: dict = config or {}
@property
@abstractmethod
def name(self) -> str:
"""Identifiant unique et stable du moteur."""
@abstractmethod
def version(self) -> str:
"""Retourne la version du moteur (ex : '5.3.0')."""
@abstractmethod
def _run_ocr(self, image_path: Path) -> str:
"""Exécute l'OCR et retourne le texte brut extrait."""
def run(self, image_path: str | Path) -> EngineResult:
"""Point d'entrée public : exécute l'OCR et mesure le temps d'exécution."""
image_path = Path(image_path)
start = time.perf_counter()
try:
text = self._run_ocr(image_path)
error = None
except Exception as exc: # noqa: BLE001
text = ""
error = str(exc)
duration = time.perf_counter() - start
return EngineResult(
engine_name=self.name,
image_path=str(image_path),
text=text,
duration_seconds=round(duration, 4),
error=error,
metadata={"engine_version": self._safe_version()},
)
def _safe_version(self) -> str:
try:
return self.version()
except Exception: # noqa: BLE001
return "unknown"
def __repr__(self) -> str:
return f"{self.__class__.__name__}(name={self.name!r})"
|