





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					:

   '''simple docstring'''







   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'''







    lowerCamelCase__              =	parent
    lowerCamelCase__              =	batch_size
    lowerCamelCase__              =	image_size
    lowerCamelCase__              =	num_channels
    lowerCamelCase__              =	patch_size
    lowerCamelCase__              =	tubelet_size
    lowerCamelCase__              =	num_frames
    lowerCamelCase__              =	is_training
    lowerCamelCase__              =	use_labels
    lowerCamelCase__              =	hidden_size
    lowerCamelCase__              =	num_hidden_layers
    lowerCamelCase__              =	num_attention_heads
    lowerCamelCase__              =	intermediate_size
    lowerCamelCase__              =	hidden_act
    lowerCamelCase__              =	hidden_dropout_prob
    lowerCamelCase__              =	attention_probs_dropout_prob
    lowerCamelCase__              =	type_sequence_label_size
    lowerCamelCase__              =	initializer_range
    lowerCamelCase__              =	mask_ratio
    lowerCamelCase__              =	scope

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

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





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__              =	floats_tensor(
        [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]							)

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

    lowerCamelCase__              =	self.get_config()

    return config, pixel_values, labels





   def 	__lowerCamelCase       (					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 	__lowerCamelCase       (					self ,						__lowerCAmelCase ,						__lowerCAmelCase ,						__lowerCAmelCase							):



    '''simple docstring'''







    lowerCamelCase__              =	VideoMAEModel(config=__lowerCAmelCase							)
    model.to(__lowerCAmelCase							)
    model.eval()
    lowerCamelCase__              =	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							):



    '''simple docstring'''







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

    lowerCamelCase__              =	model(__lowerCAmelCase ,						__lowerCAmelCase							)
    # model only returns predictions for masked patches
    lowerCamelCase__              =	mask.sum().item()
    lowerCamelCase__              =	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							):



    '''simple docstring'''







    lowerCamelCase__              =	self.prepare_config_and_inputs()
    lowerCamelCase__       ,			lowerCamelCase__       ,			lowerCamelCase__              =	config_and_inputs
    lowerCamelCase__              =	{'''pixel_values''': pixel_values}
    return config, inputs_dict






@require_torch
class       __A					(					lowerCAmelCase    ,       lowerCAmelCase    ,       unittest.TestCase				):

   '''simple docstring'''







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

   lowerCAmelCase_							  =		False
   lowerCAmelCase_							  =		False
   lowerCAmelCase_							  =		False
   lowerCAmelCase_							  =		False





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__              =	VideoMAEModelTester(self							)
    lowerCamelCase__              =	ConfigTester(self ,						config_class=__lowerCAmelCase ,						has_text_modality=__lowerCAmelCase ,						hidden_size=3_7							)





   def 	__lowerCamelCase       (					self ,						__lowerCAmelCase ,						__lowerCAmelCase ,						__lowerCAmelCase=False							):



    '''simple docstring'''







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

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

    return inputs_dict





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    self.config_tester.run_common_tests()





   @unittest.skip(reason='''VideoMAE does not use inputs_embeds'''							)
   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    pass





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__       ,			lowerCamelCase__              =	self.model_tester.prepare_config_and_inputs_for_common()

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





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__       ,			lowerCamelCase__              =	self.model_tester.prepare_config_and_inputs_for_common()

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

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





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__              =	self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*__lowerCAmelCase							)





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    lowerCamelCase__              =	self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_pretraining(*__lowerCAmelCase							)





   @slow
   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
     lowerCamelCase__              =	VideoMAEModel.from_pretrained(__lowerCAmelCase							)
     self.assertIsNotNone(__lowerCAmelCase							)





   def 	__lowerCamelCase       (					self							):



    '''simple docstring'''







    if not self.has_attentions:
     pass

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

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

      lowerCamelCase__              =	True
      lowerCamelCase__              =	False
      lowerCamelCase__              =	True
      lowerCamelCase__              =	model_class(__lowerCAmelCase							)
      model.to(__lowerCAmelCase							)
      model.eval()
      with torch.no_grad():
       lowerCamelCase__              =	model(**self._prepare_for_class(__lowerCAmelCase ,						__lowerCAmelCase							)							)
      lowerCamelCase__              =	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__              =	True
      lowerCamelCase__              =	model_class(__lowerCAmelCase							)
      model.to(__lowerCAmelCase							)
      model.eval()
      with torch.no_grad():
       lowerCamelCase__              =	model(**self._prepare_for_class(__lowerCAmelCase ,						__lowerCAmelCase							)							)
      lowerCamelCase__              =	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__              =	len(__lowerCAmelCase							)

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

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

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



    '''simple docstring'''







    def check_hidden_states_output(__lowerCAmelCase ,						__lowerCAmelCase ,						__lowerCAmelCase							):
     lowerCamelCase__              =	model_class(__lowerCAmelCase							)
     model.to(__lowerCAmelCase							)
     model.eval()

     with torch.no_grad():
      lowerCamelCase__              =	model(**self._prepare_for_class(__lowerCAmelCase ,						__lowerCAmelCase							)							)

     lowerCamelCase__              =	outputs.hidden_states
     lowerCamelCase__              =	self.model_tester.num_hidden_layers + 1
     self.assertEqual(len(__lowerCAmelCase							) ,						__lowerCAmelCase							)

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

    for model_class in self.all_model_classes:
     lowerCamelCase__              =	True
     check_hidden_states_output(__lowerCAmelCase ,						__lowerCAmelCase ,						__lowerCAmelCase							)

     # check that output_hidden_states also work using config
     del inputs_dict["output_hidden_states"]
     lowerCamelCase__              =	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							):



    '''simple docstring'''







    pass





def lowerCAmelCase__()       ->   int:







 '''simple docstring'''







 lowerCamelCase__              =	hf_hub_download(
     repo_id='''hf-internal-testing/spaghetti-video''' ,filename='''eating_spaghetti.npy''' ,repo_type='''dataset'''							)
 lowerCamelCase__              =	np.load(__snake_case							)
 return list(__snake_case							)






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

   '''simple docstring'''







   @cached_property
   def 	__lowerCamelCase       (					self							):



    '''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							):



    '''simple docstring'''







    lowerCamelCase__              =	VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''							).to(
        __lowerCAmelCase							)

    lowerCamelCase__              =	self.default_image_processor
    lowerCamelCase__              =	prepare_video()
    lowerCamelCase__              =	image_processor(__lowerCAmelCase ,						return_tensors='''pt'''							).to(__lowerCAmelCase							)

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

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

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



    '''simple docstring'''







    lowerCamelCase__              =	VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''							).to(__lowerCAmelCase							)

    lowerCamelCase__              =	self.default_image_processor
    lowerCamelCase__              =	prepare_video()
    lowerCamelCase__              =	image_processor(__lowerCAmelCase ,						return_tensors='''pt'''							).to(__lowerCAmelCase							)

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

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

    # verify the logits
    lowerCamelCase__              =	torch.Size([1, 1_4_0_8, 1_5_3_6]							)
    lowerCamelCase__              =	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__              =	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__              =	VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short''' ,						norm_pix_loss=__lowerCAmelCase							).to(
        __lowerCAmelCase							)

    with torch.no_grad():
     lowerCamelCase__              =	model(**__lowerCAmelCase							)

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










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






class       __A					(					lowerCAmelCase				):

			'''simple docstring'''







			lowerCAmelCase_							  =		"""ClapFeatureExtractor"""
			lowerCAmelCase_							  =		("""RobertaTokenizer""", """RobertaTokenizerFast""")





			def __init__(					self ,						__lowerCAmelCase ,						__lowerCAmelCase							):



				'''simple docstring'''







				super().__init__(__lowerCAmelCase ,						__lowerCAmelCase							)





			def __call__(					self ,						__lowerCAmelCase=None ,						__lowerCAmelCase=None ,						__lowerCAmelCase=None ,						**__lowerCAmelCase							):



				'''simple docstring'''







				lowerCamelCase__              =	kwargs.pop('''sampling_rate''' ,						__lowerCAmelCase							)

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

				if text is not None:
					lowerCamelCase__              =	self.tokenizer(__lowerCAmelCase ,						return_tensors=__lowerCAmelCase ,						**__lowerCAmelCase							)

				if audios is not None:
					lowerCamelCase__              =	self.feature_extractor(
					    __lowerCAmelCase ,						sampling_rate=__lowerCAmelCase ,						return_tensors=__lowerCAmelCase ,						**__lowerCAmelCase							)

				if text is not None and audios is not None:
					lowerCamelCase__              =	audio_features.input_features
					return encoding
				elif text is not None:
					return encoding
				else:
					return BatchEncoding(data=dict(**__lowerCAmelCase							) ,						tensor_type=__lowerCAmelCase							)





			def 	__lowerCamelCase       (					self ,						*__lowerCAmelCase ,						**__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.batch_decode(*__lowerCAmelCase ,						**__lowerCAmelCase							)





			def 	__lowerCamelCase       (					self ,						*__lowerCAmelCase ,						**__lowerCAmelCase							):



				'''simple docstring'''







				return self.tokenizer.decode(*__lowerCAmelCase ,						**__lowerCAmelCase							)





			@property
			def 	__lowerCamelCase       (					self							):



				'''simple docstring'''







				lowerCamelCase__              =	self.tokenizer.model_input_names
				lowerCamelCase__              =	self.feature_extractor.model_input_names
				return list(dict.fromkeys(tokenizer_input_names + feature_extractor_input_names							)							)



