




'''simple docstring'''







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       ,__a       ,__a=2       ,__a=True       ,__a=True       ,__a=10       ,__a=10       ,__a=8       ,__a=[2, 4, 4]       ,__a=3       ,__a=768       ,__a=5       ,__a=4       ,__a=37       ,__a="gelu_fast"       ,__a=0.0       ,__a=0.0       ,__a=0.02       ,__a=1E-06       ,__a=True       ,__a=None       ,)			->							Tuple:
          snake_case :			Tuple					   = parent
          snake_case :			List[Any]					   = batch_size
          snake_case :			Dict					   = is_training
          snake_case :			Optional[Any]					   = use_labels
          snake_case :			Dict					   = num_labels
          snake_case :			str					   = image_size
          snake_case :			Optional[Any]					   = num_frames
          snake_case :			str					   = tubelet_size
          snake_case :			Optional[Any]					   = num_channels
          snake_case :			List[str]					   = hidden_size
          snake_case :			Optional[int]					   = num_hidden_layers
          snake_case :			Dict					   = num_attention_heads
          snake_case :			str					   = intermediate_size
          snake_case :			int					   = hidden_act
          snake_case :			Tuple					   = hidden_dropout_prob
          snake_case :			str					   = attention_probs_dropout_prob
          snake_case :			str					   = initializer_range
          snake_case :			List[Any]					   = layer_norm_eps
          snake_case :			Tuple					   = qkv_bias
          snake_case :			Optional[int]					   = scope

          snake_case :			str					   = (
              (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  snake_case_			(							self       )			->							Optional[Any]:
          snake_case :			Dict					   = floats_tensor(
              [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]       )

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

          snake_case :			List[Any]					   = self.get_config()

          return config, pixel_values, labels







     def  snake_case_			(							self       )			->							List[str]:
          snake_case :			str					   = 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       ,)
          snake_case :			Any					   = self.num_labels
          return config







     def  snake_case_			(							self       ,__a       ,__a       ,__a       )			->							List[str]:
          snake_case :			Union[str, Any]					   = VivitModel(config=__a       )
          model.to(__a       )
          model.eval()
          snake_case :			Optional[int]					   = model(__a       )
          self.parent.assertEqual(result.last_hidden_state.shape       ,(self.batch_size, self.seq_length, self.hidden_size)       )







     def  snake_case_			(							self       ,__a       ,__a       ,__a       )			->							Union[str, Any]:
          snake_case :			int					   = VivitForVideoClassification(__a       )
          model.to(__a       )
          model.eval()

          snake_case :			str					   = model(__a       )

          # verify the logits shape
          snake_case :			List[Any]					   = torch.Size((self.batch_size, self.num_labels)       )
          self.parent.assertEqual(result.logits.shape       ,__a       )







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







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





     '''simple docstring'''



     lowerCAmelCase_     :    List[str]									=				(VivitModel, VivitForVideoClassification) if is_torch_available() else ()

     lowerCAmelCase_     :    List[Any]									=				False
     lowerCAmelCase_     :    Optional[Any]									=				False
     lowerCAmelCase_     :    Dict									=				False
     lowerCAmelCase_     :    Dict									=				False







     def  snake_case_			(							self       )			->							List[str]:
          snake_case :			Union[str, Any]					   = VivitModelTester(self       )
          snake_case :			Any					   = ConfigTester(self       ,config_class=__a       ,has_text_modality=__a       ,hidden_size=37       )







     def  snake_case_			(							self       ,__a       ,__a       ,__a=False       )			->							Tuple:
          snake_case :			Any					   = copy.deepcopy(__a       )

          if return_labels:
               if model_class in get_values(__a       ):
                    snake_case :			str					   = torch.zeros(
                        self.model_tester.batch_size       ,dtype=torch.long       ,device=__a       )

          return inputs_dict







     def  snake_case_			(							self       )			->							str:
          self.config_tester.run_common_tests()







     @unittest.skip(reason="""Vivit does not use inputs_embeds"""       )
     def  snake_case_			(							self       )			->							Dict:
          pass







     def  snake_case_			(							self       )			->							int:
          snake_case :			Optional[int]					   = self.model_tester.prepare_config_and_inputs_for_common()

          for model_class in self.all_model_classes:
               snake_case :			Dict					   = model_class(__a       )
               self.assertIsInstance(model.get_input_embeddings()       ,(nn.Module)       )
               snake_case :			Union[str, Any]					   = model.get_output_embeddings()
               self.assertTrue(x is None or isinstance(__a       ,nn.Linear       )       )







     def  snake_case_			(							self       )			->							Tuple:
          snake_case :			Union[str, Any]					   = self.model_tester.prepare_config_and_inputs_for_common()

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

               snake_case :			Optional[int]					   = ["pixel_values", "head_mask"]
               self.assertListEqual(arg_names[:2]       ,__a       )







     def  snake_case_			(							self       )			->							int:
          snake_case :			List[str]					   = self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_model(*__a       )







     def  snake_case_			(							self       )			->							List[Any]:
          snake_case :			Dict					   = self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_for_video_classification(*__a       )







     @slow
     def  snake_case_			(							self       )			->							Union[str, Any]:






          for model_name in VIVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               snake_case :			Dict					   = VivitModel.from_pretrained(__a       )
               self.assertIsNotNone(__a       )







     def  snake_case_			(							self       )			->							Optional[Any]:
          snake_case :			str					   = self.model_tester.prepare_config_and_inputs_for_common()
          snake_case :			Tuple					   = True

          for model_class in self.all_model_classes:
               snake_case :			int					   = self.model_tester.seq_length

               snake_case :			Optional[Any]					   = True
               snake_case :			Optional[int]					   = False
               snake_case :			int					   = True
               snake_case :			Tuple					   = model_class(__a       )
               model.to(__a       )
               model.eval()
               with torch.no_grad():
                    snake_case :			Optional[int]					   = model(**self._prepare_for_class(__a       ,__a       )       )
               snake_case :			Any					   = outputs.attentions
               self.assertEqual(len(__a       )       ,self.model_tester.num_hidden_layers       )

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

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

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

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

               snake_case :			Dict					   = outputs.attentions

               self.assertEqual(len(__a       )       ,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  snake_case_			(							self       )			->							Optional[int]:






          def check_hidden_states_output(__a       ,__a       ,__a       ):
               snake_case :			Optional[int]					   = model_class(__a       )
               model.to(__a       )
               model.eval()

               with torch.no_grad():
                    snake_case :			Dict					   = model(**self._prepare_for_class(__a       ,__a       )       )

               snake_case :			Dict					   = outputs.hidden_states
               snake_case :			int					   = self.model_tester.num_hidden_layers + 1
               self.assertEqual(len(__a       )       ,__a       )

               snake_case :			Union[str, Any]					   = self.model_tester.seq_length

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

          snake_case :			Optional[int]					   = self.model_tester.prepare_config_and_inputs_for_common()

          for model_class in self.all_model_classes:
               snake_case :			Optional[int]					   = True
               check_hidden_states_output(__a       ,__a       ,__a       )

               # check that output_hidden_states also work using config
               del inputs_dict["output_hidden_states"]
               snake_case :			Any					   = True

               check_hidden_states_output(__a       ,__a       ,__a       )







     @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""       )
     def  snake_case_			(							self       )			->							Optional[int]:
          pass



def 					lowerCamelCase__							(			):
     snake_case :			List[str]					   = hf_hub_download(
         repo_id="""hf-internal-testing/spaghetti-video"""    ,      filename="""eating_spaghetti_32_frames.npy"""    ,      repo_type="""dataset"""							)
     snake_case :			Optional[int]					   = np.load(__lowerCAmelCase							)
     return list(__lowerCAmelCase							)







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





     '''simple docstring'''



     @cached_property
     def  snake_case_			(							self       )			->							Optional[int]:
          return VivitImageProcessor() if is_vision_available() else None







     @slow
     def  snake_case_			(							self       )			->							Union[str, Any]:
          snake_case :			Union[str, Any]					   = VivitForVideoClassification.from_pretrained("""google/vivit-b-16x2-kinetics400"""       ).to(__a       )

          snake_case :			List[Any]					   = self.default_image_processor
          snake_case :			List[str]					   = prepare_video()
          snake_case :			Dict					   = image_processor(__a       ,return_tensors="""pt"""       ).to(__a       )

          # forward pass
          with torch.no_grad():
               snake_case :			Optional[int]					   = model(**__a       )

          # verify the logits
          snake_case :			int					   = torch.Size((1, 400)       )
          self.assertEqual(outputs.logits.shape       ,__a       )

          # taken from original model
          snake_case :			Optional[Any]					   = torch.tensor([-1.0_543, 2.0_764, -0.2_104, 0.4_439, -0.9_658]       ).to(__a       )

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









'''simple docstring'''







from __future__ import annotations

from collections.abc import Iterable, Iterator
from typing import Any, Generic, TypeVar

lowercase			:      Union[str, Any]			=					TypeVar("""T""", bound=bool)







class 				_a   (Generic[T]  ):





     '''simple docstring'''



     def __init__(							self       ,__a       )			->							None:
          snake_case :			T					   = value
          snake_case :			SkewNode[T] | None					   = None
          snake_case :			SkewNode[T] | None					   = None







     @property
     def  snake_case_			(							self       )			->							T:
          return self._value







     @staticmethod
     def  snake_case_			(							__a       ,__a       )			->							SkewNode[T] | None:
          if not roota:
               return roota

          if not roota:
               return roota

          if roota.value > roota.value:
               snake_case   ,		snake_case :			Tuple					   = roota, roota

          snake_case :			Tuple					   = roota
          snake_case :			Union[str, Any]					   = roota.right
          snake_case :			Any					   = roota.left
          snake_case :			Union[str, Any]					   = SkewNode.merge(__a       ,__a       )

          return result







class 				_a   (Generic[T]  ):





     '''simple docstring'''



     def __init__(							self       ,__a = ()       )			->							None:
          snake_case :			SkewNode[T] | None					   = None
          if data:
               for item in data:
                    self.insert(__a       )







     def __bool__(							self       )			->							bool:
          return self._root is not None







     def __iter__(							self       )			->							Iterator[T]:
          snake_case :			list[Any]					   = []
          while self:
               result.append(self.pop()       )

          # Pushing items back to the heap not to clear it.
          for item in result:
               self.insert(__a       )

          return iter(__a       )







     def  snake_case_			(							self       ,__a       )			->							None:
          snake_case :			List[Any]					   = SkewNode.merge(self._root       ,SkewNode(__a       )       )







     def  snake_case_			(							self       )			->							T | None:
          snake_case :			Tuple					   = self.top()
          snake_case :			int					   = (
              SkewNode.merge(self._root.left       ,self._root.right       ) if self._root else None
          )

          return result







     def  snake_case_			(							self       )			->							T:
          if not self._root:
               raise IndexError("""Can't get top element for the empty heap."""       )
          return self._root.value







     def  snake_case_			(							self       )			->							None:
          snake_case :			Any					   = None


if __name__ == "__main__":
      import doctest

      doctest.testmod()



