



'''simple docstring'''





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    ,__UpperCAmelCase    ,__UpperCAmelCase=13    ,__UpperCAmelCase=10    ,__UpperCAmelCase=3    ,__UpperCAmelCase=2    ,__UpperCAmelCase=2    ,__UpperCAmelCase=True    ,__UpperCAmelCase=True    ,__UpperCAmelCase=32    ,__UpperCAmelCase=5    ,__UpperCAmelCase=4    ,__UpperCAmelCase=37    ,__UpperCAmelCase="gelu"    ,__UpperCAmelCase=0.1    ,__UpperCAmelCase=0.1    ,__UpperCAmelCase=10    ,__UpperCAmelCase=0.0_2    ,__UpperCAmelCase="divided_space_time"    ,__UpperCAmelCase=None    ,)       ->						Any:
      lowerCAmelCase__					:    int				 =   parent
      lowerCAmelCase__					:    List[str]				 =   batch_size
      lowerCAmelCase__					:    Any				 =   image_size
      lowerCAmelCase__					:    List[str]				 =   num_channels
      lowerCAmelCase__					:    Optional[Any]				 =   patch_size
      lowerCAmelCase__					:    Tuple				 =   num_frames
      lowerCAmelCase__					:    List[Any]				 =   is_training
      lowerCAmelCase__					:    List[Any]				 =   use_labels
      lowerCAmelCase__					:    int				 =   hidden_size
      lowerCAmelCase__					:    Any				 =   num_hidden_layers
      lowerCAmelCase__					:    int				 =   num_attention_heads
      lowerCAmelCase__					:    List[Any]				 =   intermediate_size
      lowerCAmelCase__					:    Optional[Any]				 =   hidden_act
      lowerCAmelCase__					:    Tuple				 =   hidden_dropout_prob
      lowerCAmelCase__					:    List[str]				 =   attention_probs_dropout_prob
      lowerCAmelCase__					:    List[Any]				 =   attention_type
      lowerCAmelCase__					:    List[str]				 =   initializer_range
      lowerCAmelCase__					:    Any				 =   scope
      lowerCAmelCase__					:    Dict				 =   num_labels

      # in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
      lowerCAmelCase__					:    List[Any]				 =   (image_size // patch_size) ** 2
      lowerCAmelCase__					:    Tuple				 =   (num_frames) * self.num_patches_per_frame + 1




  def 				UpperCAmelCase_      (       self )       ->						Any:
      lowerCAmelCase__					:    int				 =   floats_tensor(
          [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size] )

      lowerCAmelCase__					:    int				 =   None
      if self.use_labels:
          lowerCAmelCase__					:    List[Any]				 =   ids_tensor([self.batch_size]    ,self.num_labels )

      lowerCAmelCase__					:    Any				 =   self.get_config()

      return config, pixel_values, labels




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




  def 				UpperCAmelCase_      (       self    ,__UpperCAmelCase    ,__UpperCAmelCase    ,__UpperCAmelCase )       ->						str:
      lowerCAmelCase__					:    Tuple				 =   TimesformerModel(config=_lowerCAmelCase )
      model.to(_lowerCAmelCase )
      model.eval()
      lowerCAmelCase__					:    List[str]				 =   model(_lowerCAmelCase )
      self.parent.assertEqual(result.last_hidden_state.shape    ,(self.batch_size, self.seq_length, self.hidden_size) )




  def 				UpperCAmelCase_      (       self    ,__UpperCAmelCase    ,__UpperCAmelCase    ,__UpperCAmelCase )       ->						List[Any]:
      lowerCAmelCase__					:    List[Any]				 =   TimesformerForVideoClassification(_lowerCAmelCase )
      model.to(_lowerCAmelCase )
      model.eval()

      lowerCAmelCase__					:    Optional[int]				 =   model(_lowerCAmelCase )

      # verify the logits shape
      lowerCAmelCase__					:    Optional[Any]				 =   torch.Size((self.batch_size, self.num_labels) )
      self.parent.assertEqual(result.logits.shape    ,_lowerCAmelCase )






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







@require_torch
class 			lowerCAmelCase_(							SCREAMING_SNAKE_CASE_		,					SCREAMING_SNAKE_CASE_		,					unittest.TestCase	):
  '''simple docstring'''




  __lowercase				: int           = (TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
  __lowercase				: List[Any]           = (
      {'''feature-extraction''': TimesformerModel, '''video-classification''': TimesformerForVideoClassification}
      if is_torch_available()
      else {}
  )

  __lowercase				: Optional[int]           = False
  __lowercase				: Any           = False
  __lowercase				: List[Any]           = False
  __lowercase				: Any           = False




  def 				UpperCAmelCase_      (       self )       ->						str:
      lowerCAmelCase__					:    List[str]				 =   TimesformerModelTester(self )
      lowerCAmelCase__					:    str				 =   ConfigTester(
          self    ,config_class=_lowerCAmelCase    ,has_text_modality=_lowerCAmelCase    ,hidden_size=37 )




  def 				UpperCAmelCase_      (       self    ,__UpperCAmelCase    ,__UpperCAmelCase    ,__UpperCAmelCase=False )       ->						int:
      lowerCAmelCase__					:    List[Any]				 =   copy.deepcopy(_lowerCAmelCase )

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

      return inputs_dict




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




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




  def 				UpperCAmelCase_      (       self )       ->						Tuple:
      lowerCAmelCase__ ,  lowerCAmelCase__					:    List[str]				 =   self.model_tester.prepare_config_and_inputs_for_common()

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




  def 				UpperCAmelCase_      (       self )       ->						str:
      lowerCAmelCase__ ,  lowerCAmelCase__					:    List[Any]				 =   self.model_tester.prepare_config_and_inputs_for_common()

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

          lowerCAmelCase__					:    List[str]				 =   ["""pixel_values"""]
          self.assertListEqual(arg_names[:1]    ,_lowerCAmelCase )




  def 				UpperCAmelCase_      (       self )       ->						Dict:
      lowerCAmelCase__					:    List[Any]				 =   self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*_lowerCAmelCase )




  def 				UpperCAmelCase_      (       self )       ->						Optional[int]:
      lowerCAmelCase__					:    Dict				 =   self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_video_classification(*_lowerCAmelCase )




  @slow
  def 				UpperCAmelCase_      (       self )       ->						int:
      for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          lowerCAmelCase__					:    Optional[int]				 =   TimesformerModel.from_pretrained(_lowerCAmelCase )
          self.assertIsNotNone(_lowerCAmelCase )




  def 				UpperCAmelCase_      (       self )       ->						int:
      if not self.has_attentions:
          pass

      else:
          lowerCAmelCase__ ,  lowerCAmelCase__					:    Union[str, Any]				 =   self.model_tester.prepare_config_and_inputs_for_common()
          lowerCAmelCase__					:    str				 =   True

          for model_class in self.all_model_classes:
              lowerCAmelCase__					:    Tuple				 =   self.model_tester.seq_length
              lowerCAmelCase__					:    Optional[Any]				 =   self.model_tester.num_frames

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

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

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

              lowerCAmelCase__					:    Tuple				 =   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 				UpperCAmelCase_      (       self )       ->						List[str]:
      def check_hidden_states_output(__UpperCAmelCase    ,__UpperCAmelCase    ,__UpperCAmelCase ):
          lowerCAmelCase__					:    Optional[int]				 =   model_class(_lowerCAmelCase )
          model.to(_lowerCAmelCase )
          model.eval()

          with torch.no_grad():
              lowerCAmelCase__					:    List[str]				 =   model(**self._prepare_for_class(_lowerCAmelCase    ,_lowerCAmelCase ) )

          lowerCAmelCase__					:    List[str]				 =   outputs.hidden_states
          lowerCAmelCase__					:    Optional[Any]				 =   self.model_tester.num_hidden_layers + 1
          self.assertEqual(len(_lowerCAmelCase )    ,_lowerCAmelCase )

          lowerCAmelCase__					:    List[Any]				 =   self.model_tester.seq_length

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

      lowerCAmelCase__ ,  lowerCAmelCase__					:    Optional[Any]				 =   self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
          lowerCAmelCase__					:    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__					:    Optional[Any]				 =   True

          check_hidden_states_output(_lowerCAmelCase    ,_lowerCAmelCase    ,_lowerCAmelCase )







def   _SCREAMING_SNAKE_CASE    (			):



    """simple docstring"""





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






@require_torch
@require_vision
class 			lowerCAmelCase_(							unittest.TestCase	):
  '''simple docstring'''




  @cached_property
  def 				UpperCAmelCase_      (       self )       ->						List[str]:
      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 				UpperCAmelCase_      (       self )       ->						List[Any]:
      lowerCAmelCase__					:    Optional[int]				 =   TimesformerForVideoClassification.from_pretrained("""facebook/timesformer-base-finetuned-k400""" ).to(
          _lowerCAmelCase )

      lowerCAmelCase__					:    Dict				 =   self.default_image_processor
      lowerCAmelCase__					:    Tuple				 =   prepare_video()
      lowerCAmelCase__					:    List[str]				 =   image_processor(video[:8]    ,return_tensors="""pt""" ).to(_lowerCAmelCase )

      # forward pass
      with torch.no_grad():
          lowerCAmelCase__					:    Optional[Any]				 =   model(**_lowerCAmelCase )

      # verify the logits
      lowerCAmelCase__					:    List[Any]				 =   torch.Size((1, 400) )
      self.assertEqual(outputs.logits.shape    ,_lowerCAmelCase )

      lowerCAmelCase__					:    Union[str, Any]				 =   torch.tensor([-0.3_0_1_6, -0.7_7_1_3, -0.4_2_0_5] ).to(_lowerCAmelCase )

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






from __future__ import annotations



def       _lowerCAmelCase    (				lowerCAmelCase_				:int		,    lowerCAmelCase_				:int				)->list[str]:
   '''simple docstring'''






   if partitions <= 0:
      raise ValueError("partitions must be a positive number!"				)
   if partitions > number_of_bytes:
      raise ValueError("partitions can not > number_of_bytes!"				)
   snake_case_            =       number_of_bytes // partitions
   snake_case_            =       []
   for i in range(lowerCAmelCase_				):
      snake_case_            =       i * bytes_per_partition + 1
      snake_case_            =       (
          number_of_bytes if i == partitions - 1 else (i + 1) * bytes_per_partition
      )
      allocation_list.append(F'''{start_bytes}-{end_bytes}'''				)
   return allocation_list


if __name__ == "__main__":
   import doctest

   doctest.testmod()




