

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

    def __init__(			self :      Optional[int]							,    snake_case__ :      List[str]							,    snake_case__ :      Union[str, Any]=1_3							,    snake_case__ :      Optional[int]=1_0							,    snake_case__ :      List[Any]=3							,    snake_case__ :      List[str]=2							,    snake_case__ :      List[str]=2							,    snake_case__ :      Union[str, Any]=2							,    snake_case__ :      Union[str, Any]=True							,    snake_case__ :      int=True							,    snake_case__ :      int=3_2							,    snake_case__ :      Optional[Any]=5							,    snake_case__ :      Optional[Any]=4							,    snake_case__ :      str=3_7							,    snake_case__ :      str="gelu"							,    snake_case__ :      Optional[int]=0.1							,    snake_case__ :      int=0.1							,    snake_case__ :      Any=1_0							,    snake_case__ :      List[Any]=0.02							,    snake_case__ :      Union[str, Any]=0.9							,    snake_case__ :      List[Any]=None							,    ):

     """simple docstring"""






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

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

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

    def UpperCamelCase			(			self :      str    ):

     """simple docstring"""






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

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

     SCREAMING_SNAKE_CASE								=			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=snake_case__							,    initializer_range=self.initializer_range							,    )

    def UpperCamelCase			(			self :      Any							,    snake_case__ :      List[str]							,    snake_case__ :      Optional[Any]							,    snake_case__ :      List[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			VideoMAEModel(config=snake_case__    )
     model.to(snake_case__    )
     model.eval()
     SCREAMING_SNAKE_CASE								=			model(snake_case__    )
     self.parent.assertEqual(result.last_hidden_state.shape							,    (self.batch_size, self.seq_length, self.hidden_size)    )

    def UpperCamelCase			(			self :      Dict							,    snake_case__ :      Optional[int]							,    snake_case__ :      Optional[int]							,    snake_case__ :      Union[str, Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			VideoMAEForPreTraining(snake_case__    )
     model.to(snake_case__    )
     model.eval()
     # important: each video needs to have the same number of masked patches
     # hence we define a single mask, which we then repeat for each example in the batch
     SCREAMING_SNAKE_CASE								=			torch.ones((self.num_masks,)    )
     SCREAMING_SNAKE_CASE								=			torch.cat([mask, torch.zeros(self.seq_length - mask.size(0    )    )]    )
     SCREAMING_SNAKE_CASE								=			mask.expand(self.batch_size							,    -1    ).bool()

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

     """simple docstring"""






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




@require_torch
class      UpperCamelCase (     SCREAMING_SNAKE_CASE	,			SCREAMING_SNAKE_CASE	,			unittest.TestCase   ):
    __UpperCamelCase         =(
        (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
    )
    __UpperCamelCase         =(
        {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
        if is_torch_available()
        else {}
    )

    __UpperCamelCase         =False
    __UpperCamelCase         =False
    __UpperCamelCase         =False
    __UpperCamelCase         =False

    def UpperCamelCase			(			self :      List[str]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			VideoMAEModelTester(self    )
     SCREAMING_SNAKE_CASE								=			ConfigTester(self							,    config_class=snake_case__							,    has_text_modality=snake_case__							,    hidden_size=3_7    )

    def UpperCamelCase			(			self :      Optional[Any]							,    snake_case__ :      int							,    snake_case__ :      Dict							,    snake_case__ :      Optional[int]=False    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			copy.deepcopy(snake_case__    )

     if model_class == VideoMAEForPreTraining:
      # important: each video needs to have the same number of masked patches
      # hence we define a single mask, which we then repeat for each example in the batch
      SCREAMING_SNAKE_CASE								=			torch.ones((self.model_tester.num_masks,)    )
      SCREAMING_SNAKE_CASE								=			torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0    )    )]    )
      SCREAMING_SNAKE_CASE								=			mask.expand(self.model_tester.batch_size							,    -1    ).bool()
      SCREAMING_SNAKE_CASE								=			bool_masked_pos.to(snake_case__    )

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

     return inputs_dict

    def UpperCamelCase			(			self :      List[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 :      Any    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE								=			self.model_tester.prepare_config_and_inputs_for_common()

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

    def UpperCamelCase			(			self :      int    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE								=			self.model_tester.prepare_config_and_inputs_for_common()

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

      SCREAMING_SNAKE_CASE								=			['pixel_values']
      self.assertListEqual(arg_names[:1]							,    snake_case__    )

    def UpperCamelCase			(			self :      str    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*snake_case__    )

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

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_pretraining(*snake_case__    )

    @slow
    def UpperCamelCase			(			self :      int    ):

     """simple docstring"""






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

    def UpperCamelCase			(			self :      Tuple    ):

     """simple docstring"""






     if not self.has_attentions:
      pass

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

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

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

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

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

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

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

       SCREAMING_SNAKE_CASE								=			outputs.attentions

       self.assertEqual(len(snake_case__    )							,    self.model_tester.num_hidden_layers    )
       self.assertListEqual(
           list(self_attentions[0].shape[-3:]    )							,    [self.model_tester.num_attention_heads, seq_len, seq_len]							,    )

    def UpperCamelCase			(			self :      Dict    ):

     """simple docstring"""






     def check_hidden_states_output(snake_case__ :      Union[str, Any]							,    snake_case__ :      int							,    snake_case__ :      str    ):
      SCREAMING_SNAKE_CASE								=			model_class(snake_case__    )
      model.to(snake_case__    )
      model.eval()

      with torch.no_grad():
       SCREAMING_SNAKE_CASE								=			model(**self._prepare_for_class(snake_case__							,    snake_case__    )    )

      SCREAMING_SNAKE_CASE								=			outputs.hidden_states
      SCREAMING_SNAKE_CASE								=			self.model_tester.num_hidden_layers + 1
      self.assertEqual(len(snake_case__    )							,    snake_case__    )

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

     SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE								=			self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
      SCREAMING_SNAKE_CASE								=			True
      check_hidden_states_output(snake_case__							,    snake_case__							,    snake_case__    )

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

      check_hidden_states_output(snake_case__							,    snake_case__							,    snake_case__    )




    @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'    )
    def UpperCamelCase			(			self :      Union[str, Any]    ):

     """simple docstring"""






     pass


def        __lowerCAmelCase    (       )		->							Optional[Any]:
 '''simple docstring'''


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




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

    @cached_property
    def UpperCamelCase			(			self :      int    ):

     """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 :      Tuple    ):

     """simple docstring"""






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

     SCREAMING_SNAKE_CASE								=			self.default_image_processor
     SCREAMING_SNAKE_CASE								=			prepare_video()
     SCREAMING_SNAKE_CASE								=			image_processor(snake_case__							,    return_tensors='pt'    ).to(snake_case__    )

     # forward pass
     with torch.no_grad():
      SCREAMING_SNAKE_CASE								=			model(**snake_case__    )

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

     SCREAMING_SNAKE_CASE								=			torch.tensor([0.3_669, -0.0_688, -0.2_421]    ).to(snake_case__    )

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




    @slow
    def UpperCamelCase			(			self :      Union[str, Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			VideoMAEForPreTraining.from_pretrained('MCG-NJU/videomae-base-short'    ).to(snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.default_image_processor
     SCREAMING_SNAKE_CASE								=			prepare_video()
     SCREAMING_SNAKE_CASE								=			image_processor(snake_case__							,    return_tensors='pt'    ).to(snake_case__    )

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

     # forward pass
     with torch.no_grad():
      SCREAMING_SNAKE_CASE								=			model(**snake_case__    )

     # verify the logits
     SCREAMING_SNAKE_CASE								=			torch.Size([1, 1_4_0_8, 1_5_3_6]    )
     SCREAMING_SNAKE_CASE								=			torch.tensor(
         [[0.7_994, 0.9_612, 0.8_508], [0.7_401, 0.8_958, 0.8_302], [0.5_862, 0.7_468, 0.7_325]]							,    device=snake_case__    )
     self.assertEqual(outputs.logits.shape							,    snake_case__    )
     self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]							,    snake_case__							,    atol=1E-4    )    )

     # verify the loss (`config.norm_pix_loss` = `True`)
     SCREAMING_SNAKE_CASE								=			torch.tensor([0.5_142]							,    device=snake_case__    )
     self.assertTrue(torch.allclose(outputs.loss							,    snake_case__							,    atol=1E-4    )    )

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

     with torch.no_grad():
      SCREAMING_SNAKE_CASE								=			model(**snake_case__    )

     SCREAMING_SNAKE_CASE								=			torch.tensor(torch.tensor([0.6_469]    )							,    device=snake_case__    )
     self.assertTrue(torch.allclose(outputs.loss							,    snake_case__							,    atol=1E-4    )    )




def        __lowerCAmelCase    (       _UpperCamelCase		:  str		,							_UpperCamelCase		:  int	)		->							list[str]:
 '''simple docstring'''


 return [sentence[i : i + ngram_size] for i in range(len(_UpperCamelCase	) - ngram_size + 1	)]


if __name__ == "__main__":
      from doctest import testmod

      testmod()

