



from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL

from ...utils import BaseOutput, OptionalDependencyNotAvailable, is_torch_available, is_transformers_available
from .timesteps import (
    fastaa_timesteps,
    smartaa_timesteps,
    smartaa_timesteps,
    smartaaa_timesteps,
    smartaaa_timesteps,
    superaa_timesteps,
    superaa_timesteps,
    superaaa_timesteps,
)

@dataclass
class 				snake_case__(_UpperCamelCase      ):

					"""simple docstring"""






					lowercase_								=		42
					lowercase_								=		42
					lowercase_								=		42


try:
				if not (is_transformers_available() and is_torch_available()):
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
				from .pipeline_if import IFPipeline
				from .pipeline_if_imgaimg import IFImgaImgPipeline
				from .pipeline_if_imgaimg_superresolution import IFImgaImgSuperResolutionPipeline
				from .pipeline_if_inpainting import IFInpaintingPipeline
				from .pipeline_if_inpainting_superresolution import IFInpaintingSuperResolutionPipeline
				from .pipeline_if_superresolution import IFSuperResolutionPipeline
				from .safety_checker import IFSafetyChecker
				from .watermark import IFWatermarker









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

					"""simple docstring"""






					def __init__(							self :     Any				,       SCREAMING_SNAKE_CASE :     str				,       SCREAMING_SNAKE_CASE :     List[str]=13				,       SCREAMING_SNAKE_CASE :     Union[str, Any]=10				,       SCREAMING_SNAKE_CASE :     Union[str, Any]=3				,       SCREAMING_SNAKE_CASE :     int=2				,       SCREAMING_SNAKE_CASE :     int=2				,       SCREAMING_SNAKE_CASE :     Optional[int]=True				,       SCREAMING_SNAKE_CASE :     str=True				,       SCREAMING_SNAKE_CASE :     int=32				,       SCREAMING_SNAKE_CASE :     Dict=5				,       SCREAMING_SNAKE_CASE :     str=4				,       SCREAMING_SNAKE_CASE :     Optional[int]=37				,       SCREAMING_SNAKE_CASE :     Any="gelu"				,       SCREAMING_SNAKE_CASE :     List[Any]=0.1				,       SCREAMING_SNAKE_CASE :     Optional[Any]=0.1				,       SCREAMING_SNAKE_CASE :     Dict=10				,       SCREAMING_SNAKE_CASE :     Optional[int]=0.02				,       SCREAMING_SNAKE_CASE :     str="divided_space_time"				,       SCREAMING_SNAKE_CASE :     Tuple=None				,       ):
									lowercase__					:   List[str]													=					parent
									lowercase__					:   Optional[int]													=					batch_size
									lowercase__					:   List[Any]													=					image_size
									lowercase__					:   Optional[Any]													=					num_channels
									lowercase__					:   List[str]													=					patch_size
									lowercase__					:   str													=					num_frames
									lowercase__					:   List[str]													=					is_training
									lowercase__					:   List[str]													=					use_labels
									lowercase__					:   int													=					hidden_size
									lowercase__					:   List[Any]													=					num_hidden_layers
									lowercase__					:   Optional[Any]													=					num_attention_heads
									lowercase__					:   Dict													=					intermediate_size
									lowercase__					:   Union[str, Any]													=					hidden_act
									lowercase__					:   str													=					hidden_dropout_prob
									lowercase__					:   Tuple													=					attention_probs_dropout_prob
									lowercase__					:   Tuple													=					attention_type
									lowercase__					:   Union[str, Any]													=					initializer_range
									lowercase__					:   Any													=					scope
									lowercase__					:   Optional[int]													=					num_labels

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







					def 			snake_case					(							self :     Optional[int]					):
									lowercase__					:   Any													=					floats_tensor(
									    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]					)

									lowercase__					:   Optional[Any]													=					None
									if self.use_labels:
													lowercase__					:   str													=					ids_tensor([self.batch_size]				,       self.num_labels					)

									lowercase__					:   List[Any]													=					self.get_config()

									return config, pixel_values, labels







					def 			snake_case					(							self :     Any					):
									lowercase__					:   Optional[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				,       )
									lowercase__					:   List[Any]													=					self.num_labels
									return config







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







					def 			snake_case					(							self :     Dict				,       SCREAMING_SNAKE_CASE :     Optional[Any]				,       SCREAMING_SNAKE_CASE :     Optional[int]				,       SCREAMING_SNAKE_CASE :     Any					):
									lowercase__					:   List[Any]													=					TimesformerForVideoClassification(SCREAMING_SNAKE_CASE					)
									model.to(SCREAMING_SNAKE_CASE					)
									model.eval()

									lowercase__					:   List[str]													=					model(SCREAMING_SNAKE_CASE					)

									# verify the logits shape
									lowercase__					:   List[str]													=					torch.Size((self.batch_size, self.num_labels)					)
									self.parent.assertEqual(result.logits.shape				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Optional[Any]					):
									lowercase__					:   List[str]													=					self.prepare_config_and_inputs()
									lowercase__							,							lowercase__							,							lowercase__					:   int													=					config_and_inputs
									lowercase__					:   Optional[int]													=					{"pixel_values": pixel_values}
									return config, inputs_dict






@require_torch
class 				snake_case__(_UpperCamelCase    ,      _UpperCamelCase    ,      unittest.TestCase      ):

					"""simple docstring"""






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

					lowercase_								=		False
					lowercase_								=		False
					lowercase_								=		False
					lowercase_								=		False







					def 			snake_case					(							self :     Dict					):
									lowercase__					:   Tuple													=					TimesformerModelTester(self					)
									lowercase__					:   Any													=					ConfigTester(
									    self				,       config_class=SCREAMING_SNAKE_CASE				,       has_text_modality=SCREAMING_SNAKE_CASE				,       hidden_size=37					)







					def 			snake_case					(							self :     Dict				,       SCREAMING_SNAKE_CASE :     List[str]				,       SCREAMING_SNAKE_CASE :     Any				,       SCREAMING_SNAKE_CASE :     Tuple=False					):
									lowercase__					:   Union[str, Any]													=					copy.deepcopy(SCREAMING_SNAKE_CASE					)

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

									return inputs_dict







					def 			snake_case					(							self :     Optional[Any]					):
									self.config_tester.run_common_tests()







					@unittest.skip(reason="TimeSformer does not use inputs_embeds"					)
					def 			snake_case					(							self :     Any					):
									pass







					def 			snake_case					(							self :     Tuple					):
									lowercase__							,							lowercase__					:   str													=					self.model_tester.prepare_config_and_inputs_for_common()

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







					def 			snake_case					(							self :     Union[str, Any]					):
									lowercase__							,							lowercase__					:   Optional[Any]													=					self.model_tester.prepare_config_and_inputs_for_common()

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

													lowercase__					:   Dict													=					["pixel_values"]
													self.assertListEqual(arg_names[:1]				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Tuple					):
									lowercase__					:   Union[str, Any]													=					self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     str					):
									lowercase__					:   List[str]													=					self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_video_classification(*SCREAMING_SNAKE_CASE					)







					@slow
					def 			snake_case					(							self :     Optional[int]					):
									for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													lowercase__					:   Optional[int]													=					TimesformerModel.from_pretrained(SCREAMING_SNAKE_CASE					)
													self.assertIsNotNone(SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Any					):
									if not self.has_attentions:
													pass

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

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

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

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

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

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

																	lowercase__					:   Optional[Any]													=					outputs.attentions

																	self.assertEqual(len(SCREAMING_SNAKE_CASE					)				,       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 :     List[Any]					):
									def check_hidden_states_output(SCREAMING_SNAKE_CASE :     str				,       SCREAMING_SNAKE_CASE :     int				,       SCREAMING_SNAKE_CASE :     Union[str, Any]					):
													lowercase__					:   Optional[int]													=					model_class(SCREAMING_SNAKE_CASE					)
													model.to(SCREAMING_SNAKE_CASE					)
													model.eval()

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

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

													lowercase__					:   int													=					self.model_tester.seq_length

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

									lowercase__							,							lowercase__					:   str													=					self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
													lowercase__					:   Dict													=					True
													check_hidden_states_output(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)

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

													check_hidden_states_output(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)






def  __lowerCamelCase (       ):

				"""simple docstring"""


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

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

					"""simple docstring"""






					@cached_property
					def 			snake_case					(							self :     Dict					):
									# 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 :     List[Any]					):
									lowercase__					:   List[Any]													=					TimesformerForVideoClassification.from_pretrained("facebook/timesformer-base-finetuned-k400"					).to(
									    SCREAMING_SNAKE_CASE					)

									lowercase__					:   int													=					self.default_image_processor
									lowercase__					:   List[str]													=					prepare_video()
									lowercase__					:   str													=					image_processor(video[:8]				,       return_tensors="pt"					).to(SCREAMING_SNAKE_CASE					)

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

									# verify the logits
									lowercase__					:   Union[str, Any]													=					torch.Size((1, 400)					)
									self.assertEqual(outputs.logits.shape				,       SCREAMING_SNAKE_CASE					)

									lowercase__					:   Optional[Any]													=					torch.tensor([-0.3_016, -0.7_713, -0.4_205]					).to(SCREAMING_SNAKE_CASE					)

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




