
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()
A     :	Any    = logging.get_logger(__name__)


def     __lowerCAmelCase			(			a__  )       ->  str:
				__a										=    ASTConfig()

				if "10-10" in model_name:
								pass
				elif "speech-commands" in model_name:
								__a										=    128
				elif "12-12" in model_name:
								__a										=    12
								__a										=    12
				elif "14-14" in model_name:
								__a										=    14
								__a										=    14
				elif "16-16" in model_name:
								__a										=    16
								__a										=    16
				else:
								raise ValueError('''Model not supported'''  )

				__a										=    'huggingface/label-files'
				if "speech-commands" in model_name:
								__a										=    35
								__a										=    'speech-commands-v2-id2label.json'
				else:
								__a										=    527
								__a										=    'audioset-id2label.json'

				__a										=    json.load(open(hf_hub_download(__A						,    __A						,    repo_type='''dataset'''  )						,    '''r'''  )  )
				__a										=    {int(__A  ): v for k, v in idalabel.items()}
				__a										=    idalabel
				__a										=    {v: k for k, v in idalabel.items()}

				return config


def     __lowerCAmelCase			(			a__  )       ->  Optional[int]:

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

				return name


def     __lowerCAmelCase			(			a__						,    a__  )       ->  Optional[int]:

				for key in orig_state_dict.copy().keys():
								__a										=    orig_state_dict.pop(__A  )

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

				return orig_state_dict


def     __lowerCAmelCase			(			a__  )       ->  Union[str, Any]:
				__a										=    [
				    '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			(			a__						,    a__						,    a__=False  )       ->  Optional[int]:
				__a										=    get_audio_spectrogram_transformer_config(__A  )

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

				# load 🤗 model
				__a										=    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										=    -4.2_677_393 if 'speech-commands' not in model_name else -6.845_978
				__a										=    4.5_689_974 if 'speech-commands' not in model_name else 5.5_654_526
				__a										=    1024 if 'speech-commands' not in model_name else 128
				__a										=    ASTFeatureExtractor(mean=__A						,    std=__A						,    max_length=__A  )

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

								__a										=    torchaudio.load(__A  )
								__a										=    waveform.squeeze().numpy()

				__a										=    feature_extractor(__A						,    sampling_rate=1_6000						,    return_tensors='''pt'''  )

				# forward pass
				__a										=    model(**__A  )
				__a										=    outputs.logits

				if model_name == "ast-finetuned-audioset-10-10-0.4593":
								__a										=    torch.tensor([-0.8_760, -7.0_042, -8.6_602]  )
				elif model_name == "ast-finetuned-audioset-10-10-0.450":
								__a										=    torch.tensor([-1.1_986, -7.0_903, -8.2_718]  )
				elif model_name == "ast-finetuned-audioset-10-10-0.448":
								__a										=    torch.tensor([-2.6_128, -8.0_080, -9.4_344]  )
				elif model_name == "ast-finetuned-audioset-10-10-0.448-v2":
								__a										=    torch.tensor([-1.5_080, -7.4_534, -8.8_917]  )
				elif model_name == "ast-finetuned-audioset-12-12-0.447":
								__a										=    torch.tensor([-0.5_050, -6.5_833, -8.0_843]  )
				elif model_name == "ast-finetuned-audioset-14-14-0.443":
								__a										=    torch.tensor([-0.3_826, -7.0_336, -8.2_413]  )
				elif model_name == "ast-finetuned-audioset-16-16-0.442":
								__a										=    torch.tensor([-1.2_113, -6.9_101, -8.3_470]  )
				elif model_name == "ast-finetuned-speech-commands-v2":
								__a										=    torch.tensor([6.1_589, -8.0_566, -8.7_984]  )
				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__":
					A     :	str    = 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.'
					)

					A     :	str    = parser.parse_args()
					convert_audio_spectrogram_transformer_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)



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  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      snake_case__				:     Optional[Any]        =						TextToVideoSDPipeline
      snake_case__				:     Optional[int]        =						TEXT_TO_IMAGE_PARAMS
      snake_case__				:     str        =						TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__				:     Optional[Any]        =						frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]			)
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  UNetaDConditionModel(
                 block_out_channels=(3_2, 6_4, 6_4, 6_4)     ,							layers_per_block=2     ,							sample_size=3_2     ,							in_channels=4     ,							out_channels=4     ,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')     ,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')     ,							cross_attention_dim=3_2     ,							attention_head_dim=4     ,							)
             a_					:		int							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)
             torch.manual_seed(0     )
             a_					:		int							=  AutoencoderKL(
                 block_out_channels=[3_2, 6_4]     ,							in_channels=3     ,							out_channels=3     ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']     ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']     ,							latent_channels=4     ,							sample_size=1_2_8     ,							)
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							num_attention_heads=4     ,							num_hidden_layers=5     ,							pad_token_id=1     ,							vocab_size=1_0_0_0     ,							hidden_act='gelu'     ,							projection_dim=5_1_2     ,							)
             a_					:		Dict							=  CLIPTextModel(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             a_					:		Union[str, Any]							=  {
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0     )		->			List[str]:
             if str(SCREAMING_SNAKE_CASE__     ).startswith('mps'     ):
                    a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             else:
                    a_					:		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		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        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Tuple:
             a_					:		List[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.get_dummy_components()
             a_					:		str							=  TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  'np'
             a_					:		Dict							=  sd_pipe(**SCREAMING_SNAKE_CASE__     ).frames
             a_					:		int							=  frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_					:		Union[str, Any]							=  np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			List[Any]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							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        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			str:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=1E-2     )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			List[str]:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     )		->			List[Any]:
             pass

      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             return super().test_progress_bar()
@slow
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             a_					:		str							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'     )

             a_					:		Any							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Optional[Any]							=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
             a_					:		Optional[Any]							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[Any]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		Optional[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2_5     ,							output_type='pt'     ).frames
             a_					:		str							=  video_frames.cpu().numpy()

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

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Any:
             a_					:		Dict							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'     )

             a_					:		Tuple							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Tuple							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[str]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		List[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2     ,							output_type='pt'     ).frames
             a_					:		List[str]							=  video_frames.cpu().numpy()

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


