


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	,       _lowercase	,       _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	,       _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	,       _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	,       _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	,       _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 dataclasses
import json
import warnings
from dataclasses import dataclass, field
from time import time
from typing import List

from ..utils import logging


SCREAMING_SNAKE_CASE :   Dict							      =				logging.get_logger(__name__)



def        UpperCamelCase_(   lowerCamelCase_=None							,				lowerCamelCase_=None				)					->		Tuple:
    return field(default_factory=lambda: default							,				metadata=lowerCamelCase_				)






@dataclass
class _lowerCamelCase:
       lowercase_				:			List[str]									=    list_field(
           default=[],				metadata={
               """help""": (
                   """Model checkpoints to be provided to the AutoModel classes. Leave blank to benchmark the base version"""
                   """ of all available models"""
               )
           },				)

       lowercase_				:			List[int]									=    list_field(
           default=[8],				metadata={"""help""": """List of batch sizes for which memory and time performance will be evaluated"""}     )

       lowercase_				:			List[int]									=    list_field(
           default=[8, 32, 1_28, 5_12],				metadata={"""help""": """List of sequence lengths for which memory and time performance will be evaluated"""},				)

       lowercase_				:			bool									=    field(
           default=_a,				metadata={"""help""": """Whether to benchmark inference of model. Inference can be disabled via --no-inference."""},				)
       lowercase_				:			bool									=    field(
           default=_a,				metadata={"""help""": """Whether to run on available cuda devices. Cuda can be disabled via --no-cuda."""},				)
       lowercase_				:			bool									=    field(
           default=_a,				metadata={"""help""": """Whether to run on available tpu devices. TPU can be disabled via --no-tpu."""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Use FP16 to accelerate inference."""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Benchmark training of model"""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Verbose memory tracing"""}     )
       lowercase_				:			bool									=    field(
           default=_a,				metadata={"""help""": """Whether to perform speed measurements. Speed measurements can be disabled via --no-speed."""},				)
       lowercase_				:			bool									=    field(
           default=_a,				metadata={
               """help""": """Whether to perform memory measurements. Memory measurements can be disabled via --no-memory"""
           },				)
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Trace memory line by line"""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Save result to a CSV file"""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Save all print statements in a log file"""}     )
       lowercase_				:			bool									=    field(default=_a,				metadata={"""help""": """Whether to print environment information"""}     )
       lowercase_				:			bool									=    field(
           default=_a,				metadata={
               """help""": (
                   """Whether to use multiprocessing for memory and speed measurement. It is highly recommended to use"""
                   """ multiprocessing for accurate CPU and GPU memory measurements. This option should only be disabled"""
                   """ for debugging / testing and on TPU."""
               )
           },				)
       lowercase_				:			str									=    field(
           default=F'''inference_time_{round(time()     )}.csv''',				metadata={"""help""": """CSV filename used if saving time results to csv."""},				)
       lowercase_				:			str									=    field(
           default=F'''inference_memory_{round(time()     )}.csv''',				metadata={"""help""": """CSV filename used if saving memory results to csv."""},				)
       lowercase_				:			str									=    field(
           default=F'''train_time_{round(time()     )}.csv''',				metadata={"""help""": """CSV filename used if saving time results to csv for training."""},				)
       lowercase_				:			str									=    field(
           default=F'''train_memory_{round(time()     )}.csv''',				metadata={"""help""": """CSV filename used if saving memory results to csv for training."""},				)
       lowercase_				:			str									=    field(
           default=F'''env_info_{round(time()     )}.csv''',				metadata={"""help""": """CSV filename used if saving environment information."""},				)
       lowercase_				:			str									=    field(
           default=F'''log_{round(time()     )}.csv''',				metadata={"""help""": """Log filename used if print statements are saved in log."""},				)
       lowercase_				:			int									=    field(default=3,				metadata={"""help""": """Times an experiment will be run."""}     )
       lowercase_				:			bool									=    field(
           default=_a,				metadata={
               """help""": (
                   """Instead of loading the model as defined in `config.architectures` if exists, just load the pretrain"""
                   """ model weights."""
               )
           },				)
       def     UpperCamelCase  (      self) ->     Union[str, Any]:




           """simple docstring"""

           warnings.warn(
               F'''The class {self.__class__} is deprecated. Hugging Face Benchmarking utils'''
               ' are deprecated in general and it is advised to use external Benchmarking libraries '
               ' to benchmark Transformer models.',   lowerCamelCase,   )
       def     UpperCamelCase  (      self) ->     Optional[Any]:




           """simple docstring"""

           return json.dumps(dataclasses.asdict(self),   indent=2)
       @property
       def     UpperCamelCase  (      self) ->     List[str]:




           """simple docstring"""

           if len(self.models) <= 0:
               raise ValueError(
                   'Please make sure you provide at least one model name / model identifier, *e.g.* `--models'
                   ' bert-base-cased` or `args.models = [\'bert-base-cased\'].')
           return self.models




       @property
       def     UpperCamelCase  (      self) ->     Dict:




           """simple docstring"""

           if not self.multi_process:
               return False
           elif self.is_tpu:
               logger.info('Multiprocessing is currently not possible on TPU.')
               return False
           else:
               return True


