'''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       :       str  ,     __a       :       Any  ,     __a       :       str=13  ,     __a       :       List[str]=2  ,     __a       :       Any=24  ,     __a       :       Any=16  ,     __a       :       List[str]=True  ,     __a       :       Tuple=True  ,     __a       :       Any=32  ,     __a       :       str=5  ,     __a       :       Optional[int]=4  ,     __a       :       List[Any]=37  ,     __a       :       Optional[Any]="gelu"  ,     __a       :       str=0.1  ,     __a       :       Tuple=0.1  ,     __a       :       Tuple=10  ,     __a       :       Any=0.02  ,     __a       :       Any=None  ,     __a       :       Union[str, Any]=2  ,     __a       :       Tuple=2  ,     ):
         UpperCAmelCase_			       =  parent
         UpperCAmelCase_			       =  batch_size
         UpperCAmelCase_			       =  patch_size
         UpperCAmelCase_			       =  max_length
         UpperCAmelCase_			       =  num_mel_bins
         UpperCAmelCase_			       =  is_training
         UpperCAmelCase_			       =  use_labels
         UpperCAmelCase_			       =  hidden_size
         UpperCAmelCase_			       =  num_hidden_layers
         UpperCAmelCase_			       =  num_attention_heads
         UpperCAmelCase_			       =  intermediate_size
         UpperCAmelCase_			       =  hidden_act
         UpperCAmelCase_			       =  hidden_dropout_prob
         UpperCAmelCase_			       =  attention_probs_dropout_prob
         UpperCAmelCase_			       =  type_sequence_label_size
         UpperCAmelCase_			       =  initializer_range
         UpperCAmelCase_			       =  scope
         UpperCAmelCase_			       =  frequency_stride
         UpperCAmelCase_			       =  time_stride

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



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

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

         UpperCAmelCase_			       =  self.get_config()

         return config, input_values, labels



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



    def    _lowercase		(self       :       str  ,     __a       :       Any  ,     __a       :       str  ,     __a       :       str					):
         UpperCAmelCase_			       =  ASTModel(config=__a					)
         model.to(__a					)
         model.eval()
         UpperCAmelCase_			       =  model(__a					)
         self.parent.assertEqual(result.last_hidden_state.shape  ,     (self.batch_size, self.seq_length, self.hidden_size)					)






    def    _lowercase		(self       :       Dict					):
         UpperCAmelCase_			       =  self.prepare_config_and_inputs()
         (
             (
             UpperCAmelCase_
         )       ,					(
             UpperCAmelCase_
         )       ,					(
             UpperCAmelCase_
         )       ,					
         )			       =  config_and_inputs
         UpperCAmelCase_			       =  {"input_values": input_values}
         return config, inputs_dict




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



    def    _lowercase		(self       :       Optional[Any]  ,     __a       :       str  ,     __a       :       List[Any]  ,     __a       :       Optional[Any]  ,     __a       :       str  ,     __a       :       str					):
         if pipeline_test_casse_name == "AudioClassificationPipelineTests":
              return True

         return False



    def    _lowercase		(self       :       Optional[Any]					):
         UpperCAmelCase_			       =  ASTModelTester(self					)
         UpperCAmelCase_			       =  ConfigTester(self  ,     config_class=__a  ,     has_text_modality=__a  ,     hidden_size=37					)



    def    _lowercase		(self       :       List[str]					):
         self.config_tester.run_common_tests()



    @unittest.skip(reason="AST does not use inputs_embeds"					)
    def    _lowercase		(self       :       List[str]					):
         pass



    def    _lowercase		(self       :       Optional[Any]					):
         UpperCAmelCase_       ,					UpperCAmelCase_			       =  self.model_tester.prepare_config_and_inputs_for_common()

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



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

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

              UpperCAmelCase_			       =  ["input_values"]
              self.assertListEqual(arg_names[:1]  ,     __a					)



    def    _lowercase		(self       :       Union[str, Any]					):
         UpperCAmelCase_			       =  self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*__a					)






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





def       lowerCAmelCase_							(   )       -> Optional[int]:


     '''simple docstring'''




     UpperCAmelCase_			       =  hf_hub_download(
         repo_id="nielsr/audio-spectogram-transformer-checkpoint"   ,       filename="sample_audio.flac"   ,       repo_type="dataset"     )

     UpperCAmelCase_       ,					UpperCAmelCase_			       =  torchaudio.load(snake_case_     )

     return audio, sampling_rate






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



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






    @slow
    def    _lowercase		(self       :       Optional[int]					):
         UpperCAmelCase_			       =  self.default_feature_extractor
         UpperCAmelCase_			       =  ASTForAudioClassification.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593"					).to(__a					)

         UpperCAmelCase_			       =  self.default_feature_extractor
         UpperCAmelCase_       ,					UpperCAmelCase_			       =  prepare_audio()
         UpperCAmelCase_			       =  audio.squeeze().numpy()
         UpperCAmelCase_			       =  feature_extractor(__a  ,     sampling_rate=__a  ,     return_tensors="pt"					).to(__a					)

         # forward pass
         with torch.no_grad():
              UpperCAmelCase_			       =  model(**__a					)

         # verify the logits
         UpperCAmelCase_			       =  torch.Size((1, 527)					)
         self.assertEqual(outputs.logits.shape  ,     __a					)

         UpperCAmelCase_			       =  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					)					)










"""simple docstring"""




import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase				:     Union[str, Any]					=				logging.get_logger(__name__)

lowerCAmelCase				:     Optional[int]					=				{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

lowerCAmelCase				:     Optional[Any]					=				{
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

lowerCAmelCase				:     Optional[Any]					=				{
    """ctrl""": 256,
}

lowerCAmelCase				:     int					=				{
    """Pregnancy""": 16_8629,
    """Christianity""": 7675,
    """Explain""": 10_6423,
    """Fitness""": 6_3440,
    """Saving""": 6_3163,
    """Ask""": 2_7171,
    """Ass""": 9_5985,
    """Joke""": 16_3509,
    """Questions""": 4_5622,
    """Thoughts""": 4_9605,
    """Retail""": 5_2342,
    """Feminism""": 16_4338,
    """Writing""": 1_1992,
    """Atheism""": 19_2263,
    """Netflix""": 4_8616,
    """Computing""": 3_9639,
    """Opinion""": 4_3213,
    """Alone""": 4_4967,
    """Funny""": 5_8917,
    """Gaming""": 4_0358,
    """Human""": 4088,
    """India""": 1331,
    """Joker""": 7_7138,
    """Diet""": 3_6206,
    """Legal""": 1_1859,
    """Norman""": 4939,
    """Tip""": 7_2689,
    """Weight""": 5_2343,
    """Movies""": 4_6273,
    """Running""": 2_3425,
    """Science""": 2090,
    """Horror""": 3_7793,
    """Confession""": 6_0572,
    """Finance""": 1_2250,
    """Politics""": 1_6360,
    """Scary""": 19_1985,
    """Support""": 1_2654,
    """Technologies""": 3_2516,
    """Teenage""": 6_6160,
    """Event""": 3_2769,
    """Learned""": 6_7460,
    """Notion""": 18_2770,
    """Wikipedia""": 3_7583,
    """Books""": 6665,
    """Extract""": 7_6050,
    """Confessions""": 10_2701,
    """Conspiracy""": 7_5932,
    """Links""": 6_3674,
    """Narcissus""": 15_0425,
    """Relationship""": 5_4766,
    """Relationships""": 13_4796,
    """Reviews""": 4_1671,
    """News""": 4256,
    """Translation""": 2_6820,
    """multilingual""": 12_8406,
}



def        a__					(							snake_case__							)		->   Optional[Any]:
			lowerCamelCase      =   set()
			lowerCamelCase      =   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)							)
						lowerCamelCase      =   char

			lowerCamelCase      =   set(snake_case__							)
			return pairs






class        __magic_name__				(			UpperCAmelCase__							):






				'''simple docstring'''
				__UpperCamelCase         =		VOCAB_FILES_NAMES
				__UpperCamelCase         =		PRETRAINED_VOCAB_FILES_MAP
				__UpperCamelCase         =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__UpperCamelCase         =		CONTROL_CODES





				def __init__(			self						,					_a						,					_a						,					_a="<unk>"						,					**_a					):





							"""simple docstring"""
							super().__init__(unk_token=_a						,					**_a					)

							with open(_a						,					encoding="""utf-8"""					) as vocab_handle:
										lowerCamelCase      =   json.load(_a					)
							lowerCamelCase      =   {v: k for k, v in self.encoder.items()}
							with open(_a						,					encoding="""utf-8"""					) as merges_handle:
										lowerCamelCase      =   merges_handle.read().split("""\n"""					)[1:-1]
							lowerCamelCase      =   [tuple(merge.split()					) for merge in merges]
							lowerCamelCase      =   dict(zip(_a						,					range(len(_a					)					)					)					)
							lowerCamelCase      =   {}





				@property
				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return len(self.encoder					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return dict(self.encoder						,					**self.added_tokens_encoder					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							if token in self.cache:
										return self.cache[token]
							lowerCamelCase      =   tuple(_a					)
							lowerCamelCase      =   tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
							lowerCamelCase      =   get_pairs(_a					)

							if not pairs:
										return token

							while True:
										lowerCamelCase      =   min(_a						,					key=lambda _a					: self.bpe_ranks.get(_a						,					float("""inf"""					)					)					)
										if bigram not in self.bpe_ranks:
													break
										lowerCamelCase ,  lowerCamelCase      =   bigram
										lowerCamelCase      =   []
										lowerCamelCase      =   0
										while i < len(_a					):
													try:
																lowerCamelCase      =   word.index(_a						,					_a					)
													except ValueError:
																new_word.extend(word[i:]					)
																break
													else:
																new_word.extend(word[i:j]					)
																lowerCamelCase      =   j

													if word[i] == first and i < len(_a					) - 1 and word[i + 1] == second:
																new_word.append(first + second					)
																i += 2
													else:
																new_word.append(word[i]					)
																i += 1
										lowerCamelCase      =   tuple(_a					)
										lowerCamelCase      =   new_word
										if len(_a					) == 1:
													break
										else:
													lowerCamelCase      =   get_pairs(_a					)
							lowerCamelCase      =   """@@ """.join(_a					)
							lowerCamelCase      =   word[:-4]
							lowerCamelCase      =   word
							return word





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   []

							lowerCamelCase      =   re.findall(r"""\S+\n?"""						,					_a					)

							for token in words:
										split_tokens.extend(list(self.bpe(_a					).split(""" """					)					)					)
							return split_tokens





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.encoder.get(_a						,					self.encoder.get(self.unk_token					)					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.decoder.get(_a						,					self.unk_token					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   """ """.join(_a					).replace("""@@ """						,					""""""					).strip()
							return out_string





				def   _lowerCAmelCase							(			self						,					_a						,					_a = None					):





							"""simple docstring"""
							if not os.path.isdir(_a					):
										logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
										return
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]					)
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]					)

							with open(_a						,					"""w"""						,					encoding="""utf-8"""					) as f:
										f.write(json.dumps(self.encoder						,					indent=2						,					sort_keys=_a						,					ensure_ascii=_a					) + """\n"""					)

							lowerCamelCase      =   0
							with open(_a						,					"""w"""						,					encoding="""utf-8"""					) as writer:
										writer.write("""#version: 0.2\n"""					)
										for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,					key=lambda _a					: kv[1]					):
													if index != token_index:
																logger.warning(
																    f'Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'
																    """ Please check that the tokenizer is not corrupted!"""					)
																lowerCamelCase      =   token_index
													writer.write(""" """.join(_a					) + """\n"""					)
													index += 1

							return vocab_file, merge_file

				# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
				#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
				#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
				#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
				#     return ''.join(tokens_generated_so_far)



