# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..models.auto import AutoProcessor
from ..models.vision_encoder_decoder import VisionEncoderDecoderModel
from ..utils import is_vision_available
from .base import PipelineTool


if is_vision_available():
    from PIL import Image




class 	__snake_case	( __lowerCAmelCase		):





      '''simple docstring'''






      _snake_case    		=       'naver-clova-ix/donut-base-finetuned-docvqa'
      _snake_case    		=       (
          'This is a tool that answers a question about an document (pdf). It takes an input named `document` which '
          'should be the document containing the information, as well as a `question` that is the question about the '
          'document. It returns a text that contains the answer to the question.'
      )
      _snake_case    		=       'document_qa'
      _snake_case    		=       AutoProcessor
      _snake_case    		=       VisionEncoderDecoderModel

      _snake_case    		=       ['image', 'text']
      _snake_case    		=       ['text']



      def __init__(					self     :				Union[str, Any]    ,  *_UpperCamelCase     :				Optional[Any]    ,  **_UpperCamelCase     :				int)				->Dict:


          """simple docstring"""


          if not is_vision_available():
              raise ValueError("""Pillow must be installed to use the DocumentQuestionAnsweringTool.""")

          super().__init__(*_UpperCamelCase    ,  **_UpperCamelCase)



      def 	_SCREAMING_SNAKE_CASE    (					self     :				List[str]    ,  _UpperCamelCase     :				"Image"    ,  _UpperCamelCase     :				str)				->Optional[Any]:


          """simple docstring"""


          _lowerCamelCase     :		List[str] 						=						"""<s_docvqa><s_question>{user_input}</s_question><s_answer>"""
          _lowerCamelCase     :		Optional[int] 						=						task_prompt.replace("""{user_input}"""    ,  _UpperCamelCase)
          _lowerCamelCase     :		Union[str, Any] 						=						self.pre_processor.tokenizer(
              _UpperCamelCase    ,  add_special_tokens=_UpperCamelCase    ,  return_tensors="""pt""").input_ids
          _lowerCamelCase     :		Dict 						=						self.pre_processor(_UpperCamelCase    ,  return_tensors="""pt""").pixel_values

          return {"decoder_input_ids": decoder_input_ids, "pixel_values": pixel_values}



      def 	_SCREAMING_SNAKE_CASE    (					self     :				int    ,  _UpperCamelCase     :				Optional[Any])				->Optional[int]:


          """simple docstring"""


          return self.model.generate(
              inputs["""pixel_values"""].to(self.device)    ,  decoder_input_ids=inputs["""decoder_input_ids"""].to(self.device)    ,  max_length=self.model.decoder.config.max_position_embeddings    ,  early_stopping=_UpperCamelCase    ,  pad_token_id=self.pre_processor.tokenizer.pad_token_id    ,  eos_token_id=self.pre_processor.tokenizer.eos_token_id    ,  use_cache=_UpperCamelCase    ,  num_beams=1    ,  bad_words_ids=[[self.pre_processor.tokenizer.unk_token_id]]    ,  return_dict_in_generate=_UpperCamelCase    ,  ).sequences




      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[int]    ,  _UpperCamelCase     :				List[Any])				->Dict:


          """simple docstring"""


          _lowerCamelCase     :		Optional[int] 						=						self.pre_processor.batch_decode(_UpperCamelCase)[0]
          _lowerCamelCase     :		Dict 						=						sequence.replace(self.pre_processor.tokenizer.eos_token    ,  """""")
          _lowerCamelCase     :		Optional[int] 						=						sequence.replace(self.pre_processor.tokenizer.pad_token    ,  """""")
          _lowerCamelCase     :		Optional[int] 						=						re.sub(R"""<.*?>"""    ,  """"""    ,  _UpperCamelCase    ,  count=1).strip()  # remove first task start token
          _lowerCamelCase     :		List[str] 						=						self.pre_processor.tokenajson(_UpperCamelCase)

          return sequence["answer"]





from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
    from PIL import Image

    from ..image_utils import load_image

if is_tf_available():
    import tensorflow as tf

    from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
    from ..tf_utils import stable_softmax

if is_torch_available():
    from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

lowerCAmelCase			:							Optional[Any]     =logging.get_logger(__name__)




@add_end_docstrings(__lowerCAmelCase		)
class 	__snake_case	( __lowerCAmelCase		):





      '''simple docstring'''






      def __init__(					self     :				str    ,  *_UpperCamelCase     :				int    ,  **_UpperCamelCase     :				List[str])				->Tuple:


          """simple docstring"""


          super().__init__(*_UpperCamelCase    ,  **_UpperCamelCase)
          requires_backends(self    ,  """vision""")
          self.check_model_type(
              TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == """tf"""
              else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING)



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Dict    ,  _UpperCamelCase     :				List[str]=None)				->Optional[int]:


          """simple docstring"""


          _lowerCamelCase     :		Optional[int] 						=						{}
          if top_k is not None:
              _lowerCamelCase     :		str 						=						top_k
          return {}, {}, postprocess_params



      def __call__(					self     :				Optional[int]    ,  _UpperCamelCase     :				Union[str, List[str], "Image.Image", List["Image.Image"]]    ,  **_UpperCamelCase     :				Optional[int])				->Dict:


          """simple docstring"""


          return super().__call__(_UpperCamelCase    ,  **_UpperCamelCase)



      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[int]    ,  _UpperCamelCase     :				Optional[int])				->str:


          """simple docstring"""


          _lowerCamelCase     :		Tuple 						=						load_image(_UpperCamelCase)
          _lowerCamelCase     :		Any 						=						self.image_processor(images=_UpperCamelCase    ,  return_tensors=self.framework)
          return model_inputs



      def 	_SCREAMING_SNAKE_CASE    (					self     :				str    ,  _UpperCamelCase     :				Union[str, Any])				->List[str]:


          """simple docstring"""


          _lowerCamelCase     :		Any 						=						self.model(**_UpperCamelCase)
          return model_outputs




      def 	_SCREAMING_SNAKE_CASE    (					self     :				Optional[Any]    ,  _UpperCamelCase     :				List[str]    ,  _UpperCamelCase     :				List[str]=5)				->str:


          """simple docstring"""


          if top_k > self.model.config.num_labels:
              _lowerCamelCase     :		Union[str, Any] 						=						self.model.config.num_labels

          if self.framework == "pt":
              _lowerCamelCase     :		Optional[Any] 						=						model_outputs.logits.softmax(-1)[0]
              _lowerCamelCase						,							_lowerCamelCase     :		Dict 						=						probs.topk(_UpperCamelCase)
          elif self.framework == "tf":
              _lowerCamelCase     :		List[Any] 						=						stable_softmax(model_outputs.logits    ,  axis=-1)[0]
              _lowerCamelCase     :		List[Any] 						=						tf.math.top_k(_UpperCamelCase    ,  k=_UpperCamelCase)
              _lowerCamelCase						,							_lowerCamelCase     :		str 						=						topk.values.numpy(), topk.indices.numpy()
          else:
              raise ValueError(F"""Unsupported framework: {self.framework}""")

          _lowerCamelCase     :		str 						=						scores.tolist()
          _lowerCamelCase     :		str 						=						ids.tolist()
          return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(_UpperCamelCase    ,  _UpperCamelCase)]




