handler.py

from typing import Dict, List, Any
from transformers import NougatProcessor, VisionEncoderDecoderModel
import torch


# check for GPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


class EndpointHandler:
    def __init__(self, path=""):
        # load the model
        self.processor = NougatProcessor.from_pretrained(path)
        self.model = VisionEncoderDecoderModel.from_pretrained(path)
        # move model to device
        self.model.to(device)
        # self.decoder_input_ids = self.processor.tokenizer(
        #     "<s_cord-v2>", add_special_tokens=False, return_tensors="pt"
        # ).input_ids

    def __call__(self, data):

        inputs = data.pop("inputs", data)


        # preprocess the input
        pixel_values = self.processor(inputs, return_tensors="pt").pixel_values
        print(type(pixel_values))
        # forward pass
        outputs = self.model.generate(
            pixel_values.to(device),
            min_length = 1,
            # decoder_input_ids=self.decoder_input_ids.to(device),
            max_length=3584,
            # early_stopping=True,
            # pad_token_id=self.processor.tokenizer.pad_token_id,
            # eos_token_id=self.processor.tokenizer.eos_token_id,
            # use_cache=True,
            # num_beams=1,
            bad_words_ids=[[self.processor.tokenizer.unk_token_id]],
            # return_dict_in_generate=True,
        )
        # process output
        prediction = self.processor.batch_decode(outputs, skip_special_tokens=True)[0]
        prediction = self.processor.post_process_generation(prediction, fix_markdown=False)

        return prediction