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







				def __init__( self    :       Dict			,					lowerCamelCase_    :       Optional[int]			,					lowerCamelCase_    :       str=1_3			,					lowerCamelCase_    :       Any=2			,					lowerCamelCase_    :       Dict=2_4			,					lowerCamelCase_    :       Optional[int]=1_6			,					lowerCamelCase_    :       Any=True			,					lowerCamelCase_    :       Optional[int]=True			,					lowerCamelCase_    :       Optional[int]=3_2			,					lowerCamelCase_    :       List[str]=5			,					lowerCamelCase_    :       Dict=4			,					lowerCamelCase_    :       Optional[int]=3_7			,					lowerCamelCase_    :       Union[str, Any]="gelu"			,					lowerCamelCase_    :       Dict=0.1			,					lowerCamelCase_    :       int=0.1			,					lowerCamelCase_    :       Any=1_0			,					lowerCamelCase_    :       Union[str, Any]=0.02			,					lowerCamelCase_    :       str=None			,					lowerCamelCase_    :       Tuple=2			,					lowerCamelCase_    :       Union[str, Any]=2			,					):

					"""simple docstring"""






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

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







				def 							__UpperCAmelCase					( self    :       str						):

					"""simple docstring"""






					_lowercase :    Union[str, Any]							=							floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]						)

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

					_lowercase :    List[Any]							=							self.get_config()

					return config, input_values, labels







				def 							__UpperCAmelCase					( self    :       List[str]						):

					"""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 							__UpperCAmelCase					( self    :       int			,					lowerCamelCase_    :       int			,					lowerCamelCase_    :       Dict			,					lowerCamelCase_    :       str						):

					"""simple docstring"""






					_lowercase :    Optional[int]							=							ASTModel(config=A_						)
					model.to(A_						)
					model.eval()
					_lowercase :    Dict							=							model(A_						)
					self.parent.assertEqual(result.last_hidden_state.shape			,					(self.batch_size, self.seq_length, self.hidden_size)						)







				def 							__UpperCAmelCase					( self    :       List[str]						):

					"""simple docstring"""






					_lowercase :    Dict							=							self.prepare_config_and_inputs()
					(
					    (
					    _lowercase	
					)				,      (
					    _lowercase	
					)				,      (
					    _lowercase	
					)				,      	
					) :    Optional[Any]							=							config_and_inputs
					_lowercase :    Tuple							=							{'input_values': input_values}
					return config, inputs_dict



@require_torch
class  _lowerCamelCase			(_UpperCAmelCase						,							_UpperCAmelCase						,							unittest.TestCase     ):
				_snake_case     =  (
				    (
				        ASTModel,
				        ASTForAudioClassification,
				    )
				    if is_torch_available()
				    else ()
				)
				_snake_case     =  (
				    {'''audio-classification''': ASTForAudioClassification, '''feature-extraction''': ASTModel}
				    if is_torch_available()
				    else {}
				)
				_snake_case     =  False
				_snake_case     =  False
				_snake_case     =  False
				_snake_case     =  False







				def 							__UpperCAmelCase					( self    :       Tuple			,					lowerCamelCase_    :       Optional[int]			,					lowerCamelCase_    :       Any			,					lowerCamelCase_    :       Tuple			,					lowerCamelCase_    :       int			,					lowerCamelCase_    :       Union[str, Any]						):

					"""simple docstring"""






					if pipeline_test_casse_name == "AudioClassificationPipelineTests":
						return True

					return False







				def 							__UpperCAmelCase					( self    :       Dict						):

					"""simple docstring"""






					_lowercase :    str							=							ASTModelTester(self						)
					_lowercase :    Union[str, Any]							=							ConfigTester(self			,					config_class=A_			,					has_text_modality=A_			,					hidden_size=3_7						)







				def 							__UpperCAmelCase					( self    :       Optional[int]						):

					"""simple docstring"""






					self.config_tester.run_common_tests()







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

					"""simple docstring"""






					pass







				def 							__UpperCAmelCase					( self    :       Dict						):

					"""simple docstring"""






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

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







				def 							__UpperCAmelCase					( self    :       List[str]						):

					"""simple docstring"""






					_lowercase				,      _lowercase :    Dict							=							self.model_tester.prepare_config_and_inputs_for_common()

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

						_lowercase :    Any							=							['input_values']
						self.assertListEqual(arg_names[:1]			,					A_						)







				def 							__UpperCAmelCase					( self    :       Tuple						):

					"""simple docstring"""






					_lowercase :    List[Any]							=							self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*A_						)







				@slow
				def 							__UpperCAmelCase					( self    :       Tuple						):

					"""simple docstring"""






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



def      __lowerCAmelCase(		):






	"""simple docstring"""





	_lowercase :    List[Any]							=							hf_hub_download(
	    repo_id='nielsr/audio-spectogram-transformer-checkpoint'     ,filename='sample_audio.flac'     ,repo_type='dataset'			)

	_lowercase				,      _lowercase :    str							=							torchaudio.load(__UpperCAmelCase			)

	return audio, sampling_rate







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







				@cached_property
				def 							__UpperCAmelCase					( self    :       List[Any]						):

					"""simple docstring"""






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







				@slow
				def 							__UpperCAmelCase					( self    :       Dict						):

					"""simple docstring"""






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

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

					# forward pass
					with torch.no_grad():
						_lowercase :    int							=							model(**A_						)

					# verify the logits
					_lowercase :    Tuple							=							torch.Size((1, 5_2_7)						)
					self.assertEqual(outputs.logits.shape			,					A_						)

					_lowercase :    Tuple							=							torch.tensor([-0.8760, -7.0042, -8.6602]						).to(A_						)

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







import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  ""
				_snake_case     =  (
				    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
				)
				_snake_case     =  None  # compression type in fsspec. ex: "gzip"
				_snake_case     =  None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz







				def __init__( self    :       str			,					lowerCamelCase_    :       str = ""			,					lowerCamelCase_    :       Optional[str] = None			,					lowerCamelCase_    :       Optional[dict] = None			,					**lowerCamelCase_    :       List[str]						):

					"""simple docstring"""






					super().__init__(self			,					**lowerCamelCase_						)
					# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
					_lowercase :    Union[str, Any]							=							fsspec.open(
					    lowerCamelCase_			,					mode='rb'			,					protocol=lowerCamelCase_			,					compression=self.compression			,					client_kwargs={
					        'requote_redirect_url': False,  # see https://github.com/huggingface/datasets/pull/5459
					        'trust_env': True,  # Enable reading proxy env variables.
					        **(target_options or {}).pop('client_kwargs'			,					{}						),  # To avoid issues if it was already passed.
					    }			,					**(target_options or {})			,					)
					_lowercase :    str							=							os.path.basename(self.file.path.split('::'						)[0]						)
					_lowercase :    Optional[int]							=							(
					    self.compressed_name[: self.compressed_name.rindex('.'						)]
					    if '.' in self.compressed_name
					    else self.compressed_name
					)
					_lowercase :    str							=							None







				@classmethod
				def 							__UpperCAmelCase					( cls    :       int			,					lowerCamelCase_    :       List[str]						):

					"""simple docstring"""






					return super()._strip_protocol(lowerCamelCase_						).lstrip('/'						)







				def 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






					if self.dir_cache is None:
						_lowercase :    Tuple							=							{**self.file.fs.info(self.file.path						), 'name': self.uncompressed_name}
						_lowercase :    int							=							{f['name']: f}







				def 							__UpperCAmelCase					( self    :       List[str]			,					lowerCamelCase_    :       str						):

					"""simple docstring"""






					return self.file.open().read()







				def 							__UpperCAmelCase					( self    :       Dict			,					lowerCamelCase_    :       str			,					lowerCamelCase_    :       str = "rb"			,					lowerCamelCase_    :       str=None			,					lowerCamelCase_    :       str=True			,					lowerCamelCase_    :       Union[str, Any]=None			,					**lowerCamelCase_    :       Optional[int]			,					):

					"""simple docstring"""






					_lowercase :    Union[str, Any]							=							self._strip_protocol(lowerCamelCase_						)
					if mode != "rb":
						raise ValueError(F'''Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\''''						)
					return self.file.open()



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  "bz2"
				_snake_case     =  "bz2"
				_snake_case     =  ".bz2"



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  "gzip"
				_snake_case     =  "gzip"
				_snake_case     =  ".gz"



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  "lz4"
				_snake_case     =  "lz4"
				_snake_case     =  ".lz4"



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  "xz"
				_snake_case     =  "xz"
				_snake_case     =  ".xz"



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  "zstd"
				_snake_case     =  "zstd"
				_snake_case     =  ".zst"







				def __init__( self    :       Union[str, Any]			,					lowerCamelCase_    :       str			,					lowerCamelCase_    :       str = "rb"			,					lowerCamelCase_    :       Optional[str] = None			,					lowerCamelCase_    :       Optional[dict] = None			,					lowerCamelCase_    :       int = DEFAULT_BLOCK_SIZE			,					**lowerCamelCase_    :       int			,					):

					"""simple docstring"""






					super().__init__(
					    fo=lowerCamelCase_			,					mode=lowerCamelCase_			,					target_protocol=lowerCamelCase_			,					target_options=lowerCamelCase_			,					block_size=lowerCamelCase_			,					**lowerCamelCase_			,					)
					# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
					#
					# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
					#     out.close = close
					# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
					#
					# see https://github.com/intake/filesystem_spec/issues/725
					_lowercase :    Any							=							self.file.__enter__



					class  _lowerCamelCase			:







									def __init__( self    :       Any			,					lowerCamelCase_    :       List[Any]						):

										"""simple docstring"""






										_lowercase :    Tuple							=							file_







									def __enter__( self    :       str						):

										"""simple docstring"""






										self._file.__enter__()
										return self







									def __exit__( self    :       Union[str, Any]			,					*lowerCamelCase_    :       Optional[Any]			,					**lowerCamelCase_    :       Union[str, Any]						):

										"""simple docstring"""






										self._file.__exit__(*lowerCamelCase_			,					**lowerCamelCase_						)







									def __iter__( self    :       Optional[int]						):

										"""simple docstring"""






										return iter(self._file						)







									def 							__UpperCAmelCase					( self    :       int						):

										"""simple docstring"""






										return next(self._file						)







									def __getattr__( self    :       List[str]			,					lowerCamelCase_    :       str						):

										"""simple docstring"""






										return getattr(self._file			,					lowerCamelCase_						)

					def fixed_enter(*lowerCamelCase_    :       List[Any]			,					**lowerCamelCase_    :       int						):
						return WrappedFile(_enter(*lowerCamelCase_			,					**lowerCamelCase_						)						)

					_lowercase :    Optional[int]							=							fixed_enter



