


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=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.0_2,   lowerCamelCase=0.9,   lowerCamelCase=None,   ) ->     Any:




           """simple docstring"""

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

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

           # use this variable to define bool_masked_pos
           _lowercase :							Optional[int]   	=	int(mask_ratio * self.seq_length)
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

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

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

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

           return config, pixel_values, labels
       def     UpperCamelCase  (      self) ->     List[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     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase) ->     Any:




           """simple docstring"""

           _lowercase :							List[str]   	=	VideoMAEModel(config=lowerCamelCase)
           model.to(lowerCamelCase)
           model.eval()
           _lowercase :							Optional[Any]   	=	model(lowerCamelCase)
           self.parent.assertEqual(result.last_hidden_state.shape,   (self.batch_size, self.seq_length, self.hidden_size))
       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase) ->     Optional[Any]:




           """simple docstring"""

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

           _lowercase :							str   	=	model(lowerCamelCase,   lowerCamelCase)
           # model only returns predictions for masked patches
           _lowercase :							Optional[int]   	=	mask.sum().item()
           _lowercase :							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     UpperCamelCase  (      self) ->     Tuple:




           """simple docstring"""

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






@require_torch
class _lowerCamelCase(		_a,				_a,				unittest.TestCase     ):
       lowercase_				:			Optional[int]									=    (
           (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
       )
       lowercase_				:			List[str]									=    (
           {"""feature-extraction""": VideoMAEModel, """video-classification""": VideoMAEForVideoClassification}
           if is_torch_available()
           else {}
       )

       lowercase_				:			str									=    False
       lowercase_				:			Tuple									=    False
       lowercase_				:			Tuple									=    False
       lowercase_				:			str									=    False
       def     UpperCamelCase  (      self) ->     Any:




           """simple docstring"""

           _lowercase :							Union[str, Any]   	=	VideoMAEModelTester(self)
           _lowercase :							List[str]   	=	ConfigTester(self,   config_class=lowerCamelCase,   has_text_modality=lowerCamelCase,   hidden_size=37)
       def     UpperCamelCase  (      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase=False) ->     Optional[int]:




           """simple docstring"""

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

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

           return inputs_dict
       def     UpperCamelCase  (      self) ->     str:




           """simple docstring"""

           self.config_tester.run_common_tests()
       @unittest.skip(reason='VideoMAE does not use inputs_embeds')
       def     UpperCamelCase  (      self) ->     Tuple:




           """simple docstring"""

           pass
       def     UpperCamelCase  (      self) ->     Union[str, Any]:




           """simple docstring"""

           _lowercase :							List[str]   	=	self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               _lowercase :							List[Any]   	=	model_class(lowerCamelCase)
               self.assertIsInstance(model.get_input_embeddings(),   (nn.Module))
               _lowercase :							List[str]   	=	model.get_output_embeddings()
               self.assertTrue(x is None or isinstance(lowerCamelCase,   nn.Linear))
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           _lowercase :							Optional[Any]   	=	self.model_tester.prepare_config_and_inputs_for_common()

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

               _lowercase :							Tuple   	=	['pixel_values']
               self.assertListEqual(arg_names[:1],   lowerCamelCase)
       def     UpperCamelCase  (      self) ->     List[str]:




           """simple docstring"""

           _lowercase :							Dict   	=	self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*lowerCamelCase)
       def     UpperCamelCase  (      self) ->     Tuple:




           """simple docstring"""

           _lowercase :							List[Any]   	=	self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_for_pretraining(*lowerCamelCase)
       @slow
       def     UpperCamelCase  (      self) ->     Any:




           """simple docstring"""

           for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               _lowercase :							Any   	=	VideoMAEModel.from_pretrained(lowerCamelCase)
               self.assertIsNotNone(lowerCamelCase)
       def     UpperCamelCase  (      self) ->     Optional[int]:




           """simple docstring"""

           if not self.has_attentions:
               pass

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

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

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

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

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

                   _lowercase :							Union[str, Any]   	=	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     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           def check_hidden_states_output(lowerCamelCase,   lowerCamelCase,   lowerCamelCase):
               _lowercase :							Tuple   	=	model_class(lowerCamelCase)
               model.to(lowerCamelCase)
               model.eval()

               with torch.no_grad():
                   _lowercase :							Dict   	=	model(**self._prepare_for_class(lowerCamelCase,   lowerCamelCase))

               _lowercase :							Dict   	=	outputs.hidden_states
               _lowercase :							Tuple   	=	self.model_tester.num_hidden_layers + 1
               self.assertEqual(len(lowerCamelCase),   lowerCamelCase)

               _lowercase :							Dict   	=	self.model_tester.seq_length - self.model_tester.num_masks
               _lowercase :							Optional[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],   )

           _lowercase :							List[Any]   	=	self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               _lowercase :							List[Any]   	=	True
               check_hidden_states_output(lowerCamelCase,   lowerCamelCase,   lowerCamelCase)

               # check that output_hidden_states also work using config
               del inputs_dict["output_hidden_states"]
               _lowercase :							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     UpperCamelCase  (      self) ->     List[Any]:




           """simple docstring"""

           pass



def        UpperCamelCase_(   )					->		List[Any]:
    _lowercase :							Dict   	=	hf_hub_download(
        repo_id='hf-internal-testing/spaghetti-video'							,				filename='eating_spaghetti.npy'							,				repo_type='dataset'				)
    _lowercase :							Dict   	=	np.load(lowerCamelCase_				)
    return list(lowerCamelCase_				)






@require_torch
@require_vision
class _lowerCamelCase(		unittest.TestCase     ):
       @cached_property
       def     UpperCamelCase  (      self) ->     Union[str, 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     UpperCamelCase  (      self) ->     Any:




           """simple docstring"""

           _lowercase :							Tuple   	=	VideoMAEForVideoClassification.from_pretrained('MCG-NJU/videomae-base-finetuned-kinetics').to(
               lowerCamelCase)

           _lowercase :							int   	=	self.default_image_processor
           _lowercase :							str   	=	prepare_video()
           _lowercase :							Tuple   	=	image_processor(lowerCamelCase,   return_tensors='pt').to(lowerCamelCase)

           # forward pass
           with torch.no_grad():
               _lowercase :							int   	=	model(**lowerCamelCase)

           # verify the logits
           _lowercase :							List[str]   	=	torch.Size((1, 4_00))
           self.assertEqual(outputs.logits.shape,   lowerCamelCase)

           _lowercase :							str   	=	torch.tensor([0.3_6_6_9, -0.0_6_8_8, -0.2_4_2_1]).to(lowerCamelCase)

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




       @slow
       def     UpperCamelCase  (      self) ->     str:




           """simple docstring"""

           _lowercase :							Dict   	=	VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short').to(lowerCamelCase)

           _lowercase :							Tuple   	=	self.default_image_processor
           _lowercase :							str   	=	prepare_video()
           _lowercase :							Optional[int]   	=	image_processor(lowerCamelCase,   return_tensors='pt').to(lowerCamelCase)

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

           # forward pass
           with torch.no_grad():
               _lowercase :							Optional[int]   	=	model(**lowerCamelCase)

           # verify the logits
           _lowercase :							int   	=	torch.Size([1, 14_08, 15_36])
           _lowercase :							List[str]   	=	torch.tensor(
               [[0.7_9_9_4, 0.9_6_1_2, 0.8_5_0_8], [0.7_4_0_1, 0.8_9_5_8, 0.8_3_0_2], [0.5_8_6_2, 0.7_4_6_8, 0.7_3_2_5]],   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`)
           _lowercase :							Union[str, Any]   	=	torch.tensor([0.5_1_4_2],   device=lowerCamelCase)
           self.assertTrue(torch.allclose(outputs.loss,   lowerCamelCase,   atol=1E-4))

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

           with torch.no_grad():
               _lowercase :							Union[str, Any]   	=	model(**lowerCamelCase)

           _lowercase :							str   	=	torch.tensor(torch.tensor([0.6_4_6_9]),   device=lowerCamelCase)
           self.assertTrue(torch.allclose(outputs.loss,   lowerCamelCase,   atol=1E-4))






import warnings

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






class _lowerCamelCase(		_a     ):
       lowercase_				:			int									=    ["""image_processor""", """tokenizer"""]
       lowercase_				:			List[str]									=    """CLIPImageProcessor"""
       lowercase_				:			Union[str, Any]									=    ("""CLIPTokenizer""", """CLIPTokenizerFast""")
       def __init__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   **lowerCamelCase) ->     Dict:




           """simple docstring"""

           _lowercase :							List[Any]   	=	None
           if "feature_extractor" in kwargs:
               warnings.warn(
                   'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                   ' instead.',   lowerCamelCase,   )
               _lowercase :							Dict   	=	kwargs.pop('feature_extractor')

           _lowercase :							int   	=	image_processor if image_processor is not None else feature_extractor
           if image_processor is None:
               raise ValueError('You need to specify an `image_processor`.')
           if tokenizer is None:
               raise ValueError('You need to specify a `tokenizer`.')

           super().__init__(lowerCamelCase,   lowerCamelCase)
       def __call__(      self,   lowerCamelCase=None,   lowerCamelCase=None,   lowerCamelCase=None,   **lowerCamelCase) ->     Tuple:




           """simple docstring"""

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

           if text is not None:
               _lowercase :							Optional[Any]   	=	self.tokenizer(lowerCamelCase,   return_tensors=lowerCamelCase,   **lowerCamelCase)

           if images is not None:
               _lowercase :							List[Any]   	=	self.image_processor(lowerCamelCase,   return_tensors=lowerCamelCase,   **lowerCamelCase)

           if text is not None and images is not None:
               _lowercase :							int   	=	image_features.pixel_values
               return encoding
           elif text is not None:
               return encoding
           else:
               return BatchEncoding(data=dict(**lowerCamelCase),   tensor_type=lowerCamelCase)
       def     UpperCamelCase  (      self,   *lowerCamelCase,   **lowerCamelCase) ->     Union[str, Any]:




           """simple docstring"""

           return self.tokenizer.batch_decode(*lowerCamelCase,   **lowerCamelCase)
       def     UpperCamelCase  (      self,   *lowerCamelCase,   **lowerCamelCase) ->     int:




           """simple docstring"""

           return self.tokenizer.decode(*lowerCamelCase,   **lowerCamelCase)
       @property
       def     UpperCamelCase  (      self) ->     int:




           """simple docstring"""

           _lowercase :							int   	=	self.tokenizer.model_input_names
           _lowercase :							Dict   	=	self.image_processor.model_input_names
           return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))
       @property
       def     UpperCamelCase  (      self) ->     List[str]:




           """simple docstring"""

           warnings.warn(
               '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.',   lowerCamelCase,   )
           return self.image_processor_class




       @property
       def     UpperCamelCase  (      self) ->     Optional[Any]:




           """simple docstring"""

           warnings.warn(
               '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.',   lowerCamelCase,   )
           return self.image_processor


