






'''simple docstring'''

import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


UpperCamelCase      					=logging.get_logger(__name__)




class      A     (   __SCREAMING_SNAKE_CASE    ):






 """simple docstring"""





 __a      :    Any			  =    ["""input_features""", """is_longer"""]
 def __init__(				self      ,  __lowerCAmelCase=64      ,  __lowerCAmelCase=4_80_00      ,  __lowerCAmelCase=4_80      ,  __lowerCAmelCase=10      ,  __lowerCAmelCase=10_24      ,  __lowerCAmelCase=0.0      ,  __lowerCAmelCase=False      ,  __lowerCAmelCase = 0      ,  __lowerCAmelCase = 1_40_00      ,  __lowerCAmelCase = None      ,  __lowerCAmelCase = "fusion"      ,  __lowerCAmelCase = "repeatpad"      ,  **__lowerCAmelCase      ,  ):
       super().__init__(
           feature_size=_A      ,  sampling_rate=_A      ,  padding_value=_A      ,  return_attention_mask=_A      ,  **_A      ,  )
       UpperCamelCase_					: Optional[int]            =				top_db
       UpperCamelCase_					: Tuple            =				truncation
       UpperCamelCase_					: str            =				padding
       UpperCamelCase_					: str            =				fft_window_size
       UpperCamelCase_					: int            =				(fft_window_size >> 1) + 1
       UpperCamelCase_					: Optional[int]            =				hop_length
       UpperCamelCase_					: Tuple            =				max_length_s
       UpperCamelCase_					: Optional[Any]            =				max_length_s * sampling_rate
       UpperCamelCase_					: List[str]            =				sampling_rate
       UpperCamelCase_					: List[Any]            =				frequency_min
       UpperCamelCase_					: Union[str, Any]            =				frequency_max
       UpperCamelCase_					: Optional[int]            =				mel_filter_bank(
           num_frequency_bins=self.nb_frequency_bins      ,  num_mel_filters=_A      ,  min_frequency=_A      ,  max_frequency=_A      ,  sampling_rate=_A      ,  norm=_A      ,  mel_scale="""htk"""      ,  )
       UpperCamelCase_					: str            =				mel_filter_bank(
           num_frequency_bins=self.nb_frequency_bins      ,  num_mel_filters=_A      ,  min_frequency=_A      ,  max_frequency=_A      ,  sampling_rate=_A      ,  norm="""slaney"""      ,  mel_scale="""slaney"""      ,  )
 def 	_UpperCAmelCase							(				self			):
       UpperCamelCase_					: Tuple            =				copy.deepcopy(self.__dict__			)
       UpperCamelCase_					: Dict            =				self.__class__.__name__
       if "mel_filters" in output:
             del output["mel_filters"]
       if "mel_filters_slaney" in output:
             del output["mel_filters_slaney"]
       return output
 def 	_UpperCAmelCase							(				self      ,  __lowerCAmelCase      ,  __lowerCAmelCase = None			):
       UpperCamelCase_					: Dict            =				spectrogram(
           _A      ,  window_function(self.fft_window_size      ,  """hann"""			)      ,  frame_length=self.fft_window_size      ,  hop_length=self.hop_length      ,  power=2.0      ,  mel_filters=_A      ,  log_mel="""dB"""      ,  )
       return log_mel_spectrogram.T
 def 	_UpperCAmelCase							(				self      ,  __lowerCAmelCase      ,  __lowerCAmelCase      ,  __lowerCAmelCase			):
       UpperCamelCase_					: int            =				np.array_split(list(range(0      ,  total_frames - chunk_frames + 1			)			)      ,  3			)
       if len(ranges[1]			) == 0:
             # if the audio is too short, we just use the first chunk
             UpperCamelCase_					: int            =				[0]
       if len(ranges[2]			) == 0:
             # if the audio is too short, we just use the first chunk
             UpperCamelCase_					: Union[str, Any]            =				[0]
       # randomly choose index for each part
       UpperCamelCase_					: Optional[int]            =				np.random.choice(ranges[0]			)
       UpperCamelCase_					: List[Any]            =				np.random.choice(ranges[1]			)
       UpperCamelCase_					: int            =				np.random.choice(ranges[2]			)

       UpperCamelCase_					: str            =				mel[idx_front : idx_front + chunk_frames, :]
       UpperCamelCase_					: List[Any]            =				mel[idx_middle : idx_middle + chunk_frames, :]
       UpperCamelCase_					: int            =				mel[idx_back : idx_back + chunk_frames, :]

       UpperCamelCase_					: int            =				torch.tensor(mel[None, None, :]			)
       UpperCamelCase_					: Union[str, Any]            =				torch.nn.functional.interpolate(
           _A      ,  size=[chunk_frames, 64]      ,  mode="""bilinear"""      ,  align_corners=_A			)
       UpperCamelCase_					: List[Any]            =				mel_shrink[0][0].numpy()
       UpperCamelCase_					: Optional[int]            =				np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]      ,  axis=0			)
       return mel_fusion
 def 	_UpperCAmelCase							(				self      ,  __lowerCAmelCase      ,  __lowerCAmelCase      ,  __lowerCAmelCase      ,  __lowerCAmelCase			):







       if waveform.shape[0] > max_length:
             if truncation == "rand_trunc":
                   UpperCamelCase_					: Any            =				True
                   # random crop to max_length (for compatibility) -> this should be handled by self.pad
                   UpperCamelCase_					: Union[str, Any]            =				len(_A			) - max_length
                   UpperCamelCase_					: Optional[int]            =				np.random.randint(0      ,  overflow + 1			)
                   UpperCamelCase_					: Optional[int]            =				waveform[idx : idx + max_length]
                   UpperCamelCase_					: Union[str, Any]            =				self._np_extract_fbank_features(_A      ,  self.mel_filters_slaney			)[None, :]
             elif truncation == "fusion":
                   UpperCamelCase_					: List[Any]            =				self._np_extract_fbank_features(_A      ,  self.mel_filters			)
                   UpperCamelCase_					: Union[str, Any]            =				max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                   UpperCamelCase_					: Any            =				mel.shape[0]
                   if chunk_frames == total_frames:
                         # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                         # In this case, we just use the whole audio.
                         UpperCamelCase_					: Optional[Any]            =				np.stack([mel, mel, mel, mel]      ,  axis=0			)
                         UpperCamelCase_					: Optional[Any]            =				False
                   else:
                         UpperCamelCase_					: Optional[int]            =				self._random_mel_fusion(_A      ,  _A      ,  _A			)
                         UpperCamelCase_					: Any            =				True
             else:
                   raise NotImplementedError(F"data_truncating {truncation} not implemented"			)

       else:
             UpperCamelCase_					: Union[str, Any]            =				False
             # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
             if waveform.shape[0] < max_length:
                   if padding == "repeat":
                         UpperCamelCase_					: int            =				int(max_length / len(_A			)			)
                         UpperCamelCase_					: Tuple            =				np.stack(np.tile(_A      ,  n_repeat + 1			)			)[:max_length]
                   if padding == "repeatpad":
                         UpperCamelCase_					: Any            =				int(max_length / len(_A			)			)
                         UpperCamelCase_					: str            =				np.stack(np.tile(_A      ,  _A			)			)
                   UpperCamelCase_					: Optional[int]            =				np.pad(_A      ,  (0, max_length - waveform.shape[0])      ,  mode="""constant"""      ,  constant_values=0			)

             if truncation == "fusion":
                   UpperCamelCase_					: List[Any]            =				self._np_extract_fbank_features(_A      ,  self.mel_filters			)
                   UpperCamelCase_					: Any            =				np.stack([input_mel, input_mel, input_mel, input_mel]      ,  axis=0			)
             else:
                   UpperCamelCase_					: Union[str, Any]            =				self._np_extract_fbank_features(_A      ,  self.mel_filters_slaney			)[None, :]

       return input_mel, longer





 def __call__(				self      ,  __lowerCAmelCase      ,  __lowerCAmelCase = None      ,  __lowerCAmelCase = None      ,  __lowerCAmelCase = None      ,  __lowerCAmelCase = None      ,  __lowerCAmelCase = None      ,  **__lowerCAmelCase      ,  ):
       UpperCamelCase_					: str            =				truncation if truncation is not None else self.truncation
       UpperCamelCase_					: int            =				padding if padding else self.padding

       if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                   raise ValueError(
                       F"The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"
                       F" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"
                       F" was sampled with {self.sampling_rate} and not {sampling_rate}."			)
       else:
             logger.warning(
                 """It is strongly recommended to pass the `sampling_rate` argument to this function. """
                 """Failing to do so can result in silent errors that might be hard to debug."""			)

       UpperCamelCase_					: int            =				isinstance(_A      ,  np.ndarray			) and len(raw_speech.shape			) > 1
       if is_batched_numpy and len(raw_speech.shape			) > 2:
             raise ValueError(F"Only mono-channel audio is supported for input to {self}"			)
       UpperCamelCase_					: int            =				is_batched_numpy or (
           isinstance(_A      ,  (list, tuple)			) and (isinstance(raw_speech[0]      ,  (np.ndarray, tuple, list)			))
       )

       if is_batched:
             UpperCamelCase_					: Any            =				[np.asarray(_A      ,  dtype=np.floataa			) for speech in raw_speech]
       elif not is_batched and not isinstance(_A      ,  np.ndarray			):
             UpperCamelCase_					: Any            =				np.asarray(_A      ,  dtype=np.floataa			)
       elif isinstance(_A      ,  np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
             UpperCamelCase_					: List[Any]            =				raw_speech.astype(np.floataa			)

       # always return batch
       if not is_batched:
             UpperCamelCase_					: Optional[int]            =				[np.asarray(_A			)]

       # convert to mel spectrogram, truncate and pad if needed.
       UpperCamelCase_					: List[str]            =				[
           self._get_input_mel(_A      ,  max_length if max_length else self.nb_max_samples      ,  _A      ,  _A			)
           for waveform in raw_speech
       ]

       UpperCamelCase_					: Union[str, Any]            =				[]
       UpperCamelCase_					: Optional[int]            =				[]
       for mel, longer in padded_inputs:
             input_mel.append(_A			)
             is_longer.append(_A			)

       if truncation == "fusion" and sum(_A			) == 0:
             # if no audio is longer than 10s, then randomly select one audio to be longer
             UpperCamelCase_					: int            =				np.random.randint(0      ,  len(_A			)			)
             UpperCamelCase_					: List[str]            =				True

       if isinstance(input_mel[0]      ,  _A			):
             UpperCamelCase_					: Dict            =				[np.asarray(_A      ,  dtype=np.floataa			) for feature in input_mel]

       # is_longer is a list of bool
       UpperCamelCase_					: Optional[int]            =				[[longer] for longer in is_longer]

       UpperCamelCase_					: Union[str, Any]            =				{"""input_features""": input_mel, """is_longer""": is_longer}
       UpperCamelCase_					: List[Any]            =				BatchFeature(_A			)

       if return_tensors is not None:
             UpperCamelCase_					: str            =				input_features.convert_to_tensors(_A			)

       return input_features













"""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 							lowercase__			:

			"""simple docstring"""







			def __init__(      self						,   _A						,   _A=1_3						,   _A=2						,   _A=2_4						,   _A=1_6						,   _A=True						,   _A=True						,   _A=3_2						,   _A=5						,   _A=4						,   _A=3_7						,   _A="gelu"						,   _A=0.1						,   _A=0.1						,   _A=1_0						,   _A=0.02						,   _A=None						,   _A=2						,   _A=2						,   ):






									'''simple docstring'''







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

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







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		Optional[int]				   =			floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]    )

									UpperCamelCase				:		int				   =			None
									if self.use_labels:
															UpperCamelCase				:		Tuple				   =			ids_tensor([self.batch_size]						,   self.type_sequence_label_size    )

									UpperCamelCase				:		Dict				   =			self.get_config()

									return config, input_values, labels







			def   _a      (      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=_A						,   initializer_range=self.initializer_range						,   frequency_stride=self.frequency_stride						,   time_stride=self.time_stride						,   )







			def   _a      (      self						,   _A						,   _A						,   _A    ):






									'''simple docstring'''







									UpperCamelCase				:		List[str]				   =			ASTModel(config=_A    )
									model.to(_A    )
									model.eval()
									UpperCamelCase				:		int				   =			model(_A    )
									self.parent.assertEqual(result.last_hidden_state.shape						,   (self.batch_size, self.seq_length, self.hidden_size)    )







			def   _a      (      self    ):






									'''simple docstring'''







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






@require_torch
class 							lowercase__			(   __SCREAMING_SNAKE_CASE					,     __SCREAMING_SNAKE_CASE					,     unittest.TestCase	):

			"""simple docstring"""
			__lowerCAmelCase     :				List[Any]										=  (
			    (
			        ASTModel,
			        ASTForAudioClassification,
			    )
			    if is_torch_available()
			    else ()
			)
			__lowerCAmelCase     :				Any										=  (
			    {"""audio-classification""": ASTForAudioClassification, """feature-extraction""": ASTModel}
			    if is_torch_available()
			    else {}
			)
			__lowerCAmelCase     :				Union[str, Any]										=  False
			__lowerCAmelCase     :				Any										=  False
			__lowerCAmelCase     :				List[Any]										=  False
			__lowerCAmelCase     :				Union[str, Any]										=  False







			def   _a      (      self						,   _A						,   _A						,   _A						,   _A						,   _A    ):






									'''simple docstring'''







									if pipeline_test_casse_name == "AudioClassificationPipelineTests":
															return True

									return False







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		str				   =			ASTModelTester(self    )
									UpperCamelCase				:		int				   =			ConfigTester(self						,   config_class=_A						,   has_text_modality=_A						,   hidden_size=3_7    )







			def   _a      (      self    ):






									'''simple docstring'''







									self.config_tester.run_common_tests()







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






									'''simple docstring'''







									pass







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase   ,						UpperCamelCase				:		int				   =			self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
															UpperCamelCase				:		Tuple				   =			model_class(_A    )
															self.assertIsInstance(model.get_input_embeddings()						,   (nn.Module)    )
															UpperCamelCase				:		Any				   =			model.get_output_embeddings()
															self.assertTrue(x is None or isinstance(_A						,   nn.Linear    )    )







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase   ,						UpperCamelCase				:		Union[str, Any]				   =			self.model_tester.prepare_config_and_inputs_for_common()

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

															UpperCamelCase				:		Any				   =			["""input_values"""]
															self.assertListEqual(arg_names[:1]						,   _A    )







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		int				   =			self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*_A    )







			@slow
			def   _a      (      self    ):






									'''simple docstring'''







									for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
															UpperCamelCase				:		int				   =			ASTModel.from_pretrained(_A    )
															self.assertIsNotNone(_A    )


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

						UpperCamelCase   ,						UpperCamelCase				:		Dict				   =			torchaudio.load(SCREAMING_SNAKE_CASE     )

						return audio, sampling_rate






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

			"""simple docstring"""







			@cached_property
			def   _a      (      self    ):






									'''simple docstring'''







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







			@slow
			def   _a      (      self    ):






									'''simple docstring'''







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

									UpperCamelCase				:		Union[str, Any]				   =			self.default_feature_extractor
									UpperCamelCase   ,						UpperCamelCase				:		List[Any]				   =			prepare_audio()
									UpperCamelCase				:		Optional[Any]				   =			audio.squeeze().numpy()
									UpperCamelCase				:		Union[str, Any]				   =			feature_extractor(_A						,   sampling_rate=_A						,   return_tensors="""pt"""    ).to(_A    )

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

									# verify the logits
									UpperCamelCase				:		List[Any]				   =			torch.Size((1, 5_2_7)    )
									self.assertEqual(outputs.logits.shape						,   _A    )

									UpperCamelCase				:		Optional[Any]				   =			torch.tensor([-0.87_60, -7.00_42, -8.66_02]    ).to(_A    )

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



