




# 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     a__					(      UpperCAmelCase_   ):
						_a							:					List[Any]      =					"naver-clova-ix/donut-base-finetuned-docvqa"
						_a							:					Optional[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."
						)
						_a							:					Tuple      =					"document_qa"
						_a							:					str      =					AutoProcessor
						_a							:					Union[str, Any]      =					VisionEncoderDecoderModel

						_a							:					int      =					["image", "text"]
						_a							:					Any      =					["text"]






						def __init__(     self	,    *_A	,    **_A    ):




											"""simple docstring"""



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

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






						def        __SCREAMING_SNAKE_CASE(     self	,    _A	,    _A    ):




											"""simple docstring"""



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

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






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""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        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



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

											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

UpperCamelCase__					    =    logging.get_logger(__name__)
@add_end_docstrings(snake_case__   )
class     a__					(      snake_case__   ):






						def __init__(     self	,    *_A	,    **_A    ):




											"""simple docstring"""



											super().__init__(*_A	,    **_A    )
											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	,    _A=None    ):




											"""simple docstring"""



											__lowerCAmelCase            = {}
											if top_k is not None:
																__lowerCAmelCase            = top_k
											return {}, {}, postprocess_params






						def __call__(     self	,    _A	,    **_A    ):




											"""simple docstring"""



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






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



											__lowerCAmelCase            = load_image(_A    )
											__lowerCAmelCase            = self.image_processor(images=_A	,    return_tensors=self.framework    )
											return model_inputs






						def        __SCREAMING_SNAKE_CASE(     self	,    _A    ):




											"""simple docstring"""



											__lowerCAmelCase            = self.model(**_A    )
											return model_outputs






						def        __SCREAMING_SNAKE_CASE(     self	,    _A	,    _A=5    ):




											"""simple docstring"""



											if top_k > self.model.config.num_labels:
																__lowerCAmelCase            = self.model.config.num_labels

											if self.framework == "pt":
																__lowerCAmelCase            = model_outputs.logits.softmax(-1    )[0]
																__lowerCAmelCase     ,							__lowerCAmelCase            = probs.topk(_A    )
											elif self.framework == "tf":
																__lowerCAmelCase            = stable_softmax(model_outputs.logits	,    axis=-1    )[0]
																__lowerCAmelCase            = tf.math.top_k(_A	,    k=_A    )
																__lowerCAmelCase     ,							__lowerCAmelCase            = topk.values.numpy(), topk.indices.numpy()
											else:
																raise ValueError(f"""Unsupported framework: {self.framework}"""    )

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




