



"""simple docstring"""







import torch
from torch import nn
class        __A       (	nn.Module    ):







		'''simple docstring'''





		def __init__(	self						:  Optional[int]			,_snake_case						:  Any			,_snake_case						:  str			,_snake_case						:  List[Any]			,_snake_case						:  str			,_snake_case						:  Optional[Any]=1			,_snake_case						:  List[str]=False						)					->		Optional[Any]:






				"""simple docstring"""



				super().__init__()

				lowercase__   :    Optional[Any]				=			n_token
				lowercase__   :    List[str]				=			d_embed
				lowercase__   :    int				=			d_proj

				lowercase__   :    Union[str, Any]				=			cutoffs + [n_token]
				lowercase__   :    Optional[Any]				=			[0] + self.cutoffs
				lowercase__   :    Optional[Any]				=			div_val

				lowercase__   :    Dict				=			self.cutoffs[0]
				lowercase__   :    str				=			len(self.cutoffs						) - 1
				lowercase__   :    Optional[Any]				=			self.shortlist_size + self.n_clusters

				if self.n_clusters > 0:
						lowercase__   :    List[Any]				=			nn.Parameter(torch.zeros(self.n_clusters			,self.d_embed						)						)
						lowercase__   :    Any				=			nn.Parameter(torch.zeros(self.n_clusters						)						)

				lowercase__   :    Dict				=			nn.ModuleList()
				lowercase__   :    Optional[Any]				=			nn.ParameterList()

				if div_val == 1:
						for i in range(len(self.cutoffs						)						):
								if d_proj != d_embed:
										self.out_projs.append(nn.Parameter(torch.FloatTensor(_snake_case			,_snake_case						)						)						)
								else:
										self.out_projs.append(_snake_case						)

						self.out_layers.append(nn.Linear(_snake_case			,_snake_case						)						)
				else:
						for i in range(len(self.cutoffs						)						):
								lowercase__		,			lowercase__   :    Tuple				=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
								lowercase__   :    List[str]				=			d_embed // (div_val**i)

								self.out_projs.append(nn.Parameter(torch.FloatTensor(_snake_case			,_snake_case						)						)						)

								self.out_layers.append(nn.Linear(_snake_case			,r_idx - l_idx						)						)

				lowercase__   :    Union[str, Any]				=			keep_order







		def 					UpperCAmelCase    (	self						:  List[Any]			,_snake_case						:  int			,_snake_case						:  Union[str, Any]			,_snake_case						:  int			,_snake_case						:  Optional[Any]						)					->		Tuple:






				"""simple docstring"""



				if proj is None:
						lowercase__   :    List[Any]				=			nn.functional.linear(_snake_case			,_snake_case			,bias=_snake_case						)
				else:
						# if CUDA_MAJOR <= 9 and CUDA_MINOR <= 1:
						lowercase__   :    List[str]				=			nn.functional.linear(_snake_case			,proj.t().contiguous()						)
						lowercase__   :    Tuple				=			nn.functional.linear(_snake_case			,_snake_case			,bias=_snake_case						)
						# else:
						#     logit = torch.einsum('bd,de,ev->bv', (hidden, proj, weight.t()))
						#     if bias is not None:
						#         logit = logit + bias

				return logit







		def 					UpperCAmelCase    (	self						:  Dict			,_snake_case						:  List[str]			,_snake_case						:  Dict=None			,_snake_case						:  Dict=False						)					->		Optional[int]:






				"""simple docstring"""



				if labels is not None:
						# Shift so that tokens < n predict n
						lowercase__   :    List[str]				=			hidden[..., :-1, :].contiguous()
						lowercase__   :    List[str]				=			labels[..., 1:].contiguous()
						lowercase__   :    str				=			hidden.view(-1			,hidden.size(-1						)						)
						lowercase__   :    int				=			labels.view(-1						)
						if hidden.size(0						) != labels.size(0						):
								raise RuntimeError('''Input and labels should have the same size in the batch dimension.'''						)
				else:
						lowercase__   :    str				=			hidden.view(-1			,hidden.size(-1						)						)

				if self.n_clusters == 0:
						lowercase__   :    int				=			self._compute_logit(_snake_case			,self.out_layers[0].weight			,self.out_layers[0].bias			,self.out_projs[0]						)
						if labels is not None:
								lowercase__   :    Dict				=			labels != -100
								lowercase__   :    Union[str, Any]				=			torch.zeros_like(_snake_case			,dtype=hidden.dtype			,device=hidden.device						)
								lowercase__   :    List[str]				=			(
								    -nn.functional.log_softmax(_snake_case			,dim=-1						)[mask].gather(1			,labels[mask].unsqueeze(1						)						).squeeze(1						)
								)
						else:
								lowercase__   :    str				=			nn.functional.log_softmax(_snake_case			,dim=-1						)
				else:
						# construct weights and biases
						lowercase__		,			lowercase__   :    Dict				=			[], []
						for i in range(len(self.cutoffs						)						):
								if self.div_val == 1:
										lowercase__		,			lowercase__   :    Optional[Any]				=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
										lowercase__   :    List[str]				=			self.out_layers[0].weight[l_idx:r_idx]
										lowercase__   :    int				=			self.out_layers[0].bias[l_idx:r_idx]
								else:
										lowercase__   :    Union[str, Any]				=			self.out_layers[i].weight
										lowercase__   :    List[str]				=			self.out_layers[i].bias

								if i == 0:
										lowercase__   :    int				=			torch.cat([weight_i, self.cluster_weight]			,dim=0						)
										lowercase__   :    Tuple				=			torch.cat([bias_i, self.cluster_bias]			,dim=0						)

								weights.append(_snake_case						)
								biases.append(_snake_case						)

						lowercase__		,			lowercase__		,			lowercase__   :    Optional[Any]				=			weights[0], biases[0], self.out_projs[0]

						lowercase__   :    Optional[int]				=			self._compute_logit(_snake_case			,_snake_case			,_snake_case			,_snake_case						)
						lowercase__   :    Tuple				=			nn.functional.log_softmax(_snake_case			,dim=1						)

						if labels is None:
								lowercase__   :    Any				=			hidden.new_empty((head_logit.size(0						), self.n_token)						)
						else:
								lowercase__   :    List[Any]				=			torch.zeros_like(_snake_case			,dtype=hidden.dtype			,device=hidden.device						)

						lowercase__   :    Any				=			0
						lowercase__   :    Optional[int]				=			[0] + self.cutoffs
						for i in range(len(_snake_case						) - 1						):
								lowercase__		,			lowercase__   :    Any				=			cutoff_values[i], cutoff_values[i + 1]

								if labels is not None:
										lowercase__   :    Dict				=			(labels >= l_idx) & (labels < r_idx)
										lowercase__   :    Optional[int]				=			mask_i.nonzero().squeeze()

										if indices_i.numel() == 0:
												continue

										lowercase__   :    Optional[int]				=			labels.index_select(0			,_snake_case						) - l_idx
										lowercase__   :    Tuple				=			head_logprob.index_select(0			,_snake_case						)
										lowercase__   :    List[Any]				=			hidden.index_select(0			,_snake_case						)
								else:
										lowercase__   :    int				=			hidden

								if i == 0:
										if labels is not None:
												lowercase__   :    str				=			head_logprob_i.gather(1			,target_i[:, None]						).squeeze(1						)
										else:
												lowercase__   :    Dict				=			head_logprob[:, : self.cutoffs[0]]
								else:
										lowercase__		,			lowercase__		,			lowercase__   :    Optional[Any]				=			weights[i], biases[i], self.out_projs[i]

										lowercase__   :    Optional[Any]				=			self._compute_logit(_snake_case			,_snake_case			,_snake_case			,_snake_case						)
										lowercase__   :    Union[str, Any]				=			nn.functional.log_softmax(_snake_case			,dim=1						)
										lowercase__   :    Optional[int]				=			self.cutoffs[0] + i - 1  # No probability for the head cluster
										if labels is not None:
												lowercase__   :    Dict				=			head_logprob_i[:, cluster_prob_idx] + tail_logprob_i.gather(
												    1			,target_i[:, None]						).squeeze(1						)
										else:
												lowercase__   :    List[str]				=			head_logprob[:, cluster_prob_idx, None] + tail_logprob_i
												lowercase__   :    Optional[Any]				=			logprob_i

								if labels is not None:
										if (hasattr(self			,'''keep_order'''						) and self.keep_order) or keep_order:
												out.index_copy_(0			,_snake_case			,-logprob_i						)
										else:
												out[offset : offset + logprob_i.size(0						)].copy_(-logprob_i						)
										offset += logprob_i.size(0						)

				return out







		def 					UpperCAmelCase    (	self						:  Optional[Any]			,_snake_case						:  List[str]						)					->		int:






				"""simple docstring"""



				if self.n_clusters == 0:
						lowercase__   :    List[Any]				=			self._compute_logit(_snake_case			,self.out_layers[0].weight			,self.out_layers[0].bias			,self.out_projs[0]						)
						return nn.functional.log_softmax(_snake_case			,dim=-1						)
				else:
						# construct weights and biases
						lowercase__		,			lowercase__   :    Optional[Any]				=			[], []
						for i in range(len(self.cutoffs						)						):
								if self.div_val == 1:
										lowercase__		,			lowercase__   :    List[str]				=			self.cutoff_ends[i], self.cutoff_ends[i + 1]
										lowercase__   :    List[Any]				=			self.out_layers[0].weight[l_idx:r_idx]
										lowercase__   :    List[Any]				=			self.out_layers[0].bias[l_idx:r_idx]
								else:
										lowercase__   :    Optional[int]				=			self.out_layers[i].weight
										lowercase__   :    int				=			self.out_layers[i].bias

								if i == 0:
										lowercase__   :    str				=			torch.cat([weight_i, self.cluster_weight]			,dim=0						)
										lowercase__   :    Dict				=			torch.cat([bias_i, self.cluster_bias]			,dim=0						)

								weights.append(_snake_case						)
								biases.append(_snake_case						)

						lowercase__		,			lowercase__		,			lowercase__   :    List[str]				=			weights[0], biases[0], self.out_projs[0]
						lowercase__   :    Optional[Any]				=			self._compute_logit(_snake_case			,_snake_case			,_snake_case			,_snake_case						)

						lowercase__   :    Optional[Any]				=			hidden.new_empty((head_logit.size(0						), self.n_token)						)
						lowercase__   :    List[Any]				=			nn.functional.log_softmax(_snake_case			,dim=1						)

						lowercase__   :    Optional[Any]				=			[0] + self.cutoffs
						for i in range(len(_snake_case						) - 1						):
								lowercase__		,			lowercase__   :    Union[str, Any]				=			cutoff_values[i], cutoff_values[i + 1]

								if i == 0:
										lowercase__   :    Dict				=			head_logprob[:, : self.cutoffs[0]]
								else:
										lowercase__		,			lowercase__		,			lowercase__   :    List[str]				=			weights[i], biases[i], self.out_projs[i]

										lowercase__   :    Any				=			self._compute_logit(_snake_case			,_snake_case			,_snake_case			,_snake_case						)
										lowercase__   :    Any				=			nn.functional.log_softmax(_snake_case			,dim=1						)

										lowercase__   :    Any				=			head_logprob[:, -i] + tail_logprob_i
										lowercase__   :    str				=			logprob_i

						return out




"""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 						_A  (  _a  ):







    """simple docstring"""




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

    UpperCAmelCase       :							int	=       ["""image""", """text"""]
    UpperCAmelCase       :							int	=       ["""text"""]






    def __init__(					self				:    Tuple				, *__UpperCAmelCase				:    Union[str, Any]				, **__UpperCAmelCase				:    Any):
      if not is_vision_available():
        raise ValueError("Pillow must be installed to use the DocumentQuestionAnsweringTool.")

      super().__init__(*__UpperCAmelCase				, **__UpperCAmelCase)






    def 	__snake_case							(					self				:    Tuple				, __UpperCAmelCase				:    "Image"				, __UpperCAmelCase				:    str):
      a						:		Any      							=			"<s_docvqa><s_question>{user_input}</s_question><s_answer>"
      a						:		Union[str, Any]      							=			task_prompt.replace("{user_input}"				, __UpperCAmelCase)
      a						:		Optional[Any]      							=			self.pre_processor.tokenizer(
          __UpperCAmelCase				, add_special_tokens=__UpperCAmelCase				, return_tensors="pt").input_ids
      a						:		Any      							=			self.pre_processor(__UpperCAmelCase				, return_tensors="pt").pixel_values

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






    def 	__snake_case							(					self				:    int				, __UpperCAmelCase				:    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=__UpperCAmelCase				, pad_token_id=self.pre_processor.tokenizer.pad_token_id				, eos_token_id=self.pre_processor.tokenizer.eos_token_id				, use_cache=__UpperCAmelCase				, num_beams=1				, bad_words_ids=[[self.pre_processor.tokenizer.unk_token_id]]				, return_dict_in_generate=__UpperCAmelCase				, ).sequences






    def 	__snake_case							(					self				:    str				, __UpperCAmelCase				:    List[Any]):
      a						:		Union[str, Any]      							=			self.pre_processor.batch_decode(__UpperCAmelCase)[0]
      a						:		Optional[Any]      							=			sequence.replace(self.pre_processor.tokenizer.eos_token				, "")
      a						:		Any      							=			sequence.replace(self.pre_processor.tokenizer.pad_token				, "")
      a						:		Optional[Any]      							=			re.sub(r"<.*?>"				, ""				, __UpperCAmelCase				, count=1).strip()  # remove first task start token
      a						:		List[str]      							=			self.pre_processor.tokenajson(__UpperCAmelCase)

      return sequence["answer"]






