

'''simple docstring'''





from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
      import torch

__a						:Dict            = logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
      from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):

      def 			__snake_case	(   cls  :   Any						):



            """simple docstring"""

            return cls

def 			__snake_case	(   __UpperCamelCase  :   Union[Tuple, List]							,__UpperCamelCase  :   Optional[Union[List["torch.Generator"], "torch.Generator"]] = None							,__UpperCamelCase  :   Optional["torch.device"] = None							,__UpperCamelCase  :   Optional["torch.dtype"] = None							,__UpperCamelCase  :   Optional["torch.layout"] = None							,):



      """simple docstring"""

      A_	     =							device
      A_	     =							shape[0]

      A_	     =							layout or torch.strided
      A_	     =							device or torch.device("cpu"						)

      if generator is not None:
            A_	     =							generator.device.type if not isinstance(__UpperCamelCase							,__UpperCamelCase						) else generator[0].device.type
            if gen_device_type != device.type and gen_device_type == "cpu":
                  A_	     =							"cpu"
                  if device != "mps":
                        logger.info(
                            f'''The passed generator was created on \'cpu\' even though a tensor on {device} was expected.'''
                            f''' Tensors will be created on \'cpu\' and then moved to {device}. Note that one can probably'''
                            f''' slighly speed up this function by passing a generator that was created on the {device} device.'''						)
            elif gen_device_type != device.type and gen_device_type == "cuda":
                  raise ValueError(f'''Cannot generate a {device} tensor from a generator of type {gen_device_type}.'''						)

      if isinstance(__UpperCamelCase							,__UpperCamelCase						):
            A_	     =							(1,) + shape[1:]
            A_	     =							[
                torch.randn(__UpperCamelCase							,generator=generator[i]							,device=__UpperCamelCase							,dtype=__UpperCamelCase							,layout=__UpperCamelCase						)
                for i in range(__UpperCamelCase						)
            ]
            A_	     =							torch.cat(__UpperCamelCase							,dim=0						).to(__UpperCamelCase						)
      else:
            A_	     =							torch.randn(__UpperCamelCase							,generator=__UpperCamelCase							,device=__UpperCamelCase							,dtype=__UpperCamelCase							,layout=__UpperCamelCase						).to(__UpperCamelCase						)

      return latents



def 			__snake_case	(   __UpperCamelCase  :   Union[str, Any]						):



      """simple docstring"""

      if is_torch_version("<"							,"2.0.0"						) or not hasattr(__UpperCamelCase							,"_dynamo"						):
            return False
      return isinstance(__UpperCamelCase							,torch._dynamo.eval_frame.OptimizedModule						)


import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VivitConfig
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


if is_torch_available():
      import torch
      from torch import nn

      from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VivitForVideoClassification, VivitModel
      from transformers.models.vivit.modeling_vivit import VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
      from transformers import VivitImageProcessor

class      _a				:




      """simple docstring"""

      def __init__(      self       :							List[str]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							List[str]=2	, UpperCAmelCase       :							str=True	, UpperCAmelCase       :							Optional[Any]=True	, UpperCAmelCase       :							Optional[int]=10	, UpperCAmelCase       :							List[Any]=10	, UpperCAmelCase       :							Optional[Any]=8	, UpperCAmelCase       :							Dict=[2, 4, 4]	, UpperCAmelCase       :							Dict=3	, UpperCAmelCase       :							Optional[int]=768	, UpperCAmelCase       :							Union[str, Any]=5	, UpperCAmelCase       :							List[str]=4	, UpperCAmelCase       :							Union[str, Any]=37	, UpperCAmelCase       :							Dict="gelu_fast"	, UpperCAmelCase       :							int=0.0	, UpperCAmelCase       :							Dict=0.0	, UpperCAmelCase       :							List[Any]=0.02	, UpperCAmelCase       :							Any=1E-06	, UpperCAmelCase       :							str=True	, UpperCAmelCase       :							str=None	, ):
            A_	     =							parent
            A_	     =							batch_size
            A_	     =							is_training
            A_	     =							use_labels
            A_	     =							num_labels
            A_	     =							image_size
            A_	     =							num_frames
            A_	     =							tubelet_size
            A_	     =							num_channels
            A_	     =							hidden_size
            A_	     =							num_hidden_layers
            A_	     =							num_attention_heads
            A_	     =							intermediate_size
            A_	     =							hidden_act
            A_	     =							hidden_dropout_prob
            A_	     =							attention_probs_dropout_prob
            A_	     =							initializer_range
            A_	     =							layer_norm_eps
            A_	     =							qkv_bias
            A_	     =							scope

            A_	     =							(
                (self.image_size // self.tubelet_size[2])
                * (self.image_size // self.tubelet_size[1])
                * (self.num_frames // self.tubelet_size[0])
            ) + 1  # CLS token





      def 		__A  (      self       :							List[Any]  ):
            A_	     =							floats_tensor(
                [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

            A_	     =							None
            if self.use_labels:
                  A_	     =							ids_tensor([self.batch_size]	, self.num_labels  )

            A_	     =							self.get_config()

            return config, pixel_values, labels





      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							VivitConfig(
                num_frames=self.num_frames	, image_size=self.image_size	, tubelet_size=self.tubelet_size	, num_channels=self.num_channels	, 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	, initializer_range=self.initializer_range	, layer_norm_eps=self.layer_norm_eps	, qkv_bias=self.qkv_bias	, )
            A_	     =							self.num_labels
            return config





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							Optional[int]  ):
            A_	     =							VivitModel(config=UpperCAmelCase  )
            model.to(UpperCAmelCase  )
            model.eval()
            A_	     =							model(UpperCAmelCase  )
            self.parent.assertEqual(result.last_hidden_state.shape	, (self.batch_size, self.seq_length, self.hidden_size)  )





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							Dict	, UpperCAmelCase       :							str	, UpperCAmelCase       :							List[Any]  ):
            A_	     =							VivitForVideoClassification(UpperCAmelCase  )
            model.to(UpperCAmelCase  )
            model.eval()

            A_	     =							model(UpperCAmelCase  )

            # verify the logits shape
            A_	     =							torch.Size((self.batch_size, self.num_labels)  )
            self.parent.assertEqual(result.logits.shape	, UpperCAmelCase  )







      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							self.prepare_config_and_inputs()
            A_     ,    A_     ,    A_	     =							config_and_inputs
            A_	     =							{"pixel_values": pixel_values}
            return config, inputs_dict




@require_torch
class      _a				( snake_case_    ,							unittest.TestCase							):




      """simple docstring"""

      _lowerCamelCase					: List[Any]					=					(VivitModel, VivitForVideoClassification) if is_torch_available() else ()

      _lowerCamelCase					: Union[str, Any]					=					False
      _lowerCamelCase					: List[Any]					=					False
      _lowerCamelCase					: Any					=					False
      _lowerCamelCase					: int					=					False





      def 		__A  (      self       :							Any  ):
            A_	     =							VivitModelTester(self  )
            A_	     =							ConfigTester(self	, config_class=UpperCAmelCase	, has_text_modality=UpperCAmelCase	, hidden_size=37  )





      def 		__A  (      self       :							Union[str, Any]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Dict	, UpperCAmelCase       :							Tuple=False  ):
            A_	     =							copy.deepcopy(UpperCAmelCase  )

            if return_labels:
                  if model_class in get_values(UpperCAmelCase  ):
                        A_	     =							torch.zeros(
                            self.model_tester.batch_size	, dtype=torch.long	, device=UpperCAmelCase  )

            return inputs_dict





      def 		__A  (      self       :							int  ):
            self.config_tester.run_common_tests()





      @unittest.skip(reason="Vivit does not use inputs_embeds"  )
      def 		__A  (      self       :							Any  ):
            pass





      def 		__A  (      self       :							Optional[Any]  ):
            A_     ,    A_	     =							self.model_tester.prepare_config_and_inputs_for_common()

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





      def 		__A  (      self       :							Any  ):
            A_     ,    A_	     =							self.model_tester.prepare_config_and_inputs_for_common()

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

                  A_	     =							["pixel_values", "head_mask"]
                  self.assertListEqual(arg_names[:2]	, UpperCAmelCase  )





      def 		__A  (      self       :							Optional[Any]  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*UpperCAmelCase  )





      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_for_video_classification(*UpperCAmelCase  )





      @slow
      def 		__A  (      self       :							Optional[int]  ):
            for model_name in VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  A_	     =							VivitModel.from_pretrained(UpperCAmelCase  )
                  self.assertIsNotNone(UpperCAmelCase  )





      def 		__A  (      self       :							List[str]  ):
            A_     ,    A_	     =							self.model_tester.prepare_config_and_inputs_for_common()
            A_	     =							True

            for model_class in self.all_model_classes:
                  A_	     =							self.model_tester.seq_length

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

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

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

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

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

                  A_	     =							outputs.attentions

                  self.assertEqual(len(UpperCAmelCase  )	, 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 		__A  (      self       :							Any  ):
            def check_hidden_states_output(UpperCAmelCase       :							str	, UpperCAmelCase       :							str	, UpperCAmelCase       :							List[Any]  ):
                  A_	     =							model_class(UpperCAmelCase  )
                  model.to(UpperCAmelCase  )
                  model.eval()

                  with torch.no_grad():
                        A_	     =							model(**self._prepare_for_class(UpperCAmelCase	, UpperCAmelCase  )  )

                  A_	     =							outputs.hidden_states
                  A_	     =							self.model_tester.num_hidden_layers + 1
                  self.assertEqual(len(UpperCAmelCase  )	, UpperCAmelCase  )

                  A_	     =							self.model_tester.seq_length

                  self.assertListEqual(
                      list(hidden_states[0].shape[-2:]  )	, [seq_length, self.model_tester.hidden_size]	, )

            A_     ,    A_	     =							self.model_tester.prepare_config_and_inputs_for_common()

            for model_class in self.all_model_classes:
                  A_	     =							True
                  check_hidden_states_output(UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase  )

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

                  check_hidden_states_output(UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase  )







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




def 			__snake_case	(   ):



      """simple docstring"""

      A_	     =							hf_hub_download(
          repo_id="hf-internal-testing/spaghetti-video"							,filename="eating_spaghetti_32_frames.npy"							,repo_type="dataset"						)
      A_	     =							np.load(__UpperCamelCase						)
      return list(__UpperCamelCase						)



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




      """simple docstring"""

      @cached_property
      def 		__A  (      self       :							int  ):
            return VivitImageProcessor() if is_vision_available() else None







      @slow
      def 		__A  (      self       :							Union[str, Any]  ):
            A_	     =							VivitForVideoClassification.from_pretrained("google/vivit-b-16x2-kinetics400"  ).to(UpperCAmelCase  )

            A_	     =							self.default_image_processor
            A_	     =							prepare_video()
            A_	     =							image_processor(UpperCAmelCase	, return_tensors="pt"  ).to(UpperCAmelCase  )

            # forward pass
            with torch.no_grad():
                  A_	     =							model(**UpperCAmelCase  )

            # verify the logits
            A_	     =							torch.Size((1, 400)  )
            self.assertEqual(outputs.logits.shape	, UpperCAmelCase  )

            # taken from original model
            A_	     =							torch.tensor([-1.0_543, 2.0_764, -0.2_104, 0.4_439, -0.9_658]  ).to(UpperCAmelCase  )

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