






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







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

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

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







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:Dict   =   floats_tensor(
								    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

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

								__magic_name__		:Dict   =   self.get_config()

								return config, pixel_values, labels







				def       A						(    self  ):
								"""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       A						(    self							,       __lowerCAmelCase							,       __lowerCAmelCase							,       __lowerCAmelCase  ):
								"""simple docstring"""
								__magic_name__		:Union[str, Any]   =   VideoMAEModel(config=__lowerCAmelCase  )
								model.to(__lowerCAmelCase  )
								model.eval()
								__magic_name__		:Tuple   =   model(__lowerCAmelCase  )
								self.parent.assertEqual(result.last_hidden_state.shape							,       (self.batch_size, self.seq_length, self.hidden_size)  )







				def       A						(    self							,       __lowerCAmelCase							,       __lowerCAmelCase							,       __lowerCAmelCase  ):
								"""simple docstring"""
								__magic_name__		:Any   =   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
								__magic_name__		:Any   =   torch.ones((self.num_masks,)  )
								__magic_name__		:int   =   torch.cat([mask, torch.zeros(self.seq_length - mask.size(0  )  )]  )
								__magic_name__		:List[str]   =   mask.expand(self.batch_size							,       -1  ).bool()

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







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:Union[str, Any]   =   self.prepare_config_and_inputs()
								__magic_name__  ,					__magic_name__  ,					__magic_name__		:List[Any]   =   config_and_inputs
								__magic_name__		:Optional[int]   =   {'''pixel_values''': pixel_values}
								return config, inputs_dict







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

				a__       							=	False
				a__       							=	False
				a__       							=	False
				a__       							=	False







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:str   =   VideoMAEModelTester(self  )
								__magic_name__		:List[Any]   =   ConfigTester(self							,       config_class=__lowerCAmelCase							,       has_text_modality=__lowerCAmelCase							,       hidden_size=3_7  )







				def       A						(    self							,       __lowerCAmelCase							,       __lowerCAmelCase							,       __lowerCAmelCase=False  ):
								"""simple docstring"""
								__magic_name__		:Optional[Any]   =   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
												__magic_name__		:Optional[int]   =   torch.ones((self.model_tester.num_masks,)  )
												__magic_name__		:str   =   torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0  )  )]  )
												__magic_name__		:List[str]   =   mask.expand(self.model_tester.batch_size							,       -1  ).bool()
												__magic_name__		:List[str]   =   bool_masked_pos.to(__lowerCAmelCase  )

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

								return inputs_dict







				def       A						(    self  ):
								"""simple docstring"""
								self.config_tester.run_common_tests()







				@unittest.skip(reason='''VideoMAE does not use inputs_embeds'''  )
				def       A						(    self  ):
								"""simple docstring"""
								pass







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__  ,					__magic_name__		:str   =   self.model_tester.prepare_config_and_inputs_for_common()

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







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__  ,					__magic_name__		:Any   =   self.model_tester.prepare_config_and_inputs_for_common()

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

												__magic_name__		:Any   =   ['''pixel_values''']
												self.assertListEqual(arg_names[:1]							,       __lowerCAmelCase  )







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:Union[str, Any]   =   self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*__lowerCAmelCase  )







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:Any   =   self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_pretraining(*__lowerCAmelCase  )







				@slow
				def       A						(    self  ):
								"""simple docstring"""
								for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												__magic_name__		:List[Any]   =   VideoMAEModel.from_pretrained(__lowerCAmelCase  )
												self.assertIsNotNone(__lowerCAmelCase  )







				def       A						(    self  ):
								"""simple docstring"""
								if not self.has_attentions:
												pass

								else:
												__magic_name__  ,					__magic_name__		:List[str]   =   self.model_tester.prepare_config_and_inputs_for_common()
												__magic_name__		:Tuple   =   True

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

																__magic_name__		:Optional[int]   =   True
																__magic_name__		:Any   =   False
																__magic_name__		:Union[str, Any]   =   True
																__magic_name__		:Tuple   =   model_class(__lowerCAmelCase  )
																model.to(__lowerCAmelCase  )
																model.eval()
																with torch.no_grad():
																				__magic_name__		:str   =   model(**self._prepare_for_class(__lowerCAmelCase							,       __lowerCAmelCase  )  )
																__magic_name__		:List[Any]   =   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"]
																__magic_name__		:Tuple   =   True
																__magic_name__		:Any   =   model_class(__lowerCAmelCase  )
																model.to(__lowerCAmelCase  )
																model.eval()
																with torch.no_grad():
																				__magic_name__		:Any   =   model(**self._prepare_for_class(__lowerCAmelCase							,       __lowerCAmelCase  )  )
																__magic_name__		:Optional[Any]   =   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]							,       )
																__magic_name__		:Any   =   len(__lowerCAmelCase  )

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

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

																__magic_name__		: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       A						(    self  ):
								"""simple docstring"""
								def check_hidden_states_output(__lowerCAmelCase							,       __lowerCAmelCase							,       __lowerCAmelCase  ):
												__magic_name__		:Tuple   =   model_class(__lowerCAmelCase  )
												model.to(__lowerCAmelCase  )
												model.eval()

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

												__magic_name__		:Optional[int]   =   outputs.hidden_states
												__magic_name__		:Optional[int]   =   self.model_tester.num_hidden_layers + 1
												self.assertEqual(len(__lowerCAmelCase  )							,       __lowerCAmelCase  )

												__magic_name__		:Tuple   =   self.model_tester.seq_length - self.model_tester.num_masks
												__magic_name__		: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]							,       )

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

								for model_class in self.all_model_classes:
												__magic_name__		: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"]
												__magic_name__		:Any   =   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       A						(    self  ):
								"""simple docstring"""
								pass







def   __lowercase       (			):







				"""simple docstring"""






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

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







				@cached_property
				def       A						(    self  ):
								"""simple docstring"""
								# 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       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:Tuple   =   VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''  ).to(
								    __lowerCAmelCase  )

								__magic_name__		:List[str]   =   self.default_image_processor
								__magic_name__		:Optional[int]   =   prepare_video()
								__magic_name__		:int   =   image_processor(__lowerCAmelCase							,       return_tensors='''pt'''  ).to(__lowerCAmelCase  )

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

								# verify the logits
								__magic_name__		:Dict   =   torch.Size((1, 4_0_0)  )
								self.assertEqual(outputs.logits.shape							,       __lowerCAmelCase  )

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

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







				@slow
				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:int   =   VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''  ).to(__lowerCAmelCase  )

								__magic_name__		:Tuple   =   self.default_image_processor
								__magic_name__		:List[str]   =   prepare_video()
								__magic_name__		:Any   =   image_processor(__lowerCAmelCase							,       return_tensors='''pt'''  ).to(__lowerCAmelCase  )

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

								# forward pass
								with torch.no_grad():
												__magic_name__		:Tuple   =   model(**__lowerCAmelCase  )

								# verify the logits
								__magic_name__		:List[str]   =   torch.Size([1, 1_4_0_8, 1_5_3_6]  )
								__magic_name__		:str   =   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`)
								__magic_name__		:Optional[Any]   =   torch.tensor([0.5142]							,       device=__lowerCAmelCase  )
								self.assertTrue(torch.allclose(outputs.loss							,       __lowerCAmelCase							,       atol=1E-4  )  )

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

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

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














from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


SCREAMING_SNAKE_CASE__   :      Dict				=		{
    """configuration_canine""": ["""CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP""", """CanineConfig"""],
    """tokenization_canine""": ["""CanineTokenizer"""],
}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					SCREAMING_SNAKE_CASE__   :      str				=		[
					    """CANINE_PRETRAINED_MODEL_ARCHIVE_LIST""",
					    """CanineForMultipleChoice""",
					    """CanineForQuestionAnswering""",
					    """CanineForSequenceClassification""",
					    """CanineForTokenClassification""",
					    """CanineLayer""",
					    """CanineModel""",
					    """CaninePreTrainedModel""",
					    """load_tf_weights_in_canine""",
					]


if TYPE_CHECKING:
					from .configuration_canine import CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP, CanineConfig
					from .tokenization_canine import CanineTokenizer

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_canine import (
										    CANINE_PRETRAINED_MODEL_ARCHIVE_LIST,
										    CanineForMultipleChoice,
										    CanineForQuestionAnswering,
										    CanineForSequenceClassification,
										    CanineForTokenClassification,
										    CanineLayer,
										    CanineModel,
										    CaninePreTrainedModel,
										    load_tf_weights_in_canine,
										)


else:
					import sys

					SCREAMING_SNAKE_CASE__   :      Union[str, Any]				=		_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)






