






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 ,						_lowerCAmelCase ,						_lowerCAmelCase=1_3 ,						_lowerCAmelCase=2 ,						_lowerCAmelCase=2_4 ,						_lowerCAmelCase=1_6 ,						_lowerCAmelCase=True ,						_lowerCAmelCase=True ,						_lowerCAmelCase=3_2 ,						_lowerCAmelCase=5 ,						_lowerCAmelCase=4 ,						_lowerCAmelCase=3_7 ,						_lowerCAmelCase="gelu" ,						_lowerCAmelCase=0.1 ,						_lowerCAmelCase=0.1 ,						_lowerCAmelCase=1_0 ,						_lowerCAmelCase=0.02 ,						_lowerCAmelCase=None ,						_lowerCAmelCase=2 ,						_lowerCAmelCase=2 ,						):
											_lowercase							:   List[Any] 							=   parent
											_lowercase							:   Dict 							=   batch_size
											_lowercase							:   int 							=   patch_size
											_lowercase							:   int 							=   max_length
											_lowercase							:   Optional[Any] 							=   num_mel_bins
											_lowercase							:   Any 							=   is_training
											_lowercase							:   int 							=   use_labels
											_lowercase							:   Dict 							=   hidden_size
											_lowercase							:   int 							=   num_hidden_layers
											_lowercase							:   Any 							=   num_attention_heads
											_lowercase							:   str 							=   intermediate_size
											_lowercase							:   Tuple 							=   hidden_act
											_lowercase							:   Optional[int] 							=   hidden_dropout_prob
											_lowercase							:   List[Any] 							=   attention_probs_dropout_prob
											_lowercase							:   Dict 							=   type_sequence_label_size
											_lowercase							:   List[str] 							=   initializer_range
											_lowercase							:   List[Any] 							=   scope
											_lowercase							:   str 							=   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[Any] 							=   (self.num_mel_bins - self.patch_size) // self.frequency_stride + 1
											_lowercase							:   int 							=   (self.max_length - self.patch_size) // self.time_stride + 1
											_lowercase							:   List[Any] 							=   frequency_out_dimension * time_out_dimension
											_lowercase							:   Tuple 							=   num_patches + 2


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]      )

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

											_lowercase							:   Union[str, Any] 							=   self.get_config()

											return config, input_values, labels


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


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											_lowercase							:   Dict 							=   ASTModel(config=_lowerCAmelCase      )
											model.to(_lowerCAmelCase      )
											model.eval()
											_lowercase							:   List[str] 							=   model(_lowerCAmelCase      )
											self.parent.assertEqual(result.last_hidden_state.shape ,						(self.batch_size, self.seq_length, self.hidden_size)      )


				def 						__a					(					self      ):
											_lowercase							:   Optional[int] 							=   self.prepare_config_and_inputs()
											(
											    (
											    _lowercase
											)							,							(
											    _lowercase
											)							,							(
											    _lowercase
											)							,							
											)							:   Union[str, Any] 							=   config_and_inputs
											_lowercase							:   Optional[int] 							=   {'input_values': input_values}
											return config, inputs_dict




@require_torch
class 				lowerCAmelCase_	(		__snake_case   ,		__snake_case   ,		unittest.TestCase			):
				_UpperCamelCase					:       List[Any]							       = (
				    (
				        ASTModel,
				        ASTForAudioClassification,
				    )
				    if is_torch_available()
				    else ()
				)
				_UpperCamelCase					:       Tuple							       = (
				    {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
				    if is_torch_available()
				    else {}
				)
				_UpperCamelCase					:       int							       = False
				_UpperCamelCase					:       Optional[Any]							       = False
				_UpperCamelCase					:       Any							       = False
				_UpperCamelCase					:       str							       = False


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											if pipeline_test_casse_name == "AudioClassificationPipelineTests":
																		return True

											return False


				def 						__a					(					self      ):
											_lowercase							:   Optional[Any] 							=   ASTModelTester(self      )
											_lowercase							:   List[Any] 							=   ConfigTester(self ,						config_class=_lowerCAmelCase ,						has_text_modality=_lowerCAmelCase ,						hidden_size=3_7      )


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


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


				def 						__a					(					self      ):
											_lowercase							,							_lowercase							:   List[str] 							=   self.model_tester.prepare_config_and_inputs_for_common()

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


				def 						__a					(					self      ):
											_lowercase							,							_lowercase							:   Union[str, Any] 							=   self.model_tester.prepare_config_and_inputs_for_common()

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

																		_lowercase							:   Tuple 							=   ['input_values']
																		self.assertListEqual(arg_names[:1] ,						_lowerCAmelCase      )


				def 						__a					(					self      ):
											_lowercase							:   Optional[int] 							=   self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*_lowerCAmelCase      )


				@slow
				def 						__a					(					self      ):
											for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																		_lowercase							:   Any 							=   ASTModel.from_pretrained(_lowerCAmelCase      )
																		self.assertIsNotNone(_lowerCAmelCase      )






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

							_lowercase							,							_lowercase							:   Tuple 							=   torchaudio.load(SCREAMING_SNAKE_CASE				)

							return audio, sampling_rate






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


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


				@slow
				def 						__a					(					self      ):
											_lowercase							:   List[Any] 							=   self.default_feature_extractor
											_lowercase							:   List[Any] 							=   ASTForAudioClassification.from_pretrained('MIT/ast-finetuned-audioset-10-10-0.4593'      ).to(_lowerCAmelCase      )

											_lowercase							:   List[Any] 							=   self.default_feature_extractor
											_lowercase							,							_lowercase							:   Any 							=   prepare_audio()
											_lowercase							:   int 							=   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[str] 							=   model(**_lowerCAmelCase      )

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

											_lowercase							:   Dict 							=   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      )      )










'''simple docstring'''


import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_huggingface_suite,
    require_multi_gpu,
    require_single_gpu,
)
from accelerate.utils import patch_environment





@require_huggingface_suite
class  __lowercase    (	unittest.TestCase			):







   def 		UpperCamelCase__      (						self   )					->     Tuple:
     __a								=       inspect.getfile(accelerate.test_utils   )
     __a								=       os.path.sep.join(
         mod_file.split(os.path.sep   )[:-1] + ['scripts', 'external_deps', 'test_metrics.py']   )

     from accelerate.test_utils.scripts.external_deps import test_metrics  # noqa: F401

     __a								=       test_metrics







   @require_cpu
   def 		UpperCamelCase__      (						self   )					->     Tuple:
     debug_launcher(self.test_metrics.main   ,				num_processes=1   )







   @require_cpu
   def 		UpperCamelCase__      (						self   )					->     str:
     debug_launcher(self.test_metrics.main   )







   @require_single_gpu
   def 		UpperCamelCase__      (						self   )					->     Dict:
     self.test_metrics.main()







   @require_multi_gpu
   def 		UpperCamelCase__      (						self   )					->     int:
     print(f"Found {torch.cuda.device_count()} devices."   )
     __a								=       ['torchrun', f"--nproc_per_node={torch.cuda.device_count()}", self.test_file_path]
     with patch_environment(omp_num_threads=1   ):
       execute_subprocess_async(UpperCamelCase   ,				env=os.environ.copy()   )



