





'''simple docstring'''






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

		def __init__(   self					:   List[Any]  ,      __magic_name__					:   Optional[int]  ,      __magic_name__					:   str=13  ,      __magic_name__					:   Tuple=10  ,      __magic_name__					:   Union[str, Any]=3  ,      __magic_name__					:   Union[str, Any]=2  ,      __magic_name__					:   Optional[Any]=2  ,      __magic_name__					:   str=2  ,      __magic_name__					:   Dict=True  ,      __magic_name__					:   Union[str, Any]=True  ,      __magic_name__					:   Optional[int]=32  ,      __magic_name__					:   str=5  ,      __magic_name__					:   Tuple=4  ,      __magic_name__					:   Tuple=37  ,      __magic_name__					:   Union[str, Any]="gelu"  ,      __magic_name__					:   Any=0.1  ,      __magic_name__					:   int=0.1  ,      __magic_name__					:   Tuple=10  ,      __magic_name__					:   Optional[int]=0.02  ,      __magic_name__					:   Optional[Any]=0.9  ,      __magic_name__					:   str=None  ,      )					->    Any:


					"""simple docstring"""

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

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

					# use this variable to define bool_masked_pos
					__snake_case					:      Optional[Any]							      =     int(mask_ratio * self.seq_length			)

		def 	lowercase__     (   self					:   Optional[int]			)					->    str:


					"""simple docstring"""

					__snake_case					:      Optional[Any]							      =     floats_tensor(
					    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

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

					__snake_case					:      List[str]							      =     self.get_config()

					return config, pixel_values, labels

		def 	lowercase__     (   self					:   Union[str, Any]			)					->    int:


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

		def 	lowercase__     (   self					:   Optional[Any]  ,      __magic_name__					:   Any  ,      __magic_name__					:   str  ,      __magic_name__					:   Tuple			)					->    int:


					"""simple docstring"""

					__snake_case					:      Tuple							      =     VideoMAEModel(config=__magic_name__			)
					model.to(__magic_name__			)
					model.eval()
					__snake_case					:      Optional[int]							      =     model(__magic_name__			)
					self.parent.assertEqual(result.last_hidden_state.shape  ,      (self.batch_size, self.seq_length, self.hidden_size)			)

		def 	lowercase__     (   self					:   Tuple  ,      __magic_name__					:   str  ,      __magic_name__					:   Tuple  ,      __magic_name__					:   Any			)					->    int:


					"""simple docstring"""

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

					__snake_case					:      Union[str, Any]							      =     model(__magic_name__  ,      __magic_name__			)
					# model only returns predictions for masked patches
					__snake_case					:      List[Any]							      =     mask.sum().item()
					__snake_case					:      Any							      =     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					:   int			)					->    Any:


					"""simple docstring"""

					__snake_case					:      List[str]							      =     self.prepare_config_and_inputs()
					__snake_case      ,     __snake_case      ,     __snake_case					:      Dict							      =     config_and_inputs
					__snake_case					:      Optional[Any]							      =     {"""pixel_values""": pixel_values}
					return config, inputs_dict





@require_torch
class  _A (					__lowercase						,					__lowercase						,					unittest.TestCase		):
		lowercase__:   Union[str, Any]				     =   (
		    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
		)
		lowercase__:   Any				     =   (
		    {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
		    if is_torch_available()
		    else {}
		)

		lowercase__:   int				     =   False
		lowercase__:   Optional[Any]				     =   False
		lowercase__:   int				     =   False
		lowercase__:   List[Any]				     =   False

		def 	lowercase__     (   self					:   Tuple			)					->    Optional[int]:


					"""simple docstring"""

					__snake_case					:      str							      =     VideoMAEModelTester(self			)
					__snake_case					:      str							      =     ConfigTester(self  ,      config_class=__magic_name__  ,      has_text_modality=__magic_name__  ,      hidden_size=37			)

		def 	lowercase__     (   self					:   int  ,      __magic_name__					:   Union[str, Any]  ,      __magic_name__					:   Union[str, Any]  ,      __magic_name__					:   str=False			)					->    List[str]:


					"""simple docstring"""

					__snake_case					:      Dict							      =     copy.deepcopy(__magic_name__			)

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

					if return_labels:
								if model_class in [
								    *get_values(__magic_name__			),
								]:
											__snake_case					:      int							      =     torch.zeros(
											    self.model_tester.batch_size  ,      dtype=torch.long  ,      device=__magic_name__			)

					return inputs_dict

		def 	lowercase__     (   self					:   str			)					->    Any:


					"""simple docstring"""

					self.config_tester.run_common_tests()

		@unittest.skip(reason="""VideoMAE does not use inputs_embeds"""			)
		def 	lowercase__     (   self					:   Optional[Any]			)					->    List[Any]:


					"""simple docstring"""

					pass

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


					"""simple docstring"""

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

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

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


					"""simple docstring"""

					__snake_case      ,     __snake_case					:      Union[str, Any]							      =     self.model_tester.prepare_config_and_inputs_for_common()

					for model_class in self.all_model_classes:
								__snake_case					:      Any							      =     model_class(__magic_name__			)
								__snake_case					:      Dict							      =     inspect.signature(model.forward			)
								# signature.parameters is an OrderedDict => so arg_names order is deterministic
								__snake_case					:      Any							      =     [*signature.parameters.keys()]

								__snake_case					:      str							      =     ["""pixel_values"""]
								self.assertListEqual(arg_names[:1]  ,      __magic_name__			)

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


					"""simple docstring"""

					__snake_case					:      Dict							      =     self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*__magic_name__			)

		def 	lowercase__     (   self					:   int			)					->    List[Any]:


					"""simple docstring"""

					__snake_case					:      Tuple							      =     self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_for_pretraining(*__magic_name__			)

		@slow
		def 	lowercase__     (   self					:   Dict			)					->    List[str]:


					"""simple docstring"""

					for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
								__snake_case					:      List[Any]							      =     VideoMAEModel.from_pretrained(__magic_name__			)
								self.assertIsNotNone(__magic_name__			)

		def 	lowercase__     (   self					:   Any			)					->    List[str]:


					"""simple docstring"""

					if not self.has_attentions:
								pass

					else:
								__snake_case      ,     __snake_case					:      List[Any]							      =     self.model_tester.prepare_config_and_inputs_for_common()
								__snake_case					:      Tuple							      =     True

								for model_class in self.all_model_classes:
											__snake_case					:      List[str]							      =     self.model_tester.seq_length - self.model_tester.num_masks
											__snake_case					:      Union[str, Any]							      =     (
											    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
											)

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

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

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

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

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

											__snake_case					:      List[str]							      =     outputs.attentions

											self.assertEqual(len(__magic_name__			)  ,      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					:   Optional[int]			)					->    Optional[Any]:


					"""simple docstring"""

					def check_hidden_states_output(__magic_name__					:   Optional[int]  ,      __magic_name__					:   Optional[int]  ,      __magic_name__					:   Optional[Any]			):
								__snake_case					:      Optional[int]							      =     model_class(__magic_name__			)
								model.to(__magic_name__			)
								model.eval()

								with torch.no_grad():
											__snake_case					:      Dict							      =     model(**self._prepare_for_class(__magic_name__  ,      __magic_name__			)			)

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

								__snake_case					:      Any							      =     self.model_tester.seq_length - self.model_tester.num_masks
								__snake_case					:      List[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]  ,      )

					__snake_case      ,     __snake_case					:      str							      =     self.model_tester.prepare_config_and_inputs_for_common()

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

								# check that output_hidden_states also work using config
								del inputs_dict["output_hidden_states"]
								__snake_case					:      Optional[int]							      =     True

								check_hidden_states_output(__magic_name__  ,      __magic_name__  ,      __magic_name__			)

		@unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""			)
		def 	lowercase__     (   self					:   Any			)					->    str:


					"""simple docstring"""

					pass







def 					_a						(       ) ->				int:



			"""simple docstring"""


			__snake_case					:      Union[str, Any]							      =     hf_hub_download(
			    repo_id="""hf-internal-testing/spaghetti-video"""    ,	filename="""eating_spaghetti.npy"""    ,	repo_type="""dataset""" )
			__snake_case					:      Union[str, Any]							      =     np.load(_lowerCamelCase )
			return list(_lowerCamelCase )




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

		@cached_property
		def 	lowercase__     (   self					:   List[Any]			)					->    Optional[int]:


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


					"""simple docstring"""

					__snake_case					:      Dict							      =     VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""			).to(
					    __magic_name__			)

					__snake_case					:      Optional[Any]							      =     self.default_image_processor
					__snake_case					:      List[Any]							      =     prepare_video()
					__snake_case					:      Union[str, Any]							      =     image_processor(__magic_name__  ,      return_tensors="""pt"""			).to(__magic_name__			)

					# forward pass
					with torch.no_grad():
								__snake_case					:      int							      =     model(**__magic_name__			)

					# verify the logits
					__snake_case					:      Optional[int]							      =     torch.Size((1, 4_00)			)
					self.assertEqual(outputs.logits.shape  ,      __magic_name__			)

					__snake_case					:      str							      =     torch.tensor([0.3669, -0.0688, -0.2421]			).to(__magic_name__			)

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

		@slow
		def 	lowercase__     (   self					:   List[Any]			)					->    List[str]:


					"""simple docstring"""

					__snake_case					:      Union[str, Any]							      =     VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""			).to(__magic_name__			)

					__snake_case					:      List[Any]							      =     self.default_image_processor
					__snake_case					:      List[str]							      =     prepare_video()
					__snake_case					:      Union[str, Any]							      =     image_processor(__magic_name__  ,      return_tensors="""pt"""			).to(__magic_name__			)

					# add boolean mask, indicating which patches to mask
					__snake_case					:      str							      =     hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""  ,      filename="""bool_masked_pos.pt"""			)
					__snake_case					:      Tuple							      =     torch.load(__magic_name__			)

					# forward pass
					with torch.no_grad():
								__snake_case					:      str							      =     model(**__magic_name__			)

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

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

					# verify the loss (`config.norm_pix_loss` = `False`)
					__snake_case					:      Union[str, Any]							      =     VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""  ,      norm_pix_loss=__magic_name__			).to(
					    __magic_name__			)

					with torch.no_grad():
								__snake_case					:      Optional[Any]							      =     model(**__magic_name__			)

					__snake_case					:      int							      =     torch.tensor(torch.tensor([0.6469]			)  ,      device=__magic_name__			)
					self.assertTrue(torch.allclose(outputs.loss  ,      __magic_name__  ,      atol=1E-4			)			)



from typing import List

from .keymap import KEYMAP, get_character
def 					_snake_case	(  __snake_case						):

					def decorator(__snake_case						):
										_UpperCamelCase			     =     getattr(__snake_case  ,     '''handle_key'''  ,     []						)
										handle += [key]
										setattr(__snake_case  ,     '''handle_key'''  ,     __snake_case						)
										return func

					return decorator
def 					_snake_case	(  *__snake_case						):

					def decorator(__snake_case						):
										_UpperCamelCase			     =     getattr(__snake_case  ,     '''handle_key'''  ,     []						)
										handle += keys
										setattr(__snake_case  ,     '''handle_key'''  ,     __snake_case						)
										return func

					return decorator
class     lowerCAmelCase_     (       __lowercase ):







							def __new__(   cls      :     Optional[Any] ,			_A      :     Optional[Any] ,			_A      :     Optional[int] ,			_A      :     Union[str, Any]		):
												_UpperCamelCase			     =     super().__new__(cls ,			_A ,			_A ,			_A		)
												if not hasattr(_A ,			'''key_handler'''		):
																	setattr(_A ,			'''key_handler''' ,			{}		)
												setattr(_A ,			'''handle_input''' ,			KeyHandler.handle_input		)

												for value in attrs.values():
																	_UpperCamelCase			     =     getattr(_A ,			'''handle_key''' ,			[]		)
																	for key in handled_keys:
																						_UpperCamelCase			     =     value
												return new_cls







							@staticmethod
							def  UpperCamelCase_						(   cls      :     str		):
												_UpperCamelCase			     =     get_character()
												if char != KEYMAP["undefined"]:
																	_UpperCamelCase			     =     ord(_A		)
												_UpperCamelCase			     =     cls.key_handler.get(_A		)
												if handler:
																	_UpperCamelCase			     =     char
																	return handler(cls		)
												else:
																	return None




def 					_snake_case	(  cls						):
					return KeyHandler(cls.__name__  ,     cls.__bases__  ,     cls.__dict__.copy()						)
