


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  snake_case_     :
					def __init__(   self     :List[Any]    ,__snake_case     :Optional[int]    ,__snake_case     :List[str]=13    ,__snake_case     :Optional[int]=10    ,__snake_case     :List[Any]=3    ,__snake_case     :List[Any]=2    ,__snake_case     :Optional[int]=2    ,__snake_case     :int=2    ,__snake_case     :str=True    ,__snake_case     :Any=True    ,__snake_case     :int=32    ,__snake_case     :List[str]=5    ,__snake_case     :Optional[int]=4    ,__snake_case     :Optional[Any]=37    ,__snake_case     :Optional[Any]="gelu"    ,__snake_case     :int=0.1    ,__snake_case     :Tuple=0.1    ,__snake_case     :Tuple=10    ,__snake_case     :Dict=0.02    ,__snake_case     :List[str]=0.9    ,__snake_case     :Optional[int]=None    ,)      ->					str:
												a__    =  parent
												a__    =  batch_size
												a__    =  image_size
												a__    =  num_channels
												a__    =  patch_size
												a__    =  tubelet_size
												a__    =  num_frames
												a__    =  is_training
												a__    =  use_labels
												a__    =  hidden_size
												a__    =  num_hidden_layers
												a__    =  num_attention_heads
												a__    =  intermediate_size
												a__    =  hidden_act
												a__    =  hidden_dropout_prob
												a__    =  attention_probs_dropout_prob
												a__    =  type_sequence_label_size
												a__    =  initializer_range
												a__    =  mask_ratio
												a__    =  scope

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

												# use this variable to define bool_masked_pos
												a__    =  int(mask_ratio * self.seq_length   )
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					Optional[int]:
												a__    =  floats_tensor(
												    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]   )

												a__    =  None
												if self.use_labels:
																			a__    =  ids_tensor([self.batch_size]    ,self.type_sequence_label_size   )

												a__    =  self.get_config()

												return config, pixel_values, labels
					def 				lowerCamelCase__(   self     :Any   )      ->					Optional[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=__snake_case    ,initializer_range=self.initializer_range    ,)
					def 				lowerCamelCase__(   self     :Tuple    ,__snake_case     :List[Any]    ,__snake_case     :Any    ,__snake_case     :Union[str, Any]   )      ->					Any:
												a__    =  VideoMAEModel(config=__snake_case   )
												model.to(__snake_case   )
												model.eval()
												a__    =  model(__snake_case   )
												self.parent.assertEqual(result.last_hidden_state.shape    ,(self.batch_size, self.seq_length, self.hidden_size)   )
					def 				lowerCamelCase__(   self     :str    ,__snake_case     :List[str]    ,__snake_case     :Optional[int]    ,__snake_case     :Dict   )      ->					Union[str, Any]:
												a__    =  VideoMAEForPreTraining(__snake_case   )
												model.to(__snake_case   )
												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__    =  torch.ones((self.num_masks,)   )
												a__    =  torch.cat([mask, torch.zeros(self.seq_length - mask.size(0   )   )]   )
												a__    =  mask.expand(self.batch_size    ,-1   ).bool()

												a__    =  model(__snake_case    ,__snake_case   )
												# model only returns predictions for masked patches
												a__    =  mask.sum().item()
												a__    =  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     :Any   )      ->					Optional[int]:
												a__    =  self.prepare_config_and_inputs()
												a__   ,      a__   ,      a__    =  config_and_inputs
												a__    =  {'pixel_values': pixel_values}
												return config, inputs_dict







@require_torch
class  snake_case_     (lowerCamelCase_			,		lowerCamelCase_			,		unittest.TestCase   ):
					UpperCAmelCase__      :					int         =    (
					    (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
					)
					UpperCAmelCase__      :					Dict         =    (
					    {'''feature-extraction''': VideoMAEModel, '''video-classification''': VideoMAEForVideoClassification}
					    if is_torch_available()
					    else {}
					)

					UpperCAmelCase__      :					Optional[int]         =    False
					UpperCAmelCase__      :					List[Any]         =    False
					UpperCAmelCase__      :					Union[str, Any]         =    False
					UpperCAmelCase__      :					Any         =    False
					def 				lowerCamelCase__(   self     :Optional[int]   )      ->					Optional[int]:
												a__    =  VideoMAEModelTester(self   )
												a__    =  ConfigTester(self    ,config_class=__snake_case    ,has_text_modality=__snake_case    ,hidden_size=37   )
					def 				lowerCamelCase__(   self     :Any    ,__snake_case     :List[Any]    ,__snake_case     :List[str]    ,__snake_case     :List[Any]=False   )      ->					Optional[int]:
												a__    =  copy.deepcopy(__snake_case   )

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

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

												return inputs_dict
					def 				lowerCamelCase__(   self     :Dict   )      ->					int:
												self.config_tester.run_common_tests()
					@unittest.skip(reason='VideoMAE does not use inputs_embeds'   )
					def 				lowerCamelCase__(   self     :List[Any]   )      ->					Tuple:
												pass
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					int:
												a__   ,      a__    =  self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																			a__    =  model_class(__snake_case   )
																			self.assertIsInstance(model.get_input_embeddings()    ,(nn.Module)   )
																			a__    =  model.get_output_embeddings()
																			self.assertTrue(x is None or isinstance(__snake_case    ,nn.Linear   )   )
					def 				lowerCamelCase__(   self     :Tuple   )      ->					Any:
												a__   ,      a__    =  self.model_tester.prepare_config_and_inputs_for_common()

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

																			a__    =  ['pixel_values']
																			self.assertListEqual(arg_names[:1]    ,__snake_case   )
					def 				lowerCamelCase__(   self     :int   )      ->					List[str]:
												a__    =  self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*__snake_case   )
					def 				lowerCamelCase__(   self     :List[str]   )      ->					Optional[int]:
												a__    =  self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_pretraining(*__snake_case   )
					@slow
					def 				lowerCamelCase__(   self     :str   )      ->					Optional[Any]:
												for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																			a__    =  VideoMAEModel.from_pretrained(__snake_case   )
																			self.assertIsNotNone(__snake_case   )
					def 				lowerCamelCase__(   self     :Tuple   )      ->					Any:
												if not self.has_attentions:
																			pass

												else:
																			a__   ,      a__    =  self.model_tester.prepare_config_and_inputs_for_common()
																			a__    =  True

																			for model_class in self.all_model_classes:
																										a__    =  self.model_tester.seq_length - self.model_tester.num_masks
																										a__    =  (
																										    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
																										)

																										a__    =  True
																										a__    =  False
																										a__    =  True
																										a__    =  model_class(__snake_case   )
																										model.to(__snake_case   )
																										model.eval()
																										with torch.no_grad():
																																	a__    =  model(**self._prepare_for_class(__snake_case    ,__snake_case   )   )
																										a__    =  outputs.attentions
																										self.assertEqual(len(__snake_case   )    ,self.model_tester.num_hidden_layers   )

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

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

																										# Check attention is always last and order is fine
																										a__    =  True
																										a__    =  True
																										a__    =  model_class(__snake_case   )
																										model.to(__snake_case   )
																										model.eval()
																										with torch.no_grad():
																																	a__    =  model(**self._prepare_for_class(__snake_case    ,__snake_case   )   )

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

																										a__    =  outputs.attentions

																										self.assertEqual(len(__snake_case   )    ,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]   )      ->					Optional[Any]:
												def check_hidden_states_output(__snake_case     :Dict    ,__snake_case     :str    ,__snake_case     :Union[str, Any]   ):
																			a__    =  model_class(__snake_case   )
																			model.to(__snake_case   )
																			model.eval()

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

																			a__    =  outputs.hidden_states
																			a__    =  self.model_tester.num_hidden_layers + 1
																			self.assertEqual(len(__snake_case   )    ,__snake_case   )

																			a__    =  self.model_tester.seq_length - self.model_tester.num_masks
																			a__    =  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__    =  self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																			a__    =  True
																			check_hidden_states_output(__snake_case    ,__snake_case    ,__snake_case   )

																			# check that output_hidden_states also work using config
																			del inputs_dict["output_hidden_states"]
																			a__    =  True

																			check_hidden_states_output(__snake_case    ,__snake_case    ,__snake_case   )



					@unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'   )
					def 				lowerCamelCase__(   self     :Dict   )      ->					Optional[int]:
												pass






def 							__lowercase     (				):
							a__    =  hf_hub_download(
							    repo_id='hf-internal-testing/spaghetti-video'      ,					filename='eating_spaghetti.npy'      ,					repo_type='dataset'     )
							a__    =  np.load(__lowerCAmelCase     )
							return list(__lowerCAmelCase     )







@require_torch
@require_vision
class  snake_case_     (unittest.TestCase   ):
					@cached_property
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					Union[str, Any]:
												# 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 				lowerCamelCase__(   self     :List[str]   )      ->					List[Any]:
												a__    =  VideoMAEForVideoClassification.from_pretrained('MCG-NJU/videomae-base-finetuned-kinetics'   ).to(
												    __snake_case   )

												a__    =  self.default_image_processor
												a__    =  prepare_video()
												a__    =  image_processor(__snake_case    ,return_tensors='pt'   ).to(__snake_case   )

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

												# verify the logits
												a__    =  torch.Size((1, 4_00)   )
												self.assertEqual(outputs.logits.shape    ,__snake_case   )

												a__    =  torch.tensor([0.36_69, -0.06_88, -0.24_21]   ).to(__snake_case   )

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



					@slow
					def 				lowerCamelCase__(   self     :Union[str, Any]   )      ->					Optional[int]:
												a__    =  VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'   ).to(__snake_case   )

												a__    =  self.default_image_processor
												a__    =  prepare_video()
												a__    =  image_processor(__snake_case    ,return_tensors='pt'   ).to(__snake_case   )

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

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

												# verify the logits
												a__    =  torch.Size([1, 14_08, 15_36]   )
												a__    =  torch.tensor(
												    [[0.79_94, 0.96_12, 0.85_08], [0.74_01, 0.89_58, 0.83_02], [0.58_62, 0.74_68, 0.73_25]]    ,device=__snake_case   )
												self.assertEqual(outputs.logits.shape    ,__snake_case   )
												self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]    ,__snake_case    ,atol=1E-4   )   )

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

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

												with torch.no_grad():
																			a__    =  model(**__snake_case   )

												a__    =  torch.tensor(torch.tensor([0.64_69]   )    ,device=__snake_case   )
												self.assertTrue(torch.allclose(outputs.loss    ,__snake_case    ,atol=1E-4   )   )








from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_torch_available,
    is_vision_available,
)


snake_case							: Any      =   {
    '''configuration_mobilevit''': ['''MOBILEVIT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''MobileViTConfig''', '''MobileViTOnnxConfig'''],
}

try:
						if not is_vision_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						snake_case							: Union[str, Any]      =   ['''MobileViTFeatureExtractor''']
						snake_case							: int      =   ['''MobileViTImageProcessor''']

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						snake_case							: Dict      =   [
						    '''MOBILEVIT_PRETRAINED_MODEL_ARCHIVE_LIST''',
						    '''MobileViTForImageClassification''',
						    '''MobileViTForSemanticSegmentation''',
						    '''MobileViTModel''',
						    '''MobileViTPreTrainedModel''',
						]

try:
						if not is_tf_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						snake_case							: Tuple      =   [
						    '''TF_MOBILEVIT_PRETRAINED_MODEL_ARCHIVE_LIST''',
						    '''TFMobileViTForImageClassification''',
						    '''TFMobileViTForSemanticSegmentation''',
						    '''TFMobileViTModel''',
						    '''TFMobileViTPreTrainedModel''',
						]

if TYPE_CHECKING:
						from .configuration_mobilevit import MOBILEVIT_PRETRAINED_CONFIG_ARCHIVE_MAP, MobileViTConfig, MobileViTOnnxConfig

						try:
												if not is_vision_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .feature_extraction_mobilevit import MobileViTFeatureExtractor
												from .image_processing_mobilevit import MobileViTImageProcessor

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_mobilevit import (
												    MOBILEVIT_PRETRAINED_MODEL_ARCHIVE_LIST,
												    MobileViTForImageClassification,
												    MobileViTForSemanticSegmentation,
												    MobileViTModel,
												    MobileViTPreTrainedModel,
												)

						try:
												if not is_tf_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_tf_mobilevit import (
												    TF_MOBILEVIT_PRETRAINED_MODEL_ARCHIVE_LIST,
												    TFMobileViTForImageClassification,
												    TFMobileViTForSemanticSegmentation,
												    TFMobileViTModel,
												    TFMobileViTPreTrainedModel,
												)


else:
						import sys

						snake_case							: Optional[int]      =   _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)




