
import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ASTConfig
from transformers.testing_utils import require_torch, require_torchaudio, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_torchaudio_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 ASTForAudioClassification, ASTModel
					from transformers.models.audio_spectrogram_transformer.modeling_audio_spectrogram_transformer import (
					    AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
					)


if is_torchaudio_available():
					import torchaudio

					from transformers import ASTFeatureExtractor





class 				UpperCAmelCase :




	'''simple docstring'''





	def __init__(			self   :					int				,A   :					List[str]				,A   :					List[Any]=13				,A   :					str=2				,A   :					int=24				,A   :					Any=16				,A   :					Tuple=True				,A   :					Optional[int]=True				,A   :					int=32				,A   :					int=5				,A   :					Optional[Any]=4				,A   :					str=37				,A   :					Union[str, Any]="gelu"				,A   :					Union[str, Any]=0.1				,A   :					List[str]=0.1				,A   :					Tuple=10				,A   :					str=0.02				,A   :					Tuple=None				,A   :					List[str]=2				,A   :					str=2				,):
					__A       =					parent
					__A       =					batch_size
					__A       =					patch_size
					__A       =					max_length
					__A       =					num_mel_bins
					__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       =					type_sequence_label_size
					__A       =					initializer_range
					__A       =					scope
					__A       =					frequency_stride
					__A       =					time_stride

					# in AST, the seq length equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
					__A       =					(self.num_mel_bins - self.patch_size) // self.frequency_stride + 1
					__A       =					(self.max_length - self.patch_size) // self.time_stride + 1
					__A       =					frequency_out_dimension * time_out_dimension
					__A       =					num_patches + 2






	def 							UpperCamelCase_    (			self   :					Optional[Any]					):
					__A       =					floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]					)

					__A       =					None
					if self.use_labels:
									__A       =					ids_tensor([self.batch_size]				,self.type_sequence_label_size					)

					__A       =					self.get_config()

					return config, input_values, labels






	def 							UpperCamelCase_    (			self   :					str					):
					return ASTConfig(
					    patch_size=self.patch_size				,max_length=self.max_length				,num_mel_bins=self.num_mel_bins				,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				,is_decoder=A				,initializer_range=self.initializer_range				,frequency_stride=self.frequency_stride				,time_stride=self.time_stride				,)






	def 							UpperCamelCase_    (			self   :					Tuple				,A   :					Dict				,A   :					str				,A   :					List[str]					):
					__A       =					ASTModel(config=A					)
					model.to(A					)
					model.eval()
					__A       =					model(A					)
					self.parent.assertEqual(result.last_hidden_state.shape				,(self.batch_size, self.seq_length, self.hidden_size)					)







	def 							UpperCamelCase_    (			self   :					int					):
					__A       =					self.prepare_config_and_inputs()
					(
					    (
					    __A
					)     ,				(
					    __A
					)     ,				(
					    __A
					)     ,				
					)       =					config_and_inputs
					__A       =					{"input_values": input_values}
					return config, inputs_dict





@require_torch
class 				UpperCAmelCase (    __SCREAMING_SNAKE_CASE     ,     __SCREAMING_SNAKE_CASE     ,     unittest.TestCase    ):




	'''simple docstring'''





	snake_case_          =	(
	    (
	        ASTModel,
	        ASTForAudioClassification,
	    )
	    if is_torch_available()
	    else ()
	)
	snake_case_          =	(
	    {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
	    if is_torch_available()
	    else {}
	)
	snake_case_          =	False
	snake_case_          =	False
	snake_case_          =	False
	snake_case_          =	False






	def 							UpperCamelCase_    (			self   :					str				,A   :					Union[str, Any]				,A   :					Optional[Any]				,A   :					List[str]				,A   :					List[str]				,A   :					List[Any]					):
					if pipeline_test_casse_name == "AudioClassificationPipelineTests":
									return True

					return False






	def 							UpperCamelCase_    (			self   :					Optional[int]					):
					__A       =					ASTModelTester(self					)
					__A       =					ConfigTester(self				,config_class=A				,has_text_modality=A				,hidden_size=37					)






	def 							UpperCamelCase_    (			self   :					List[Any]					):
					self.config_tester.run_common_tests()






	@unittest.skip(reason="AST does not use inputs_embeds"					)
	def 							UpperCamelCase_    (			self   :					int					):
					pass






	def 							UpperCamelCase_    (			self   :					Optional[int]					):
					__A     ,				__A       =					self.model_tester.prepare_config_and_inputs_for_common()

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






	def 							UpperCamelCase_    (			self   :					Optional[int]					):
					__A     ,				__A       =					self.model_tester.prepare_config_and_inputs_for_common()

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

									__A       =					["input_values"]
									self.assertListEqual(arg_names[:1]				,A					)






	def 							UpperCamelCase_    (			self   :					Tuple					):
					__A       =					self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*A					)







	@slow
	def 							UpperCamelCase_    (			self   :					str					):
					for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									__A       =					ASTModel.from_pretrained(A					)
									self.assertIsNotNone(A					)





def 		UpperCAmelCase     (      )						->      int:







				"""simple docstring"""



				__A       =					hf_hub_download(
				    repo_id="nielsr/audio-spectogram-transformer-checkpoint"						,  filename="sample_audio.flac"						,  repo_type="dataset"		)

				__A     ,				__A       =					torchaudio.load(a_		)

				return audio, sampling_rate





@require_torch
@require_torchaudio
class 				UpperCAmelCase (    unittest.TestCase    ):




	'''simple docstring'''





	@cached_property
	def 							UpperCamelCase_    (			self   :					int					):
					return (
					    ASTFeatureExtractor.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593"					)
					    if is_torchaudio_available()
					    else None
					)







	@slow
	def 							UpperCamelCase_    (			self   :					Optional[Any]					):
					__A       =					self.default_feature_extractor
					__A       =					ASTForAudioClassification.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593"					).to(A					)

					__A       =					self.default_feature_extractor
					__A     ,				__A       =					prepare_audio()
					__A       =					audio.squeeze().numpy()
					__A       =					feature_extractor(A				,sampling_rate=A				,return_tensors="pt"					).to(A					)

					# forward pass
					with torch.no_grad():
									__A       =					model(**A					)

					# verify the logits
					__A       =					torch.Size((1, 5_27)					)
					self.assertEqual(outputs.logits.shape				,A					)

					__A       =					torch.tensor([-0.87_60, -7.00_42, -8.66_02]					).to(A					)

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





"""simple docstring"""







import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







@skip_mps
class 					A_(SCREAMING_SNAKE_CASE_   ,   unittest.TestCase      ):







			"""simple docstring"""



			a_       :			Dict												=			TextToVideoSDPipeline
			a_       :			Dict												=			TEXT_TO_IMAGE_PARAMS
			a_       :			Dict												=			TEXT_TO_IMAGE_BATCH_PARAMS
			# No `output_type`.
			a_       :			str												=			frozenset(
			    [
			        """num_inference_steps""",
			        """generator""",
			        """latents""",
			        """return_dict""",
			        """callback""",
			        """callback_steps""",
			    ]      )
			def    _lowerCAmelCase		(	self							):
						torch.manual_seed(0							)
						_lowerCamelCase					:					List[Any]	       =    UNetaDConditionModel(
						    block_out_channels=(32, 64, 64, 64)				,   layers_per_block=2				,   sample_size=32				,   in_channels=4				,   out_channels=4				,   down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')				,   up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')				,   cross_attention_dim=32				,   attention_head_dim=4				,   )
						_lowerCamelCase					:					Optional[int]	       =    DDIMScheduler(
						    beta_start=0.0_0_0_8_5				,   beta_end=0.0_1_2				,   beta_schedule='scaled_linear'				,   clip_sample=A				,   set_alpha_to_one=A				,   )
						torch.manual_seed(0							)
						_lowerCamelCase					:					Dict	       =    AutoencoderKL(
						    block_out_channels=[32, 64]				,   in_channels=3				,   out_channels=3				,   down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']				,   up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']				,   latent_channels=4				,   sample_size=128				,   )
						torch.manual_seed(0							)
						_lowerCamelCase					:					Optional[int]	       =    CLIPTextConfig(
						    bos_token_id=0				,   eos_token_id=2				,   hidden_size=32				,   intermediate_size=37				,   layer_norm_eps=1E-05				,   num_attention_heads=4				,   num_hidden_layers=5				,   pad_token_id=1				,   vocab_size=1000				,   hidden_act='gelu'				,   projection_dim=512				,   )
						_lowerCamelCase					:					str	       =    CLIPTextModel(A							)
						_lowerCamelCase					:					List[str]	       =    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

						_lowerCamelCase					:					Any	       =    {
						    'unet': unet,
						    'scheduler': scheduler,
						    'vae': vae,
						    'text_encoder': text_encoder,
						    'tokenizer': tokenizer,
						}
						return components
			def    _lowerCAmelCase		(	self				,   A				,   A=0							):
						if str(A							).startswith('mps'							):
									_lowerCamelCase					:					Tuple	       =    torch.manual_seed(A							)
						else:
									_lowerCamelCase					:					Optional[int]	       =    torch.Generator(device=A							).manual_seed(A							)
						_lowerCamelCase					:					Optional[int]	       =    {
						    'prompt': 'A painting of a squirrel eating a burger',
						    'generator': generator,
						    'num_inference_steps': 2,
						    'guidance_scale': 6.0,
						    'output_type': 'pt',
						}
						return inputs
			def    _lowerCAmelCase		(	self							):
						_lowerCamelCase					:					List[Any]	       =    'cpu'  # ensure determinism for the device-dependent torch.Generator
						_lowerCamelCase					:					Tuple	       =    self.get_dummy_components()
						_lowerCamelCase					:					Dict	       =    TextToVideoSDPipeline(**A							)
						_lowerCamelCase					:					Optional[int]	       =    sd_pipe.to(A							)
						sd_pipe.set_progress_bar_config(disable=A							)

						_lowerCamelCase					:					Union[str, Any]	       =    self.get_dummy_inputs(A							)
						_lowerCamelCase					:					Union[str, Any]	       =    'np'
						_lowerCamelCase					:					Optional[int]	       =    sd_pipe(**A							).frames
						_lowerCamelCase					:					Dict	       =    frames[0][-3:, -3:, -1]

						assert frames[0].shape == (64, 64, 3)
						_lowerCamelCase					:					Tuple	       =    np.array([1_5_8.0, 1_6_0.0, 1_5_3.0, 1_2_5.0, 1_0_0.0, 1_2_1.0, 1_1_1.0, 9_3.0, 1_1_3.0]							)

						assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
			def    _lowerCAmelCase		(	self							):
						self._test_attention_slicing_forward_pass(test_mean_pixel_difference=A				,   expected_max_diff=3E-3							)
			@unittest.skipIf(
			    torch_device != 'cuda' or not is_xformers_available()				,   reason='XFormers attention is only available with CUDA and `xformers` installed'				,   )
			def    _lowerCAmelCase		(	self							):
						self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A				,   expected_max_diff=1E-2							)
			@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'							)
			def    _lowerCAmelCase		(	self							):
						pass
			@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'							)
			def    _lowerCAmelCase		(	self							):
						pass
			@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'							)
			def    _lowerCAmelCase		(	self							):
						pass

			def    _lowerCAmelCase		(	self							):
						return super().test_progress_bar()







@slow
@skip_mps
class 					A_(unittest.TestCase      ):







			"""simple docstring"""



			def    _lowerCAmelCase		(	self							):
						_lowerCamelCase					:					Tuple	       =    load_numpy(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'							)

						_lowerCamelCase					:					Dict	       =    TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'							)
						_lowerCamelCase					:					Dict	       =    DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
						_lowerCamelCase					:					Tuple	       =    pipe.to('cuda'							)

						_lowerCamelCase					:					str	       =    'Spiderman is surfing'
						_lowerCamelCase					:					Any	       =    torch.Generator(device='cpu'							).manual_seed(0							)

						_lowerCamelCase					:					Union[str, Any]	       =    pipe(A				,   generator=A				,   num_inference_steps=25				,   output_type='pt'							).frames
						_lowerCamelCase					:					Any	       =    video_frames.cpu().numpy()

						assert np.abs(expected_video - video							).mean() < 5E-2

			def    _lowerCAmelCase		(	self							):
						_lowerCamelCase					:					Optional[Any]	       =    load_numpy(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'							)

						_lowerCamelCase					:					int	       =    TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'							)
						_lowerCamelCase					:					Optional[Any]	       =    pipe.to('cuda'							)

						_lowerCamelCase					:					Tuple	       =    'Spiderman is surfing'
						_lowerCamelCase					:					Union[str, Any]	       =    torch.Generator(device='cpu'							).manual_seed(0							)

						_lowerCamelCase					:					Tuple	       =    pipe(A				,   generator=A				,   num_inference_steps=2				,   output_type='pt'							).frames
						_lowerCamelCase					:					Optional[int]	       =    video_frames.cpu().numpy()

						assert np.abs(expected_video - video							).mean() < 5E-2





