




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






       def __init__(      self:    List[Any]				,    _lowerCAmelCase:    Any				,    _lowerCAmelCase:    List[Any]=13				,    _lowerCAmelCase:    List[Any]=2				,    _lowerCAmelCase:    Tuple=24				,    _lowerCAmelCase:    List[str]=16				,    _lowerCAmelCase:    Dict=True				,    _lowerCAmelCase:    List[Any]=True				,    _lowerCAmelCase:    Optional[int]=32				,    _lowerCAmelCase:    Tuple=5				,    _lowerCAmelCase:    int=4				,    _lowerCAmelCase:    Tuple=37				,    _lowerCAmelCase:    List[str]="gelu"				,    _lowerCAmelCase:    Tuple=0.1				,    _lowerCAmelCase:    Tuple=0.1				,    _lowerCAmelCase:    Union[str, Any]=10				,    _lowerCAmelCase:    Tuple=0.02				,    _lowerCAmelCase:    Tuple=None				,    _lowerCAmelCase:    Any=2				,    _lowerCAmelCase:    Dict=2				,    ):
         lowercase			:Optional[Any]							=  parent
         lowercase			:Dict							=  batch_size
         lowercase			:Union[str, Any]							=  patch_size
         lowercase			:List[str]							=  max_length
         lowercase			:Tuple							=  num_mel_bins
         lowercase			:List[Any]							=  is_training
         lowercase			:Optional[Any]							=  use_labels
         lowercase			:Tuple							=  hidden_size
         lowercase			:Tuple							=  num_hidden_layers
         lowercase			:Any							=  num_attention_heads
         lowercase			:List[str]							=  intermediate_size
         lowercase			:List[Any]							=  hidden_act
         lowercase			:Optional[int]							=  hidden_dropout_prob
         lowercase			:Optional[Any]							=  attention_probs_dropout_prob
         lowercase			:Optional[Any]							=  type_sequence_label_size
         lowercase			:Optional[int]							=  initializer_range
         lowercase			:str							=  scope
         lowercase			:Tuple							=  frequency_stride
         lowercase			: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			:Union[str, Any]							=  (self.max_length - self.patch_size) // self.time_stride + 1
         lowercase			:str							=  frequency_out_dimension * time_out_dimension
         lowercase			:Optional[int]							=  num_patches + 2






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

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

         lowercase			:int							=  self.get_config()

         return config, input_values, labels






       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         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=_lowerCAmelCase				,    initializer_range=self.initializer_range				,    frequency_stride=self.frequency_stride				,    time_stride=self.time_stride				,    )






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






       def     SCREAMING_SNAKE_CASE			(      self:    Tuple ):
         lowercase			:Optional[Any]							=  self.prepare_config_and_inputs()
         (
             (
             lowercase       
         )		,			(
             lowercase       
         )		,			(
             lowercase       
         )		,			       
         )			:int							=  config_and_inputs
         lowercase			:Union[str, Any]							=  {"input_values": input_values}
         return config, inputs_dict







@require_torch
class       __lowerCAmelCase				(							SCREAMING_SNAKE_CASE__					,       SCREAMING_SNAKE_CASE__					,       unittest.TestCase):
       _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     SCREAMING_SNAKE_CASE			(      self:    Tuple				,    _lowerCAmelCase:    Union[str, Any]				,    _lowerCAmelCase:    Optional[Any]				,    _lowerCAmelCase:    Dict				,    _lowerCAmelCase:    Optional[int]				,    _lowerCAmelCase:    Tuple ):
         if pipeline_test_casse_name == "AudioClassificationPipelineTests":
           return True

         return False






       def     SCREAMING_SNAKE_CASE			(      self:    Optional[Any] ):
         lowercase			:Dict							=  ASTModelTester(self )
         lowercase			:Tuple							=  ConfigTester(self				,    config_class=_lowerCAmelCase				,    has_text_modality=_lowerCAmelCase				,    hidden_size=37 )






       def     SCREAMING_SNAKE_CASE			(      self:    Union[str, Any] ):
         self.config_tester.run_common_tests()






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






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

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






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any] ):
         lowercase		,			lowercase			:List[str]							=  self.model_tester.prepare_config_and_inputs_for_common()

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

           lowercase			:Tuple							=  ["input_values"]
           self.assertListEqual(arg_names[:1]				,    _lowerCAmelCase )






       def     SCREAMING_SNAKE_CASE			(      self:    int ):
         lowercase			:Any							=  self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*_lowerCAmelCase )






       @slow
       def     SCREAMING_SNAKE_CASE			(      self:    int ):
         for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           lowercase			:Tuple							=  ASTModel.from_pretrained(_lowerCAmelCase )
           self.assertIsNotNone(_lowerCAmelCase )







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

  lowercase		,			lowercase			:List[Any]							=  torchaudio.load(__UpperCamelCase      )

  return audio, sampling_rate







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






       @cached_property
       def     SCREAMING_SNAKE_CASE			(      self:    List[Any] ):
         return (
             ASTFeatureExtractor.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593" )
             if is_torchaudio_available()
             else None
         )






       @slow
       def     SCREAMING_SNAKE_CASE			(      self:    Union[str, Any] ):
         lowercase			:Optional[Any]							=  self.default_feature_extractor
         lowercase			:List[Any]							=  ASTForAudioClassification.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593" ).to(_lowerCAmelCase )

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

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

         # verify the logits
         lowercase			:Any							=  torch.Size((1, 5_27) )
         self.assertEqual(outputs.logits.shape				,    _lowerCAmelCase )

         lowercase			:Optional[int]							=  torch.tensor([-0.87_60, -7.00_42, -8.66_02] ).to(_lowerCAmelCase )

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




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


A   :     Tuple       =				get_tests_dir("fixtures")


class      lowerCamelCase    (unittest.TestCase    ):
  """simple docstring"""







  def       __A   (							self       :				Union[str, Any]      )     ->		Union[str, Any]:
        # A mock response for an HTTP head request to emulate server down
        SCREAMING_SNAKE_CASE_      =     mock.Mock()
        SCREAMING_SNAKE_CASE_      =     500
        SCREAMING_SNAKE_CASE_      =     {}
        SCREAMING_SNAKE_CASE_      =     HTTPError
        SCREAMING_SNAKE_CASE_      =     {}

        # Download this model to make sure it's in the cache.
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"      )
        # Under the mock environment we get a 500 error when trying to reach the model.
        with mock.patch("requests.Session.request"     ,   return_value=__magic_name__      ) as mock_head:
              SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"      )
              # This check we did call the fake head request
              mock_head.assert_called()







  def       __A   (							self       :				Optional[int]      )     ->		Tuple:
        # This test is for deprecated behavior and can be removed in v5
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(
            "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"      )



@is_staging_test
class      lowerCamelCase    (unittest.TestCase    ):
  """simple docstring"""







  @classmethod
  def       __A   (							cls       :				Any      )     ->		Union[str, Any]:
        SCREAMING_SNAKE_CASE_      =     TOKEN
        HfFolder.save_token(__magic_name__      )







  @classmethod
  def       __A   (							cls       :				int      )     ->		Optional[Any]:
        try:
              delete_repo(token=cls._token     ,   repo_id="test-feature-extractor"      )
        except HTTPError:
              pass

        try:
              delete_repo(token=cls._token     ,   repo_id="valid_org/test-feature-extractor-org"      )
        except HTTPError:
              pass

        try:
              delete_repo(token=cls._token     ,   repo_id="test-dynamic-feature-extractor"      )
        except HTTPError:
              pass







  def       __A   (							self       :				str      )     ->		List[Any]:
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(__magic_name__      )
        feature_extractor.push_to_hub("test-feature-extractor"     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )

        # Reset repo
        delete_repo(token=self._token     ,   repo_id="test-feature-extractor"      )

        # Push to hub via save_pretrained
        with tempfile.TemporaryDirectory() as tmp_dir:
              feature_extractor.save_pretrained(
                  __magic_name__     ,   repo_id="test-feature-extractor"     ,   push_to_hub=__magic_name__     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(F'''{USER}/test-feature-extractor'''      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )







  def       __A   (							self       :				Dict      )     ->		str:
        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained(__magic_name__      )
        feature_extractor.push_to_hub("valid_org/test-feature-extractor"     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )

        # Reset repo
        delete_repo(token=self._token     ,   repo_id="valid_org/test-feature-extractor"      )

        # Push to hub via save_pretrained
        with tempfile.TemporaryDirectory() as tmp_dir:
              feature_extractor.save_pretrained(
                  __magic_name__     ,   repo_id="valid_org/test-feature-extractor-org"     ,   push_to_hub=__magic_name__     ,   use_auth_token=self._token      )

        SCREAMING_SNAKE_CASE_      =     WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"      )
        for k, v in feature_extractor.__dict__.items():
              self.assertEqual(__magic_name__     ,   getattr(__magic_name__     ,   __magic_name__      )      )







  def       __A   (							self       :				List[Any]      )     ->		Dict:
        CustomFeatureExtractor.register_for_auto_class()
        SCREAMING_SNAKE_CASE_      =     CustomFeatureExtractor.from_pretrained(__magic_name__      )

        feature_extractor.push_to_hub("test-dynamic-feature-extractor"     ,   use_auth_token=self._token      )

        # This has added the proper auto_map field to the config
        self.assertDictEqual(
            feature_extractor.auto_map     ,   {"AutoFeatureExtractor": "custom_feature_extraction.CustomFeatureExtractor"}     ,   )

        SCREAMING_SNAKE_CASE_      =     AutoFeatureExtractor.from_pretrained(
            F'''{USER}/test-dynamic-feature-extractor'''     ,   trust_remote_code=__magic_name__      )
        # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
        self.assertEqual(new_feature_extractor.__class__.__name__     ,   "CustomFeatureExtractor"      )


