
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()
lowerCamelCase__	=							logging.get_logger(__name__)


def 							__A(lowerCAmelCase )		->				int:

		"""simple docstring"""






		_UpperCamelCase						    =	ASTConfig()

		if "10-10" in model_name:
				pass
		elif "speech-commands" in model_name:
				_UpperCamelCase						    =	1_2_8
		elif "12-12" in model_name:
				_UpperCamelCase						    =	1_2
				_UpperCamelCase						    =	1_2
		elif "14-14" in model_name:
				_UpperCamelCase						    =	1_4
				_UpperCamelCase						    =	1_4
		elif "16-16" in model_name:
				_UpperCamelCase						    =	1_6
				_UpperCamelCase						    =	1_6
		else:
				raise ValueError("""Model not supported""" )

		_UpperCamelCase						    =	"""huggingface/label-files"""
		if "speech-commands" in model_name:
				_UpperCamelCase						    =	3_5
				_UpperCamelCase						    =	"""speech-commands-v2-id2label.json"""
		else:
				_UpperCamelCase						    =	5_2_7
				_UpperCamelCase						    =	"""audioset-id2label.json"""

		_UpperCamelCase						    =	json.load(open(hf_hub_download(lowerCAmelCase						,  lowerCAmelCase						,  repo_type="""dataset""" )						,  """r""" ) )
		_UpperCamelCase						    =	{int(lowerCAmelCase ): v for k, v in idalabel.items()}
		_UpperCamelCase						    =	idalabel
		_UpperCamelCase						    =	{v: k for k, v in idalabel.items()}

		return config


def 							__A(lowerCAmelCase )		->				Union[str, Any]:

		"""simple docstring"""






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

		return name


def 							__A(lowerCAmelCase						,  lowerCAmelCase )		->				Dict:

		"""simple docstring"""






		for key in orig_state_dict.copy().keys():
				_UpperCamelCase						    =	orig_state_dict.pop(lowerCAmelCase )

				if "qkv" in key:
						_UpperCamelCase						    =	key.split(""".""" )
						_UpperCamelCase						    =	int(key_split[3] )
						_UpperCamelCase						    =	config.hidden_size
						if "weight" in key:
								_UpperCamelCase						    =	val[:dim, :]
								_UpperCamelCase						    =	val[dim : dim * 2, :]
								_UpperCamelCase						    =	val[-dim:, :]
						else:
								_UpperCamelCase						    =	val[:dim]
								_UpperCamelCase						    =	val[dim : dim * 2]
								_UpperCamelCase						    =	val[-dim:]
				else:
						_UpperCamelCase						    =	val

		return orig_state_dict


def 							__A(lowerCAmelCase )		->				Tuple:

		"""simple docstring"""






		_UpperCamelCase						    =	[
		    """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(lowerCAmelCase						,  lowerCAmelCase )


@torch.no_grad()
def 							__A(lowerCAmelCase						,  lowerCAmelCase						,  lowerCAmelCase=False )		->				List[str]:

		"""simple docstring"""






		_UpperCamelCase						    =	get_audio_spectrogram_transformer_config(lowerCAmelCase )

		_UpperCamelCase						    =	{
		    """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
		_UpperCamelCase						    =	model_name_to_url[model_name]
		_UpperCamelCase						    =	torch.hub.load_state_dict_from_url(lowerCAmelCase						,  map_location="""cpu""" )
		# remove some keys
		remove_keys(lowerCAmelCase )
		# rename some keys
		_UpperCamelCase						    =	convert_state_dict(lowerCAmelCase						,  lowerCAmelCase )

		# load 🤗 model
		_UpperCamelCase						    =	ASTForAudioClassification(lowerCAmelCase )
		model.eval()

		model.load_state_dict(lowerCAmelCase )

		# verify outputs on dummy input
		# source: https://github.com/YuanGongND/ast/blob/79e873b8a54d0a3b330dd522584ff2b9926cd581/src/run.py#L62
		_UpperCamelCase						    =	-4.2677393 if """speech-commands""" not in model_name else -6.845978
		_UpperCamelCase						    =	4.5689974 if """speech-commands""" not in model_name else 5.5654526
		_UpperCamelCase						    =	1_0_2_4 if """speech-commands""" not in model_name else 1_2_8
		_UpperCamelCase						    =	ASTFeatureExtractor(mean=lowerCAmelCase						,  std=lowerCAmelCase						,  max_length=lowerCAmelCase )

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

				_UpperCamelCase    ,	_UpperCamelCase						    =	torchaudio.load(lowerCAmelCase )
				_UpperCamelCase						    =	waveform.squeeze().numpy()

		_UpperCamelCase						    =	feature_extractor(lowerCAmelCase						,  sampling_rate=1_6_0_0_0						,  return_tensors="""pt""" )

		# forward pass
		_UpperCamelCase						    =	model(**lowerCAmelCase )
		_UpperCamelCase						    =	outputs.logits

		if model_name == "ast-finetuned-audioset-10-10-0.4593":
				_UpperCamelCase						    =	torch.tensor([-0.8760, -7.0042, -8.6602] )
		elif model_name == "ast-finetuned-audioset-10-10-0.450":
				_UpperCamelCase						    =	torch.tensor([-1.1986, -7.0903, -8.2718] )
		elif model_name == "ast-finetuned-audioset-10-10-0.448":
				_UpperCamelCase						    =	torch.tensor([-2.6128, -8.0080, -9.4344] )
		elif model_name == "ast-finetuned-audioset-10-10-0.448-v2":
				_UpperCamelCase						    =	torch.tensor([-1.5080, -7.4534, -8.8917] )
		elif model_name == "ast-finetuned-audioset-12-12-0.447":
				_UpperCamelCase						    =	torch.tensor([-0.5050, -6.5833, -8.0843] )
		elif model_name == "ast-finetuned-audioset-14-14-0.443":
				_UpperCamelCase						    =	torch.tensor([-0.3826, -7.0336, -8.2413] )
		elif model_name == "ast-finetuned-audioset-16-16-0.442":
				_UpperCamelCase						    =	torch.tensor([-1.2113, -6.9101, -8.3470] )
		elif model_name == "ast-finetuned-speech-commands-v2":
				_UpperCamelCase						    =	torch.tensor([6.1589, -8.0566, -8.7984] )
		else:
				raise ValueError("""Unknown model name""" )
		if not torch.allclose(logits[0, :3]						,  lowerCAmelCase						,  atol=1e-4 ):
				raise ValueError("""Logits don't match""" )
		print("""Looks ok!""" )

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

		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__":
		lowerCamelCase__	=							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."
		)

		lowerCamelCase__	=							parser.parse_args()
		convert_audio_spectrogram_transformer_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)





from typing import List, Optional, Tuple, Union

import torch

from ...utils import logging, randn_tensor
from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline


lowerCamelCase__	=							logging.get_logger(__name__)  # pylint: disable=invalid-name







class        lowerCAmelCase__      (   __lowercase	):

						def __init__(			self       ,							a       ,							a				)					->      Any:


								'''simple docstring'''

								super().__init__()
								self.register_modules(unet=a       ,							scheduler=a				)







						@torch.no_grad()
						def __call__(			self       ,							a = 1       ,							a = 1_00       ,							a = None       ,							a = None       ,							a = True       ,							)					->      Union[AudioPipelineOutput, Tuple]:


								'''simple docstring'''

								if audio_length_in_s is None:
										_UpperCamelCase						    =	self.unet.config.sample_size / self.unet.config.sample_rate

								_UpperCamelCase						    =	audio_length_in_s * self.unet.config.sample_rate

								_UpperCamelCase						    =	2 ** len(self.unet.up_blocks				)
								if sample_size < 3 * down_scale_factor:
										raise ValueError(
										    F'{audio_length_in_s} is too small. Make sure it\'s bigger or equal to'
										    F' {3 * down_scale_factor / self.unet.config.sample_rate}.'				)

								_UpperCamelCase						    =	int(a				)
								if sample_size % down_scale_factor != 0:
										_UpperCamelCase						    =	(
										    (audio_length_in_s * self.unet.config.sample_rate) // down_scale_factor + 1
										) * down_scale_factor
										logger.info(
										    F'{audio_length_in_s} is increased to {sample_size / self.unet.config.sample_rate} so that it can be handled'
										    F' by the model. It will be cut to {original_sample_size / self.unet.config.sample_rate} after the denoising'
										    """ process."""				)
								_UpperCamelCase						    =	int(a				)

								_UpperCamelCase						    =	next(iter(self.unet.parameters()				)				).dtype
								_UpperCamelCase						    =	(batch_size, self.unet.config.in_channels, sample_size)
								if isinstance(a       ,							a				) and len(a				) != batch_size:
										raise ValueError(
										    F'You have passed a list of generators of length {len(a				)}, but requested an effective batch'
										    F' size of {batch_size}. Make sure the batch size matches the length of the generators.'				)

								_UpperCamelCase						    =	randn_tensor(a       ,							generator=a       ,							device=self.device       ,							dtype=a				)

								# set step values
								self.scheduler.set_timesteps(a       ,							device=audio.device				)
								_UpperCamelCase						    =	self.scheduler.timesteps.to(a				)

								for t in self.progress_bar(self.scheduler.timesteps				):
										# 1. predict noise model_output
										_UpperCamelCase						    =	self.unet(a       ,							a				).sample

										# 2. compute previous image: x_t -> t_t-1
										_UpperCamelCase						    =	self.scheduler.step(a       ,							a       ,							a				).prev_sample

								_UpperCamelCase						    =	audio.clamp(-1       ,							1				).float().cpu().numpy()

								_UpperCamelCase						    =	audio[:, :, :original_sample_size]

								if not return_dict:
										return (audio,)

								return AudioPipelineOutput(audios=a				)



