# 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		( UpperCamelCase_     ):
    __a						=   "naver-clova-ix/donut-base-finetuned-docvqa"
    __a						=   (
        "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."
    )
    __a						=   "document_qa"
    __a						=   AutoProcessor
    __a						=   VisionEncoderDecoderModel

    __a						=   ["image", "text"]
    __a						=   ["text"]
    def __init__(	self			,			*lowerCAmelCase			,			**lowerCAmelCase )							->       Union[str, Any]:
     if not is_vision_available():
      raise ValueError('''Pillow must be installed to use the DocumentQuestionAnsweringTool.''' )

     super().__init__(*lowerCAmelCase			,			**lowerCAmelCase )
    def     UpperCamelCase_					(	self			,			lowerCAmelCase			,			lowerCAmelCase )							->       Any:
     SCREAMING_SNAKE_CASE__: Optional[int]=   '''<s_docvqa><s_question>{user_input}</s_question><s_answer>'''
     SCREAMING_SNAKE_CASE__: Optional[Any]=   task_prompt.replace('''{user_input}'''			,			lowerCAmelCase )
     SCREAMING_SNAKE_CASE__: Optional[int]=   self.pre_processor.tokenizer(
         lowerCAmelCase			,			add_special_tokens=lowerCAmelCase			,			return_tensors='''pt''' ).input_ids
     SCREAMING_SNAKE_CASE__: Any=   self.pre_processor(lowerCAmelCase			,			return_tensors='''pt''' ).pixel_values

     return {"decoder_input_ids": decoder_input_ids, "pixel_values": pixel_values}
    def     UpperCamelCase_					(	self			,			lowerCAmelCase )							->       int:
     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=lowerCAmelCase			,			pad_token_id=self.pre_processor.tokenizer.pad_token_id			,			eos_token_id=self.pre_processor.tokenizer.eos_token_id			,			use_cache=lowerCAmelCase			,			num_beams=1			,			bad_words_ids=[[self.pre_processor.tokenizer.unk_token_id]]			,			return_dict_in_generate=lowerCAmelCase			,			).sequences



    def     UpperCamelCase_					(	self			,			lowerCAmelCase )							->       Optional[int]:
     SCREAMING_SNAKE_CASE__: int=   self.pre_processor.batch_decode(lowerCAmelCase )[0]
     SCREAMING_SNAKE_CASE__: Optional[Any]=   sequence.replace(self.pre_processor.tokenizer.eos_token			,			'''''' )
     SCREAMING_SNAKE_CASE__: Any=   sequence.replace(self.pre_processor.tokenizer.pad_token			,			'''''' )
     SCREAMING_SNAKE_CASE__: int=   re.sub(r'''<.*?>'''			,			''''''			,			lowerCAmelCase			,			count=1 ).strip()  # remove first task start token
     SCREAMING_SNAKE_CASE__: List[str]=   self.pre_processor.tokenajson(lowerCAmelCase )

     return sequence["answer"]





import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate
@dataclass(frozen=lowerCAmelCase__					)
class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					field(default="automatic-speech-recognition"  ,  metadata={"include_in_asdict_even_if_is_default": True}					)
					__UpperCAmelCase							  =					Features({"audio": Audio()}					)
					__UpperCAmelCase							  =					Features({"transcription": Value("string"					)}					)
					__UpperCAmelCase							  =					"audio"
					__UpperCAmelCase							  =					"transcription"






					def A   (					self			,							snake_case_    ) ->      Any:







								'''simple docstring'''

								if self.audio_column not in features:
											raise ValueError(F'Column {self.audio_column} is not present in features.'    )
								if not isinstance(features[self.audio_column]			,							snake_case_    ):
											raise ValueError(F'Column {self.audio_column} is not an Audio type.'    )
								__lowercase			       = copy.deepcopy(self    )
								__lowercase			       = self.input_schema.copy()
								__lowercase			       = features[self.audio_column]
								__lowercase			       = input_schema
								return task_template






					@property
					def A   (					self    ) ->      Dict[str, str]:







								'''simple docstring'''

								return {self.audio_column: "audio", self.transcription_column: "transcription"}


