




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


			def __init__(      self      :    Any      ,  UpperCamelCase_      :    int      ,  UpperCamelCase_      :    Any=13      ,  UpperCamelCase_      :    str=2      ,  UpperCamelCase_      :    Optional[Any]=24      ,  UpperCamelCase_      :    str=16      ,  UpperCamelCase_      :    str=True      ,  UpperCamelCase_      :    str=True      ,  UpperCamelCase_      :    List[str]=32      ,  UpperCamelCase_      :    Optional[Any]=5      ,  UpperCamelCase_      :    List[str]=4      ,  UpperCamelCase_      :    Any=37      ,  UpperCamelCase_      :    List[Any]="gelu"      ,  UpperCamelCase_      :    int=0.1      ,  UpperCamelCase_      :    List[Any]=0.1      ,  UpperCamelCase_      :    Tuple=10      ,  UpperCamelCase_      :    str=0.02      ,  UpperCamelCase_      :    Dict=None      ,  UpperCamelCase_      :    str=2      ,  UpperCamelCase_      :    str=2      ,  ):


						"""simple docstring"""




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

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


			def       a_     (      self      :    str):


						"""simple docstring"""




						__UpperCAmelCase	: Any		=					floats_tensor([self.batch_size, self.max_length, self.num_mel_bins])

						__UpperCAmelCase	: Optional[int]		=					None
						if self.use_labels:
									__UpperCAmelCase	: List[str]		=					ids_tensor([self.batch_size]      ,  self.type_sequence_label_size)

						__UpperCAmelCase	: Any		=					self.get_config()

						return config, input_values, labels


			def       a_     (      self      :    Union[str, Any]):


						"""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=UpperCamelCase_      ,  initializer_range=self.initializer_range      ,  frequency_stride=self.frequency_stride      ,  time_stride=self.time_stride      ,  )


			def       a_     (      self      :    Dict      ,  UpperCamelCase_      :    Optional[int]      ,  UpperCamelCase_      :    Optional[Any]      ,  UpperCamelCase_      :    Optional[int]):


						"""simple docstring"""




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


			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




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






@require_torch
class      a__				(	__magic_name__							,    __magic_name__							,    unittest.TestCase   ):
			lowercase_    =	(
			    (
			        ASTModel,
			        ASTForAudioClassification,
			    )
			    if is_torch_available()
			    else ()
			)
			lowercase_    =	(
			    {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
			    if is_torch_available()
			    else {}
			)
			lowercase_    =	False
			lowercase_    =	False
			lowercase_    =	False
			lowercase_    =	False


			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    Union[str, Any]      ,  UpperCamelCase_      :    Dict      ,  UpperCamelCase_      :    Any      ,  UpperCamelCase_      :    Tuple      ,  UpperCamelCase_      :    Any):


						"""simple docstring"""




						if pipeline_test_casse_name == "AudioClassificationPipelineTests":
									return True

						return False


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					ASTModelTester(self)
						__UpperCAmelCase	: str		=					ConfigTester(self      ,  config_class=UpperCamelCase_      ,  has_text_modality=UpperCamelCase_      ,  hidden_size=37)


			def       a_     (      self      :    Tuple):


						"""simple docstring"""




						self.config_tester.run_common_tests()


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


						"""simple docstring"""




						pass


			def       a_     (      self      :    int):


						"""simple docstring"""




						__UpperCAmelCase	,     __UpperCAmelCase	: List[str]		=					self.model_tester.prepare_config_and_inputs_for_common()

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


			def       a_     (      self      :    int):


						"""simple docstring"""




						__UpperCAmelCase	,     __UpperCAmelCase	: Optional[Any]		=					self.model_tester.prepare_config_and_inputs_for_common()

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

									__UpperCAmelCase	: Optional[Any]		=					["input_values"]
									self.assertListEqual(arg_names[:1]      ,  UpperCamelCase_)


			def       a_     (      self      :    Dict):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*UpperCamelCase_)


			@slow
			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




						for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									__UpperCAmelCase	: Optional[Any]		=					ASTModel.from_pretrained(UpperCamelCase_)
									self.assertIsNotNone(UpperCamelCase_)






def 					_UpperCamelCase    (			)			->			Any:


			"""simple docstring"""
			__UpperCAmelCase	: Union[str, Any]		=					hf_hub_download(
			    repo_id="nielsr/audio-spectogram-transformer-checkpoint"				,       filename="sample_audio.flac"				,       repo_type="dataset"							)

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

			return audio, sampling_rate





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


			@cached_property
			def       a_     (      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       a_     (      self      :    int):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					self.default_feature_extractor
						__UpperCAmelCase	: str		=					ASTForAudioClassification.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593").to(UpperCamelCase_)

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

						# forward pass
						with torch.no_grad():
									__UpperCAmelCase	: Dict		=					model(**UpperCamelCase_)

						# verify the logits
						__UpperCAmelCase	: Tuple		=					torch.Size((1, 527))
						self.assertEqual(outputs.logits.shape      ,  UpperCamelCase_)

						__UpperCAmelCase	: Union[str, Any]		=					torch.tensor([-0.8760, -7.0042, -8.6602]).to(UpperCamelCase_)

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








"""simple docstring"""





import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format="""%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s""",
    datefmt="""%m/%d/%Y %H:%M:%S""",
    level=logging.INFO,
)
A   =       logging.getLogger(__name__)

def 					_UpperCamelCase    (			UpperCamelCase							)			->			Any:


			"""simple docstring"""
			__UpperCAmelCase	: Optional[Any]		=					git.Repo(search_parent_directories=UpperCamelCase							)
			__UpperCAmelCase	: Any		=					{
			    "repo_id": str(UpperCamelCase							),
			    "repo_sha": str(repo.head.object.hexsha							),
			    "repo_branch": str(repo.active_branch							),
			}

			with open(os.path.join(UpperCamelCase				,       "git_log.json"							)				,       "w"							) as f:
						json.dump(UpperCamelCase				,       UpperCamelCase				,       indent=4							)

def 					_UpperCamelCase    (			UpperCamelCase							)			->			str:


			"""simple docstring"""
			if params.n_gpu <= 0:
						__UpperCAmelCase	: str		=					0
						__UpperCAmelCase	: Dict		=					-1
						__UpperCAmelCase	: Tuple		=					True
						__UpperCAmelCase	: List[str]		=					False
						return

			assert torch.cuda.is_available()

			logger.info("Initializing GPUs"							)
			if params.n_gpu > 1:
						assert params.local_rank != -1

						__UpperCAmelCase	: Optional[int]		=					int(os.environ["WORLD_SIZE"]							)
						__UpperCAmelCase	: Union[str, Any]		=					int(os.environ["N_GPU_NODE"]							)
						__UpperCAmelCase	: Optional[int]		=					int(os.environ["RANK"]							)

						# number of nodes / node ID
						__UpperCAmelCase	: int		=					params.world_size // params.n_gpu_per_node
						__UpperCAmelCase	: Optional[int]		=					params.global_rank // params.n_gpu_per_node
						__UpperCAmelCase	: Tuple		=					True

						assert params.n_nodes == int(os.environ["N_NODES"]							)
						assert params.node_id == int(os.environ["NODE_RANK"]							)

			# local job (single GPU)
			else:
						assert params.local_rank == -1

						__UpperCAmelCase	: Union[str, Any]		=					1
						__UpperCAmelCase	: Dict		=					0
						__UpperCAmelCase	: List[Any]		=					0
						__UpperCAmelCase	: Tuple		=					0
						__UpperCAmelCase	: Any		=					1
						__UpperCAmelCase	: str		=					1
						__UpperCAmelCase	: Dict		=					False

			# sanity checks
			assert params.n_nodes >= 1
			assert 0 <= params.node_id < params.n_nodes
			assert 0 <= params.local_rank <= params.global_rank < params.world_size
			assert params.world_size == params.n_nodes * params.n_gpu_per_node

			# define whether this is the master process / if we are in multi-node distributed mode
			__UpperCAmelCase	: Union[str, Any]		=					params.node_id == 0 and params.local_rank == 0
			__UpperCAmelCase	: Dict		=					params.n_nodes > 1

			# summary
			__UpperCAmelCase	: Any		=					f"--- Global rank: {params.global_rank} - "
			logger.info(PREFIX + "Number of nodes: %i" % params.n_nodes							)
			logger.info(PREFIX + "Node ID        : %i" % params.node_id							)
			logger.info(PREFIX + "Local rank     : %i" % params.local_rank							)
			logger.info(PREFIX + "World size     : %i" % params.world_size							)
			logger.info(PREFIX + "GPUs per node  : %i" % params.n_gpu_per_node							)
			logger.info(PREFIX + "Master         : %s" % str(params.is_master							)							)
			logger.info(PREFIX + "Multi-node     : %s" % str(params.multi_node							)							)
			logger.info(PREFIX + "Multi-GPU      : %s" % str(params.multi_gpu							)							)
			logger.info(PREFIX + "Hostname       : %s" % socket.gethostname()							)

			# set GPU device
			torch.cuda.set_device(params.local_rank							)

			# initialize multi-GPU
			if params.multi_gpu:
						logger.info("Initializing PyTorch distributed"							)
						torch.distributed.init_process_group(
						    init_method="env://"				,       backend="nccl"				,       )





def 					_UpperCamelCase    (			UpperCamelCase							)			->			Tuple:


			"""simple docstring"""
			np.random.seed(args.seed							)
			torch.manual_seed(args.seed							)
			if args.n_gpu > 0:
						torch.cuda.manual_seed_all(args.seed							)


