

'''simple docstring'''






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




  """simple docstring"""




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

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

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






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any			):
   __lowercase							  = floats_tensor(
       [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

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

   __lowercase							  = self.get_config()

   return config, pixel_values, labels






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   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=lowercase__					,initializer_range=self.initializer_range					,)






  def 					SCREAMING_SNAKE_CASE      (      self   :     int					,lowercase__   :     int					,lowercase__   :     Any					,lowercase__   :     Optional[Any]			):
   __lowercase							  = VideoMAEModel(config=lowercase__			)
   model.to(lowercase__			)
   model.eval()
   __lowercase							  = model(lowercase__			)
   self.parent.assertEqual(result.last_hidden_state.shape					,(self.batch_size, self.seq_length, self.hidden_size)			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Dict					,lowercase__   :     Optional[Any]					,lowercase__   :     Any					,lowercase__   :     Any			):
   __lowercase							  = VideoMAEForPreTraining(lowercase__			)
   model.to(lowercase__			)
   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							  = torch.ones((self.num_masks,)			)
   __lowercase							  = torch.cat([mask, torch.zeros(self.seq_length - mask.size(0			)			)]			)
   __lowercase							  = mask.expand(self.batch_size					,-1			).bool()

   __lowercase							  = model(lowercase__					,lowercase__			)
   # model only returns predictions for masked patches
   __lowercase							  = mask.sum().item()
   __lowercase							  = 3 * self.tubelet_size * self.patch_size**2
   self.parent.assertEqual(result.logits.shape					,(self.batch_size, num_masked_patches, decoder_num_labels)			)






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






@require_torch
class 			lowercase_	(lowerCamelCase__				,       lowerCamelCase__				,       unittest.TestCase      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  Optional[Any]							=			(
      (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
  )
  SCREAMING_SNAKE_CASE    :  int							=			(
      {'feature-extraction': VideoMAEModel, 'video-classification': VideoMAEForVideoClassification}
      if is_torch_available()
      else {}
  )

  SCREAMING_SNAKE_CASE    :  str							=			False
  SCREAMING_SNAKE_CASE    :  Any							=			False
  SCREAMING_SNAKE_CASE    :  str							=			False
  SCREAMING_SNAKE_CASE    :  Union[str, Any]							=			False






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any			):
   __lowercase							  = VideoMAEModelTester(self			)
   __lowercase							  = ConfigTester(self					,config_class=lowercase__					,has_text_modality=lowercase__					,hidden_size=3_7			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     int					,lowercase__   :     Tuple					,lowercase__   :     List[str]					,lowercase__   :     List[str]=False			):
   __lowercase							  = copy.deepcopy(lowercase__			)

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

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

   return inputs_dict






  def 					SCREAMING_SNAKE_CASE      (      self   :     str			):
   self.config_tester.run_common_tests()






  @unittest.skip(reason='''VideoMAE does not use inputs_embeds'''			)
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]			):
   pass






  def 					SCREAMING_SNAKE_CASE      (      self   :     str			):
   __lowercase  ,	__lowercase							  = self.model_tester.prepare_config_and_inputs_for_common()

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






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]			):
   __lowercase  ,	__lowercase							  = self.model_tester.prepare_config_and_inputs_for_common()

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

    __lowercase							  = ['''pixel_values''']
    self.assertListEqual(arg_names[:1]					,lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]			):
   __lowercase							  = self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_model(*lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     str			):
   __lowercase							  = self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_for_pretraining(*lowercase__			)






  @slow
  def 					SCREAMING_SNAKE_CASE      (      self   :     Any			):
   for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
    __lowercase							  = VideoMAEModel.from_pretrained(lowercase__			)
    self.assertIsNotNone(lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]			):
   if not self.has_attentions:
    pass

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

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

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

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

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

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

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

     __lowercase							  = outputs.attentions

     self.assertEqual(len(lowercase__			)					,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 					SCREAMING_SNAKE_CASE      (      self   :     Dict			):
   def check_hidden_states_output(lowercase__   :     List[str]					,lowercase__   :     int					,lowercase__   :     int			):
    __lowercase							  = model_class(lowercase__			)
    model.to(lowercase__			)
    model.eval()

    with torch.no_grad():
     __lowercase							  = model(**self._prepare_for_class(lowercase__					,lowercase__			)			)

    __lowercase							  = outputs.hidden_states
    __lowercase							  = self.model_tester.num_hidden_layers + 1
    self.assertEqual(len(lowercase__			)					,lowercase__			)

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

   for model_class in self.all_model_classes:
    __lowercase							  = True
    check_hidden_states_output(lowercase__					,lowercase__					,lowercase__			)

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

    check_hidden_states_output(lowercase__					,lowercase__					,lowercase__			)






  @unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''			)
  def 					SCREAMING_SNAKE_CASE      (      self   :     Tuple			):
   pass


def 		_A     (    ):






 """simple docstring"""



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






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




  """simple docstring"""




  @cached_property
  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   # logits were tested with a different mean and std, so we use the same here
   return (
       VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]					,image_std=[0.5, 0.5, 0.5]			)
       if is_vision_available()
       else None
   )






  @slow
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]			):
   __lowercase							  = VideoMAEForVideoClassification.from_pretrained('''MCG-NJU/videomae-base-finetuned-kinetics'''			).to(
       lowercase__			)

   __lowercase							  = self.default_image_processor
   __lowercase							  = prepare_video()
   __lowercase							  = image_processor(lowercase__					,return_tensors='''pt'''			).to(lowercase__			)

   # forward pass
   with torch.no_grad():
    __lowercase							  = model(**lowercase__			)

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

   __lowercase							  = torch.tensor([0.3_6_6_9, -0.0_6_8_8, -0.2_4_2_1]			).to(lowercase__			)

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






  @slow
  def 					SCREAMING_SNAKE_CASE      (      self   :     Tuple			):
   __lowercase							  = VideoMAEForPreTraining.from_pretrained('''MCG-NJU/videomae-base-short'''			).to(lowercase__			)

   __lowercase							  = self.default_image_processor
   __lowercase							  = prepare_video()
   __lowercase							  = image_processor(lowercase__					,return_tensors='''pt'''			).to(lowercase__			)

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

   # forward pass
   with torch.no_grad():
    __lowercase							  = model(**lowercase__			)

   # verify the logits
   __lowercase							  = torch.Size([1, 1_4_0_8, 1_5_3_6]			)
   __lowercase							  = 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=lowercase__			)
   self.assertEqual(outputs.logits.shape					,lowercase__			)
   self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]					,lowercase__					,atol=1e-4			)			)

   # verify the loss (`config.norm_pix_loss` = `True`)
   __lowercase							  = torch.tensor([0.5_1_4_2]					,device=lowercase__			)
   self.assertTrue(torch.allclose(outputs.loss					,lowercase__					,atol=1e-4			)			)

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

   with torch.no_grad():
    __lowercase							  = model(**lowercase__			)

   __lowercase							  = torch.tensor(torch.tensor([0.6_4_6_9]			)					,device=lowercase__			)
   self.assertTrue(torch.allclose(outputs.loss					,lowercase__					,atol=1e-4			)			)




'''simple docstring'''






import warnings

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






class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  Optional[Any]							=			['image_processor', 'tokenizer']
  SCREAMING_SNAKE_CASE    :  Union[str, Any]							=			'CLIPImageProcessor'
  SCREAMING_SNAKE_CASE    :  Union[str, Any]							=			('CLIPTokenizer', 'CLIPTokenizerFast')






  def __init__(      self   :     List[Any]					,lowercase__   :     Dict=None					,lowercase__   :     Union[str, Any]=None					,**lowercase__   :     Tuple			):
   __lowercase							  = None
   if "feature_extractor" in kwargs:
    warnings.warn(
        '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
        ''' instead.'''					,lowercase__					,)
    __lowercase							  = kwargs.pop('''feature_extractor'''			)

   __lowercase							  = 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__(lowercase__					,lowercase__			)






  def __call__(      self   :     List[Any]					,lowercase__   :     str=None					,lowercase__   :     List[Any]=None					,lowercase__   :     Optional[Any]=None					,**lowercase__   :     int			):

   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							  = self.tokenizer(lowercase__					,return_tensors=lowercase__					,**lowercase__			)

   if images is not None:
    __lowercase							  = self.image_processor(lowercase__					,return_tensors=lowercase__					,**lowercase__			)

   if text is not None and images is not None:
    __lowercase							  = image_features.pixel_values
    return encoding
   elif text is not None:
    return encoding
   else:
    return BatchEncoding(data=dict(**lowercase__			)					,tensor_type=lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]					,*lowercase__   :     List[str]					,**lowercase__   :     int			):
   return self.tokenizer.batch_decode(*lowercase__					,**lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[Any]					,*lowercase__   :     Optional[int]					,**lowercase__   :     Union[str, Any]			):
   return self.tokenizer.decode(*lowercase__					,**lowercase__			)






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]			):
   __lowercase							  = self.tokenizer.model_input_names
   __lowercase							  = self.image_processor.model_input_names
   return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names			)			)






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   warnings.warn(
       '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''					,lowercase__					,)
   return self.image_processor_class






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]			):
   warnings.warn(
       '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''					,lowercase__					,)
   return self.image_processor

