


'''simple docstring'''
import unittest

from transformers import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
		from PIL import Image
else:
		class      lowercase			:

				"""simple docstring"""

				@staticmethod
				def 					lowerCAmelCase__       (	*UpperCamelCase_					,     **UpperCamelCase_							):

									'''simple docstring'''


									pass
@is_pipeline_test
@require_torch
@require_vision
class      lowercase			(							unittest.TestCase      ):

		"""simple docstring"""

		_a									=       MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING




		def 					lowerCAmelCase__       (	self					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_							):

							'''simple docstring'''


							UpperCamelCase__     :str							      =				pipeline('''visual-question-answering'''					,     model='''hf-internal-testing/tiny-vilt-random-vqa'''							)
							UpperCamelCase__     :Any							      =				[
							    {
							        '''image''': Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							),
							        '''question''': '''How many cats are there?''',
							    },
							    {
							        '''image''': '''./tests/fixtures/tests_samples/COCO/000000039769.png''',
							        '''question''': '''How many cats are there?''',
							    },
							]
							return vqa_pipeline, examples




		def 					lowerCAmelCase__       (	self					,     UpperCamelCase_					,     UpperCamelCase_							):

							'''simple docstring'''


							UpperCamelCase__     :Tuple							      =				vqa_pipeline(UpperCamelCase_					,     top_k=1							)
							self.assertEqual(
							    UpperCamelCase_					,     [
							        [{'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}],
							        [{'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}],
							    ]					,     )




		@require_torch
		def 					lowerCAmelCase__       (	self							):

							'''simple docstring'''


							UpperCamelCase__     :Union[str, Any]							      =				pipeline('''visual-question-answering'''					,     model='''hf-internal-testing/tiny-vilt-random-vqa'''							)
							UpperCamelCase__     :Optional[int]							      =				'''./tests/fixtures/tests_samples/COCO/000000039769.png'''
							UpperCamelCase__     :Optional[Any]							      =				'''How many cats are there?'''

							UpperCamelCase__     :Any							      =				vqa_pipeline(image=UpperCamelCase_					,     question='''How many cats are there?'''					,     top_k=2							)
							self.assertEqual(
							    UpperCamelCase_					,     [{'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}, {'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}]							)

							UpperCamelCase__     :List[Any]							      =				vqa_pipeline({'''image''': image, '''question''': question}					,     top_k=2							)
							self.assertEqual(
							    UpperCamelCase_					,     [{'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}, {'''score''': ANY(UpperCamelCase_							), '''answer''': ANY(UpperCamelCase_							)}]							)




		@slow
		@require_torch
		def 					lowerCAmelCase__       (	self							):

							'''simple docstring'''


							UpperCamelCase__     :int							      =				pipeline('''visual-question-answering'''					,     model='''dandelin/vilt-b32-finetuned-vqa'''							)
							UpperCamelCase__     :Optional[int]							      =				'''./tests/fixtures/tests_samples/COCO/000000039769.png'''
							UpperCamelCase__     :Tuple							      =				'''How many cats are there?'''

							UpperCamelCase__     :Optional[int]							      =				vqa_pipeline(image=UpperCamelCase_					,     question=UpperCamelCase_					,     top_k=2							)
							self.assertEqual(
							    nested_simplify(UpperCamelCase_					,     decimals=4							)					,     [{'''score''': 0.8799, '''answer''': '''2'''}, {'''score''': 0.296, '''answer''': '''1'''}]							)

							UpperCamelCase__     :List[str]							      =				vqa_pipeline({'''image''': image, '''question''': question}					,     top_k=2							)
							self.assertEqual(
							    nested_simplify(UpperCamelCase_					,     decimals=4							)					,     [{'''score''': 0.8799, '''answer''': '''2'''}, {'''score''': 0.296, '''answer''': '''1'''}]							)

							UpperCamelCase__     :Tuple							      =				vqa_pipeline(
							    [{'''image''': image, '''question''': question}, {'''image''': image, '''question''': question}]					,     top_k=2							)
							self.assertEqual(
							    nested_simplify(UpperCamelCase_					,     decimals=4							)					,     [[{'''score''': 0.8799, '''answer''': '''2'''}, {'''score''': 0.296, '''answer''': '''1'''}]] * 2					,     )




		@require_tf
		@unittest.skip('''Visual question answering not implemented in TF'''							)
		def 					lowerCAmelCase__       (	self							):

							'''simple docstring'''


							pass



'''simple docstring'''
import torch
from torch import nn
class      lowercase			(							nn.Module      ):

		"""simple docstring"""

		def __init__(	self					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_=1					,     UpperCamelCase_=False							):

							'''simple docstring'''


							super().__init__()

							UpperCamelCase__     :Dict							      =				n_token
							UpperCamelCase__     :List[Any]							      =				d_embed
							UpperCamelCase__     :Dict							      =				d_proj

							UpperCamelCase__     :Dict							      =				cutoffs + [n_token]
							UpperCamelCase__     :Union[str, Any]							      =				[0] + self.cutoffs
							UpperCamelCase__     :Any							      =				div_val

							UpperCamelCase__     :int							      =				self.cutoffs[0]
							UpperCamelCase__     :List[Any]							      =				len(self.cutoffs							) - 1
							UpperCamelCase__     :List[Any]							      =				self.shortlist_size + self.n_clusters

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

							UpperCamelCase__     :Union[str, Any]							      =				nn.ModuleList()
							UpperCamelCase__     :str							      =				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(UpperCamelCase_					,     UpperCamelCase_							)							)							)
																	else:
																						self.out_projs.append(UpperCamelCase_							)

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

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

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

							UpperCamelCase__     :Tuple							      =				keep_order




		def 					lowerCAmelCase__       (	self					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_							):

							'''simple docstring'''


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

							return logit




		def 					lowerCAmelCase__       (	self					,     UpperCamelCase_					,     UpperCamelCase_=None					,     UpperCamelCase_=False							):

							'''simple docstring'''


							if labels is not None:
												# Shift so that tokens < n predict n
												UpperCamelCase__     :Optional[Any]							      =				hidden[..., :-1, :].contiguous()
												UpperCamelCase__     :Optional[Any]							      =				labels[..., 1:].contiguous()
												UpperCamelCase__     :Optional[Any]							      =				hidden.view(-1					,     hidden.size(-1							)							)
												UpperCamelCase__     :str							      =				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:
												UpperCamelCase__     :int							      =				hidden.view(-1					,     hidden.size(-1							)							)

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

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

																	weights.append(UpperCamelCase_							)
																	biases.append(UpperCamelCase_							)

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

												UpperCamelCase__     :str							      =				self._compute_logit(UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_							)
												UpperCamelCase__     :List[Any]							      =				nn.functional.log_softmax(UpperCamelCase_					,     dim=1							)

												if labels is None:
																	UpperCamelCase__     :Union[str, Any]							      =				hidden.new_empty((head_logit.size(0							), self.n_token)							)
												else:
																	UpperCamelCase__     :Any							      =				torch.zeros_like(UpperCamelCase_					,     dtype=hidden.dtype					,     device=hidden.device							)

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

																	if labels is not None:
																						UpperCamelCase__     :Any							      =				(labels >= l_idx) & (labels < r_idx)
																						UpperCamelCase__     :int							      =				mask_i.nonzero().squeeze()

																						if indices_i.numel() == 0:
																											continue

																						UpperCamelCase__     :Tuple							      =				labels.index_select(0					,     UpperCamelCase_							) - l_idx
																						UpperCamelCase__     :str							      =				head_logprob.index_select(0					,     UpperCamelCase_							)
																						UpperCamelCase__     :int							      =				hidden.index_select(0					,     UpperCamelCase_							)
																	else:
																						UpperCamelCase__     :Dict							      =				hidden

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

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

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

							return out




		def 					lowerCAmelCase__       (	self					,     UpperCamelCase_							):

							'''simple docstring'''


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

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

																	weights.append(UpperCamelCase_							)
																	biases.append(UpperCamelCase_							)

												UpperCamelCase__      ,       UpperCamelCase__      ,       UpperCamelCase__     :str							      =				weights[0], biases[0], self.out_projs[0]
												UpperCamelCase__     :Optional[int]							      =				self._compute_logit(UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_							)

												UpperCamelCase__     :int							      =				hidden.new_empty((head_logit.size(0							), self.n_token)							)
												UpperCamelCase__     :Union[str, Any]							      =				nn.functional.log_softmax(UpperCamelCase_					,     dim=1							)

												UpperCamelCase__     :int							      =				[0] + self.cutoffs
												for i in range(len(UpperCamelCase_							) - 1							):
																	UpperCamelCase__      ,       UpperCamelCase__     :Union[str, Any]							      =				cutoff_values[i], cutoff_values[i + 1]

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

																						UpperCamelCase__     :List[str]							      =				self._compute_logit(UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_					,     UpperCamelCase_							)
																						UpperCamelCase__     :Tuple							      =				nn.functional.log_softmax(UpperCamelCase_					,     dim=1							)

																						UpperCamelCase__     :Optional[Any]							      =				head_logprob[:, -i] + tail_logprob_i
																						UpperCamelCase__     :Union[str, Any]							      =				logprob_i

												return out