




'''simple docstring'''





# 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  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: Optional[int]    =					"naver-clova-ix/donut-base-finetuned-docvqa"
      lowerCAmelCase__			: Dict    =					(
          "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."
      )
      lowerCAmelCase__			: Any    =					"document_qa"
      lowerCAmelCase__			: Tuple    =					AutoProcessor
      lowerCAmelCase__			: int    =					VisionEncoderDecoderModel

      lowerCAmelCase__			: Optional[int]    =					["image", "text"]
      lowerCAmelCase__			: Optional[Any]    =					["text"]







      def __init__(     self					: int ,       *snake_case					: List[Any] ,       **snake_case					: Optional[Any]				):
          if not is_vision_available():
              raise ValueError('''Pillow must be installed to use the DocumentQuestionAnsweringTool.'''				)

          super().__init__(*snake_case ,       **snake_case				)







      def       snake_case	(     self					: Union[str, Any] ,       snake_case					: "Image" ,       snake_case					: str				):
          __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       snake_case	(     self					: str ,       snake_case					: Union[str, Any]				):
          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       snake_case	(     self					: int ,       snake_case					: List[str]				):
          __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 pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url

class  _lowerCamelCase						(			_SCREAMING_SNAKE_CASE   ):

      """simple docstring"""

      lowerCAmelCase__			: Any    =					""
      lowerCAmelCase__			: Any    =					"hf-legacy"  # "hf://"" is reserved for hffs







      def __init__(     self					: str ,       snake_case					: Optional[DatasetInfo] = None ,       snake_case					: Optional[str] = None ,       **snake_case					: List[Any] ,       ):
          super().__init__(self ,       **snake_case				)
          __UpperCamelCase  		=						repo_info
          __UpperCamelCase  		=						token
          __UpperCamelCase  		=						None







      def       snake_case	(     self					: List[Any]				):
          if self.dir_cache is None:
              __UpperCamelCase  		=						{}
              for hf_file in self.repo_info.siblings:
                  # TODO(QL): add sizes
                  __UpperCamelCase  		=						{
                      '''name''': hf_file.rfilename,
                      '''size''': None,
                      '''type''': '''file''',
                  }
                  self.dir_cache.update(
                      {
                          str(snake_case				): {'''name''': str(snake_case				), '''size''': None, '''type''': '''directory'''}
                          for d in list(PurePosixPath(hf_file.rfilename				).parents				)[:-1]
                      }				)







      def       snake_case	(     self					: Dict ,       snake_case					: str ,       snake_case					: str = "rb" ,       **snake_case					: Union[str, Any] ,       ):
          if not isinstance(self.repo_info ,       snake_case				):
              raise NotImplementedError(F"Open is only implemented for dataset repositories, but got {self.repo_info}"				)
          __UpperCamelCase  		=						hf_hub_url(self.repo_info.id ,       snake_case ,       revision=self.repo_info.sha				)
          return fsspec.open(
              snake_case ,       mode=snake_case ,       headers=get_authentication_headers_for_url(snake_case ,       use_auth_token=self.token				) ,       client_kwargs={'''trust_env''': True} ,       ).open()







      def       snake_case	(     self					: Optional[Any] ,       snake_case					: Tuple ,       **snake_case					: List[Any]				):
          self._get_dirs()
          __UpperCamelCase  		=						self._strip_protocol(snake_case				)
          if path in self.dir_cache:
              return self.dir_cache[path]
          else:
              raise FileNotFoundError(snake_case				)







      def       snake_case	(     self					: List[str] ,       snake_case					: int ,       snake_case					: Tuple=False ,       **snake_case					: Dict				):
          self._get_dirs()
          __UpperCamelCase  		=						PurePosixPath(path.strip('''/'''				)				)
          __UpperCamelCase  		=						{}
          for p, f in self.dir_cache.items():
              __UpperCamelCase  		=						PurePosixPath(p.strip('''/'''				)				)
              __UpperCamelCase  		=						p.parent
              if root == path:
                  __UpperCamelCase  		=						f
          __UpperCamelCase  		=						list(paths.values()				)
          if detail:
              return out
          else:
              return sorted(f['''name'''] for f in out				)






