

import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig
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,
        TimesformerForVideoClassification,
        TimesformerModel,
    )
    from transformers.models.timesformer.modeling_timesformer import TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
    from transformers import VideoMAEImageProcessor





class 			_SCREAMING_SNAKE_CASE :







   '''simple docstring'''




   def __init__(       self					:			List[str]     , __lowerCamelCase					:			Union[str, Any]     , __lowerCamelCase					:			str=13     , __lowerCamelCase					:			List[Any]=10     , __lowerCamelCase					:			Optional[Any]=3     , __lowerCamelCase					:			int=2     , __lowerCamelCase					:			List[str]=2     , __lowerCamelCase					:			List[str]=True     , __lowerCamelCase					:			Optional[int]=True     , __lowerCamelCase					:			List[str]=32     , __lowerCamelCase					:			Dict=5     , __lowerCamelCase					:			Union[str, Any]=4     , __lowerCamelCase					:			List[Any]=37     , __lowerCamelCase					:			Tuple="gelu"     , __lowerCamelCase					:			Any=0.1     , __lowerCamelCase					:			Tuple=0.1     , __lowerCamelCase					:			Any=10     , __lowerCamelCase					:			List[Any]=0.02     , __lowerCamelCase					:			List[str]="divided_space_time"     , __lowerCamelCase					:			List[Any]=None     , ):
       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										=   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										=   attention_type
       SCREAMING_SNAKE_CASE										=   initializer_range
       SCREAMING_SNAKE_CASE										=   scope
       SCREAMING_SNAKE_CASE										=   num_labels

       # in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
       SCREAMING_SNAKE_CASE										=   (image_size // patch_size) ** 2
       SCREAMING_SNAKE_CASE										=   (num_frames) * self.num_patches_per_frame + 1





   def      _snake_case		(       self					:			int			):
       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.num_labels			)

       SCREAMING_SNAKE_CASE										=   self.get_config()

       return config, pixel_values, labels





   def      _snake_case		(       self					:			Dict			):
       SCREAMING_SNAKE_CASE										=   TimesformerConfig(
           image_size=self.image_size     , patch_size=self.patch_size     , num_channels=self.num_channels     , num_frames=self.num_frames     , 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     , attention_type=self.attention_type     , )
       SCREAMING_SNAKE_CASE										=   self.num_labels
       return config





   def      _snake_case		(       self					:			str     , __lowerCamelCase					:			Optional[int]     , __lowerCamelCase					:			Tuple     , __lowerCamelCase					:			Optional[Any]			):
       SCREAMING_SNAKE_CASE										=   TimesformerModel(config=__lowerCamelCase			)
       model.to(__lowerCamelCase			)
       model.eval()
       SCREAMING_SNAKE_CASE										=   model(__lowerCamelCase			)
       self.parent.assertEqual(result.last_hidden_state.shape     , (self.batch_size, self.seq_length, self.hidden_size)			)





   def      _snake_case		(       self					:			Optional[int]     , __lowerCamelCase					:			Union[str, Any]     , __lowerCamelCase					:			List[Any]     , __lowerCamelCase					:			Optional[Any]			):
       SCREAMING_SNAKE_CASE										=   TimesformerForVideoClassification(__lowerCamelCase			)
       model.to(__lowerCamelCase			)
       model.eval()

       SCREAMING_SNAKE_CASE										=   model(__lowerCamelCase			)

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





   def      _snake_case		(       self					:			Tuple			):
       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 			_SCREAMING_SNAKE_CASE (   __snake_case					,       __snake_case					,       unittest.TestCase		):







   '''simple docstring'''




   lowerCamelCase__				=      (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
   lowerCamelCase__				=      (
       {"feature-extraction": TimesformerModel, "video-classification": TimesformerForVideoClassification}
       if is_torch_available()
       else {}
   )

   lowerCamelCase__				=      False
   lowerCamelCase__				=      False
   lowerCamelCase__				=      False
   lowerCamelCase__				=      False





   def      _snake_case		(       self					:			Optional[Any]			):
       SCREAMING_SNAKE_CASE										=   TimesformerModelTester(self			)
       SCREAMING_SNAKE_CASE										=   ConfigTester(
           self     , config_class=__lowerCamelCase     , has_text_modality=__lowerCamelCase     , hidden_size=37			)





   def      _snake_case		(       self					:			Union[str, Any]     , __lowerCamelCase					:			Any     , __lowerCamelCase					:			str     , __lowerCamelCase					:			str=False			):
       SCREAMING_SNAKE_CASE										=   copy.deepcopy(__lowerCamelCase			)

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

       return inputs_dict





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





   @unittest.skip(reason="TimeSformer does not use inputs_embeds"			)
   def      _snake_case		(       self					:			Optional[Any]			):
       pass





   def      _snake_case		(       self					:			int			):
       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(__lowerCamelCase			)
           self.assertIsInstance(model.get_input_embeddings()     , (nn.Module)			)
           SCREAMING_SNAKE_CASE										=   model.get_output_embeddings()
           self.assertTrue(x is None or isinstance(__lowerCamelCase     , nn.Linear			)			)





   def      _snake_case		(       self					:			Optional[int]			):
       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(__lowerCamelCase			)
           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]     , __lowerCamelCase			)





   def      _snake_case		(       self					:			Any			):
       SCREAMING_SNAKE_CASE										=   self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*__lowerCamelCase			)





   def      _snake_case		(       self					:			Dict			):
       SCREAMING_SNAKE_CASE										=   self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_video_classification(*__lowerCamelCase			)





   @slow
   def      _snake_case		(       self					:			str			):
       for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           SCREAMING_SNAKE_CASE										=   TimesformerModel.from_pretrained(__lowerCamelCase			)
           self.assertIsNotNone(__lowerCamelCase			)





   def      _snake_case		(       self					:			List[Any]			):
       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
               SCREAMING_SNAKE_CASE										=   self.model_tester.num_frames

               SCREAMING_SNAKE_CASE										=   True
               SCREAMING_SNAKE_CASE										=   False
               SCREAMING_SNAKE_CASE										=   True
               SCREAMING_SNAKE_CASE										=   model_class(__lowerCamelCase			)
               model.to(__lowerCamelCase			)
               model.eval()
               with torch.no_grad():
                   SCREAMING_SNAKE_CASE										=   model(**self._prepare_for_class(__lowerCamelCase     , __lowerCamelCase			)			)
               SCREAMING_SNAKE_CASE										=   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"]
               SCREAMING_SNAKE_CASE										=   True
               SCREAMING_SNAKE_CASE										=   model_class(__lowerCamelCase			)
               model.to(__lowerCamelCase			)
               model.eval()
               with torch.no_grad():
                   SCREAMING_SNAKE_CASE										=   model(**self._prepare_for_class(__lowerCamelCase     , __lowerCamelCase			)			)
               SCREAMING_SNAKE_CASE										=   outputs.attentions
               self.assertEqual(len(__lowerCamelCase			)     , self.model_tester.num_hidden_layers			)

               # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
               self.assertListEqual(
                   list(attentions[0].shape[-3:]			)     , [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]     , )
               SCREAMING_SNAKE_CASE										=   len(__lowerCamelCase			)

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

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

               SCREAMING_SNAKE_CASE										=   outputs.attentions

               self.assertEqual(len(__lowerCamelCase			)     , self.model_tester.num_hidden_layers			)

               # attentions has shape (batch_size x num_frames) x num_heads x (num_patches per frame + 1) x (num_patches per frame + 1)
               self.assertListEqual(
                   list(self_attentions[0].shape[-3:]			)     , [self.model_tester.num_attention_heads, seq_len // num_frames + 1, seq_len // num_frames + 1]     , )





   def      _snake_case		(       self					:			Any			):
       def check_hidden_states_output(__lowerCamelCase					:			Optional[Any]     , __lowerCamelCase					:			Optional[Any]     , __lowerCamelCase					:			Dict			):
           SCREAMING_SNAKE_CASE										=   model_class(__lowerCamelCase			)
           model.to(__lowerCamelCase			)
           model.eval()

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

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

           SCREAMING_SNAKE_CASE										=   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(__lowerCamelCase     , __lowerCamelCase     , __lowerCamelCase			)

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

           check_hidden_states_output(__lowerCamelCase     , __lowerCamelCase     , __lowerCamelCase			)





def 	__a     (      ):
    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(A__       )
    return list(A__       )





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







   '''simple docstring'''




   @cached_property
   def      _snake_case		(       self					:			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      _snake_case		(       self					:			Any			):
       SCREAMING_SNAKE_CASE										=   TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"			).to(
           __lowerCamelCase			)

       SCREAMING_SNAKE_CASE										=   self.default_image_processor
       SCREAMING_SNAKE_CASE										=   prepare_video()
       SCREAMING_SNAKE_CASE										=   image_processor(video[:8]     , return_tensors="pt"			).to(__lowerCamelCase			)

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

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

       SCREAMING_SNAKE_CASE										=   torch.tensor([-0.3_016, -0.7_713, -0.4_205]			).to(__lowerCamelCase			)

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


from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
    import tensorflow as tf


__A			:					Dict						=			logging.get_logger(__name__)





@dataclass
class 			_SCREAMING_SNAKE_CASE (   __snake_case		):







   '''simple docstring'''




   lowerCamelCase__				=      [
       "no_inference",
       "no_cuda",
       "no_tpu",
       "no_speed",
       "no_memory",
       "no_env_print",
       "no_multi_process",
   ]





   def __init__(       self					:			List[Any]     , **__lowerCamelCase					:			Any			):
       for deprecated_arg in self.deprecated_args:
           if deprecated_arg in kwargs:
               SCREAMING_SNAKE_CASE										=   deprecated_arg[3:]
               SCREAMING_SNAKE_CASE										=   not kwargs.pop(__lowerCamelCase			)
               logger.warning(
                   f"{deprecated_arg} is depreciated. Please use --no-{positive_arg} or"
                   f" {positive_arg}={kwargs[positive_arg]}"			)
       SCREAMING_SNAKE_CASE										=   kwargs.pop("tpu_name"     , self.tpu_name			)
       SCREAMING_SNAKE_CASE										=   kwargs.pop("device_idx"     , self.device_idx			)
       SCREAMING_SNAKE_CASE										=   kwargs.pop("eager_mode"     , self.eager_mode			)
       SCREAMING_SNAKE_CASE										=   kwargs.pop("use_xla"     , self.use_xla			)
       super().__init__(**__lowerCamelCase			)

   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={"help": "Name of TPU"}					,       )
   lowerCamelCase__				=      field(
       default=0					,       metadata={"help": "CPU / GPU device index. Defaults to 0."}					,       )
   lowerCamelCase__				=      field(default=__snake_case					,       metadata={"help": "Benchmark models in eager model."}		)
   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={
           "help": "Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`."
       }					,       )





   @cached_property
   def      _snake_case		(       self					:			Optional[int]			):
       requires_backends(self     , ["tf"]			)
       SCREAMING_SNAKE_CASE										=   None
       if self.tpu:
           try:
               if self.tpu_name:
                   SCREAMING_SNAKE_CASE										=   tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name			)
               else:
                   SCREAMING_SNAKE_CASE										=   tf.distribute.cluster_resolver.TPUClusterResolver()
           except ValueError:
               SCREAMING_SNAKE_CASE										=   None
       return tpu





   @cached_property
   def      _snake_case		(       self					:			Any			):
       requires_backends(self     , ["tf"]			)
       if self.is_tpu:
           tf.config.experimental_connect_to_cluster(self._setup_tpu			)
           tf.tpu.experimental.initialize_tpu_system(self._setup_tpu			)

           SCREAMING_SNAKE_CASE										=   tf.distribute.TPUStrategy(self._setup_tpu			)
       else:
           # currently no multi gpu is allowed
           if self.is_gpu:
               # TODO: Currently only single GPU is supported
               tf.config.set_visible_devices(self.gpu_list[self.device_idx]     , "GPU"			)
               SCREAMING_SNAKE_CASE										=   tf.distribute.OneDeviceStrategy(device=f"/gpu:{self.device_idx}"			)
           else:
               tf.config.set_visible_devices([]     , "GPU"			)  # disable GPU
               SCREAMING_SNAKE_CASE										=   tf.distribute.OneDeviceStrategy(device=f"/cpu:{self.device_idx}"			)

       return strategy





   @property
   def      _snake_case		(       self					:			Any			):
       requires_backends(self     , ["tf"]			)
       return self._setup_tpu is not None





   @property
   def      _snake_case		(       self					:			Optional[Any]			):
       requires_backends(self     , ["tf"]			)
       return self._setup_strategy





   @property
   def      _snake_case		(       self					:			List[str]			):
       requires_backends(self     , ["tf"]			)
       return tf.config.list_physical_devices("GPU"			)





   @property
   def      _snake_case		(       self					:			Any			):
       requires_backends(self     , ["tf"]			)
       if self.cuda:
           return len(self.gpu_list			)
       return 0





   @property
   def      _snake_case		(       self					:			Dict			):
       return self.n_gpu > 0