



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

   def __init__(      self			:       Dict							,   __lowerCAmelCase			:       Tuple							,   __lowerCAmelCase			:       Union[str, Any]=13							,   __lowerCAmelCase			:       Any=10							,   __lowerCAmelCase			:       Tuple=3							,   __lowerCAmelCase			:       str=2							,   __lowerCAmelCase			:       Any=2							,   __lowerCAmelCase			:       Any=True							,   __lowerCAmelCase			:       List[str]=True							,   __lowerCAmelCase			:       Any=32							,   __lowerCAmelCase			:       Dict=5							,   __lowerCAmelCase			:       Any=4							,   __lowerCAmelCase			:       Any=37							,   __lowerCAmelCase			:       Optional[Any]="gelu"							,   __lowerCAmelCase			:       Optional[int]=0.1							,   __lowerCAmelCase			:       Tuple=0.1							,   __lowerCAmelCase			:       Union[str, Any]=10							,   __lowerCAmelCase			:       Union[str, Any]=0.0_2							,   __lowerCAmelCase			:       Any="divided_space_time"							,   __lowerCAmelCase			:       Union[str, Any]=None							,   )       ->				str:


    """simple docstring"""






    a						=		parent
    a						=		batch_size
    a						=		image_size
    a						=		num_channels
    a						=		patch_size
    a						=		num_frames
    a						=		is_training
    a						=		use_labels
    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						=		attention_type
    a						=		initializer_range
    a						=		scope
    a						=		num_labels

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

   def 		A							(      self			:       str							)       ->				int:


    """simple docstring"""






    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			:       List[Any]							)       ->				List[str]:


    """simple docstring"""






    a						=		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							,   )
    a						=		self.num_labels
    return config

   def 		A							(      self			:       List[Any]							,   __lowerCAmelCase			:       Any							,   __lowerCAmelCase			:       Optional[int]							,   __lowerCAmelCase			:       Optional[int]							)       ->				str:


    """simple docstring"""






    a						=		TimesformerModel(config=__lowerCAmelCase							)
    model.to(__lowerCAmelCase							)
    model.eval()
    a						=		model(__lowerCAmelCase							)
    self.parent.assertEqual(result.last_hidden_state.shape							,   (self.batch_size, self.seq_length, self.hidden_size)							)

   def 		A							(      self			:       str							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       Any							,   __lowerCAmelCase			:       Optional[int]							)       ->				Optional[Any]:


    """simple docstring"""






    a						=		TimesformerForVideoClassification(__lowerCAmelCase							)
    model.to(__lowerCAmelCase							)
    model.eval()

    a						=		model(__lowerCAmelCase							)

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




   def 		A							(      self			:       str							)       ->				Dict:


    """simple docstring"""






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



@require_torch
class 		_lowercase		(       UpperCAmelCase__,   UpperCAmelCase__,   unittest.TestCase	):
   _UpperCAmelCase     	=			(TimesformerModel, TimesformerForVideoClassification) if is_torch_available() else ()
   _UpperCAmelCase     	=			(
       {'''feature-extraction''': TimesformerModel, '''video-classification''': TimesformerForVideoClassification}
       if is_torch_available()
       else {}
   )

   _UpperCAmelCase     	=			False
   _UpperCAmelCase     	=			False
   _UpperCAmelCase     	=			False
   _UpperCAmelCase     	=			False

   def 		A							(      self			:       Optional[int]							)       ->				List[Any]:


    """simple docstring"""






    a						=		TimesformerModelTester(self							)
    a						=		ConfigTester(
        self							,   config_class=__lowerCAmelCase							,   has_text_modality=__lowerCAmelCase							,   hidden_size=37							)

   def 		A							(      self			:       Tuple							,   __lowerCAmelCase			:       Any							,   __lowerCAmelCase			:       List[Any]							,   __lowerCAmelCase			:       Union[str, Any]=False							)       ->				List[str]:


    """simple docstring"""






    a						=		copy.deepcopy(__lowerCAmelCase							)

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

    return inputs_dict

   def 		A							(      self			:       List[Any]							)       ->				Tuple:


    """simple docstring"""






    self.config_tester.run_common_tests()

   @unittest.skip(reason="TimeSformer does not use inputs_embeds"							)
   def 		A							(      self			:       int							)       ->				List[str]:


    """simple docstring"""






    pass

   def 		A							(      self			:       Dict							)       ->				List[Any]:


    """simple docstring"""






    a		,  a						=		self.model_tester.prepare_config_and_inputs_for_common()

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

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


    """simple docstring"""






    a		,  a						=		self.model_tester.prepare_config_and_inputs_for_common()

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

     a						=		["pixel_values"]
     self.assertListEqual(arg_names[:1]							,   __lowerCAmelCase							)

   def 		A							(      self			:       Any							)       ->				Optional[Any]:


    """simple docstring"""






    a						=		self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*__lowerCAmelCase							)

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


    """simple docstring"""






    a						=		self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_video_classification(*__lowerCAmelCase							)

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


    """simple docstring"""






    for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
     a						=		TimesformerModel.from_pretrained(__lowerCAmelCase							)
     self.assertIsNotNone(__lowerCAmelCase							)

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


    """simple docstring"""






    if not self.has_attentions:
     pass

    else:
     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						=		self.model_tester.num_frames

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

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

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

      a						=		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 		A							(      self			:       Optional[int]							)       ->				int:


    """simple docstring"""






    def check_hidden_states_output(__lowerCAmelCase			:       Union[str, Any]							,   __lowerCAmelCase			:       Optional[int]							,   __lowerCAmelCase			:       str							):
     a						=		model_class(__lowerCAmelCase							)
     model.to(__lowerCAmelCase							)
     model.eval()

     with torch.no_grad():
      a						=		model(**self._prepare_for_class(__lowerCAmelCase							,   __lowerCAmelCase							)							)

     a						=		outputs.hidden_states
     a						=		self.model_tester.num_hidden_layers + 1
     self.assertEqual(len(__lowerCAmelCase							)							,   __lowerCAmelCase							)

     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(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							)

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

     check_hidden_states_output(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							)


def 	UpperCAmelCase__      (      ):






 '''simple docstring'''

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




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

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


    """simple docstring"""






    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 		A							(      self			:       int							)       ->				int:


    """simple docstring"""






    a						=		TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"							).to(
        __lowerCAmelCase							)

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

    # forward pass
    with torch.no_grad():
     a						=		model(**__lowerCAmelCase							)

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

    a						=		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 math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


A_		:     List[str]        =     logging.get_logger(__name__)



class 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			['''audio_values''', '''audio_mask''']

   def __init__(      self			:       List[Any]							,   __lowerCAmelCase			:       Dict=2048							,   __lowerCAmelCase			:       List[Any]=1							,   __lowerCAmelCase			:       Dict=[16, 16]							,   __lowerCAmelCase			:       str=128							,   __lowerCAmelCase			:       Optional[int]=4_4100							,   __lowerCAmelCase			:       int=86							,   __lowerCAmelCase			:       Optional[Any]=2048							,   __lowerCAmelCase			:       str=0.0							,   **__lowerCAmelCase			:       Optional[int]							,   )       ->				Union[str, Any]:


    """simple docstring"""






    super().__init__(
        feature_size=__lowerCAmelCase							,   sampling_rate=__lowerCAmelCase							,   padding_value=__lowerCAmelCase							,   **__lowerCAmelCase							,   )

    a						=		spectrogram_length
    a						=		num_channels
    a						=		patch_size
    a						=		feature_size // self.patch_size[1]
    a						=		n_fft
    a						=		sampling_rate // hop_length_to_sampling_rate
    a						=		sampling_rate
    a						=		padding_value
    a						=		mel_filter_bank(
        num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=__lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=2_2_0_5_0.0							,   sampling_rate=__lowerCAmelCase							,   norm="slaney"							,   mel_scale="slaney"							,   ).T

   def 		A							(      self			:       List[str]							,   __lowerCAmelCase			:       np.array							)       ->				np.ndarray:


    """simple docstring"""






    a						=		spectrogram(
        __lowerCAmelCase							,   window_function(self.n_fft							,   "hann"							)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters.T							,   log_mel="dB"							,   db_range=8_0.0							,   )
    a						=		log_spec[:, :-1]
    a						=		log_spec - 2_0.0
    a						=		np.clip(log_spec / 4_0.0							,   -2.0							,   0.0							) + 1.0
    return log_spec




   def __call__(      self			:       Union[str, Any]							,   __lowerCAmelCase			:       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,   __lowerCAmelCase			:       Optional[Union[str, TensorType]] = None							,   __lowerCAmelCase			:       Optional[bool] = True							,   __lowerCAmelCase			:       Optional[int] = None							,   __lowerCAmelCase			:       bool = False							,   __lowerCAmelCase			:       bool = False							,   **__lowerCAmelCase			:       Optional[int]							,   )       ->				BatchFeature:


    """simple docstring"""






    if sampling_rate is not None:
     if sampling_rate != self.sampling_rate:
      raise ValueError(
          "This feature extractor is set to support sampling rate"
          f""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
          f""" with {self.sampling_rate} and not {sampling_rate}."""							)
    else:
     logger.warning(
         "It is strongly recommended to pass the `sampling_rate` argument to this function. "
         "Failing to do so can result in silent errors that might be hard to debug."							)

    a						=		isinstance(__lowerCAmelCase							,   np.ndarray							) and len(raw_speech.shape							) > 1
    if is_batched_numpy and len(raw_speech.shape							) > 2:
     raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""							)
    a						=		is_batched_numpy or (
        isinstance(__lowerCAmelCase							,   (list, tuple)							) and (isinstance(raw_speech[0]							,   (np.ndarray, tuple, list)							))
    )
    if is_batched:
     a						=		[np.asarray([speech]							,   dtype=np.floataa							).T for speech in raw_speech]
    elif not is_batched and not isinstance(__lowerCAmelCase							,   np.ndarray							):
     a						=		np.asarray(__lowerCAmelCase							,   dtype=np.floataa							)
    elif isinstance(__lowerCAmelCase							,   np.ndarray							) and raw_speech.dtype is np.dtype(np.floataa							):
     a						=		raw_speech.astype(np.floataa							)
    # always return batch
    if not is_batched:
     a						=		[np.asarray([raw_speech]							).T]

    # Convert audio signals to log mel spectrograms, truncate by time axis
    a						=		[
        self._np_extract_fbank_features(waveform.squeeze()							).T[: self.spectrogram_length] for waveform in raw_speech
    ]
    if isinstance(audio_features[0]							,   __lowerCAmelCase							):
     a						=		[np.asarray(__lowerCAmelCase							,   dtype=np.floataa							) for feature in audio_features]

    # Create audio attention mask
    a						=		max(
        [ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len for feature in audio_features]							)  # The maximum number of audio patches in a batch
    if return_attention_mask:
     a						=		[
         (ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len) * [1]
         + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]							) * self.freq_len) * [0]
         for feature in audio_features
     ]
     a						=		np.array(__lowerCAmelCase							).astype(np.floataa							)

    # convert into correct format for padding
    a						=		max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
    a						=		np.ones([len(__lowerCAmelCase							), 1, max_time_len, self.feature_size]							).astype(np.floataa							)
    a						=		padded_audio_features * self.padding_value
    for i in range(len(__lowerCAmelCase							)							):
     a						=		audio_features[i]
     a						=		feature

    # return as BatchFeature
    if return_attention_mask:
     a						=		{"audio_values": padded_audio_features, "audio_mask": audio_mask}
    else:
     a						=		{"audio_values": padded_audio_features}

    a						=		BatchFeature(data=__lowerCAmelCase							,   tensor_type=__lowerCAmelCase							)
    return encoded_inputs

