
"""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        _UpperCAmelCase:



		def __init__(   self							,						__a							,						__a=13							,						__a=10							,						__a=3							,						__a=2							,						__a=2							,						__a=True							,						__a=True							,						__a=32							,						__a=5							,						__a=4							,						__a=37							,						__a="gelu"							,						__a=0.1							,						__a=0.1							,						__a=10							,						__a=0.02							,						__a="divided_space_time"							,						__a=None							,						)			->			List[str]:






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

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



		def 		UpperCAmelCase				(   self)			->			Tuple:






							'''simple docstring'''
							_UpperCamelCase									=    floats_tensor(
							    [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size])

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

							_UpperCamelCase									=    self.get_config()

							return config, pixel_values, labels



		def 		UpperCAmelCase				(   self)			->			Tuple:






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



		def 		UpperCAmelCase				(   self							,						__a							,						__a							,						__a)			->			Optional[Any]:






							'''simple docstring'''
							_UpperCamelCase									=    TimesformerModel(config=__a)
							model.to(__a)
							model.eval()
							_UpperCamelCase									=    model(__a)
							self.parent.assertEqual(result.last_hidden_state.shape							,						(self.batch_size, self.seq_length, self.hidden_size))



		def 		UpperCAmelCase				(   self							,						__a							,						__a							,						__a)			->			Dict:






							'''simple docstring'''
							_UpperCamelCase									=    TimesformerForVideoClassification(__a)
							model.to(__a)
							model.eval()

							_UpperCamelCase									=    model(__a)

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




		def 		UpperCAmelCase				(   self)			->			Optional[int]:






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






@require_torch
class        _UpperCAmelCase(       lowerCamelCase ,	lowerCamelCase ,	unittest.TestCase      ):
		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 		UpperCAmelCase				(   self)			->			Union[str, Any]:






							'''simple docstring'''
							_UpperCamelCase									=    TimesformerModelTester(self)
							_UpperCamelCase									=    ConfigTester(
							    self							,						config_class=__a							,						has_text_modality=__a							,						hidden_size=37)



		def 		UpperCAmelCase				(   self							,						__a							,						__a							,						__a=False)			->			List[str]:






							'''simple docstring'''
							_UpperCamelCase									=    copy.deepcopy(__a)

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

							return inputs_dict



		def 		UpperCAmelCase				(   self)			->			Dict:






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



		@unittest.skip(reason='''TimeSformer does not use inputs_embeds''')
		def 		UpperCAmelCase				(   self)			->			Any:






							'''simple docstring'''
							pass



		def 		UpperCAmelCase				(   self)			->			str:






							'''simple docstring'''
							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

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



		def 		UpperCAmelCase				(   self)			->			str:






							'''simple docstring'''
							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

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

												_UpperCamelCase									=    ['''pixel_values''']
												self.assertListEqual(arg_names[:1]							,						__a)



		def 		UpperCAmelCase				(   self)			->			Dict:






							'''simple docstring'''
							_UpperCamelCase									=    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*__a)



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






							'''simple docstring'''
							_UpperCamelCase									=    self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_video_classification(*__a)



		@slow
		def 		UpperCAmelCase				(   self)			->			int:






							'''simple docstring'''
							for model_name in TIMESFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												_UpperCamelCase									=    TimesformerModel.from_pretrained(__a)
												self.assertIsNotNone(__a)



		def 		UpperCAmelCase				(   self)			->			Optional[int]:






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

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

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

																	_UpperCamelCase									=    True
																	_UpperCamelCase									=    False
																	_UpperCamelCase									=    True
																	_UpperCamelCase									=    model_class(__a)
																	model.to(__a)
																	model.eval()
																	with torch.no_grad():
																						_UpperCamelCase									=    model(**self._prepare_for_class(__a							,						__a))
																	_UpperCamelCase									=    outputs.attentions
																	self.assertEqual(len(__a)							,						self.model_tester.num_hidden_layers)

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

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

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

																	_UpperCamelCase									=    outputs.attentions

																	self.assertEqual(len(__a)							,						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)			->			int:






							'''simple docstring'''
							def check_hidden_states_output(__a							,						__a							,						__a):
												_UpperCamelCase									=    model_class(__a)
												model.to(__a)
												model.eval()

												with torch.no_grad():
																	_UpperCamelCase									=    model(**self._prepare_for_class(__a							,						__a))

												_UpperCamelCase									=    outputs.hidden_states
												_UpperCamelCase									=    self.model_tester.num_hidden_layers + 1
												self.assertEqual(len(__a)							,						__a)

												_UpperCamelCase									=    self.model_tester.seq_length

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

							_UpperCamelCase					,    _UpperCamelCase									=    self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												_UpperCamelCase									=    True
												check_hidden_states_output(__a							,						__a							,						__a)

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

												check_hidden_states_output(__a							,						__a							,						__a)





def  lowerCamelCase__						(     )				->						List[Any]:


					"""simple docstring"""





					_UpperCamelCase									=    hf_hub_download(
					    repo_id='''hf-internal-testing/spaghetti-video''',	filename='''eating_spaghetti.npy''',	repo_type='''dataset'''      )
					_UpperCamelCase									=    np.load(__snake_case      )
					return list(__snake_case      )






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



		@cached_property
		def 		UpperCAmelCase				(   self)			->			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 		UpperCAmelCase				(   self)			->			List[str]:






							'''simple docstring'''
							_UpperCamelCase									=    TimesformerForVideoClassification.from_pretrained('''facebook/timesformer-base-finetuned-k400''').to(
							    __a)

							_UpperCamelCase									=    self.default_image_processor
							_UpperCamelCase									=    prepare_video()
							_UpperCamelCase									=    image_processor(video[:8]							,						return_tensors='''pt''').to(__a)

							# forward pass
							with torch.no_grad():
												_UpperCamelCase									=    model(**__a)

							# verify the logits
							_UpperCamelCase									=    torch.Size((1, 4_00))
							self.assertEqual(outputs.logits.shape							,						__a)

							_UpperCamelCase									=    torch.tensor([-0.3016, -0.7713, -0.4205]).to(__a)

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








import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 					_lowerCamelCase			(   UpperCamelCase_						,				unittest.TestCase			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	DanceDiffusionPipeline
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE_		 =	PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_		 =	False
     SCREAMING_SNAKE_CASE_		 =	False






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					UNetaDModel(
               block_out_channels=(3_2, 3_2, 6_4)    ,    extra_in_channels=1_6    ,    sample_size=5_1_2    ,    sample_rate=1_6_0_0_0    ,    in_channels=2    ,    out_channels=2    ,    flip_sin_to_cos=__SCREAMING_SNAKE_CASE    ,    use_timestep_embedding=__SCREAMING_SNAKE_CASE    ,    time_embedding_type='''fourier'''    ,    mid_block_type='''UNetMidBlock1D'''    ,    down_block_types=('''DownBlock1DNoSkip''', '''DownBlock1D''', '''AttnDownBlock1D''')    ,    up_block_types=('''AttnUpBlock1D''', '''UpBlock1D''', '''UpBlock1DNoSkip''')    ,    )
           UpperCamelCase__     : Union[str, Any]											=					IPNDMScheduler()

           UpperCamelCase__     : List[str]											=					{
               '''unet''': unet,
               '''scheduler''': scheduler,
           }
           return components






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Any:




           """simple docstring"""







           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Optional[Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[Any]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : int											=					{
               '''batch_size''': 1,
               '''generator''': generator,
               '''num_inference_steps''': 4,
           }
           return inputs






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : str											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Dict											=					self.get_dummy_components()
           UpperCamelCase__     : str											=					DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					pipe(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					output.audios

           UpperCamelCase__     : Dict											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, components["unet"].sample_size)
           UpperCamelCase__     : List[Any]											=					np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]	)
           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           return super().test_save_load_local()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3	)






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return super().test_save_load_optional_components()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           return super().test_attention_slicing_forward_pass()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           super().test_inference_batch_single_identical(expected_max_diff=3e-3	)




@slow
@require_torch_gpu
class 					_lowerCamelCase			(   unittest.TestCase			):





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






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




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Any											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''	)
           UpperCamelCase__     : int											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : str											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Tuple											=					np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






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




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Union[str, Any]											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''    ,    torch_dtype=torch.floataa	)
           UpperCamelCase__     : Tuple											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : List[Any]											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Optional[Any]											=					np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2


