


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







    """simple docstring"""



    def __init__(       self   :  List[str]   ,						lowerCamelCase   :  List[str]   ,						lowerCamelCase   :  str=13   ,						lowerCamelCase   :  List[str]=10   ,						lowerCamelCase   :  Optional[Any]=3   ,						lowerCamelCase   :  Union[str, Any]=2   ,						lowerCamelCase   :  Optional[Any]=2   ,						lowerCamelCase   :  Dict=2   ,						lowerCamelCase   :  Optional[int]=True   ,						lowerCamelCase   :  int=True   ,						lowerCamelCase   :  Optional[Any]=32   ,						lowerCamelCase   :  Optional[int]=5   ,						lowerCamelCase   :  Optional[int]=4   ,						lowerCamelCase   :  str=37   ,						lowerCamelCase   :  Optional[Any]="gelu"   ,						lowerCamelCase   :  Tuple=0.1   ,						lowerCamelCase   :  int=0.1   ,						lowerCamelCase   :  Optional[Any]=10   ,						lowerCamelCase   :  Optional[Any]=0.02   ,						lowerCamelCase   :  List[str]=0.9   ,						lowerCamelCase   :  Any=None   ,						)       ->  Dict:
      lowerCAmelCase_      :				Optional[int]      =       parent
      lowerCAmelCase_      :				int      =       batch_size
      lowerCAmelCase_      :				Any      =       image_size
      lowerCAmelCase_      :				Optional[Any]      =       num_channels
      lowerCAmelCase_      :				str      =       patch_size
      lowerCAmelCase_      :				Any      =       tubelet_size
      lowerCAmelCase_      :				int      =       num_frames
      lowerCAmelCase_      :				Optional[Any]      =       is_training
      lowerCAmelCase_      :				Optional[int]      =       use_labels
      lowerCAmelCase_      :				Union[str, Any]      =       hidden_size
      lowerCAmelCase_      :				Optional[Any]      =       num_hidden_layers
      lowerCAmelCase_      :				str      =       num_attention_heads
      lowerCAmelCase_      :				List[str]      =       intermediate_size
      lowerCAmelCase_      :				List[Any]      =       hidden_act
      lowerCAmelCase_      :				List[str]      =       hidden_dropout_prob
      lowerCAmelCase_      :				Optional[Any]      =       attention_probs_dropout_prob
      lowerCAmelCase_      :				Dict      =       type_sequence_label_size
      lowerCAmelCase_      :				Union[str, Any]      =       initializer_range
      lowerCAmelCase_      :				int      =       mask_ratio
      lowerCAmelCase_      :				Union[str, Any]      =       scope

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

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

    def __lowercase       (       self   :  Dict		)       ->  Optional[Any]:
      lowerCAmelCase_      :				Tuple      =       floats_tensor(
          [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]		)

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

      lowerCAmelCase_      :				Dict      =       self.get_config()

      return config, pixel_values, labels

    def __lowercase       (       self   :  Optional[int]		)       ->  int:
      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 __lowercase       (       self   :  Tuple   ,						lowerCamelCase   :  Any   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  Any		)       ->  Union[str, Any]:
      lowerCAmelCase_      :				Optional[int]      =       VideoMAEModel(config=lowerCamelCase		)
      model.to(lowerCamelCase		)
      model.eval()
      lowerCAmelCase_      :				Any      =       model(lowerCamelCase		)
      self.parent.assertEqual(result.last_hidden_state.shape   ,						(self.batch_size, self.seq_length, self.hidden_size)		)

    def __lowercase       (       self   :  str   ,						lowerCamelCase   :  Dict   ,						lowerCamelCase   :  List[Any]   ,						lowerCamelCase   :  Dict		)       ->  List[Any]:
      lowerCAmelCase_      :				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
      lowerCAmelCase_      :				int      =       torch.ones((self.num_masks,)		)
      lowerCAmelCase_      :				int      =       torch.cat([mask, torch.zeros(self.seq_length - mask.size(0		)		)]		)
      lowerCAmelCase_      :				List[str]      =       mask.expand(self.batch_size   ,						-1		).bool()

      lowerCAmelCase_      :				Optional[int]      =       model(lowerCamelCase   ,						lowerCamelCase		)
      # model only returns predictions for masked patches
      lowerCAmelCase_      :				List[Any]      =       mask.sum().item()
      lowerCAmelCase_      :				List[str]      =       3 * self.tubelet_size * self.patch_size**2
      self.parent.assertEqual(result.logits.shape   ,						(self.batch_size, num_masked_patches, decoder_num_labels)		)





    def __lowercase       (       self   :  Optional[int]		)       ->  List[str]:
      lowerCAmelCase_      :				List[Any]      =       self.prepare_config_and_inputs()
      lowerCAmelCase_,							lowerCAmelCase_,							lowerCAmelCase_      :				Optional[int]      =       config_and_inputs
      lowerCAmelCase_      :				int      =       {"""pixel_values""": pixel_values}
      return config, inputs_dict







@require_torch
class        __snake_case	(       _SCREAMING_SNAKE_CASE     ,_SCREAMING_SNAKE_CASE     ,unittest.TestCase):







    """simple docstring"""



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

    lowercase       			= False
    lowercase       			= False
    lowercase       			= False
    lowercase       			= False

    def __lowercase       (       self   :  Optional[int]		)       ->  Optional[int]:
      lowerCAmelCase_      :				List[str]      =       VideoMAEModelTester(self		)
      lowerCAmelCase_      :				Dict      =       ConfigTester(self   ,						config_class=lowerCamelCase   ,						has_text_modality=lowerCamelCase   ,						hidden_size=37		)

    def __lowercase       (       self   :  Dict   ,						lowerCamelCase   :  List[str]   ,						lowerCamelCase   :  Tuple   ,						lowerCamelCase   :  List[Any]=False		)       ->  Any:
      lowerCAmelCase_      :				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
        lowerCAmelCase_      :				Union[str, Any]      =       torch.ones((self.model_tester.num_masks,)		)
        lowerCAmelCase_      :				List[str]      =       torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0		)		)]		)
        lowerCAmelCase_      :				Tuple      =       mask.expand(self.model_tester.batch_size   ,						-1		).bool()
        lowerCAmelCase_      :				Union[str, Any]      =       bool_masked_pos.to(lowerCamelCase		)

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

      return inputs_dict

    def __lowercase       (       self   :  Any		)       ->  Tuple:
      self.config_tester.run_common_tests()

    @unittest.skip(reason="""VideoMAE does not use inputs_embeds"""		)
    def __lowercase       (       self   :  List[Any]		)       ->  List[str]:
      pass

    def __lowercase       (       self   :  Dict		)       ->  Optional[int]:
      lowerCAmelCase_,							lowerCAmelCase_      :				List[Any]      =       self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        lowerCAmelCase_      :				int      =       model_class(lowerCamelCase		)
        self.assertIsInstance(model.get_input_embeddings()   ,						(nn.Module)		)
        lowerCAmelCase_      :				Optional[Any]      =       model.get_output_embeddings()
        self.assertTrue(x is None or isinstance(lowerCamelCase   ,						nn.Linear		)		)

    def __lowercase       (       self   :  Optional[Any]		)       ->  Tuple:
      lowerCAmelCase_,							lowerCAmelCase_      :				int      =       self.model_tester.prepare_config_and_inputs_for_common()

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

        lowerCAmelCase_      :				Optional[Any]      =       ["""pixel_values"""]
        self.assertListEqual(arg_names[:1]   ,						lowerCamelCase		)

    def __lowercase       (       self   :  str		)       ->  List[str]:
      lowerCAmelCase_      :				Union[str, Any]      =       self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*lowerCamelCase		)

    def __lowercase       (       self   :  List[str]		)       ->  Any:
      lowerCAmelCase_      :				List[Any]      =       self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_pretraining(*lowerCamelCase		)

    @slow
    def __lowercase       (       self   :  str		)       ->  int:
      for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        lowerCAmelCase_      :				str      =       VideoMAEModel.from_pretrained(lowerCamelCase		)
        self.assertIsNotNone(lowerCamelCase		)

    def __lowercase       (       self   :  Union[str, Any]		)       ->  Tuple:
      if not self.has_attentions:
        pass

      else:
        lowerCAmelCase_,							lowerCAmelCase_      :				List[Any]      =       self.model_tester.prepare_config_and_inputs_for_common()
        lowerCAmelCase_      :				int      =       True

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

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

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

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

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

        with torch.no_grad():
          lowerCAmelCase_      :				str      =       model(**self._prepare_for_class(lowerCamelCase   ,						lowerCamelCase		)		)

        lowerCAmelCase_      :				int      =       outputs.hidden_states
        lowerCAmelCase_      :				Union[str, Any]      =       self.model_tester.num_hidden_layers + 1
        self.assertEqual(len(lowerCamelCase		)   ,						lowerCamelCase		)

        lowerCAmelCase_      :				Dict      =       self.model_tester.seq_length - self.model_tester.num_masks
        lowerCAmelCase_      :				str      =       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_      :				Any      =       self.model_tester.prepare_config_and_inputs_for_common()

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

        # check that output_hidden_states also work using config
        del inputs_dict["output_hidden_states"]
        lowerCAmelCase_      :				List[Any]      =       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 __lowercase       (       self   :  Any		)       ->  Optional[Any]:
      pass







def     UpperCamelCase_      (   ):
  '''simple docstring'''





  lowerCAmelCase_      :				Any      =       hf_hub_download(
      repo_id="""hf-internal-testing/spaghetti-video"""							,   filename="""eating_spaghetti.npy"""							,   repo_type="""dataset""" )
  lowerCAmelCase_      :				Optional[Any]      =       np.load(A__ )
  return list(A__ )






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







    """simple docstring"""



    @cached_property
    def __lowercase       (       self   :  Union[str, Any]		)       ->  Union[str, Any]:
      # 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 __lowercase       (       self   :  List[str]		)       ->  List[str]:
      lowerCAmelCase_      :				Optional[Any]      =       VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""		).to(
          lowerCamelCase		)

      lowerCAmelCase_      :				List[str]      =       self.default_image_processor
      lowerCAmelCase_      :				str      =       prepare_video()
      lowerCAmelCase_      :				Any      =       image_processor(lowerCamelCase   ,						return_tensors="""pt"""		).to(lowerCamelCase		)

      # forward pass
      with torch.no_grad():
        lowerCAmelCase_      :				Any      =       model(**lowerCamelCase		)

      # verify the logits
      lowerCAmelCase_      :				Any      =       torch.Size((1, 4_00)		)
      self.assertEqual(outputs.logits.shape   ,						lowerCamelCase		)

      lowerCAmelCase_      :				int      =       torch.tensor([0.3_669, -0.0_688, -0.2_421]		).to(lowerCamelCase		)

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





    @slow
    def __lowercase       (       self   :  int		)       ->  List[str]:
      lowerCAmelCase_      :				Optional[Any]      =       VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""		).to(lowerCamelCase		)

      lowerCAmelCase_      :				Optional[int]      =       self.default_image_processor
      lowerCAmelCase_      :				List[Any]      =       prepare_video()
      lowerCAmelCase_      :				int      =       image_processor(lowerCamelCase   ,						return_tensors="""pt"""		).to(lowerCamelCase		)

      # add boolean mask, indicating which patches to mask
      lowerCAmelCase_      :				List[str]      =       hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""   ,						filename="""bool_masked_pos.pt"""		)
      lowerCAmelCase_      :				str      =       torch.load(lowerCamelCase		)

      # forward pass
      with torch.no_grad():
        lowerCAmelCase_      :				Any      =       model(**lowerCamelCase		)

      # verify the logits
      lowerCAmelCase_      :				Dict      =       torch.Size([1, 14_08, 15_36]		)
      lowerCAmelCase_      :				List[Any]      =       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=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_      :				int      =       torch.tensor([0.5_142]   ,						device=lowerCamelCase		)
      self.assertTrue(torch.allclose(outputs.loss   ,						lowerCamelCase   ,						atol=1E-4		)		)

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

      with torch.no_grad():
        lowerCAmelCase_      :				int      =       model(**lowerCamelCase		)

      lowerCAmelCase_      :				Tuple      =       torch.tensor(torch.tensor([0.6_469]		)   ,						device=lowerCamelCase		)
      self.assertTrue(torch.allclose(outputs.loss   ,						lowerCamelCase   ,						atol=1E-4		)		)







'''simple docstring'''






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


__A :  List[str]    							=				{
    "configuration_bigbird_pegasus": [
        "BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "BigBirdPegasusConfig",
        "BigBirdPegasusOnnxConfig",
    ],
}

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __A :  Union[str, Any]    							=				[
      "BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST",
      "BigBirdPegasusForCausalLM",
      "BigBirdPegasusForConditionalGeneration",
      "BigBirdPegasusForQuestionAnswering",
      "BigBirdPegasusForSequenceClassification",
      "BigBirdPegasusModel",
      "BigBirdPegasusPreTrainedModel",
  ]


if TYPE_CHECKING:
  from .configuration_bigbird_pegasus import (
      BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP,
      BigBirdPegasusConfig,
      BigBirdPegasusOnnxConfig,
  )

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_bigbird_pegasus import (
        BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST,
        BigBirdPegasusForCausalLM,
        BigBirdPegasusForConditionalGeneration,
        BigBirdPegasusForQuestionAnswering,
        BigBirdPegasusForSequenceClassification,
        BigBirdPegasusModel,
        BigBirdPegasusPreTrainedModel,
    )


else:
  import sys

  __A :  List[Any]    							=				_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)



