



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  lowerCamelCase__    :




		'''simple docstring'''



		def __init__(self	,__lowerCamelCase	,__lowerCamelCase=13	,__lowerCamelCase=10	,__lowerCamelCase=3	,__lowerCamelCase=2	,__lowerCamelCase=2	,__lowerCamelCase=2	,__lowerCamelCase=True	,__lowerCamelCase=True	,__lowerCamelCase=32	,__lowerCamelCase=5	,__lowerCamelCase=4	,__lowerCamelCase=37	,__lowerCamelCase="gelu"	,__lowerCamelCase=0.1	,__lowerCamelCase=0.1	,__lowerCamelCase=10	,__lowerCamelCase=0.02	,__lowerCamelCase=0.9	,__lowerCamelCase=None	,)  ->    Any:

						"""simple docstring"""






						lowerCAmelCase__  :       List[Any]				=  parent
						lowerCAmelCase__  :       List[str]				=  batch_size
						lowerCAmelCase__  :       Optional[int]				=  image_size
						lowerCAmelCase__  :       int				=  num_channels
						lowerCAmelCase__  :       Union[str, Any]				=  patch_size
						lowerCAmelCase__  :       int				=  tubelet_size
						lowerCAmelCase__  :       List[Any]				=  num_frames
						lowerCAmelCase__  :       str				=  is_training
						lowerCAmelCase__  :       int				=  use_labels
						lowerCAmelCase__  :       str				=  hidden_size
						lowerCAmelCase__  :       List[str]				=  num_hidden_layers
						lowerCAmelCase__  :       Optional[int]				=  num_attention_heads
						lowerCAmelCase__  :       Union[str, Any]				=  intermediate_size
						lowerCAmelCase__  :       Optional[Any]				=  hidden_act
						lowerCAmelCase__  :       Dict				=  hidden_dropout_prob
						lowerCAmelCase__  :       Any				=  attention_probs_dropout_prob
						lowerCAmelCase__  :       Optional[Any]				=  type_sequence_label_size
						lowerCAmelCase__  :       Union[str, Any]				=  initializer_range
						lowerCAmelCase__  :       Tuple				=  mask_ratio
						lowerCAmelCase__  :       Any				=  scope

						# in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
						lowerCAmelCase__  :       int				=  (image_size // patch_size) ** 2
						lowerCAmelCase__  :       Optional[Any]				=  (num_frames // tubelet_size) * self.num_patches_per_frame

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



		def 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__  :       Tuple				=  floats_tensor(
						    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]				)

						lowerCAmelCase__  :       List[str]				=  None
						if self.use_labels:
										lowerCAmelCase__  :       Union[str, Any]				=  ids_tensor([self.batch_size]	,self.type_sequence_label_size				)

						lowerCAmelCase__  :       Tuple				=  self.get_config()

						return config, pixel_values, labels



		def 	lowerCAmelCase__   (self				)  ->    str:

						"""simple docstring"""






						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=__lowerCamelCase	,initializer_range=self.initializer_range	,)



		def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__  :       Tuple				=  VideoMAEModel(config=__lowerCamelCase				)
						model.to(__lowerCamelCase				)
						model.eval()
						lowerCAmelCase__  :       Any				=  model(__lowerCamelCase				)
						self.parent.assertEqual(result.last_hidden_state.shape	,(self.batch_size, self.seq_length, self.hidden_size)				)



		def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__  :       int				=  VideoMAEForPreTraining(__lowerCamelCase				)
						model.to(__lowerCamelCase				)
						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
						lowerCAmelCase__  :       Optional[Any]				=  torch.ones((self.num_masks,)				)
						lowerCAmelCase__  :       Optional[int]				=  torch.cat([mask, torch.zeros(self.seq_length - mask.size(0				)				)]				)
						lowerCAmelCase__  :       int				=  mask.expand(self.batch_size	,-1				).bool()

						lowerCAmelCase__  :       Any				=  model(__lowerCamelCase	,__lowerCamelCase				)
						# model only returns predictions for masked patches
						lowerCAmelCase__  :       str				=  mask.sum().item()
						lowerCAmelCase__  :       Tuple				=  3 * self.tubelet_size * self.patch_size**2
						self.parent.assertEqual(result.logits.shape	,(self.batch_size, num_masked_patches, decoder_num_labels)				)



		def 	lowerCAmelCase__   (self				)  ->    List[Any]:

						"""simple docstring"""






						lowerCAmelCase__  :       List[Any]				=  self.prepare_config_and_inputs()
						lowerCAmelCase__		,	lowerCAmelCase__		,	lowerCAmelCase__  :       Any				=  config_and_inputs
						lowerCAmelCase__  :       Dict				=  {'''pixel_values''': pixel_values}
						return config, inputs_dict



@require_torch
class  lowerCamelCase__    (							lowerCamelCase__   ,       lowerCamelCase__   ,       unittest.TestCase):




		'''simple docstring'''



		snake_case_			=(
		    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
		)
		snake_case_			=(
		    {"""feature-extraction""": VideoMAEModel, """video-classification""": VideoMAEForVideoClassification}
		    if is_torch_available()
		    else {}
		)

		snake_case_			=False
		snake_case_			=False
		snake_case_			=False
		snake_case_			=False



		def 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__  :       str				=  VideoMAEModelTester(self				)
						lowerCAmelCase__  :       List[str]				=  ConfigTester(self	,config_class=__lowerCamelCase	,has_text_modality=__lowerCamelCase	,hidden_size=37				)



		def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase=False				)  ->    Optional[Any]:

						"""simple docstring"""






						lowerCAmelCase__  :       List[str]				=  copy.deepcopy(__lowerCamelCase				)

						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
										lowerCAmelCase__  :       List[str]				=  torch.ones((self.model_tester.num_masks,)				)
										lowerCAmelCase__  :       Tuple				=  torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0				)				)]				)
										lowerCAmelCase__  :       Optional[Any]				=  mask.expand(self.model_tester.batch_size	,-1				).bool()
										lowerCAmelCase__  :       Dict				=  bool_masked_pos.to(__lowerCamelCase				)

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

						return inputs_dict



		def 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						self.config_tester.run_common_tests()



		@unittest.skip(reason='''VideoMAE does not use inputs_embeds'''				)
		def 	lowerCAmelCase__   (self				)  ->    Union[str, Any]:

						"""simple docstring"""






						pass



		def 	lowerCAmelCase__   (self				)  ->    Optional[int]:

						"""simple docstring"""






						lowerCAmelCase__		,	lowerCAmelCase__  :       Tuple				=  self.model_tester.prepare_config_and_inputs_for_common()

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



		def 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__		,	lowerCAmelCase__  :       Optional[int]				=  self.model_tester.prepare_config_and_inputs_for_common()

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

										lowerCAmelCase__  :       List[str]				=  ['''pixel_values''']
										self.assertListEqual(arg_names[:1]	,__lowerCamelCase				)



		def 	lowerCAmelCase__   (self				)  ->    str:

						"""simple docstring"""






						lowerCAmelCase__  :       List[str]				=  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*__lowerCamelCase				)



		def 	lowerCAmelCase__   (self				)  ->    Dict:

						"""simple docstring"""






						lowerCAmelCase__  :       List[str]				=  self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_pretraining(*__lowerCamelCase				)



		@slow
		def 	lowerCAmelCase__   (self				)  ->    str:

						"""simple docstring"""






						for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										lowerCAmelCase__  :       Union[str, Any]				=  VideoMAEModel.from_pretrained(__lowerCamelCase				)
										self.assertIsNotNone(__lowerCamelCase				)



		def 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						if not self.has_attentions:
										pass

						else:
										lowerCAmelCase__		,	lowerCAmelCase__  :       Dict				=  self.model_tester.prepare_config_and_inputs_for_common()
										lowerCAmelCase__  :       List[Any]				=  True

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

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

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

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

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

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

														lowerCAmelCase__  :       Optional[int]				=  outputs.attentions

														self.assertEqual(len(__lowerCamelCase				)	,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 	lowerCAmelCase__   (self				)  ->    List[Any]:

						"""simple docstring"""






						def check_hidden_states_output(__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase				):
										lowerCAmelCase__  :       Any				=  model_class(__lowerCamelCase				)
										model.to(__lowerCamelCase				)
										model.eval()

										with torch.no_grad():
														lowerCAmelCase__  :       Optional[Any]				=  model(**self._prepare_for_class(__lowerCamelCase	,__lowerCamelCase				)				)

										lowerCAmelCase__  :       Any				=  outputs.hidden_states
										lowerCAmelCase__  :       Union[str, Any]				=  self.model_tester.num_hidden_layers + 1
										self.assertEqual(len(__lowerCamelCase				)	,__lowerCamelCase				)

										lowerCAmelCase__  :       str				=  self.model_tester.seq_length - self.model_tester.num_masks
										lowerCAmelCase__  :       int				=  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]	,)

						lowerCAmelCase__		,	lowerCAmelCase__  :       Any				=  self.model_tester.prepare_config_and_inputs_for_common()

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

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

										check_hidden_states_output(__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase				)



		@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''				)
		def 	lowerCAmelCase__   (self				)  ->    int:

						"""simple docstring"""






						pass






def 			lowerCAmelCase__	(					):





				'''simple docstring'''

				lowerCAmelCase__  :       str				=  hf_hub_download(
				    repo_id='''hf-internal-testing/spaghetti-video''' ,filename='''eating_spaghetti.npy''' ,repo_type='''dataset''')
				lowerCAmelCase__  :       Tuple				=  np.load(lowerCamelCase_)
				return list(lowerCamelCase_)



@require_torch
@require_vision
class  lowerCamelCase__    (							unittest.TestCase):




		'''simple docstring'''



		@cached_property
		def 	lowerCAmelCase__   (self				)  ->    Optional[Any]:

						"""simple docstring"""






						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 	lowerCAmelCase__   (self				)  ->    List[str]:

						"""simple docstring"""






						lowerCAmelCase__  :       int				=  VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''				).to(
						    __lowerCamelCase				)

						lowerCAmelCase__  :       Dict				=  self.default_image_processor
						lowerCAmelCase__  :       Dict				=  prepare_video()
						lowerCAmelCase__  :       str				=  image_processor(__lowerCamelCase	,return_tensors='''pt'''				).to(__lowerCamelCase				)

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

						# verify the logits
						lowerCAmelCase__  :       Any				=  torch.Size((1, 4_00)				)
						self.assertEqual(outputs.logits.shape	,__lowerCamelCase				)

						lowerCAmelCase__  :       List[str]				=  torch.tensor([0.3669, -0.0688, -0.2421]				).to(__lowerCamelCase				)

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



		@slow
		def 	lowerCAmelCase__   (self				)  ->    Any:

						"""simple docstring"""






						lowerCAmelCase__  :       Union[str, Any]				=  VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''				).to(__lowerCamelCase				)

						lowerCAmelCase__  :       Any				=  self.default_image_processor
						lowerCAmelCase__  :       str				=  prepare_video()
						lowerCAmelCase__  :       Tuple				=  image_processor(__lowerCamelCase	,return_tensors='''pt'''				).to(__lowerCamelCase				)

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

						# forward pass
						with torch.no_grad():
										lowerCAmelCase__  :       str				=  model(**__lowerCamelCase				)

						# verify the logits
						lowerCAmelCase__  :       Tuple				=  torch.Size([1, 14_08, 15_36]				)
						lowerCAmelCase__  :       Union[str, Any]				=  torch.tensor(
						    [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]	,device=__lowerCamelCase				)
						self.assertEqual(outputs.logits.shape	,__lowerCamelCase				)
						self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]	,__lowerCamelCase	,atol=1e-4				)				)

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

						# verify the loss (`config.norm_pix_loss` = `False`)
						lowerCAmelCase__  :       int				=  VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''	,norm_pix_loss=__lowerCamelCase				).to(
						    __lowerCamelCase				)

						with torch.no_grad():
										lowerCAmelCase__  :       List[Any]				=  model(**__lowerCamelCase				)

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






import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__snake_case       :	Any            =logging.get_logger(__name__)



class  lowerCamelCase__    (							lowerCamelCase__):




		'''simple docstring'''



		snake_case_			="""encoder-decoder"""
		snake_case_			=True



		def __init__(self	,**__lowerCamelCase				)  ->    Union[str, Any]:

						"""simple docstring"""






						super().__init__(**__lowerCamelCase				)
						assert (
						    "encoder" in kwargs and "decoder" in kwargs
						), "Config has to be initialized with encoder and decoder config"
						lowerCAmelCase__  :       Optional[Any]				=  kwargs.pop('''encoder'''				)
						lowerCAmelCase__  :       Any				=  encoder_config.pop('''model_type'''				)
						lowerCAmelCase__  :       str				=  kwargs.pop('''decoder'''				)
						lowerCAmelCase__  :       Tuple				=  decoder_config.pop('''model_type'''				)

						from ..auto.configuration_auto import AutoConfig

						lowerCAmelCase__  :       Tuple				=  AutoConfig.for_model(__lowerCamelCase	,**__lowerCamelCase				)
						lowerCAmelCase__  :       Any				=  AutoConfig.for_model(__lowerCamelCase	,**__lowerCamelCase				)
						lowerCAmelCase__  :       str				=  True



		@classmethod
		def 	lowerCAmelCase__   (cls	,__lowerCamelCase	,__lowerCamelCase	,**__lowerCamelCase				)  ->    PretrainedConfig:

						"""simple docstring"""






						logger.info('''Set `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config'''				)
						lowerCAmelCase__  :       Optional[int]				=  True
						lowerCAmelCase__  :       List[str]				=  True

						return cls(encoder=encoder_config.to_dict()	,decoder=decoder_config.to_dict()	,**__lowerCamelCase				)



		def 	lowerCAmelCase__   (self				)  ->    Tuple:

						"""simple docstring"""






						lowerCAmelCase__  :       int				=  copy.deepcopy(self.__dict__				)
						lowerCAmelCase__  :       Optional[Any]				=  self.encoder.to_dict()
						lowerCAmelCase__  :       str				=  self.decoder.to_dict()
						lowerCAmelCase__  :       Optional[int]				=  self.__class__.model_type
						return output

