

'''simple docstring'''



import argparse
import json
from pathlib import Path

import torch
import torchaudio
from datasets import load_dataset
from huggingface_hub import hf_hub_download

from transformers import ASTConfig, ASTFeatureExtractor, ASTForAudioClassification
from transformers.utils import logging


logging.set_verbosity_info()
__snake_case			:     Tuple										=      logging.get_logger(__name__)
def 						__lowerCamelCase	(       __snake_case					:		Optional[int]	)	->							Union[str, Any]:






						"""simple docstring"""





						A__					:       Optional[int]     =ASTConfig()

						if "10-10" in model_name:
												pass
						elif "speech-commands" in model_name:
												A__					:       Optional[Any]     =128
						elif "12-12" in model_name:
												A__					:       List[str]     =12
												A__					:       List[Any]     =12
						elif "14-14" in model_name:
												A__					:       List[Any]     =14
												A__					:       Dict     =14
						elif "16-16" in model_name:
												A__					:       List[Any]     =16
												A__					:       Tuple     =16
						else:
												raise ValueError("""Model not supported"""	)

						A__					:       Union[str, Any]     ="""huggingface/label-files"""
						if "speech-commands" in model_name:
												A__					:       Union[str, Any]     =35
												A__					:       Optional[int]     ="""speech-commands-v2-id2label.json"""
						else:
												A__					:       Dict     =527
												A__					:       int     ="""audioset-id2label.json"""

						A__					:       Dict     =json.load(open(hf_hub_download(a__,		a__,		repo_type="""dataset"""	),		"""r"""	)	)
						A__					:       str     ={int(a__	): v for k, v in idalabel.items()}
						A__					:       int     =idalabel
						A__					:       List[Any]     ={v: k for k, v in idalabel.items()}

						return config
def 						__lowerCamelCase	(       __snake_case					:		Union[str, Any]	)	->							Tuple:






						"""simple docstring"""





						if "module.v" in name:
												A__					:       str     =name.replace("""module.v""",		"""audio_spectrogram_transformer"""	)
						if "cls_token" in name:
												A__					:       Dict     =name.replace("""cls_token""",		"""embeddings.cls_token"""	)
						if "dist_token" in name:
												A__					:       str     =name.replace("""dist_token""",		"""embeddings.distillation_token"""	)
						if "pos_embed" in name:
												A__					:       Tuple     =name.replace("""pos_embed""",		"""embeddings.position_embeddings"""	)
						if "patch_embed.proj" in name:
												A__					:       Union[str, Any]     =name.replace("""patch_embed.proj""",		"""embeddings.patch_embeddings.projection"""	)
						# transformer blocks
						if "blocks" in name:
												A__					:       Union[str, Any]     =name.replace("""blocks""",		"""encoder.layer"""	)
						if "attn.proj" in name:
												A__					:       Optional[Any]     =name.replace("""attn.proj""",		"""attention.output.dense"""	)
						if "attn" in name:
												A__					:       Union[str, Any]     =name.replace("""attn""",		"""attention.self"""	)
						if "norm1" in name:
												A__					:       Union[str, Any]     =name.replace("""norm1""",		"""layernorm_before"""	)
						if "norm2" in name:
												A__					:       str     =name.replace("""norm2""",		"""layernorm_after"""	)
						if "mlp.fc1" in name:
												A__					:       Any     =name.replace("""mlp.fc1""",		"""intermediate.dense"""	)
						if "mlp.fc2" in name:
												A__					:       Dict     =name.replace("""mlp.fc2""",		"""output.dense"""	)
						# final layernorm
						if "audio_spectrogram_transformer.norm" in name:
												A__					:       Dict     =name.replace("""audio_spectrogram_transformer.norm""",		"""audio_spectrogram_transformer.layernorm"""	)
						# classifier head
						if "module.mlp_head.0" in name:
												A__					:       Tuple     =name.replace("""module.mlp_head.0""",		"""classifier.layernorm"""	)
						if "module.mlp_head.1" in name:
												A__					:       List[Any]     =name.replace("""module.mlp_head.1""",		"""classifier.dense"""	)

						return name
def 						__lowerCamelCase	(       __snake_case					:		str,		__snake_case					:		Optional[Any]	)	->							List[str]:






						"""simple docstring"""





						for key in orig_state_dict.copy().keys():
												A__					:       Optional[Any]     =orig_state_dict.pop(a__	)

												if "qkv" in key:
																		A__					:       int     =key.split("""."""	)
																		A__					:       str     =int(key_split[3]	)
																		A__					:       List[str]     =config.hidden_size
																		if "weight" in key:
																								A__					:       List[str]     =val[:dim, :]
																								A__					:       List[Any]     =val[dim : dim * 2, :]
																								A__					:       List[str]     =val[-dim:, :]
																		else:
																								A__					:       Tuple     =val[:dim]
																								A__					:       List[str]     =val[dim : dim * 2]
																								A__					:       Optional[Any]     =val[-dim:]
												else:
																		A__					:       Tuple     =val

						return orig_state_dict
def 						__lowerCamelCase	(       __snake_case					:		str	)	->							str:






						"""simple docstring"""





						A__					:       Tuple     =[
						    """module.v.head.weight""",
						    """module.v.head.bias""",
						    """module.v.head_dist.weight""",
						    """module.v.head_dist.bias""",
						]
						for k in ignore_keys:
												state_dict.pop(a__,		a__	)
@torch.no_grad()
def 						__lowerCamelCase	(       __snake_case					:		str,		__snake_case					:		Optional[Any],		__snake_case					:		List[Any]=False	)	->							Any:






						"""simple docstring"""





						A__					:       List[str]     =get_audio_spectrogram_transformer_config(a__	)

						A__					:       Any     ={
						    """ast-finetuned-audioset-10-10-0.4593""": (
						        """https://www.dropbox.com/s/ca0b1v2nlxzyeb4/audioset_10_10_0.4593.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-10-10-0.450""": (
						        """https://www.dropbox.com/s/1tv0hovue1bxupk/audioset_10_10_0.4495.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-10-10-0.448""": (
						        """https://www.dropbox.com/s/6u5sikl4b9wo4u5/audioset_10_10_0.4483.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-10-10-0.448-v2""": (
						        """https://www.dropbox.com/s/kt6i0v9fvfm1mbq/audioset_10_10_0.4475.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-12-12-0.447""": (
						        """https://www.dropbox.com/s/snfhx3tizr4nuc8/audioset_12_12_0.4467.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-14-14-0.443""": (
						        """https://www.dropbox.com/s/z18s6pemtnxm4k7/audioset_14_14_0.4431.pth?dl=1"""
						    ),
						    """ast-finetuned-audioset-16-16-0.442""": (
						        """https://www.dropbox.com/s/mdsa4t1xmcimia6/audioset_16_16_0.4422.pth?dl=1"""
						    ),
						    """ast-finetuned-speech-commands-v2""": (
						        """https://www.dropbox.com/s/q0tbqpwv44pquwy/speechcommands_10_10_0.9812.pth?dl=1"""
						    ),
						}

						# load original state_dict
						A__					:       Optional[Any]     =model_name_to_url[model_name]
						A__					:       Tuple     =torch.hub.load_state_dict_from_url(a__,		map_location="""cpu"""	)
						# remove some keys
						remove_keys(a__	)
						# rename some keys
						A__					:       List[str]     =convert_state_dict(a__,		a__	)

						# load 🤗 model
						A__					:       Any     =ASTForAudioClassification(a__	)
						model.eval()

						model.load_state_dict(a__	)

						# verify outputs on dummy input
						# source: https://github.com/YuanGongND/ast/blob/79e873b8a54d0a3b330dd522584ff2b9926cd581/src/run.py#L62
						A__					:       List[Any]     =-4.2_67_73_93 if """speech-commands""" not in model_name else -6.84_59_78
						A__					:       Optional[int]     =4.5_68_99_74 if """speech-commands""" not in model_name else 5.5_65_45_26
						A__					:       Dict     =1_024 if """speech-commands""" not in model_name else 128
						A__					:       Optional[int]     =ASTFeatureExtractor(mean=a__,		std=a__,		max_length=a__	)

						if "speech-commands" in model_name:
												A__					:       Tuple     =load_dataset("""speech_commands""",		"""v0.02""",		split="""validation"""	)
												A__					:       Tuple     =dataset[0]["""audio"""]["""array"""]
						else:
												A__					:       Tuple     =hf_hub_download(
												    repo_id="""nielsr/audio-spectogram-transformer-checkpoint""",		filename="""sample_audio.flac""",		repo_type="""dataset""",		)

												A__   ,						A__					:       Any     =torchaudio.load(a__	)
												A__					:       Optional[int]     =waveform.squeeze().numpy()

						A__					:       Optional[int]     =feature_extractor(a__,		sampling_rate=16_000,		return_tensors="""pt"""	)

						# forward pass
						A__					:       List[str]     =model(**a__	)
						A__					:       Any     =outputs.logits

						if model_name == "ast-finetuned-audioset-10-10-0.4593":
												A__					:       Optional[Any]     =torch.tensor([-0.87_60, -7.00_42, -8.66_02]	)
						elif model_name == "ast-finetuned-audioset-10-10-0.450":
												A__					:       Optional[Any]     =torch.tensor([-1.19_86, -7.09_03, -8.27_18]	)
						elif model_name == "ast-finetuned-audioset-10-10-0.448":
												A__					:       Any     =torch.tensor([-2.61_28, -8.00_80, -9.43_44]	)
						elif model_name == "ast-finetuned-audioset-10-10-0.448-v2":
												A__					:       Tuple     =torch.tensor([-1.50_80, -7.45_34, -8.89_17]	)
						elif model_name == "ast-finetuned-audioset-12-12-0.447":
												A__					:       Union[str, Any]     =torch.tensor([-0.50_50, -6.58_33, -8.08_43]	)
						elif model_name == "ast-finetuned-audioset-14-14-0.443":
												A__					:       Any     =torch.tensor([-0.38_26, -7.03_36, -8.24_13]	)
						elif model_name == "ast-finetuned-audioset-16-16-0.442":
												A__					:       List[str]     =torch.tensor([-1.21_13, -6.91_01, -8.34_70]	)
						elif model_name == "ast-finetuned-speech-commands-v2":
												A__					:       Tuple     =torch.tensor([6.15_89, -8.05_66, -8.79_84]	)
						else:
												raise ValueError("""Unknown model name"""	)
						if not torch.allclose(logits[0, :3],		a__,		atol=1E-4	):
												raise ValueError("""Logits don't match"""	)
						print("""Looks ok!"""	)

						if pytorch_dump_folder_path is not None:
												Path(a__	).mkdir(exist_ok=a__	)
												print(f"Saving model {model_name} to {pytorch_dump_folder_path}"	)
												model.save_pretrained(a__	)
												print(f"Saving feature extractor to {pytorch_dump_folder_path}"	)
												feature_extractor.save_pretrained(a__	)

						if push_to_hub:
												print("""Pushing model and feature extractor to the hub..."""	)
												model.push_to_hub(f"MIT/{model_name}"	)
												feature_extractor.push_to_hub(f"MIT/{model_name}"	)


if __name__ == "__main__":
		__snake_case			:     Union[str, Any]										=      argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    '--model_name',
		    default='ast-finetuned-audioset-10-10-0.4593',
		    type=str,
		    help='Name of the Audio Spectrogram Transformer model you\'d like to convert.',
		)
		parser.add_argument(
		    '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
		)
		parser.add_argument(
		    '--push_to_hub', action='store_true', help='Whether or not to push the converted model to the 🤗 hub.'
		)

		__snake_case			:     Any										=      parser.parse_args()
		convert_audio_spectrogram_transformer_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)




'''simple docstring'''



import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
		import jax
		import jax.numpy as jnp



@slow
@require_flax
class lowerCamelCase			(  unittest.TestCase						):



			'''simple docstring'''






			def lowercase__						(				self	:						Optional[Any]		,      lowerCAmelCase_	:						Union[str, Any]		,      lowerCAmelCase_	:						List[str]			)      ->		List[str]:





									'''simple docstring'''
									return f"gaussian_noise_s={seed}_shape={'_'.join([str(lowerCAmelCase_			) for s in shape]			)}.npy"


			def lowercase__						(				self	:						Optional[Any]			)      ->		Dict:





									'''simple docstring'''
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()


			def lowercase__						(				self	:						Tuple		,      lowerCAmelCase_	:						str=0		,      lowerCAmelCase_	:						List[str]=(4, 4, 64, 64)		,      lowerCAmelCase_	:						Optional[Any]=False			)      ->		str:





									'''simple docstring'''
									A__					:       Union[str, Any]     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       Any     =jnp.array(load_hf_numpy(self.get_file_format(lowerCAmelCase_		,      lowerCAmelCase_			)			)		,      dtype=lowerCAmelCase_			)
									return image


			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						Optional[int]=False		,      lowerCAmelCase_	:						Dict="CompVis/stable-diffusion-v1-4"			)      ->		List[Any]:





									'''simple docstring'''
									A__					:       Any     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       int     ="""bf16""" if fpaa else None

									A__   ,						A__					:       Any     =FlaxUNetaDConditionModel.from_pretrained(
									    lowerCAmelCase_		,      subfolder="""unet"""		,      dtype=lowerCAmelCase_		,      revision=lowerCAmelCase_			)
									return model, params


			def lowercase__						(				self	:						Optional[Any]		,      lowerCAmelCase_	:						int=0		,      lowerCAmelCase_	:						Optional[Any]=(4, 77, 7_68)		,      lowerCAmelCase_	:						List[str]=False			)      ->		Tuple:





									'''simple docstring'''
									A__					:       Union[str, Any]     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       Optional[int]     =jnp.array(load_hf_numpy(self.get_file_format(lowerCAmelCase_		,      lowerCAmelCase_			)			)		,      dtype=lowerCAmelCase_			)
									return hidden_states


			@parameterized.expand(
			    [
			        # fmt: off
			        [83, 4, [-0.2323, -0.1304, 0.0813, -0.3093, -0.0919, -0.1571, -0.1125, -0.5806]],
			        [17, 0.55, [-0.0831, -0.2443, 0.0901, -0.0919, 0.3396, 0.0103, -0.3743, 0.0701]],
			        [8, 0.89, [-0.4863, 0.0859, 0.0875, -0.1658, 0.9199, -0.0114, 0.4839, 0.4639]],
			        [3, 10_00, [-0.5649, 0.2402, -0.5518, 0.1248, 1.1328, -0.2443, -0.0325, -1.0078]],
			        # fmt: on
			    ]			)
			def lowercase__						(				self	:						Any		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						str			)      ->		int:





									'''simple docstring'''
									A__   ,						A__					:       Tuple     =self.get_unet_model(model_id="""CompVis/stable-diffusion-v1-4"""		,      fpaa=lowerCAmelCase_			)
									A__					:       List[Any]     =self.get_latents(lowerCAmelCase_		,      fpaa=lowerCAmelCase_			)
									A__					:       Optional[int]     =self.get_encoder_hidden_states(lowerCAmelCase_		,      fpaa=lowerCAmelCase_			)

									A__					:       Union[str, Any]     =model.apply(
									    {"""params""": params}		,      lowerCAmelCase_		,      jnp.array(lowerCAmelCase_		,      dtype=jnp.intaa			)		,      encoder_hidden_states=lowerCAmelCase_		,      ).sample

									assert sample.shape == latents.shape

									A__					:       Optional[int]     =jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())			)		,      dtype=jnp.floataa			)
									A__					:       Tuple     =jnp.array(lowerCAmelCase_		,      dtype=jnp.floataa			)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
									assert jnp.allclose(lowerCAmelCase_		,      lowerCAmelCase_		,      atol=1e-2			)



			@parameterized.expand(
			    [
			        # fmt: off
			        [83, 4, [0.1514, 0.0807, 0.1624, 0.1016, -0.1896, 0.0263, 0.0677, 0.2310]],
			        [17, 0.55, [0.1164, -0.0216, 0.0170, 0.1589, -0.3120, 0.1005, -0.0581, -0.1458]],
			        [8, 0.89, [-0.1758, -0.0169, 0.1004, -0.1411, 0.1312, 0.1103, -0.1996, 0.2139]],
			        [3, 10_00, [0.1214, 0.0352, -0.0731, -0.1562, -0.0994, -0.0906, -0.2340, -0.0539]],
			        # fmt: on
			    ]			)
			def lowercase__						(				self	:						Dict		,      lowerCAmelCase_	:						str		,      lowerCAmelCase_	:						int		,      lowerCAmelCase_	:						Optional[int]			)      ->		List[str]:





									'''simple docstring'''
									A__   ,						A__					:       List[str]     =self.get_unet_model(model_id="""stabilityai/stable-diffusion-2"""		,      fpaa=lowerCAmelCase_			)
									A__					:       Union[str, Any]     =self.get_latents(lowerCAmelCase_		,      shape=(4, 4, 96, 96)		,      fpaa=lowerCAmelCase_			)
									A__					:       Dict     =self.get_encoder_hidden_states(lowerCAmelCase_		,      shape=(4, 77, 10_24)		,      fpaa=lowerCAmelCase_			)

									A__					:       Optional[Any]     =model.apply(
									    {"""params""": params}		,      lowerCAmelCase_		,      jnp.array(lowerCAmelCase_		,      dtype=jnp.intaa			)		,      encoder_hidden_states=lowerCAmelCase_		,      ).sample

									assert sample.shape == latents.shape

									A__					:       Optional[Any]     =jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())			)		,      dtype=jnp.floataa			)
									A__					:       List[Any]     =jnp.array(lowerCAmelCase_		,      dtype=jnp.floataa			)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
									assert jnp.allclose(lowerCAmelCase_		,      lowerCAmelCase_		,      atol=1e-2			)

