




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

							# in TimeSformer, the number of spatial tokens equals num_frames * num_patches per frame + 1 CLS token
							_lowerCamelCase     :			Optional[Any]    =    (image_size // patch_size) ** 2
							_lowerCamelCase     :			Dict    =    (num_frames) * self.num_patches_per_frame + 1
			def 					A_    (  self  ):
							_lowerCamelCase     :			str    =    floats_tensor(
							    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]  )

							_lowerCamelCase     :			Union[str, Any]    =    None
							if self.use_labels:
											_lowerCamelCase     :			Dict    =    ids_tensor([self.batch_size]  ,      self.num_labels  )

							_lowerCamelCase     :			int    =    self.get_config()

							return config, pixel_values, labels
			def 					A_    (  self  ):
							_lowerCamelCase     :			str    =    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     :			List[str]    =    self.num_labels
							return config
			def 					A_    (  self  ,      lowercase  ,      lowercase  ,      lowercase  ):
							_lowerCamelCase     :			Union[str, Any]    =    TimesformerModel(config=lowercase  )
							model.to(lowercase  )
							model.eval()
							_lowerCamelCase     :			Any    =    model(lowercase  )
							self.parent.assertEqual(result.last_hidden_state.shape  ,      (self.batch_size, self.seq_length, self.hidden_size)  )
			def 					A_    (  self  ,      lowercase  ,      lowercase  ,      lowercase  ):
							_lowerCamelCase     :			Optional[Any]    =    TimesformerForVideoClassification(lowercase  )
							model.to(lowercase  )
							model.eval()

							_lowerCamelCase     :			List[Any]    =    model(lowercase  )

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






			def 					A_    (  self  ):
							_lowerCamelCase     :			Optional[int]    =    self.prepare_config_and_inputs()
							_lowerCamelCase, _lowerCamelCase, _lowerCamelCase     :			List[str]    =    config_and_inputs
							_lowerCamelCase     :			Dict    =    {'pixel_values': pixel_values}
							return config, inputs_dict






@require_torch
class lowerCAmelCase__	(     lowercase,	lowercase,	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 					A_    (  self  ):
							_lowerCamelCase     :			Union[str, Any]    =    TimesformerModelTester(self  )
							_lowerCamelCase     :			List[Any]    =    ConfigTester(
							    self  ,      config_class=lowercase  ,      has_text_modality=lowercase  ,      hidden_size=37  )
			def 					A_    (  self  ,      lowercase  ,      lowercase  ,      lowercase=False  ):
							_lowerCamelCase     :			Tuple    =    copy.deepcopy(lowercase  )

							if return_labels:
											if model_class in get_values(lowercase  ):
															_lowerCamelCase     :			Optional[int]    =    torch.zeros(
															    self.model_tester.batch_size  ,      dtype=torch.long  ,      device=lowercase  )

							return inputs_dict
			def 					A_    (  self  ):
							self.config_tester.run_common_tests()
			@unittest.skip(reason='TimeSformer does not use inputs_embeds'  )
			def 					A_    (  self  ):
							pass
			def 					A_    (  self  ):
							_lowerCamelCase, _lowerCamelCase     :			Dict    =    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
											_lowerCamelCase     :			Union[str, Any]    =    model_class(lowercase  )
											self.assertIsInstance(model.get_input_embeddings()  ,      (nn.Module)  )
											_lowerCamelCase     :			List[str]    =    model.get_output_embeddings()
											self.assertTrue(x is None or isinstance(lowercase  ,      nn.Linear  )  )
			def 					A_    (  self  ):
							_lowerCamelCase, _lowerCamelCase     :			Tuple    =    self.model_tester.prepare_config_and_inputs_for_common()

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

											_lowerCamelCase     :			Optional[int]    =    ['pixel_values']
											self.assertListEqual(arg_names[:1]  ,      lowercase  )
			def 					A_    (  self  ):
							_lowerCamelCase     :			Dict    =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*lowercase  )
			def 					A_    (  self  ):
							_lowerCamelCase     :			Optional[Any]    =    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_video_classification(*lowercase  )
			@slow
			def 					A_    (  self  ):
							for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											_lowerCamelCase     :			str    =    TimesformerModel.from_pretrained(lowercase  )
											self.assertIsNotNone(lowercase  )
			def 					A_    (  self  ):
							if not self.has_attentions:
											pass

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

											for model_class in self.all_model_classes:
															_lowerCamelCase     :			List[Any]    =    self.model_tester.seq_length
															_lowerCamelCase     :			List[str]    =    self.model_tester.num_frames

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

															# check that output_attentions also work using config
															del inputs_dict["output_attentions"]
															_lowerCamelCase     :			int    =    True
															_lowerCamelCase     :			Optional[int]    =    model_class(lowercase  )
															model.to(lowercase  )
															model.eval()
															with torch.no_grad():
																			_lowerCamelCase     :			List[Any]    =    model(**self._prepare_for_class(lowercase  ,      lowercase  )  )
															_lowerCamelCase     :			List[Any]    =    outputs.attentions
															self.assertEqual(len(lowercase  )  ,      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     :			Any    =    len(lowercase  )

															# Check attention is always last and order is fine
															_lowerCamelCase     :			Optional[Any]    =    True
															_lowerCamelCase     :			Union[str, Any]    =    True
															_lowerCamelCase     :			Union[str, Any]    =    model_class(lowercase  )
															model.to(lowercase  )
															model.eval()
															with torch.no_grad():
																			_lowerCamelCase     :			str    =    model(**self._prepare_for_class(lowercase  ,      lowercase  )  )

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

															_lowerCamelCase     :			Optional[Any]    =    outputs.attentions

															self.assertEqual(len(lowercase  )  ,      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  ):
							def check_hidden_states_output(lowercase  ,      lowercase  ,      lowercase  ):
											_lowerCamelCase     :			Any    =    model_class(lowercase  )
											model.to(lowercase  )
											model.eval()

											with torch.no_grad():
															_lowerCamelCase     :			List[Any]    =    model(**self._prepare_for_class(lowercase  ,      lowercase  )  )

											_lowerCamelCase     :			int    =    outputs.hidden_states
											_lowerCamelCase     :			str    =    self.model_tester.num_hidden_layers + 1
											self.assertEqual(len(lowercase  )  ,      lowercase  )

											_lowerCamelCase     :			Optional[Any]    =    self.model_tester.seq_length

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

							_lowerCamelCase, _lowerCamelCase     :			Any    =    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
											_lowerCamelCase     :			Tuple    =    True
											check_hidden_states_output(lowercase  ,      lowercase  ,      lowercase  )

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

											check_hidden_states_output(lowercase  ,      lowercase  ,      lowercase  )







def      _snake_case			(				):
				_lowerCamelCase     :			str    =    hf_hub_download(
				    repo_id='hf-internal-testing/spaghetti-video'							,    filename='eating_spaghetti.npy'							,    repo_type='dataset'		)
				_lowerCamelCase     :			Optional[Any]    =    np.load(lowercase__		)
				return list(lowercase__		)






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






			'''simple docstring'''



			@cached_property
			def 					A_    (  self  ):
							# 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 					A_    (  self  ):
							_lowerCamelCase     :			List[Any]    =    TimesformerForVideoClassification.from_pretrained('facebook/timesformer-base-finetuned-k400'  ).to(
							    lowercase  )

							_lowerCamelCase     :			Any    =    self.default_image_processor
							_lowerCamelCase     :			Union[str, Any]    =    prepare_video()
							_lowerCamelCase     :			Any    =    image_processor(video[:8]  ,      return_tensors='pt'  ).to(lowercase  )

							# forward pass
							with torch.no_grad():
											_lowerCamelCase     :			Any    =    model(**lowercase  )

							# verify the logits
							_lowerCamelCase     :			Dict    =    torch.Size((1, 400)  )
							self.assertEqual(outputs.logits.shape  ,      lowercase  )

							_lowerCamelCase     :			Optional[Any]    =    torch.tensor([-0.30_16, -0.77_13, -0.42_05]  ).to(lowercase  )

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






class     SCREAMING_SNAKE_CASE__ :



 def __init__(							self,__lowerCamelCase ):
        A__											=						set_counts
        A__											=						max(__lowerCamelCase )
        A__											=						len(__lowerCamelCase )
        A__											=						[1] * num_sets
        A__											=						list(range(__lowerCamelCase ) )



 def 				UpperCamelCase			(							self,__lowerCamelCase,__lowerCamelCase ):
        A__											=						self.get_parent(__lowerCamelCase )
        A__											=						self.get_parent(__lowerCamelCase )

        if src_parent == dst_parent:
               return False

        if self.ranks[dst_parent] >= self.ranks[src_parent]:
               self.set_counts[dst_parent] += self.set_counts[src_parent]
               A__											=						0
               A__											=						dst_parent
               if self.ranks[dst_parent] == self.ranks[src_parent]:
                      self.ranks[dst_parent] += 1
               A__											=						self.set_counts[dst_parent]
        else:
               self.set_counts[src_parent] += self.set_counts[dst_parent]
               A__											=						0
               A__											=						src_parent
               A__											=						self.set_counts[src_parent]

        A__											=						max(self.max_set,__lowerCamelCase )
        return True




 def 				UpperCamelCase			(							self,__lowerCamelCase ):
        if self.parents[disj_set] == disj_set:
               return disj_set
        A__											=						self.get_parent(self.parents[disj_set] )
        return self.parents[disj_set]





