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  SCREAMING_SNAKE_CASE_ :
  '''simple docstring'''


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

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







  def 		SCREAMING_SNAKE_CASE_		(				self    :					int     )    ->				List[Any]:
         A						:     Any         =floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]     )

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

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

         return config, input_values, labels







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Dict     )    ->				Optional[int]:
         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=SCREAMING_SNAKE_CASE__      ,	initializer_range=self.initializer_range      ,	frequency_stride=self.frequency_stride      ,	time_stride=self.time_stride      ,	)







  def 		SCREAMING_SNAKE_CASE_		(				self    :					str      ,	SCREAMING_SNAKE_CASE__    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					str      ,	SCREAMING_SNAKE_CASE__    :					Optional[int]     )    ->				str:
         A						:     List[str]         =ASTModel(config=SCREAMING_SNAKE_CASE__     )
         model.to(SCREAMING_SNAKE_CASE__     )
         model.eval()
         A						:     Dict         =model(SCREAMING_SNAKE_CASE__     )
         self.parent.assertEqual(result.last_hidden_state.shape      ,	(self.batch_size, self.seq_length, self.hidden_size)     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[Any]     )    ->				List[str]:
         A						:     str         =self.prepare_config_and_inputs()
         (
             A  
         )						:     Any         =config_and_inputs
         A						:     Any         ={'input_values': input_values}
         return config, inputs_dict







@require_torch
class  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_   ,						lowerCAmelCase_   ,						unittest.TestCase			):
  '''simple docstring'''


  lowercase		:     Optional[Any]            =				(
      (
          ASTModel,
          ASTForAudioClassification,
      )
      if is_torch_available()
      else ()
  )
  lowercase		:     Dict            =				(
      {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
      if is_torch_available()
      else {}
  )
  lowercase		:     Optional[int]            =				False
  lowercase		:     int            =				False
  lowercase		:     Dict            =				False
  lowercase		:     List[Any]            =				False







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Dict      ,	SCREAMING_SNAKE_CASE__    :					Union[str, Any]      ,	SCREAMING_SNAKE_CASE__    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					Union[str, Any]      ,	SCREAMING_SNAKE_CASE__    :					List[str]      ,	SCREAMING_SNAKE_CASE__    :					List[str]     )    ->				List[str]:
         if pipeline_test_casse_name == "AudioClassificationPipelineTests":
                return True

         return False







  def 		SCREAMING_SNAKE_CASE_		(				self    :					int     )    ->				Any:
         A						:     Optional[int]         =ASTModelTester(self     )
         A						:     Union[str, Any]         =ConfigTester(self      ,	config_class=SCREAMING_SNAKE_CASE__      ,	has_text_modality=SCREAMING_SNAKE_CASE__      ,	hidden_size=37     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[Any]     )    ->				Tuple:
         self.config_tester.run_common_tests()







  @unittest.skip(reason='AST does not use inputs_embeds'     )
  def 		SCREAMING_SNAKE_CASE_		(				self    :					str     )    ->				List[str]:
         pass







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Union[str, Any]     )    ->				int:
         A						:     Optional[int]         =self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
                A						:     List[str]         =model_class(SCREAMING_SNAKE_CASE__     )
                self.assertIsInstance(model.get_input_embeddings()      ,	(nn.Module)     )
                A						:     Tuple         =model.get_output_embeddings()
                self.assertTrue(x is None or isinstance(SCREAMING_SNAKE_CASE__      ,	nn.Linear     )     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[Any]     )    ->				Optional[Any]:
         A						:     Any         =self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
                A						:     Optional[Any]         =model_class(SCREAMING_SNAKE_CASE__     )
                A						:     Tuple         =inspect.signature(model.forward     )
                # signature.parameters is an OrderedDict => so arg_names order is deterministic
                A						:     Union[str, Any]         =[*signature.parameters.keys()]

                A						:     Union[str, Any]         =['input_values']
                self.assertListEqual(arg_names[:1]      ,	SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Union[str, Any]     )    ->				Tuple:
         A						:     Dict         =self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*SCREAMING_SNAKE_CASE__     )







  @slow
  def 		SCREAMING_SNAKE_CASE_		(				self    :					Any     )    ->				Tuple:
         for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                A						:     List[str]         =ASTModel.from_pretrained(SCREAMING_SNAKE_CASE__     )
                self.assertIsNotNone(SCREAMING_SNAKE_CASE__     )







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

       A						:     str         =torchaudio.load(lowercase					)

       return audio, sampling_rate



@require_torch
@require_torchaudio
class  SCREAMING_SNAKE_CASE_ (					unittest.TestCase			):
  '''simple docstring'''


  @cached_property
  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[str]     )    ->				Union[str, 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    :					str     )    ->				Dict:
         A						:     Dict         =self.default_feature_extractor
         A						:     int         =ASTForAudioClassification.from_pretrained('MIT/ast-finetuned-audioset-10-10-0.4593'     ).to(SCREAMING_SNAKE_CASE__     )

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

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

         # verify the logits
         A						:     Dict         =torch.Size((1, 5_27)     )
         self.assertEqual(outputs.logits.shape      ,	SCREAMING_SNAKE_CASE__     )

         A						:     Any         =torch.tensor([-0.8_7_6_0, -7.0_0_4_2, -8.6_6_0_2]     ).to(SCREAMING_SNAKE_CASE__     )

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







from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


_lowercase							:      List[Any]      							=logging.get_logger(__name__)



class  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_			):
  '''simple docstring'''


  def __init__(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					int      ,	SCREAMING_SNAKE_CASE__    :					int      ,	SCREAMING_SNAKE_CASE__    :					float      ,	**SCREAMING_SNAKE_CASE__    :					List[Any]     )    ->				int:
         A						:     Tuple         =feature_size
         A						:     int         =sampling_rate
         A						:     List[str]         =padding_value

         A						:     Tuple         =kwargs.pop('padding_side'      ,	'right'     )
         A						:     str         =kwargs.pop('return_attention_mask'      ,	SCREAMING_SNAKE_CASE__     )

         super().__init__(**SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					str      ,	SCREAMING_SNAKE_CASE__    :					Union[
          BatchFeature,
          List[BatchFeature],
          Dict[str, BatchFeature],
          Dict[str, List[BatchFeature]],
          List[Dict[str, BatchFeature]],
      ]      ,	SCREAMING_SNAKE_CASE__    :					Union[bool, str, PaddingStrategy] = True      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					bool = False      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[bool] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[Union[str, TensorType]] = None      ,	)    ->				BatchFeature:
         # If we have a list of dicts, let's convert it in a dict of lists
         # We do this to allow using this method as a collate_fn function in PyTorch Dataloader
         if isinstance(SCREAMING_SNAKE_CASE__      ,	(list, tuple)     ) and isinstance(processed_features[0]      ,	(dict, BatchFeature)     ):
                A						:     Tuple         ={
                    key: [example[key] for example in processed_features] for key in processed_features[0].keys()
                }

         # The model's main input name, usually `input_values`, has be passed for padding
         if self.model_input_names[0] not in processed_features:
                raise ValueError(
                    'You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`'
                    f' to this method that includes {self.model_input_names[0]}, but you provided'
                    f' {list(processed_features.keys()     )}'     )

         A						:     Dict         =processed_features[self.model_input_names[0]]
         A						:     int         =(
             return_attention_mask if return_attention_mask is not None else self.return_attention_mask
         )

         if len(SCREAMING_SNAKE_CASE__     ) == 0:
                if return_attention_mask:
                       A						:     List[Any]         =[]
                return processed_features

         # If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
         # and rebuild them afterwards if no return_tensors is specified
         # Note that we lose the specific device the tensor may be on for PyTorch

         A						:     List[str]         =required_input[0]
         if isinstance(SCREAMING_SNAKE_CASE__      ,	(list, tuple)     ):
                # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
                A						:     Any         =0
                while len(required_input[index]     ) == 0:
                       index += 1
                if index < len(SCREAMING_SNAKE_CASE__     ):
                       A						:     Dict         =required_input[index][0]

         if return_tensors is None:
                if is_tf_tensor(SCREAMING_SNAKE_CASE__     ):
                       A						:     List[Any]         ='tf'
                elif is_torch_tensor(SCREAMING_SNAKE_CASE__     ):
                       A						:     Optional[int]         ='pt'
                elif isinstance(SCREAMING_SNAKE_CASE__      ,	(int, float, list, tuple, np.ndarray)     ):
                       A						:     Union[str, Any]         ='np'
                else:
                       raise ValueError(
                           f'type of {first_element} unknown: {type(SCREAMING_SNAKE_CASE__     )}. '
                           'Should be one of a python, numpy, pytorch or tensorflow object.'     )

         for key, value in processed_features.items():
                if isinstance(value[0]      ,	(int, float)     ):
                       A						:     int         =to_numpy(SCREAMING_SNAKE_CASE__     )
                else:
                       A						:     List[Any]         =[to_numpy(SCREAMING_SNAKE_CASE__     ) for v in value]

        # Convert padding_strategy in PaddingStrategy
         A						:     List[Any]         =self._get_padding_strategies(padding=SCREAMING_SNAKE_CASE__      ,	max_length=SCREAMING_SNAKE_CASE__     )

         A						:     Optional[int]         =processed_features[self.model_input_names[0]]

         A						:     List[str]         =len(SCREAMING_SNAKE_CASE__     )
         if not all(len(SCREAMING_SNAKE_CASE__     ) == batch_size for v in processed_features.values()     ):
                raise ValueError('Some items in the output dictionary have a different batch size than others.'     )

         A						:     Tuple         =[]
         for i in range(SCREAMING_SNAKE_CASE__     ):
                A						:     int         ={k: v[i] for k, v in processed_features.items()}
                # truncation
                A						:     List[Any]         =self._truncate(
                    SCREAMING_SNAKE_CASE__      ,	max_length=SCREAMING_SNAKE_CASE__      ,	pad_to_multiple_of=SCREAMING_SNAKE_CASE__      ,	truncation=SCREAMING_SNAKE_CASE__      ,	)
                truncated_inputs.append(SCREAMING_SNAKE_CASE__     )

         if padding_strategy == PaddingStrategy.LONGEST:
                # make sure that `max_length` cannot be longer than the longest truncated length
                A						:     Any         =max(len(input_slice[self.model_input_names[0]]     ) for input_slice in truncated_inputs     )
                A						:     Optional[Any]         =PaddingStrategy.MAX_LENGTH

         A						:     List[Any]         ={}
         for i in range(SCREAMING_SNAKE_CASE__     ):
                # padding
                A						:     Optional[Any]         =self._pad(
                    truncated_inputs[i]      ,	max_length=SCREAMING_SNAKE_CASE__      ,	padding_strategy=SCREAMING_SNAKE_CASE__      ,	pad_to_multiple_of=SCREAMING_SNAKE_CASE__      ,	return_attention_mask=SCREAMING_SNAKE_CASE__      ,	)

                for key, value in outputs.items():
                       if key not in batch_outputs:
                              A						:     Dict         =[]
                       if value.dtype is np.dtype(np.floataa     ):
                              A						:     Tuple         =value.astype(np.floataa     )
                       batch_outputs[key].append(SCREAMING_SNAKE_CASE__     )

         return BatchFeature(SCREAMING_SNAKE_CASE__      ,	tensor_type=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Union[Dict[str, np.ndarray], BatchFeature]      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					PaddingStrategy = PaddingStrategy.DO_NOT_PAD      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[bool] = None      ,	)    ->				dict:
         A						:     Optional[int]         =processed_features[self.model_input_names[0]]

         if padding_strategy == PaddingStrategy.LONGEST:
                A						:     List[str]         =len(SCREAMING_SNAKE_CASE__     )

         if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
                A						:     Tuple         =((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

         A						:     int         =padding_strategy != PaddingStrategy.DO_NOT_PAD and len(SCREAMING_SNAKE_CASE__     ) < max_length

         if return_attention_mask and "attention_mask" not in processed_features:
                A						:     str         =np.ones(len(SCREAMING_SNAKE_CASE__     )      ,	dtype=np.intaa     )

         if needs_to_be_padded:
                A						:     Union[str, Any]         =max_length - len(SCREAMING_SNAKE_CASE__     )
                if self.padding_side == "right":
                       if return_attention_mask:
                              A						:     Dict         =np.pad(
                                  processed_features['attention_mask']      ,	(0, difference)     )
                       A						:     str         =((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
                       A						:     Tuple         =np.pad(
                           SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	'constant'      ,	constant_values=self.padding_value     )
                elif self.padding_side == "left":
                       if return_attention_mask:
                              A						:     List[Any]         =np.pad(
                                  processed_features['attention_mask']      ,	(difference, 0)     )
                       A						:     Union[str, Any]         =((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
                       A						:     Tuple         =np.pad(
                           SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	'constant'      ,	constant_values=self.padding_value     )
                else:
                       raise ValueError('Invalid padding strategy:' + str(self.padding_side     )     )

         return processed_features







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Union[Dict[str, np.ndarray], BatchFeature]      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[bool] = None      ,	)    ->				Optional[Any]:
         if not truncation:
                return processed_features
         elif truncation and max_length is None:
                raise ValueError('When setting ``truncation=True``, make sure that ``max_length`` is defined.'     )

         A						:     Tuple         =processed_features[self.model_input_names[0]]

         # find `max_length` that fits `pad_to_multiple_of`
         if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
                A						:     Any         =((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

         A						:     List[str]         =len(SCREAMING_SNAKE_CASE__     ) > max_length

         if needs_to_be_truncated:
                A						:     Union[str, Any]         =processed_features[self.model_input_names[0]][:max_length]
                if "attention_mask" in processed_features:
                       A						:     Dict         =processed_features['attention_mask'][:max_length]

         return processed_features







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					Any=False      ,	SCREAMING_SNAKE_CASE__    :					Dict=None     )    ->				Union[str, Any]:

         # Get padding strategy
         if padding is not False:
                if padding is True:
                       A						:     List[Any]         =PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
                elif not isinstance(SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__     ):
                       A						:     Tuple         =PaddingStrategy(SCREAMING_SNAKE_CASE__     )
                elif isinstance(SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__     ):
                       A						:     Optional[int]         =padding
         else:
                A						:     List[str]         =PaddingStrategy.DO_NOT_PAD

         # Set max length if needed
         if max_length is None:
                if padding_strategy == PaddingStrategy.MAX_LENGTH:
                       raise ValueError(
                           f'When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined'     )

        # Test if we have a padding value
         if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
                raise ValueError(
                    'Asking to pad but the feature_extractor does not have a padding value. Please select a value to use'
                    ' as `padding_value`. For example: `feature_extractor.padding_value = 0.0`.'     )

         return padding_strategy






