

import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
			import torch
			from torch import nn

			from transformers import (
			    MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
			    VideoMAEForPreTraining,
			    VideoMAEForVideoClassification,
			    VideoMAEModel,
			)
			from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
			from transformers import VideoMAEImageProcessor

class 				UpperCAmelCase_		:
							"""simple docstring"""







							def __init__(			self       ,							_a       ,							_a=1_3       ,							_a=1_0       ,							_a=3       ,							_a=2       ,							_a=2       ,							_a=2       ,							_a=True       ,							_a=True       ,							_a=3_2       ,							_a=5       ,							_a=4       ,							_a=3_7       ,							_a="gelu"       ,							_a=0.1       ,							_a=0.1       ,							_a=1_0       ,							_a=0.02       ,							_a=0.9       ,							_a=None       ,							)    ->    List[Any]:
												_a	:    Tuple         =		parent
												_a	:    Tuple         =		batch_size
												_a	:    List[str]         =		image_size
												_a	:    List[str]         =		num_channels
												_a	:    Optional[Any]         =		patch_size
												_a	:    Optional[int]         =		tubelet_size
												_a	:    Union[str, Any]         =		num_frames
												_a	:    Any         =		is_training
												_a	:    Optional[Any]         =		use_labels
												_a	:    str         =		hidden_size
												_a	:    Optional[Any]         =		num_hidden_layers
												_a	:    int         =		num_attention_heads
												_a	:    Dict         =		intermediate_size
												_a	:    List[str]         =		hidden_act
												_a	:    Optional[int]         =		hidden_dropout_prob
												_a	:    Dict         =		attention_probs_dropout_prob
												_a	:    Optional[Any]         =		type_sequence_label_size
												_a	:    List[Any]         =		initializer_range
												_a	:    Optional[int]         =		mask_ratio
												_a	:    Optional[Any]         =		scope

												# in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
												_a	:    int         =		(image_size // patch_size) ** 2
												_a	:    Tuple         =		(num_frames // tubelet_size) * self.num_patches_per_frame

												# use this variable to define bool_masked_pos
												_a	:    Any         =		int(mask_ratio * self.seq_length       )


							def 				__lowercase				(			self       )    ->    Tuple:
												_a	:    List[Any]         =		floats_tensor(
												    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

												_a	:    Any         =		None
												if self.use_labels:
																	_a	:    Optional[int]         =		ids_tensor([self.batch_size]       ,							self.type_sequence_label_size       )

												_a	:    List[Any]         =		self.get_config()

												return config, pixel_values, labels


							def 				__lowercase				(			self       )    ->    Union[str, Any]:
												return VideoMAEConfig(
												    image_size=self.image_size       ,							patch_size=self.patch_size       ,							num_channels=self.num_channels       ,							num_frames=self.num_frames       ,							tubelet_size=self.tubelet_size       ,							hidden_size=self.hidden_size       ,							num_hidden_layers=self.num_hidden_layers       ,							num_attention_heads=self.num_attention_heads       ,							intermediate_size=self.intermediate_size       ,							hidden_act=self.hidden_act       ,							hidden_dropout_prob=self.hidden_dropout_prob       ,							attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,							is_decoder=_a       ,							initializer_range=self.initializer_range       ,							)


							def 				__lowercase				(			self       ,							_a       ,							_a       ,							_a       )    ->    List[Any]:
												_a	:    str         =		VideoMAEModel(config=_a       )
												model.to(_a       )
												model.eval()
												_a	:    Any         =		model(_a       )
												self.parent.assertEqual(result.last_hidden_state.shape       ,							(self.batch_size, self.seq_length, self.hidden_size)       )


							def 				__lowercase				(			self       ,							_a       ,							_a       ,							_a       )    ->    Any:
												_a	:    Any         =		VideoMAEForPreTraining(_a       )
												model.to(_a       )
												model.eval()
												# important: each video needs to have the same number of masked patches
												# hence we define a single mask, which we then repeat for each example in the batch
												_a	:    List[str]         =		torch.ones((self.num_masks,)       )
												_a	:    List[str]         =		torch.cat([mask, torch.zeros(self.seq_length - mask.size(0       )       )]       )
												_a	:    Tuple         =		mask.expand(self.batch_size       ,							-1       ).bool()

												_a	:    Dict         =		model(_a       ,							_a       )
												# model only returns predictions for masked patches
												_a	:    Optional[Any]         =		mask.sum().item()
												_a	:    Dict         =		3 * self.tubelet_size * self.patch_size**2
												self.parent.assertEqual(result.logits.shape       ,							(self.batch_size, num_masked_patches, decoder_num_labels)       )




							def 				__lowercase				(			self       )    ->    Optional[int]:
												_a	:    List[Any]         =		self.prepare_config_and_inputs()
												_a      ,					_a      ,					_a	:    Tuple         =		config_and_inputs
												_a	:    Tuple         =		{'''pixel_values''': pixel_values}
												return config, inputs_dict



@require_torch
class 				UpperCAmelCase_		(			__lowercase				,       __lowercase				,       unittest.TestCase   ):
							"""simple docstring"""







							UpperCAmelCase__       :						Optional[int]	    =    (
							    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
							)
							UpperCAmelCase__       :						Optional[Any]	    =    (
							    {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
							    if is_torch_available()
							    else {}
							)

							UpperCAmelCase__       :						Optional[Any]	    =    False
							UpperCAmelCase__       :						Union[str, Any]	    =    False
							UpperCAmelCase__       :						int	    =    False
							UpperCAmelCase__       :						List[Any]	    =    False


							def 				__lowercase				(			self       )    ->    Any:
												_a	:    int         =		VideoMAEModelTester(self       )
												_a	:    int         =		ConfigTester(self       ,							config_class=_a       ,							has_text_modality=_a       ,							hidden_size=3_7       )


							def 				__lowercase				(			self       ,							_a       ,							_a       ,							_a=False       )    ->    List[str]:
												_a	:    List[str]         =		copy.deepcopy(_a       )

												if model_class == VideoMAEForPreTraining:
																	# important: each video needs to have the same number of masked patches
																	# hence we define a single mask, which we then repeat for each example in the batch
																	_a	:    List[str]         =		torch.ones((self.model_tester.num_masks,)       )
																	_a	:    str         =		torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0       )       )]       )
																	_a	:    Any         =		mask.expand(self.model_tester.batch_size       ,							-1       ).bool()
																	_a	:    Optional[int]         =		bool_masked_pos.to(_a       )

												if return_labels:
																	if model_class in [
																	    *get_values(_a       ),
																	]:
																						_a	:    List[str]         =		torch.zeros(
																						    self.model_tester.batch_size       ,							dtype=torch.long       ,							device=_a       )

												return inputs_dict


							def 				__lowercase				(			self       )    ->    Union[str, Any]:
												self.config_tester.run_common_tests()


							@unittest.skip(reason='''VideoMAE does not use inputs_embeds'''       )
							def 				__lowercase				(			self       )    ->    Dict:
												pass


							def 				__lowercase				(			self       )    ->    int:
												_a      ,					_a	:    Any         =		self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	_a	:    Any         =		model_class(_a       )
																	self.assertIsInstance(model.get_input_embeddings()       ,							(nn.Module)       )
																	_a	:    Optional[Any]         =		model.get_output_embeddings()
																	self.assertTrue(x is None or isinstance(_a       ,							nn.Linear       )       )


							def 				__lowercase				(			self       )    ->    Dict:
												_a      ,					_a	:    Union[str, Any]         =		self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	_a	:    Optional[int]         =		model_class(_a       )
																	_a	:    str         =		inspect.signature(model.forward       )
																	# signature.parameters is an OrderedDict => so arg_names order is deterministic
																	_a	:    Union[str, Any]         =		[*signature.parameters.keys()]

																	_a	:    Dict         =		['''pixel_values''']
																	self.assertListEqual(arg_names[:1]       ,							_a       )


							def 				__lowercase				(			self       )    ->    Optional[Any]:
												_a	:    str         =		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*_a       )


							def 				__lowercase				(			self       )    ->    List[Any]:
												_a	:    Tuple         =		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_pretraining(*_a       )


							@slow
							def 				__lowercase				(			self       )    ->    Optional[Any]:
												for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	_a	:    Union[str, Any]         =		VideoMAEModel.from_pretrained(_a       )
																	self.assertIsNotNone(_a       )


							def 				__lowercase				(			self       )    ->    Union[str, Any]:
												if not self.has_attentions:
																	pass

												else:
																	_a      ,					_a	:    Dict         =		self.model_tester.prepare_config_and_inputs_for_common()
																	_a	:    Dict         =		True

																	for model_class in self.all_model_classes:
																						_a	:    int         =		self.model_tester.seq_length - self.model_tester.num_masks
																						_a	:    Optional[Any]         =		(
																						    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
																						)

																						_a	:    str         =		True
																						_a	:    Optional[int]         =		False
																						_a	:    str         =		True
																						_a	:    Any         =		model_class(_a       )
																						model.to(_a       )
																						model.eval()
																						with torch.no_grad():
																											_a	:    str         =		model(**self._prepare_for_class(_a       ,							_a       )       )
																						_a	:    int         =		outputs.attentions
																						self.assertEqual(len(_a       )       ,							self.model_tester.num_hidden_layers       )

																						# check that output_attentions also work using config
																						del inputs_dict["output_attentions"]
																						_a	:    Optional[int]         =		True
																						_a	:    Tuple         =		model_class(_a       )
																						model.to(_a       )
																						model.eval()
																						with torch.no_grad():
																											_a	:    int         =		model(**self._prepare_for_class(_a       ,							_a       )       )
																						_a	:    int         =		outputs.attentions
																						self.assertEqual(len(_a       )       ,							self.model_tester.num_hidden_layers       )

																						self.assertListEqual(
																						    list(attentions[0].shape[-3:]       )       ,							[self.model_tester.num_attention_heads, seq_len, seq_len]       ,							)
																						_a	:    int         =		len(_a       )

																						# Check attention is always last and order is fine
																						_a	:    int         =		True
																						_a	:    int         =		True
																						_a	:    List[Any]         =		model_class(_a       )
																						model.to(_a       )
																						model.eval()
																						with torch.no_grad():
																											_a	:    Any         =		model(**self._prepare_for_class(_a       ,							_a       )       )

																						self.assertEqual(out_len + 1       ,							len(_a       )       )

																						_a	:    Tuple         =		outputs.attentions

																						self.assertEqual(len(_a       )       ,							self.model_tester.num_hidden_layers       )
																						self.assertListEqual(
																						    list(self_attentions[0].shape[-3:]       )       ,							[self.model_tester.num_attention_heads, seq_len, seq_len]       ,							)


							def 				__lowercase				(			self       )    ->    Dict:
												def check_hidden_states_output(_a       ,							_a       ,							_a       ):
																	_a	:    Optional[int]         =		model_class(_a       )
																	model.to(_a       )
																	model.eval()

																	with torch.no_grad():
																						_a	:    List[str]         =		model(**self._prepare_for_class(_a       ,							_a       )       )

																	_a	:    int         =		outputs.hidden_states
																	_a	:    List[Any]         =		self.model_tester.num_hidden_layers + 1
																	self.assertEqual(len(_a       )       ,							_a       )

																	_a	:    Optional[int]         =		self.model_tester.seq_length - self.model_tester.num_masks
																	_a	:    str         =		num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

																	self.assertListEqual(
																	    list(hidden_states[0].shape[-2:]       )       ,							[seq_length, self.model_tester.hidden_size]       ,							)

												_a      ,					_a	:    List[str]         =		self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	_a	:    List[str]         =		True
																	check_hidden_states_output(_a       ,							_a       ,							_a       )

																	# check that output_hidden_states also work using config
																	del inputs_dict["output_hidden_states"]
																	_a	:    str         =		True

																	check_hidden_states_output(_a       ,							_a       ,							_a       )




							@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''       )
							def 				__lowercase				(			self       )    ->    Union[str, Any]:
												pass



def    __UpperCAmelCase (	)			->     Tuple:

					"""simple docstring"""

					_a	:    str         =		hf_hub_download(
					    repo_id='''hf-internal-testing/spaghetti-video'''		,filename='''eating_spaghetti.npy'''		,repo_type='''dataset'''	)
					_a	:    List[Any]         =		np.load(__a	)
					return list(__a	)







@require_torch
@require_vision
class 				UpperCAmelCase_		(			unittest.TestCase   ):
							"""simple docstring"""







							@cached_property
							def 				__lowercase				(			self       )    ->    Tuple:
												# logits were tested with a different mean and std, so we use the same here
												return (
												    VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]       ,							image_std=[0.5, 0.5, 0.5]       )
												    if is_vision_available()
												    else None
												)


							@slow
							def 				__lowercase				(			self       )    ->    Tuple:
												_a	:    Optional[int]         =		VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''       ).to(
												    _a       )

												_a	:    Any         =		self.default_image_processor
												_a	:    Dict         =		prepare_video()
												_a	:    List[str]         =		image_processor(_a       ,							return_tensors='''pt'''       ).to(_a       )

												# forward pass
												with torch.no_grad():
																	_a	:    Any         =		model(**_a       )

												# verify the logits
												_a	:    List[str]         =		torch.Size((1, 4_0_0)       )
												self.assertEqual(outputs.logits.shape       ,							_a       )

												_a	:    List[Any]         =		torch.tensor([0.3669, -0.0688, -0.2421]       ).to(_a       )

												self.assertTrue(torch.allclose(outputs.logits[0, :3]       ,							_a       ,							atol=1e-4       )       )




							@slow
							def 				__lowercase				(			self       )    ->    Dict:
												_a	:    int         =		VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''       ).to(_a       )

												_a	:    Dict         =		self.default_image_processor
												_a	:    List[Any]         =		prepare_video()
												_a	:    Dict         =		image_processor(_a       ,							return_tensors='''pt'''       ).to(_a       )

												# add boolean mask, indicating which patches to mask
												_a	:    Union[str, Any]         =		hf_hub_download(repo_id='''hf-internal-testing/bool-masked-pos'''       ,							filename='''bool_masked_pos.pt'''       )
												_a	:    Optional[int]         =		torch.load(_a       )

												# forward pass
												with torch.no_grad():
																	_a	:    List[Any]         =		model(**_a       )

												# verify the logits
												_a	:    int         =		torch.Size([1, 1_4_0_8, 1_5_3_6]       )
												_a	:    Optional[Any]         =		torch.tensor(
												    [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]       ,							device=_a       )
												self.assertEqual(outputs.logits.shape       ,							_a       )
												self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]       ,							_a       ,							atol=1e-4       )       )

												# verify the loss (`config.norm_pix_loss` = `True`)
												_a	:    str         =		torch.tensor([0.5142]       ,							device=_a       )
												self.assertTrue(torch.allclose(outputs.loss       ,							_a       ,							atol=1e-4       )       )

												# verify the loss (`config.norm_pix_loss` = `False`)
												_a	:    List[str]         =		VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''       ,							norm_pix_loss=_a       ).to(
												    _a       )

												with torch.no_grad():
																	_a	:    Any         =		model(**_a       )

												_a	:    Any         =		torch.tensor(torch.tensor([0.6469]       )       ,							device=_a       )
												self.assertTrue(torch.allclose(outputs.loss       ,							_a       ,							atol=1e-4       )       )






import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding

class 				UpperCAmelCase_		(			__lowercase   ):
							"""simple docstring"""







							UpperCAmelCase__       :						str	    =    ["image_processor", "tokenizer"]
							UpperCAmelCase__       :						Optional[int]	    =    "CLIPImageProcessor"
							UpperCAmelCase__       :						Tuple	    =    ("XLMRobertaTokenizer", "XLMRobertaTokenizerFast")


							def __init__(			self       ,							_a=None       ,							_a=None       ,							**_a       )    ->    Any:
												_a	:    int         =		None
												if "feature_extractor" in kwargs:
																	warnings.warn(
																	    '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
																	    ''' instead.'''       ,							_a       ,							)
																	_a	:    int         =		kwargs.pop('''feature_extractor'''       )

												_a	:    int         =		image_processor if image_processor is not None else feature_extractor
												if image_processor is None:
																	raise ValueError('''You need to specify an `image_processor`.'''       )
												if tokenizer is None:
																	raise ValueError('''You need to specify a `tokenizer`.'''       )

												super().__init__(_a       ,							_a       )


							def __call__(			self       ,							_a=None       ,							_a=None       ,							_a=None       ,							**_a       )    ->    List[Any]:

												if text is None and images is None:
																	raise ValueError('''You have to specify either text or images. Both cannot be none.'''       )

												if text is not None:
																	_a	:    str         =		self.tokenizer(_a       ,							return_tensors=_a       ,							**_a       )

												if images is not None:
																	_a	:    Optional[Any]         =		self.image_processor(_a       ,							return_tensors=_a       ,							**_a       )

												if text is not None and images is not None:
																	_a	:    List[Any]         =		image_features.pixel_values
																	return encoding
												elif text is not None:
																	return encoding
												else:
																	return BatchEncoding(data=dict(**_a       )       ,							tensor_type=_a       )


							def 				__lowercase				(			self       ,							*_a       ,							**_a       )    ->    Optional[Any]:
												return self.tokenizer.batch_decode(*_a       ,							**_a       )


							def 				__lowercase				(			self       ,							*_a       ,							**_a       )    ->    Optional[int]:
												return self.tokenizer.decode(*_a       ,							**_a       )




							@property
							def 				__lowercase				(			self       )    ->    Optional[int]:
												_a	:    Dict         =		self.tokenizer.model_input_names
												_a	:    Dict         =		self.image_processor.model_input_names
												return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names       )       )



