



# 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"""






					A__				:		str			       =    "naver-clova-ix/donut-base-finetuned-docvqa"
					A__				:		Union[str, 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__				:		List[Any]			       =    "document_qa"
					A__				:		Tuple			       =    AutoProcessor
					A__				:		List[str]			       =    VisionEncoderDecoderModel

					A__				:		str			       =    ["image", "text"]
					A__				:		List[str]			       =    ["text"]



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






								"""simple docstring"""


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

								super().__init__(*__snake_case				,					**__snake_case					)



					def     _a							(     self					:       Dict				,					_snake_case					:       "Image"				,					_snake_case					:       str					):






								"""simple docstring"""


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

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



					def     _a							(     self					:       Optional[int]				,					_snake_case					:       Tuple					):






								"""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     _a							(     self					:       Dict				,					_snake_case					:       Optional[int]					):






								"""simple docstring"""


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

								return sequence["answer"]









import random
import unittest

import torch

from diffusers import IFInpaintingPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin


@skip_mps
class      __lowerCAmelCase     (		UpperCAmelCase_							,     UpperCAmelCase_							,     unittest.TestCase      ):
					"""simple docstring"""






					A__				:		Any			       =    IFInpaintingPipeline
					A__				:		Dict			       =    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {"width", "height"}
					A__				:		List[Any]			       =    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
					A__				:		Dict			       =    PipelineTesterMixin.required_optional_params - {"latents"}



					def     _a							(     self					:       Any					):






								"""simple docstring"""


								return self._get_dummy_components()



					def     _a							(     self					:       Optional[int]				,					_snake_case					:       Any				,					_snake_case					:       str=0					):






								"""simple docstring"""


								if str(_snake_case					).startswith('mps'					):
											A__       =    torch.manual_seed(_snake_case					)
								else:
											A__       =    torch.Generator(device=_snake_case					).manual_seed(_snake_case					)

								A__       =    floats_tensor((1, 3, 32, 32)				,					rng=random.Random(_snake_case					)					).to(_snake_case					)
								A__       =    floats_tensor((1, 3, 32, 32)				,					rng=random.Random(_snake_case					)					).to(_snake_case					)

								A__       =    {
								    'prompt': 'A painting of a squirrel eating a burger',
								    'image': image,
								    'mask_image': mask_image,
								    'generator': generator,
								    'num_inference_steps': 2,
								    'output_type': 'numpy',
								}

								return inputs



					@unittest.skipIf(
					    torch_device != 'cuda' or not is_xformers_available()				,					reason='XFormers attention is only available with CUDA and `xformers` installed'				,					)
					def     _a							(     self					:       Dict					):






								"""simple docstring"""


								self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3					)



					def     _a							(     self					:       int					):






								"""simple docstring"""


								self._test_save_load_optional_components()



					@unittest.skipIf(torch_device != 'cuda'				,					reason='float16 requires CUDA'					)
					def     _a							(     self					:       Optional[int]					):






								"""simple docstring"""


								super().test_save_load_floataa(expected_max_diff=1E-1					)



					def     _a							(     self					:       List[str]					):






								"""simple docstring"""


								self._test_attention_slicing_forward_pass(expected_max_diff=1E-2					)



					def     _a							(     self					:       Dict					):






								"""simple docstring"""


								self._test_save_load_local()



					def     _a							(     self					:       Optional[int]					):






								"""simple docstring"""


								self._test_inference_batch_single_identical(
								    expected_max_diff=1E-2				,					)




