wehub-resource-sync
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
import math
|
|
from typing import List, Optional
|
|
|
|
from tqdm import tqdm
|
|
|
|
from surya.common.predictor import BasePredictor
|
|
from surya.ocr_error.loader import OCRErrorModelLoader
|
|
from surya.ocr_error.model.config import ID2LABEL
|
|
from surya.ocr_error.schema import OCRErrorDetectionResult
|
|
from surya.settings import settings
|
|
|
|
|
|
class OCRErrorPredictor(BasePredictor):
|
|
model_loader_cls = OCRErrorModelLoader
|
|
batch_size = settings.OCR_ERROR_BATCH_SIZE
|
|
default_batch_sizes = {"cpu": 8, "mps": 8, "cuda": 64}
|
|
|
|
def __call__(self, texts: List[str], batch_size: Optional[int] = None):
|
|
return self.batch_ocr_error_detection(texts, batch_size)
|
|
|
|
def batch_ocr_error_detection(
|
|
self, texts: List[str], batch_size: Optional[int] = None
|
|
):
|
|
if batch_size is None:
|
|
batch_size = self.get_batch_size()
|
|
|
|
num_batches = math.ceil(len(texts) / batch_size)
|
|
texts_processed = self.processor(
|
|
texts, padding="longest", truncation=True, return_tensors="pt"
|
|
)
|
|
predictions = []
|
|
scores = []
|
|
for batch_idx in tqdm(
|
|
range(num_batches),
|
|
desc="Running OCR Error Detection",
|
|
disable=self.disable_tqdm,
|
|
):
|
|
start_idx, end_idx = batch_idx * batch_size, (batch_idx + 1) * batch_size
|
|
batch_input_ids = texts_processed.input_ids[start_idx:end_idx].to(
|
|
self.model.device
|
|
)
|
|
batch_attention_mask = texts_processed.attention_mask[start_idx:end_idx].to(
|
|
self.model.device
|
|
)
|
|
|
|
with settings.INFERENCE_MODE():
|
|
pred = self.model(batch_input_ids, attention_mask=batch_attention_mask)
|
|
probs = pred.logits.softmax(dim=1)
|
|
predictions.extend(probs.argmax(dim=1).cpu().tolist())
|
|
scores.extend(probs[:, 1].cpu().tolist())
|
|
|
|
return OCRErrorDetectionResult(
|
|
texts=texts,
|
|
labels=[ID2LABEL[p] for p in predictions],
|
|
scores=scores,
|
|
)
|