


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






  """simple docstring"""





  def __init__(  self							:							Tuple     ,					_lowerCAmelCase							:							Any     ,					_lowerCAmelCase							:							List[str]=1_3     ,					_lowerCAmelCase							:							Tuple=1_0     ,					_lowerCAmelCase							:							Optional[Any]=3     ,					_lowerCAmelCase							:							str=2     ,					_lowerCAmelCase							:							Optional[Any]=2     ,					_lowerCAmelCase							:							List[str]=True     ,					_lowerCAmelCase							:							List[str]=True     ,					_lowerCAmelCase							:							Union[str, Any]=3_2     ,					_lowerCAmelCase							:							List[str]=5     ,					_lowerCAmelCase							:							int=4     ,					_lowerCAmelCase							:							Union[str, Any]=3_7     ,					_lowerCAmelCase							:							Optional[int]="gelu"     ,					_lowerCAmelCase							:							Union[str, Any]=0.1     ,					_lowerCAmelCase							:							List[Any]=0.1     ,					_lowerCAmelCase							:							int=1_0     ,					_lowerCAmelCase							:							Optional[int]=0.02     ,					_lowerCAmelCase							:							int="divided_space_time"     ,					_lowerCAmelCase							:							Any=None     ,					) ->			Any:




     """simple docstring"""
     snake_case_            =       parent
     snake_case_            =       batch_size
     snake_case_            =       image_size
     snake_case_            =       num_channels
     snake_case_            =       patch_size
     snake_case_            =       num_frames
     snake_case_            =       is_training
     snake_case_            =       use_labels
     snake_case_            =       hidden_size
     snake_case_            =       num_hidden_layers
     snake_case_            =       num_attention_heads
     snake_case_            =       intermediate_size
     snake_case_            =       hidden_act
     snake_case_            =       hidden_dropout_prob
     snake_case_            =       attention_probs_dropout_prob
     snake_case_            =       attention_type
     snake_case_            =       initializer_range
     snake_case_            =       scope
     snake_case_            =       num_labels

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



  def        lowerCAmelCase__	(  self							:							int	) ->			Any:




     """simple docstring"""
     snake_case_            =       floats_tensor(
         [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]	)

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

     snake_case_            =       self.get_config()

     return config, pixel_values, labels



  def        lowerCAmelCase__	(  self							:							Union[str, Any]	) ->			List[Any]:




     """simple docstring"""
     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     ,					)
     snake_case_            =       self.num_labels
     return config



  def        lowerCAmelCase__	(  self							:							Optional[Any]     ,					_lowerCAmelCase							:							Tuple     ,					_lowerCAmelCase							:							Any     ,					_lowerCAmelCase							:							Optional[int]	) ->			str:




     """simple docstring"""
     snake_case_            =       TimesformerModel(config=_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()
     snake_case_            =       model(_lowerCAmelCase	)
     self.parent.assertEqual(result.last_hidden_state.shape     ,					(self.batch_size, self.seq_length, self.hidden_size)	)



  def        lowerCAmelCase__	(  self							:							Tuple     ,					_lowerCAmelCase							:							str     ,					_lowerCAmelCase							:							Optional[Any]     ,					_lowerCAmelCase							:							Union[str, Any]	) ->			List[Any]:




     """simple docstring"""
     snake_case_            =       TimesformerForVideoClassification(_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()

     snake_case_            =       model(_lowerCAmelCase	)

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



  def        lowerCAmelCase__	(  self							:							List[Any]	) ->			str:




     """simple docstring"""
     snake_case_            =       self.prepare_config_and_inputs()
     snake_case_     ,     snake_case_     ,     snake_case_            =       config_and_inputs
     snake_case_            =       {"pixel_values": pixel_values}
     return config, inputs_dict






@require_torch
class      __lowerCAmelCase  (			a			,	a			,	unittest.TestCase       ):






  """simple docstring"""





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

  _SCREAMING_SNAKE_CASE		     =							False
  _SCREAMING_SNAKE_CASE		     =							False
  _SCREAMING_SNAKE_CASE		     =							False
  _SCREAMING_SNAKE_CASE		     =							False



  def        lowerCAmelCase__	(  self							:							Union[str, Any]	) ->			str:




     """simple docstring"""
     snake_case_            =       TimesformerModelTester(self	)
     snake_case_            =       ConfigTester(
         self     ,					config_class=_lowerCAmelCase     ,					has_text_modality=_lowerCAmelCase     ,					hidden_size=3_7	)



  def        lowerCAmelCase__	(  self							:							Optional[Any]     ,					_lowerCAmelCase							:							Dict     ,					_lowerCAmelCase							:							List[Any]     ,					_lowerCAmelCase							:							Tuple=False	) ->			int:




     """simple docstring"""
     snake_case_            =       copy.deepcopy(_lowerCAmelCase	)

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

     return inputs_dict



  def        lowerCAmelCase__	(  self							:							Tuple	) ->			str:




     """simple docstring"""
     self.config_tester.run_common_tests()



  @unittest.skip(reason="TimeSformer does not use inputs_embeds"	)
  def        lowerCAmelCase__	(  self							:							int	) ->			Dict:




     """simple docstring"""
     pass



  def        lowerCAmelCase__	(  self							:							int	) ->			Tuple:




     """simple docstring"""
     snake_case_     ,     snake_case_            =       self.model_tester.prepare_config_and_inputs_for_common()

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



  def        lowerCAmelCase__	(  self							:							Optional[Any]	) ->			str:




     """simple docstring"""
     snake_case_     ,     snake_case_            =       self.model_tester.prepare_config_and_inputs_for_common()

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

        snake_case_            =       ["pixel_values"]
        self.assertListEqual(arg_names[:1]     ,					_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							Optional[int]	) ->			Dict:




     """simple docstring"""
     snake_case_            =       self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							Dict	) ->			Optional[int]:




     """simple docstring"""
     snake_case_            =       self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_video_classification(*_lowerCAmelCase	)



  @slow
  def        lowerCAmelCase__	(  self							:							Tuple	) ->			int:




     """simple docstring"""
     for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        snake_case_            =       TimesformerModel.from_pretrained(_lowerCAmelCase	)
        self.assertIsNotNone(_lowerCAmelCase	)



  def        lowerCAmelCase__	(  self							:							List[str]	) ->			int:




     """simple docstring"""
     if not self.has_attentions:
        pass

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

        for model_class in self.all_model_classes:
           snake_case_            =       self.model_tester.seq_length
           snake_case_            =       self.model_tester.num_frames

           snake_case_            =       True
           snake_case_            =       False
           snake_case_            =       True
           snake_case_            =       model_class(_lowerCAmelCase	)
           model.to(_lowerCAmelCase	)
           model.eval()
           with torch.no_grad():
              snake_case_            =       model(**self._prepare_for_class(_lowerCAmelCase     ,					_lowerCAmelCase	)	)
           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"]
           snake_case_            =       True
           snake_case_            =       model_class(_lowerCAmelCase	)
           model.to(_lowerCAmelCase	)
           model.eval()
           with torch.no_grad():
              snake_case_            =       model(**self._prepare_for_class(_lowerCAmelCase     ,					_lowerCAmelCase	)	)
           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]     ,					)
           snake_case_            =       len(_lowerCAmelCase	)

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

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

           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        lowerCAmelCase__	(  self							:							Union[str, Any]	) ->			List[str]:




     """simple docstring"""
     def check_hidden_states_output(_lowerCAmelCase							:							str     ,					_lowerCAmelCase							:							Optional[int]     ,					_lowerCAmelCase							:							str	):
        snake_case_            =       model_class(_lowerCAmelCase	)
        model.to(_lowerCAmelCase	)
        model.eval()

        with torch.no_grad():
           snake_case_            =       model(**self._prepare_for_class(_lowerCAmelCase     ,					_lowerCAmelCase	)	)

        snake_case_            =       outputs.hidden_states
        snake_case_            =       self.model_tester.num_hidden_layers + 1
        self.assertEqual(len(_lowerCAmelCase	)     ,					_lowerCAmelCase	)

        snake_case_            =       self.model_tester.seq_length

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

     snake_case_     ,     snake_case_            =       self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        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"]
        snake_case_            =       True

        check_hidden_states_output(_lowerCAmelCase     ,					_lowerCAmelCase     ,					_lowerCAmelCase	)






def       _lowerCAmelCase    (				)->Union[str, Any]:
   '''simple docstring'''






   snake_case_            =       hf_hub_download(
       repo_id="hf-internal-testing/spaghetti-video"		,    filename="eating_spaghetti.npy"		,    repo_type="dataset"				)
   snake_case_            =       np.load(lowerCAmelCase_				)
   return list(lowerCAmelCase_				)



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






  """simple docstring"""





  @cached_property
  def        lowerCAmelCase__	(  self							:							int	) ->			List[str]:




     """simple docstring"""
     # 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        lowerCAmelCase__	(  self							:							Any	) ->			List[Any]:




     """simple docstring"""
     snake_case_            =       TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"	).to(
         _lowerCAmelCase	)

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

     # forward pass
     with torch.no_grad():
        snake_case_            =       model(**_lowerCAmelCase	)

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

     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	)	)








def       _lowerCAmelCase    (				lowerCAmelCase_				:int		,    lowerCAmelCase_				:int				)->int:
   '''simple docstring'''






   return int((input_a, input_a).count(0				) == 0				)



def       _lowerCAmelCase    (				)->None:
   '''simple docstring'''






   assert and_gate(0		,    0				) == 0
   assert and_gate(0		,    1				) == 0
   assert and_gate(1		,    0				) == 0
   assert and_gate(1		,    1				) == 1


if __name__ == "__main__":
   test_and_gate()
   print(and_gate(1, 0))
   print(and_gate(0, 0))
   print(and_gate(0, 1))
   print(and_gate(1, 1))




