"""simple docstring"""





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 			_snake_case		:







				"""simple docstring"""






				def __init__(   self				:				Any   ,     _A				:				List[Any]   ,     _A				:				int=1_3   ,     _A				:				Optional[int]=2   ,     _A				:				Dict=2_4   ,     _A				:				int=1_6   ,     _A				:				Optional[int]=True   ,     _A				:				str=True   ,     _A				:				Dict=3_2   ,     _A				:				List[str]=5   ,     _A				:				str=4   ,     _A				:				str=3_7   ,     _A				:				Union[str, Any]="gelu"   ,     _A				:				str=0.1   ,     _A				:				Any=0.1   ,     _A				:				int=1_0   ,     _A				:				Dict=0.02   ,     _A				:				Optional[int]=None   ,     _A				:				str=2   ,     _A				:				Optional[int]=2   ,     ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Any      	=     parent
											_SCREAMING_SNAKE_CASE		:    str      	=     batch_size
											_SCREAMING_SNAKE_CASE		:    List[str]      	=     patch_size
											_SCREAMING_SNAKE_CASE		:    Dict      	=     max_length
											_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     num_mel_bins
											_SCREAMING_SNAKE_CASE		:    Any      	=     is_training
											_SCREAMING_SNAKE_CASE		:    Any      	=     use_labels
											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     hidden_size
											_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     num_hidden_layers
											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     num_attention_heads
											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     intermediate_size
											_SCREAMING_SNAKE_CASE		:    List[str]      	=     hidden_act
											_SCREAMING_SNAKE_CASE		:    str      	=     hidden_dropout_prob
											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     attention_probs_dropout_prob
											_SCREAMING_SNAKE_CASE		:    Any      	=     type_sequence_label_size
											_SCREAMING_SNAKE_CASE		:    Dict      	=     initializer_range
											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     scope
											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     frequency_stride
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     time_stride

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





				def 		_lowerCAmelCase		(   self				:				Optional[int]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Dict      	=     floats_tensor([self.batch_size, self.max_length, self.num_mel_bins])

											_SCREAMING_SNAKE_CASE		:    List[str]      	=     None
											if self.use_labels:
																		_SCREAMING_SNAKE_CASE		:    int      	=     ids_tensor([self.batch_size]   ,     self.type_sequence_label_size)

											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     self.get_config()

											return config, input_values, labels





				def 		_lowerCAmelCase		(   self				:				List[str]):



											"""simple docstring"""







											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=lowercase_   ,     initializer_range=self.initializer_range   ,     frequency_stride=self.frequency_stride   ,     time_stride=self.time_stride   ,     )





				def 		_lowerCAmelCase		(   self				:				Any   ,     _A				:				Union[str, Any]   ,     _A				:				Union[str, Any]   ,     _A				:				List[str]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    int      	=     ASTModel(config=lowercase_)
											model.to(lowercase_)
											model.eval()
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     model(lowercase_)
											self.parent.assertEqual(result.last_hidden_state.shape   ,     (self.batch_size, self.seq_length, self.hidden_size))





				def 		_lowerCAmelCase		(   self				:				Optional[int]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Any      	=     self.prepare_config_and_inputs()
											(
											    (
											    _SCREAMING_SNAKE_CASE	
											)	,					(
											    _SCREAMING_SNAKE_CASE	
											)	,					(
											    _SCREAMING_SNAKE_CASE	
											)	,						
											)		:    Union[str, Any]      	=     config_and_inputs
											_SCREAMING_SNAKE_CASE		:    Any      	=     {"""input_values""": input_values}
											return config, inputs_dict




@require_torch
class 			_snake_case		(     __a ,							__a ,							unittest.TestCase			):







				"""simple docstring"""






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





				def 		_lowerCAmelCase		(   self				:				int   ,     _A				:				Any   ,     _A				:				str   ,     _A				:				Optional[int]   ,     _A				:				List[str]   ,     _A				:				List[str]):



											"""simple docstring"""







											if pipeline_test_casse_name == "AudioClassificationPipelineTests":
																		return True

											return False





				def 		_lowerCAmelCase		(   self				:				Union[str, Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    int      	=     ASTModelTester(self)
											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     ConfigTester(self   ,     config_class=lowercase_   ,     has_text_modality=lowercase_   ,     hidden_size=3_7)





				def 		_lowerCAmelCase		(   self				:				str):



											"""simple docstring"""







											self.config_tester.run_common_tests()





				@unittest.skip(reason="""AST does not use inputs_embeds""")
				def 		_lowerCAmelCase		(   self				:				List[Any]):



											"""simple docstring"""







											pass





				def 		_lowerCAmelCase		(   self				:				Optional[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE	,					_SCREAMING_SNAKE_CASE		:    Tuple      	=     self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     model_class(lowercase_)
																		self.assertIsInstance(model.get_input_embeddings()   ,     (nn.Module))
																		_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     model.get_output_embeddings()
																		self.assertTrue(x is None or isinstance(lowercase_   ,     nn.Linear))





				def 		_lowerCAmelCase		(   self				:				Union[str, Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE	,					_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		_SCREAMING_SNAKE_CASE		:    List[str]      	=     model_class(lowercase_)
																		_SCREAMING_SNAKE_CASE		:    int      	=     inspect.signature(model.forward)
																		# signature.parameters is an OrderedDict => so arg_names order is deterministic
																		_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     [*signature.parameters.keys()]

																		_SCREAMING_SNAKE_CASE		:    Any      	=     ["""input_values"""]
																		self.assertListEqual(arg_names[:1]   ,     lowercase_)





				def 		_lowerCAmelCase		(   self				:				Dict):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*lowercase_)





				@slow
				def 		_lowerCAmelCase		(   self				:				str):



											"""simple docstring"""







											for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																		_SCREAMING_SNAKE_CASE		:    Tuple      	=     ASTModel.from_pretrained(lowercase_)
																		self.assertIsNotNone(lowercase_)



def   lowerCamelCase_()->     Tuple:
							_SCREAMING_SNAKE_CASE		:    List[Any]      	=     hf_hub_download(
							    repo_id="""nielsr/audio-spectogram-transformer-checkpoint"""  ,						filename="""sample_audio.flac"""  ,						repo_type="""dataset"""	)

							_SCREAMING_SNAKE_CASE	,					_SCREAMING_SNAKE_CASE		:    int      	=     torchaudio.load(SCREAMING_SNAKE_CASE_	)

							return audio, sampling_rate




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







				"""simple docstring"""






				@cached_property
				def 		_lowerCAmelCase		(   self				:				str):



											"""simple docstring"""







											return (
											    ASTFeatureExtractor.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593""")
											    if is_torchaudio_available()
											    else None
											)





				@slow
				def 		_lowerCAmelCase		(   self				:				List[Any]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     self.default_feature_extractor
											_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     ASTForAudioClassification.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593""").to(lowercase_)

											_SCREAMING_SNAKE_CASE		:    Dict      	=     self.default_feature_extractor
											_SCREAMING_SNAKE_CASE	,					_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     prepare_audio()
											_SCREAMING_SNAKE_CASE		:    str      	=     audio.squeeze().numpy()
											_SCREAMING_SNAKE_CASE		:    Dict      	=     feature_extractor(lowercase_   ,     sampling_rate=lowercase_   ,     return_tensors="""pt""").to(lowercase_)

											# forward pass
											with torch.no_grad():
																		_SCREAMING_SNAKE_CASE		:    List[Any]      	=     model(**lowercase_)

											# verify the logits
											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     torch.Size((1, 5_2_7))
											self.assertEqual(outputs.logits.shape   ,     lowercase_)

											_SCREAMING_SNAKE_CASE		:    Any      	=     torch.tensor([-0.8_760, -7.0_042, -8.6_602]).to(lowercase_)

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



'''simple docstring'''


import argparse

import torch
from torch import nn

from transformers import SpeechaTextConfig, SpeechaTextForConditionalGeneration

def   A_      (       SCREAMING_SNAKE_CASE_						)			->List[Any]:
  lowercase_     =       [
      """encoder.version""",
      """decoder.version""",
      """model.encoder.version""",
      """model.decoder.version""",
      """decoder.output_projection.weight""",
      """_float_tensor""",
      """encoder.embed_positions._float_tensor""",
      """decoder.embed_positions._float_tensor""",
  ]
  for k in ignore_keys:
    state_dict.pop(SCREAMING_SNAKE_CASE_ ,	SCREAMING_SNAKE_CASE_						)

def   A_      (       SCREAMING_SNAKE_CASE_						)			->Optional[int]:
  lowercase_     =       list(s_dict.keys()						)
  for key in keys:
    if "transformer_layers" in key:
      lowercase_     =       s_dict.pop(SCREAMING_SNAKE_CASE_						)
    elif "subsample" in key:
      lowercase_     =       s_dict.pop(SCREAMING_SNAKE_CASE_						)

def   A_      (       SCREAMING_SNAKE_CASE_						)			->Dict:
  lowercase_     ,	lowercase_     =       emb.weight.shape
  lowercase_     =       nn.Linear(SCREAMING_SNAKE_CASE_ ,	SCREAMING_SNAKE_CASE_ ,	bias=SCREAMING_SNAKE_CASE_						)
  lowercase_     =       emb.weight.data
  return lin_layer

def   A_      (       SCREAMING_SNAKE_CASE_ ,	SCREAMING_SNAKE_CASE_						)			->Any:
  lowercase_     =       torch.load(SCREAMING_SNAKE_CASE_ ,	map_location="""cpu"""						)
  lowercase_     =       mam_aaa["""args"""]
  lowercase_     =       mam_aaa["""model"""]
  lowercase_     =       state_dict["""decoder.output_projection.weight"""]

  remove_ignore_keys_(SCREAMING_SNAKE_CASE_						)
  rename_keys(SCREAMING_SNAKE_CASE_						)

  lowercase_     =       state_dict["""decoder.embed_tokens.weight"""].shape[0]

  lowercase_     =       args.share_decoder_input_output_embed

  lowercase_     =       [int(SCREAMING_SNAKE_CASE_						) for i in args.conv_kernel_sizes.split(""","""						)]
  lowercase_     =       SpeechaTextConfig(
      vocab_size=SCREAMING_SNAKE_CASE_ ,	max_source_positions=args.max_source_positions ,	max_target_positions=args.max_target_positions ,	encoder_layers=args.encoder_layers ,	decoder_layers=args.decoder_layers ,	encoder_attention_heads=args.encoder_attention_heads ,	decoder_attention_heads=args.decoder_attention_heads ,	encoder_ffn_dim=args.encoder_ffn_embed_dim ,	decoder_ffn_dim=args.decoder_ffn_embed_dim ,	d_model=args.encoder_embed_dim ,	dropout=args.dropout ,	attention_dropout=args.attention_dropout ,	activation_dropout=args.activation_dropout ,	activation_function="""relu""" ,	num_conv_layers=len(SCREAMING_SNAKE_CASE_						) ,	conv_channels=args.conv_channels ,	conv_kernel_sizes=SCREAMING_SNAKE_CASE_ ,	input_feat_per_channel=args.input_feat_per_channel ,	input_channels=args.input_channels ,	tie_word_embeddings=SCREAMING_SNAKE_CASE_ ,	num_beams=5 ,	max_length=2_00 ,	use_cache=SCREAMING_SNAKE_CASE_ ,	decoder_start_token_id=2 ,	early_stopping=SCREAMING_SNAKE_CASE_ ,	)

  lowercase_     =       SpeechaTextForConditionalGeneration(SCREAMING_SNAKE_CASE_						)
  lowercase_     ,	lowercase_     =       model.model.load_state_dict(SCREAMING_SNAKE_CASE_ ,	strict=SCREAMING_SNAKE_CASE_						)
  if len(SCREAMING_SNAKE_CASE_						) > 0 and not set(SCREAMING_SNAKE_CASE_						) <= {
      "encoder.embed_positions.weights",
      "decoder.embed_positions.weights",
  }:
    raise ValueError(
        """Only `encoder.embed_positions.weights` and `decoder.embed_positions.weights`  are allowed to be missing,"""
        f""" but all the following weights are missing {missing}"""						)

  if tie_embeds:
    lowercase_     =       make_linear_from_emb(model.model.decoder.embed_tokens						)
  else:
    lowercase_     =       lm_head_weights

  model.save_pretrained(SCREAMING_SNAKE_CASE_						)


if __name__ == "__main__":
       __snake_case						  =		argparse.ArgumentParser()
       # Required parameters
       parser.add_argument("""--fairseq_path""", type=str, help="""Path to the fairseq model (.pt) file.""")
       parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
       __snake_case						  =		parser.parse_args()
       convert_fairseq_sat_checkpoint_to_tfms(args.fairseq_path, args.pytorch_dump_folder_path)

