
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		:
    def __init__(  self		,snake_case		,snake_case=13		,snake_case=2		,snake_case=24		,snake_case=16		,snake_case=True		,snake_case=True		,snake_case=32		,snake_case=5		,snake_case=4		,snake_case=37		,snake_case="gelu"		,snake_case=0.1		,snake_case=0.1		,snake_case=10		,snake_case=0.02		,snake_case=None		,snake_case=2		,snake_case=2		,):



           '''simple docstring'''







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

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



           '''simple docstring'''







           lowercase							:     Any       	=      floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]      )

           lowercase							:     List[Any]       	=      None
           if self.use_labels:
                  lowercase							:     Union[str, Any]       	=      ids_tensor([self.batch_size]		,self.type_sequence_label_size      )

           lowercase							:     Tuple       	=      self.get_config()

           return config, input_values, labels
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''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=snake_case		,initializer_range=self.initializer_range		,frequency_stride=self.frequency_stride		,time_stride=self.time_stride		,)
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case      ):



           '''simple docstring'''







           lowercase							:     str       	=      ASTModel(config=snake_case      )
           model.to(snake_case      )
           model.eval()
           lowercase							:     List[str]       	=      model(snake_case      )
           self.parent.assertEqual(result.last_hidden_state.shape		,(self.batch_size, self.seq_length, self.hidden_size)      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Optional[int]       	=      self.prepare_config_and_inputs()
           (
               (
               lowercase 
           )	,			(
               lowercase 
           )	,			(
               lowercase 
           )	,			 
           )							:     List[Any]       	=      config_and_inputs
           lowercase							:     Optional[int]       	=      {"""input_values""": input_values}
           return config, inputs_dict






@require_torch
class __snake_case		(      lowerCAmelCase					,	lowerCAmelCase					,	unittest.TestCase     ):
    _a		:						Union[str, Any]=				(
        (
            ASTModel,
            ASTForAudioClassification,
        )
        if is_torch_available()
        else ()
    )
    _a		:						Any=				(
        {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
        if is_torch_available()
        else {}
    )
    _a		:						Union[str, Any]=				False
    _a		:						Tuple=				False
    _a		:						int=				False
    _a		:						Optional[Any]=				False
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case      ):



           '''simple docstring'''







           if pipeline_test_casse_name == "AudioClassificationPipelineTests":
                  return True

           return False
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     List[str]       	=      ASTModelTester(self      )
           lowercase							:     Any       	=      ConfigTester(self		,config_class=snake_case		,has_text_modality=snake_case		,hidden_size=37      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           self.config_tester.run_common_tests()
    @unittest.skip(reason="""AST does not use inputs_embeds"""      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           pass
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







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

           for model_class in self.all_model_classes:
                  lowercase							:     Union[str, Any]       	=      model_class(snake_case      )
                  self.assertIsInstance(model.get_input_embeddings()		,(nn.Module)      )
                  lowercase							:     str       	=      model.get_output_embeddings()
                  self.assertTrue(x is None or isinstance(snake_case		,nn.Linear      )      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase	,			lowercase							:     Optional[int]       	=      self.model_tester.prepare_config_and_inputs_for_common()

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

                  lowercase							:     Union[str, Any]       	=      ["""input_values"""]
                  self.assertListEqual(arg_names[:1]		,snake_case      )
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*snake_case      )
    @slow
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  lowercase							:     Dict       	=      ASTModel.from_pretrained(snake_case      )
                  self.assertIsNotNone(snake_case      )







def 						_snake_case(  ) ->    Union[str, Any]:
       lowercase							:     Optional[int]       	=      hf_hub_download(
           repo_id="""nielsr/audio-spectogram-transformer-checkpoint"""					,	filename="""sample_audio.flac"""					,	repo_type="""dataset"""				)

       lowercase	,			lowercase							:     List[str]       	=      torchaudio.load(__lowerCamelCase				)

       return audio, sampling_rate




@require_torch
@require_torchaudio
class __snake_case		(      unittest.TestCase     ):
    @cached_property
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           return (
               ASTFeatureExtractor.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593"""      )
               if is_torchaudio_available()
               else None
           )
    @slow
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           lowercase							:     Any       	=      self.default_feature_extractor
           lowercase							:     int       	=      ASTForAudioClassification.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593"""      ).to(snake_case      )

           lowercase							:     List[str]       	=      self.default_feature_extractor
           lowercase	,			lowercase							:     int       	=      prepare_audio()
           lowercase							:     Dict       	=      audio.squeeze().numpy()
           lowercase							:     Union[str, Any]       	=      feature_extractor(snake_case		,sampling_rate=snake_case		,return_tensors="""pt"""      ).to(snake_case      )

           # forward pass
           with torch.no_grad():
                  lowercase							:     Union[str, Any]       	=      model(**snake_case      )

           # verify the logits
           lowercase							:     List[Any]       	=      torch.Size((1, 527)      )
           self.assertEqual(outputs.logits.shape		,snake_case      )

           lowercase							:     str       	=      torch.tensor([-0.8_760, -7.0_042, -8.6_602]      ).to(snake_case      )

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






import argparse

import torch

from transformers import (
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaForAudioFrameClassification,
    WavaVecaForSequenceClassification,
    WavaVecaForXVector,
    logging,
)


logging.set_verbosity_info()
__UpperCamelCase					:       List[str]    					=      logging.get_logger(__name__)





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )      ->       Dict:
	a										=							WavaVecaForSequenceClassification.from_pretrained(__lowerCamelCase							,			config=__lowerCamelCase    )
	a										=							downstream_dict["""projector.weight"""]
	a										=							downstream_dict["""projector.bias"""]
	a										=							downstream_dict["""model.post_net.linear.weight"""]
	a										=							downstream_dict["""model.post_net.linear.bias"""]
	return model





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )      ->       Dict:
	a										=							WavaVecaForAudioFrameClassification.from_pretrained(__lowerCamelCase							,			config=__lowerCamelCase    )
	a										=							downstream_dict["""model.linear.weight"""]
	a										=							downstream_dict["""model.linear.bias"""]
	return model





def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )      ->       Optional[Any]:
	a										=							WavaVecaForXVector.from_pretrained(__lowerCamelCase							,			config=__lowerCamelCase    )
	a										=							downstream_dict["""connector.weight"""]
	a										=							downstream_dict["""connector.bias"""]
	for i, kernel_size in enumerate(hf_config.tdnn_kernel    ):
		a										=							downstream_dict[
		    f'model.framelevel_feature_extractor.module.{i}.kernel.weight'
		]
		a										=							downstream_dict[f'model.framelevel_feature_extractor.module.{i}.kernel.bias']

	a										=							downstream_dict["""model.utterancelevel_feature_extractor.linear1.weight"""]
	a										=							downstream_dict["""model.utterancelevel_feature_extractor.linear1.bias"""]
	a										=							downstream_dict["""model.utterancelevel_feature_extractor.linear2.weight"""]
	a										=							downstream_dict["""model.utterancelevel_feature_extractor.linear2.bias"""]
	a										=							downstream_dict["""objective.W"""]
	return model





@torch.no_grad()
def 		__A		(	__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )      ->       Dict:
	a										=							torch.load(__lowerCamelCase							,			map_location="""cpu"""    )

	a										=							checkpoint["""Downstream"""]

	a										=							WavaVecaConfig.from_pretrained(__lowerCamelCase    )
	a										=							WavaVecaFeatureExtractor.from_pretrained(
	    __lowerCamelCase							,			return_attention_mask=__lowerCamelCase							,			do_normalize=__lowerCamelCase    )

	a										=							hf_config.architectures[0]
	if arch.endswith("""ForSequenceClassification"""    ):
		a										=							convert_classification(__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )
	elif arch.endswith("""ForAudioFrameClassification"""    ):
		a										=							convert_diarization(__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )
	elif arch.endswith("""ForXVector"""    ):
		a										=							convert_xvector(__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase    )
	else:
		raise NotImplementedError(f'S3PRL weights conversion is not supported for {arch}'    )

	if hf_config.use_weighted_layer_sum:
		a										=							checkpoint["""Featurizer"""]["""weights"""]

	hf_feature_extractor.save_pretrained(__lowerCamelCase    )
	hf_model.save_pretrained(__lowerCamelCase    )


if __name__ == "__main__":
		__UpperCamelCase					:       List[str]    					=      argparse.ArgumentParser()
		parser.add_argument(
		    "--base_model_name", default=None, type=str, help="Name of the huggingface pretrained base model."
		)
		parser.add_argument("--config_path", default=None, type=str, help="Path to the huggingface classifier config.")
		parser.add_argument("--checkpoint_path", default=None, type=str, help="Path to the s3prl checkpoint.")
		parser.add_argument("--model_dump_path", default=None, type=str, help="Path to the final converted model.")
		__UpperCamelCase					:       List[Any]    					=      parser.parse_args()
		convert_saprl_checkpoint(args.base_model_name, args.config_path, args.checkpoint_path, args.model_dump_path)

