





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


   def __init__( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase=13    ,   _UpperCAmelCase=2    ,   _UpperCAmelCase=24    ,   _UpperCAmelCase=16    ,   _UpperCAmelCase=True    ,   _UpperCAmelCase=True    ,   _UpperCAmelCase=32    ,   _UpperCAmelCase=5    ,   _UpperCAmelCase=4    ,   _UpperCAmelCase=37    ,   _UpperCAmelCase="gelu"    ,   _UpperCAmelCase=0.1    ,   _UpperCAmelCase=0.1    ,   _UpperCAmelCase=10    ,   _UpperCAmelCase=0.02    ,   _UpperCAmelCase=None    ,   _UpperCAmelCase=2    ,   _UpperCAmelCase=2    ,   ):
       A       :       Tuple							=						parent
       A       :       Any							=						batch_size
       A       :       int							=						patch_size
       A       :       List[str]							=						max_length
       A       :       str							=						num_mel_bins
       A       :       Optional[int]							=						is_training
       A       :       Any							=						use_labels
       A       :       str							=						hidden_size
       A       :       Union[str, Any]							=						num_hidden_layers
       A       :       List[str]							=						num_attention_heads
       A       :       str							=						intermediate_size
       A       :       str							=						hidden_act
       A       :       List[Any]							=						hidden_dropout_prob
       A       :       Dict							=						attention_probs_dropout_prob
       A       :       Any							=						type_sequence_label_size
       A       :       Tuple							=						initializer_range
       A       :       List[str]							=						scope
       A       :       List[str]							=						frequency_stride
       A       :       Dict							=						time_stride

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


   def 					snake_case				( self					):
       A       :       Any							=						floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]					)

       A       :       Any							=						None
       if self.use_labels:
           A       :       Optional[Any]							=						ids_tensor([self.batch_size]    ,   self.type_sequence_label_size					)

       A       :       List[str]							=						self.get_config()

       return config, input_values, labels


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


   def 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase					):
       A       :       Union[str, Any]							=						ASTModel(config=__a					)
       model.to(__a					)
       model.eval()
       A       :       Union[str, Any]							=						model(__a					)
       self.parent.assertEqual(result.last_hidden_state.shape    ,   (self.batch_size, self.seq_length, self.hidden_size)					)


   def 					snake_case				( self					):
       A       :       Optional[Any]							=						self.prepare_config_and_inputs()
       (
           (
           A						
       ),	(
           A						
       ),	(
           A						
       ),							
       )       :       Union[str, Any]							=						config_and_inputs
       A       :       Dict							=						{'''input_values''': input_values}
       return config, inputs_dict







@require_torch
class   _lowercase						(							a		,					a		,					unittest.TestCase				):
   _UpperCamelCase       		=   (
       (
           ASTModel,
           ASTForAudioClassification,
       )
       if is_torch_available()
       else ()
   )
   _UpperCamelCase       		=   (
       {"""audio-classification""": ASTForAudioClassification, """feature-extraction""": ASTModel}
       if is_torch_available()
       else {}
   )
   _UpperCamelCase       		=   False
   _UpperCamelCase       		=   False
   _UpperCamelCase       		=   False
   _UpperCamelCase       		=   False


   def 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase					):
       if pipeline_test_casse_name == "AudioClassificationPipelineTests":
           return True

       return False


   def 					snake_case				( self					):
       A       :       List[Any]							=						ASTModelTester(self					)
       A       :       int							=						ConfigTester(self    ,   config_class=__a    ,   has_text_modality=__a    ,   hidden_size=37					)


   def 					snake_case				( self					):
       self.config_tester.run_common_tests()


   @unittest.skip(reason='''AST does not use inputs_embeds'''					)
   def 					snake_case				( self					):
       pass


   def 					snake_case				( self					):
       A,	A       :       List[Any]							=						self.model_tester.prepare_config_and_inputs_for_common()

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


   def 					snake_case				( self					):
       A,	A       :       int							=						self.model_tester.prepare_config_and_inputs_for_common()

       for model_class in self.all_model_classes:
           A       :       Dict							=						model_class(__a					)
           A       :       List[Any]							=						inspect.signature(model.forward					)
           # signature.parameters is an OrderedDict => so arg_names order is deterministic
           A       :       Dict							=						[*signature.parameters.keys()]

           A       :       List[Any]							=						['''input_values''']
           self.assertListEqual(arg_names[:1]    ,   __a					)


   def 					snake_case				( self					):
       A       :       List[str]							=						self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*__a					)


   @slow
   def 					snake_case				( self					):
       for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           A       :       Union[str, Any]							=						ASTModel.from_pretrained(__a					)
           self.assertIsNotNone(__a					)







def 							_lowerCamelCase(  )		->  List[str]:
    A       :       Optional[int]							=						hf_hub_download(
        repo_id='''nielsr/audio-spectogram-transformer-checkpoint'''    ,       filename='''sample_audio.flac'''    ,       repo_type='''dataset'''		)

    A,	A       :       str							=						torchaudio.load(__snake_case		)

    return audio, sampling_rate





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


   @cached_property
   def 					snake_case				( self					):
       return (
           ASTFeatureExtractor.from_pretrained('''MIT/ast-finetuned-audioset-10-10-0.4593'''					)
           if is_torchaudio_available()
           else None
       )


   @slow
   def 					snake_case				( self					):
       A       :       Optional[int]							=						self.default_feature_extractor
       A       :       Optional[int]							=						ASTForAudioClassification.from_pretrained('''MIT/ast-finetuned-audioset-10-10-0.4593'''					).to(__a					)

       A       :       str							=						self.default_feature_extractor
       A,	A       :       List[str]							=						prepare_audio()
       A       :       Union[str, Any]							=						audio.squeeze().numpy()
       A       :       Union[str, Any]							=						feature_extractor(__a    ,   sampling_rate=__a    ,   return_tensors='''pt'''					).to(__a					)

       # forward pass
       with torch.no_grad():
           A       :       Optional[Any]							=						model(**__a					)

       # verify the logits
       A       :       int							=						torch.Size((1, 527)					)
       self.assertEqual(outputs.logits.shape    ,   __a					)

       A       :       Any							=						torch.tensor([-0.8760, -7.0042, -8.6602]					).to(__a					)

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







'''simple docstring'''






# This script creates a super tiny model that is useful inside tests, when we just want to test that
# the machinery works, without needing to the check the quality of the outcomes.
#
# This version creates a tiny vocab first, and then a tiny model - so the outcome is truly tiny -
# all files ~60KB. As compared to taking a full-size model, reducing to the minimum its layers and
# emb dimensions, but keeping the full vocab + merges files, leading to ~3MB in total for all files.
# The latter is done by `fsmt-make-super-tiny-model.py`.
#
# It will be used then as "stas/tiny-wmt19-en-ru"

from pathlib import Path
import json
import tempfile

from transformers import FSMTTokenizer, FSMTConfig, FSMTForConditionalGeneration
from transformers.models.fsmt.tokenization_fsmt import VOCAB_FILES_NAMES

snake_case_     =	"""tiny-wmt19-en-ru"""

# Build

# borrowed from a test
snake_case_     =	[
    """l""",
    """o""",
    """w""",
    """e""",
    """r""",
    """s""",
    """t""",
    """i""",
    """d""",
    """n""",
    """w</w>""",
    """r</w>""",
    """t</w>""",
    """lo""",
    """low""",
    """er</w>""",
    """low</w>""",
    """lowest</w>""",
    """newer</w>""",
    """wider</w>""",
    """<unk>""",
]
snake_case_     =	dict(zip(vocab, range(len(vocab))))
snake_case_     =	["""l o 123""", """lo w 1456""", """e r</w> 1789""", """"""]

with tempfile.TemporaryDirectory() as tmpdirname:
     snake_case_     =	Path(tmpdirname)
     snake_case_     =	build_dir / VOCAB_FILES_NAMES["""src_vocab_file"""]
     snake_case_     =	build_dir / VOCAB_FILES_NAMES["""tgt_vocab_file"""]
     snake_case_     =	build_dir / VOCAB_FILES_NAMES["""merges_file"""]
     with open(src_vocab_file, """w""") as fp:
          fp.write(json.dumps(vocab_tokens))
     with open(tgt_vocab_file, """w""") as fp:
          fp.write(json.dumps(vocab_tokens))
     with open(merges_file, """w""") as fp:
          fp.write("""\n""".join(merges))

     snake_case_     =	FSMTTokenizer(
         langs=["""en""", """ru"""],
         src_vocab_size=len(vocab),
         tgt_vocab_size=len(vocab),
         src_vocab_file=src_vocab_file,
         tgt_vocab_file=tgt_vocab_file,
         merges_file=merges_file,
     )

snake_case_     =	FSMTConfig(
    langs=["""ru""", """en"""],
    src_vocab_size=10_00,
    tgt_vocab_size=10_00,
    d_model=4,
    encoder_layers=1,
    decoder_layers=1,
    encoder_ffn_dim=4,
    decoder_ffn_dim=4,
    encoder_attention_heads=1,
    decoder_attention_heads=1,
)

snake_case_     =	FSMTForConditionalGeneration(config)
print(f'''num of params {tiny_model.num_parameters()}''')

# Test
snake_case_     =	tokenizer(["""Making tiny model"""], return_tensors="""pt""")
snake_case_     =	tiny_model(**batch)

print("""test output:""", len(outputs.logits[0]))

# Save
tiny_model.half()  # makes it smaller
tiny_model.save_pretrained(mname_tiny)
tokenizer.save_pretrained(mname_tiny)

print(f'''Generated {mname_tiny}''')

# Upload
# transformers-cli upload tiny-wmt19-en-ru
