




# 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			(			a		):
   UpperCAmelCase__  :    Union[str, Any]          =  '''naver-clova-ix/donut-base-finetuned-docvqa'''
   UpperCAmelCase__  :    List[Any]          =  (
       '''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.'''
   )
   UpperCAmelCase__  :    str          =  '''document_qa'''
   UpperCAmelCase__  :    Dict          =  AutoProcessor
   UpperCAmelCase__  :    List[str]          =  VisionEncoderDecoderModel

   UpperCAmelCase__  :    str          =  ['''image''', '''text''']
   UpperCAmelCase__  :    Union[str, Any]          =  ['''text''']







   def __init__(      self		:						List[Any]					,						*_snake_case		:						Union[str, Any]					,						**_snake_case		:						List[str]):

        """simple docstring"""


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

        super().__init__(*_snake_case					,						**_snake_case)







   def lowerCamelCase						(      self		:						Optional[Any]					,						_snake_case		:						"Image"					,						_snake_case		:						str):

        """simple docstring"""


        UpperCAmelCase_          =							'''<s_docvqa><s_question>{user_input}</s_question><s_answer>'''
        UpperCAmelCase_          =							task_prompt.replace('''{user_input}'''					,						_snake_case)
        UpperCAmelCase_          =							self.pre_processor.tokenizer(
            _snake_case					,						add_special_tokens=_snake_case					,						return_tensors='''pt''').input_ids
        UpperCAmelCase_          =							self.pre_processor(_snake_case					,						return_tensors='''pt''').pixel_values

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







   def lowerCamelCase						(      self		:						Optional[int]					,						_snake_case		:						List[str]):

        """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=_snake_case					,						pad_token_id=self.pre_processor.tokenizer.pad_token_id					,						eos_token_id=self.pre_processor.tokenizer.eos_token_id					,						use_cache=_snake_case					,						num_beams=1					,						bad_words_ids=[[self.pre_processor.tokenizer.unk_token_id]]					,						return_dict_in_generate=_snake_case					,						).sequences







   def lowerCamelCase						(      self		:						Optional[Any]					,						_snake_case		:						Any):

        """simple docstring"""


        UpperCAmelCase_          =							self.pre_processor.batch_decode(_snake_case)[0]
        UpperCAmelCase_          =							sequence.replace(self.pre_processor.tokenizer.eos_token					,						'''''')
        UpperCAmelCase_          =							sequence.replace(self.pre_processor.tokenizer.pad_token					,						'''''')
        UpperCAmelCase_          =							re.sub(r'''<.*?>'''					,						''''''					,						_snake_case					,						count=1).strip()  # remove first task start token
        UpperCAmelCase_          =							self.pre_processor.tokenajson(_snake_case)

        return sequence["answer"]












from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


snake_case_ : Any    		=       logging.get_logger(__name__)


snake_case_ : Optional[int]    		=       TypeVar("DatasetType", Dataset, IterableDataset)


def    A						(__A   :    List[DatasetType]   ,				__A   :    Optional[List[float]] = None   ,				__A   :    Optional[int] = None   ,				__A   :    Optional[DatasetInfo] = None   ,				__A   :    Optional[NamedSplit] = None   ,				__A   :    Literal["first_exhausted", "all_exhausted"] = "first_exhausted"   ,				)   ->      DatasetType:







     """simple docstring"""


     from .arrow_dataset import Dataset
     from .iterable_dataset import IterableDataset

     if not datasets:
          raise ValueError('''Unable to interleave an empty list of datasets.'''       )
     for i, dataset in enumerate(__A       ):
          if not isinstance(__A   ,				(Dataset, IterableDataset)       ):
               if isinstance(__A   ,				(DatasetDict, IterableDatasetDict)       ):
                    if not dataset:
                         raise ValueError(
                             F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                             '''is an empty dataset dictionary.'''       )
                    raise ValueError(
                        F"""Dataset at position {i} has at least one split: {list(__A       )}\n"""
                        F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__A       )       )}']"""       )
               raise ValueError(
                   F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__A       ).__name__}."""       )
          if i == 0:
               UpperCAmelCase_     ,       UpperCAmelCase_          =							(
                   (Dataset, IterableDataset) if isinstance(__A   ,				__A       ) else (IterableDataset, Dataset)
               )
          elif not isinstance(__A   ,				__A       ):
               raise ValueError(
                   F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""       )
     if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
          raise ValueError(F"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""       )
     if dataset_type is Dataset:
          return _interleave_map_style_datasets(
              __A   ,				__A   ,				__A   ,				info=__A   ,				split=__A   ,				stopping_strategy=__A       )
     else:
          return _interleave_iterable_datasets(
              __A   ,				__A   ,				__A   ,				info=__A   ,				split=__A   ,				stopping_strategy=__A       )


def    A						(__A   :    List[DatasetType]   ,				__A   :    Optional[DatasetInfo] = None   ,				__A   :    Optional[NamedSplit] = None   ,				__A   :    int = 0   ,				)   ->      DatasetType:







     """simple docstring"""


     if not dsets:
          raise ValueError('''Unable to concatenate an empty list of datasets.'''       )
     for i, dataset in enumerate(__A       ):
          if not isinstance(__A   ,				(Dataset, IterableDataset)       ):
               if isinstance(__A   ,				(DatasetDict, IterableDatasetDict)       ):
                    if not dataset:
                         raise ValueError(
                             F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                             '''is an empty dataset dictionary.'''       )
                    raise ValueError(
                        F"""Dataset at position {i} has at least one split: {list(__A       )}\n"""
                        F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__A       )       )}']"""       )
               raise ValueError(
                   F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__A       ).__name__}."""       )
          if i == 0:
               UpperCAmelCase_     ,       UpperCAmelCase_          =							(
                   (Dataset, IterableDataset) if isinstance(__A   ,				__A       ) else (IterableDataset, Dataset)
               )
          elif not isinstance(__A   ,				__A       ):
               raise ValueError(
                   F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""       )
     if dataset_type is Dataset:
          return _concatenate_map_style_datasets(__A   ,				info=__A   ,				split=__A   ,				axis=__A       )
     else:
          return _concatenate_iterable_datasets(__A   ,				info=__A   ,				split=__A   ,				axis=__A       )






