





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

							snake_case  :   Any      =							["""image""", """text"""]
							snake_case  :   Union[str, Any]      =							["""text"""]






							def __init__(    self	,			*__lowerCAmelCase	,			**__lowerCAmelCase	):
														if not is_vision_available():
																					raise ValueError("""Pillow must be installed to use the DocumentQuestionAnsweringTool."""	)

														super().__init__(*__lowerCAmelCase	,			**__lowerCAmelCase	)






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

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






							def 					_lowerCamelCase		(    self	,			__lowerCAmelCase	):
														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 					_lowerCamelCase		(    self	,			__lowerCAmelCase	):
														UpperCamelCase__					  =      self.pre_processor.batch_decode(__lowerCAmelCase	)[0]
														UpperCamelCase__					  =      sequence.replace(self.pre_processor.tokenizer.eos_token	,			""""""	)
														UpperCamelCase__					  =      sequence.replace(self.pre_processor.tokenizer.pad_token	,			""""""	)
														UpperCamelCase__					  =      re.sub(r"""<.*?>"""	,			""""""	,			__lowerCAmelCase	,			count=1	).strip()  # remove first task start token
														UpperCamelCase__					  =      self.pre_processor.tokenajson(__lowerCAmelCase	)

														return sequence["answer"]






import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


__A							=logging.get_logger(__name__)




class        _SCREAMING_SNAKE_CASE	( snake_case_		):
	lowerCAmelCase__   =					CLIPConfig

	lowerCAmelCase__   =					['CLIPEncoderLayer']



	def __init__(       self		,    lowercase					)							->			int:
					super().__init__(lowercase					)

					lowerCamelCase_         =     CLIPVisionModelWithProjection(config.vision_config					)

					lowerCamelCase_         =     nn.Linear(config.vision_config.projection_dim		,    1					)
					lowerCamelCase_         =     nn.Linear(config.vision_config.projection_dim		,    1					)



	@torch.no_grad()
	def        SCREAMING_SNAKE_CASE_(       self		,    lowercase		,    lowercase		,    lowercase=0.5		,    lowercase=0.5					)							->			int:
					lowerCamelCase_         =     self.vision_model(lowercase					)[0]

					lowerCamelCase_         =     self.p_head(lowercase					)
					lowerCamelCase_         =     nsfw_detected.flatten()
					lowerCamelCase_         =     nsfw_detected > p_threshold
					lowerCamelCase_         =     nsfw_detected.tolist()

					if any(lowercase					):
									logger.warning(
									    "Potential NSFW content was detected in one or more images. A black image will be returned instead."
									    " Try again with a different prompt and/or seed."					)

					for idx, nsfw_detected_ in enumerate(lowercase					):
									if nsfw_detected_:
													lowerCamelCase_         =     np.zeros(images[idx].shape					)

					lowerCamelCase_         =     self.w_head(lowercase					)
					lowerCamelCase_         =     watermark_detected.flatten()
					lowerCamelCase_         =     watermark_detected > w_threshold
					lowerCamelCase_         =     watermark_detected.tolist()

					if any(lowercase					):
									logger.warning(
									    "Potential watermarked content was detected in one or more images. A black image will be returned instead."
									    " Try again with a different prompt and/or seed."					)

					for idx, watermark_detected_ in enumerate(lowercase					):
									if watermark_detected_:
													lowerCamelCase_         =     np.zeros(images[idx].shape					)

					return images, nsfw_detected, watermark_detected


