




import argparse
import copy
import json
import multiprocessing
import os
import time

from create_dummy_models import COMPOSITE_MODELS, create_tiny_models
from huggingface_hub import ModelFilter, hf_api

import transformers
from transformers import AutoFeatureExtractor, AutoImageProcessor, AutoTokenizer
from transformers.image_processing_utils import BaseImageProcessor




def      __magic_name__							(					):






			'''simple docstring'''







			a       =     set()
			# Each auto modeling files contains multiple mappings. Let's get them in a dynamic way.
			for module_name in ["modeling_auto", "modeling_tf_auto", "modeling_flax_auto"]:
						a       =     getattr(transformers.models.auto,    A,    A  )
						if module is None:
									continue
						# all mappings in a single auto modeling file
						a       =     [
						    x
						    for x in dir(A  )
						    if x.endswith("_MAPPING_NAMES"  )
						    and (x.startswith("MODEL_"  ) or x.startswith("TF_MODEL_"  ) or x.startswith("FLAX_MODEL_"  ))
						]
						for name in mapping_names:
									a       =     getattr(A,    A  )
									if mapping is not None:
												for v in mapping.values():
															if isinstance(A,    (list, tuple)  ):
																		model_names.update(A  )
															elif isinstance(A,    A  ):
																		model_names.add(A  )

			return sorted(A  )




def      __magic_name__							(					):






			'''simple docstring'''







			a       =     set(get_all_model_names()  )

			with open("tests/utils/tiny_model_summary.json"  ) as fp:
						a       =     json.load(A  )
			a       =     set()
			for model_base_name in tiny_model_info:
						tiny_models_names.update(tiny_model_info[model_base_name]["model_classes"]  )

			# Remove a tiny model name if one of its framework implementation hasn't yet a tiny version on the Hub.
			a       =     model_names.difference(A  )
			for model_name in copy.copy(A  ):
						if not model_name.startswith("TF"  ) and F"""TF{model_name}""" in not_on_hub:
									tiny_models_names.remove(A  )
						elif model_name.startswith("TF"  ) and model_name[2:] in not_on_hub:
									tiny_models_names.remove(A  )

			return sorted(A  )




def      __magic_name__							(					A						:   Tuple  ):






			'''simple docstring'''







			a       =     COMPOSITE_MODELS.values()

			# All tiny model base names on Hub
			a       =     get_all_model_names()
			a       =     hf_api.list_models(
			    filter=ModelFilter(
			        author="hf-internal-testing",    )  )
			a       =     set()
			for x in models:
						a       =     x.modelId
						a       ,						a       =     model.split("/"  )
						if not model.startswith("tiny-random-"  ):
									continue
						a       =     model.replace("tiny-random-",    ""  )
						if not model[0].isupper():
									continue
						if model not in model_names and model not in special_models:
									continue
						_models.add(A  )

			a       =     sorted(_models  )
			# All tiny model names on Hub
			a       =     {}
			for model in models:
						a       =     F"""hf-internal-testing/tiny-random-{model}"""
						a       =     model.split("-"  )[0]
						try:
									a       =     hf_api.repo_info(A  )
									a       =     {
									    "tokenizer_classes": set(),
									    "processor_classes": set(),
									    "model_classes": set(),
									    "sha": repo_info.sha,
									}
						except Exception:
									continue
						try:
									time.sleep(1  )
									a       =     AutoTokenizer.from_pretrained(A  )
									content["tokenizer_classes"].add(tokenizer_fast.__class__.__name__  )
						except Exception:
									pass
						try:
									time.sleep(1  )
									a       =     AutoTokenizer.from_pretrained(A,    use_fast=A  )
									content["tokenizer_classes"].add(tokenizer_slow.__class__.__name__  )
						except Exception:
									pass
						try:
									time.sleep(1  )
									a       =     AutoImageProcessor.from_pretrained(A  )
									content["processor_classes"].add(img_p.__class__.__name__  )
						except Exception:
									pass
						try:
									time.sleep(1  )
									a       =     AutoFeatureExtractor.from_pretrained(A  )
									if not isinstance(A,    A  ):
												content["processor_classes"].add(feat_p.__class__.__name__  )
						except Exception:
									pass
						try:
									time.sleep(1  )
									a       =     getattr(A,    A  )
									a       =     model_class.from_pretrained(A  )
									content["model_classes"].add(m.__class__.__name__  )
						except Exception:
									pass
						try:
									time.sleep(1  )
									a       =     getattr(A,    F"""TF{model}"""  )
									a       =     model_class.from_pretrained(A  )
									content["model_classes"].add(m.__class__.__name__  )
						except Exception:
									pass

						a       =     sorted(content["tokenizer_classes"]  )
						a       =     sorted(content["processor_classes"]  )
						a       =     sorted(content["model_classes"]  )

						a       =     content
						with open(os.path.join(A,    "hub_tiny_model_summary.json"  ),    "w"  ) as fp:
									json.dump(A,    A,    ensure_ascii=A,    indent=4  )


if __name__ == "__main__":
		__lowerCAmelCase						:							List[Any]			     =						argparse.ArgumentParser()
		parser.add_argument('--num_workers', default=1, type=int, help='The number of workers to run.')
		__lowerCAmelCase						:							Any			     =						parser.parse_args()

		# This has to be `spawn` to avoid hanging forever!
		multiprocessing.set_start_method('spawn')

		__lowerCAmelCase						:							Optional[Any]			     =						'tiny_models'
		__lowerCAmelCase						:							Any			     =						True
		__lowerCAmelCase						:							List[str]			     =						None
		__lowerCAmelCase						:							Tuple			     =						get_tiny_model_names_from_repo()
		__lowerCAmelCase						:							List[Any]			     =						True
		__lowerCAmelCase						:							List[str]			     =						True
		__lowerCAmelCase						:							Union[str, Any]			     =						'hf-internal-testing'

		create_tiny_models(
		    output_path,
		    all,
		    model_types,
		    models_to_skip,
		    no_check,
		    upload,
		    organization,
		    token=os.environ.get('TOKEN', None),
		    num_workers=args.num_workers,
		)












import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


__lowerCAmelCase						:							Optional[int]			     =						'./tests/fixtures/elise_format0.mid'




@unittest.skip("""The note-seq package currently throws an error on import"""  )
class        snake_case__				(_UpperCamelCase       ,			unittest.TestCase  ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_			:				Tuple									=   SpectrogramDiffusionPipeline
				SCREAMING_SNAKE_CASE_			:				Dict									=   PipelineTesterMixin.required_optional_params - {
				    """callback""",
				    """latents""",
				    """callback_steps""",
				    """output_type""",
				    """num_images_per_prompt""",
				}
				SCREAMING_SNAKE_CASE_			:				int									=   False
				SCREAMING_SNAKE_CASE_			:				Tuple									=   False
				SCREAMING_SNAKE_CASE_			:				str									=   TOKENS_TO_AUDIO_GENERATION_PARAMS
				SCREAMING_SNAKE_CASE_			:				List[str]									=   TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS





				def 				__UpperCAmelCase				( self							:	Dict				)   ->	str:
							torch.manual_seed(0				)
							a       =     SpectrogramNotesEncoder(
							    max_length=20_48   , vocab_size=15_36   , d_model=7_68   , dropout_rate=0.1   , num_layers=1   , num_heads=1   , d_kv=4   , d_ff=20_48   , feed_forward_proj="gated-gelu"   , )

							a       =     SpectrogramContEncoder(
							    input_dims=1_28   , targets_context_length=2_56   , d_model=7_68   , dropout_rate=0.1   , num_layers=1   , num_heads=1   , d_kv=4   , d_ff=20_48   , feed_forward_proj="gated-gelu"   , )

							a       =     TaFilmDecoder(
							    input_dims=1_28   , targets_length=2_56   , max_decoder_noise_time=20_000.0   , d_model=7_68   , num_layers=1   , num_heads=1   , d_kv=4   , d_ff=20_48   , dropout_rate=0.1   , )

							a       =     DDPMScheduler()

							a       =     {
							    "notes_encoder": notes_encoder.eval(),
							    "continuous_encoder": continuous_encoder.eval(),
							    "decoder": decoder.eval(),
							    "scheduler": scheduler,
							    "melgan": None,
							}
							return components





				def 				__UpperCAmelCase				( self							:	Any   , __lowerCamelCase							:	Optional[Any]   , __lowerCamelCase							:	Union[str, Any]=0				)   ->	Dict:
							if str(__lowerCamelCase				).startswith("mps"				):
										a       =     torch.manual_seed(__lowerCamelCase				)
							else:
										a       =     torch.Generator(device=__lowerCamelCase				).manual_seed(__lowerCamelCase				)
							a       =     {
							    "input_tokens": [
							        [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
							    ],
							    "generator": generator,
							    "num_inference_steps": 4,
							    "output_type": "mel",
							}
							return inputs





				def 				__UpperCAmelCase				( self							:	Dict				)   ->	Union[str, Any]:
							a       =     "cpu"  # ensure determinism for the device-dependent torch.Generator
							a       =     self.get_dummy_components()
							a       =     SpectrogramDiffusionPipeline(**__lowerCamelCase				)
							a       =     pipe.to(__lowerCamelCase				)
							pipe.set_progress_bar_config(disable=__lowerCamelCase				)

							a       =     self.get_dummy_inputs(__lowerCamelCase				)
							a       =     pipe(**__lowerCamelCase				)
							a       =     output.audios

							a       =     mel[0, -3:, -3:]

							assert mel_slice.shape == (3, 3)
							a       =     np.array(
							    [-11.512_925, -4.788_215, -0.46_172_905, -2.051_715, -10.539_147, -10.970_963, -9.091_634, 4.0, 4.0]				)
							assert np.abs(mel_slice.flatten() - expected_slice				).max() < 1e-2





				@skip_mps
				def 				__UpperCAmelCase				( self							:	List[Any]				)   ->	Tuple:
							return super().test_save_load_local()





				@skip_mps
				def 				__UpperCAmelCase				( self							:	str				)   ->	List[Any]:
							return super().test_dict_tuple_outputs_equivalent()





				@skip_mps
				def 				__UpperCAmelCase				( self							:	int				)   ->	Any:
							return super().test_save_load_optional_components()





				@skip_mps
				def 				__UpperCAmelCase				( self							:	List[Any]				)   ->	List[str]:
							return super().test_attention_slicing_forward_pass()





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





				def 				__UpperCAmelCase				( self							:	Tuple				)   ->	Tuple:
							pass





				@skip_mps
				def 				__UpperCAmelCase				( self							:	int				)   ->	Optional[Any]:
							return super().test_progress_bar()




@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class        snake_case__				(unittest.TestCase  ):
				"""simple docstring"""



				def 				__UpperCAmelCase				( self							:	Union[str, Any]				)   ->	Dict:
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()





				def 				__UpperCAmelCase				( self							:	Union[str, Any]				)   ->	Any:
							# TODO - test that pipeline can decode tokens in a callback
							# so that music can be played live
							a       =     torch_device

							a       =     SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"				)
							a       =     pipe.melgan
							a       =     None

							a       =     pipe.to(__lowerCamelCase				)
							pipe.set_progress_bar_config(disable=__lowerCamelCase				)

							def callback(__lowerCamelCase							:	List[str]   , __lowerCamelCase							:	Union[str, Any]				):
										# decode mel to audio
										a       =     melgan(input_features=mel_output.astype(np.floataa				)				)[0]
										assert len(audio[0]				) == 8_19_20 * (step + 1)
										# simulate that audio is played
										return audio

							a       =     MidiProcessor()
							a       =     processor(__lowerCamelCase				)

							a       =     input_tokens[:3]
							a       =     torch.manual_seed(0				)
							pipe(__lowerCamelCase   , num_inference_steps=5   , generator=__lowerCamelCase   , callback=__lowerCamelCase   , output_type="mel"				)





				def 				__UpperCAmelCase				( self							:	Any				)   ->	Union[str, Any]:
							a       =     torch_device

							a       =     SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"				)
							a       =     pipe.to(__lowerCamelCase				)
							pipe.set_progress_bar_config(disable=__lowerCamelCase				)
							a       =     MidiProcessor()

							a       =     processor(__lowerCamelCase				)
							# just run two denoising loops
							a       =     input_tokens[:2]

							a       =     torch.manual_seed(0				)
							a       =     pipe(__lowerCamelCase   , num_inference_steps=2   , generator=__lowerCamelCase				)

							a       =     output.audios[0]

							assert abs(np.abs(__lowerCamelCase				).sum() - 3_612.841				) < 1e-1





				def 				__UpperCAmelCase				( self							:	List[str]				)   ->	str:
							a       =     torch_device

							a       =     SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"				)
							a       =     pipe.to(__lowerCamelCase				)
							pipe.set_progress_bar_config(disable=__lowerCamelCase				)

							a       =     MidiProcessor()

							a       =     processor(__lowerCamelCase				)

							# just run 4 denoising loops
							a       =     input_tokens[:4]

							a       =     torch.manual_seed(0				)
							a       =     pipe(__lowerCamelCase   , num_inference_steps=1_00   , generator=__lowerCamelCase				)

							a       =     output.audios[0]
							assert abs(np.abs(__lowerCamelCase				).sum() - 9_389.1_111				) < 5e-2






