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

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






     def 					__A						(	self						):
       _lowerCAmelCase		: str						    =  floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]						)

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

       _lowerCAmelCase		: List[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=UpperCamelCase__       ,  initializer_range=self.initializer_range       ,  frequency_stride=self.frequency_stride       ,  time_stride=self.time_stride       ,  )






     def 					__A						(	self       ,  a__       ,  a__       ,  a__						):
       _lowerCAmelCase		: Optional[Any]						    =  ASTModel(config=UpperCamelCase__						)
       model.to(UpperCamelCase__						)
       model.eval()
       _lowerCAmelCase		: Union[str, Any]						    =  model(UpperCamelCase__						)
       self.parent.assertEqual(result.last_hidden_state.shape       ,  (self.batch_size, self.seq_length, self.hidden_size)						)






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







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






     def 					__A						(	self       ,  a__       ,  a__       ,  a__       ,  a__       ,  a__						):
       if pipeline_test_casse_name == "AudioClassificationPipelineTests":
         return True

       return False






     def 					__A						(	self						):
       _lowerCAmelCase		: Union[str, Any]						    =  ASTModelTester(self						)
       _lowerCAmelCase		: Dict						    =  ConfigTester(self       ,  config_class=UpperCamelCase__       ,  has_text_modality=UpperCamelCase__       ,  hidden_size=37						)






     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						):
       _lowerCAmelCase		: int						    =  self.model_tester.prepare_config_and_inputs_for_common()

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






     def 					__A						(	self						):
       _lowerCAmelCase		: Tuple						    =  self.model_tester.prepare_config_and_inputs_for_common()

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

         _lowerCAmelCase		: List[Any]						    =  ['''input_values''']
         self.assertListEqual(arg_names[:1]       ,  UpperCamelCase__						)






     def 					__A						(	self						):
       _lowerCAmelCase		: str						    =  self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*UpperCamelCase__						)






     @slow
     def 					__A						(	self						):
       for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
         _lowerCAmelCase		: int						    =  ASTModel.from_pretrained(UpperCamelCase__						)
         self.assertIsNotNone(UpperCamelCase__						)







def 							SCREAMING_SNAKE_CASE  (    ) ->   Optional[int]:
  _lowerCAmelCase		: Tuple						    =  hf_hub_download(
      repo_id="""nielsr/audio-spectogram-transformer-checkpoint"""    ,filename="""sample_audio.flac"""    ,repo_type="""dataset"""			)

  _lowerCAmelCase		: List[Any]						    =  torchaudio.load(__UpperCamelCase			)

  return audio, sampling_rate






@require_torch
@require_torchaudio
class 		__A							(				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						):
       _lowerCAmelCase		: Union[str, Any]						    =  self.default_feature_extractor
       _lowerCAmelCase		: Optional[Any]						    =  ASTForAudioClassification.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593"""						).to(UpperCamelCase__						)

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

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

       # verify the logits
       _lowerCAmelCase		: Optional[int]						    =  torch.Size((1, 527)						)
       self.assertEqual(outputs.logits.shape       ,  UpperCamelCase__						)

       _lowerCAmelCase		: str						    =  torch.tensor([-0.8_7_6_0, -7.0_0_4_2, -8.6_6_0_2]						).to(UpperCamelCase__						)

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





"""simple docstring"""




from random import shuffle

import tensorflow as tf
from numpy import array





def 							SCREAMING_SNAKE_CASE  (    _lowerCamelCase		: Union[str, Any]    ,_lowerCamelCase		: Tuple			) ->   Dict:
  _lowerCAmelCase		: List[str]						    =  int(_lowerCamelCase			)
  assert noofclusters < len(_lowerCamelCase			)

  # Find out the dimensionality
  _lowerCAmelCase		: Any						    =  len(vectors[0]			)

  # Will help select random centroids from among the available vectors
  _lowerCAmelCase		: Any						    =  list(range(len(_lowerCamelCase			)			)			)
  shuffle(_lowerCamelCase			)

  # GRAPH OF COMPUTATION
  # We initialize a new graph and set it as the default during each run
  # of this algorithm. This ensures that as this function is called
  # multiple times, the default graph doesn't keep getting crowded with
  # unused ops and Variables from previous function calls.

  _lowerCAmelCase		: List[Any]						    =  tf.Graph()

  with graph.as_default():
    # SESSION OF COMPUTATION

    _lowerCAmelCase		: str						    =  tf.Session()

    ##CONSTRUCTING THE ELEMENTS OF COMPUTATION

    ##First lets ensure we have a Variable vector for each centroid,
    ##initialized to one of the vectors from the available data points
    _lowerCAmelCase		: List[str]						    =  [
        tf.Variable(vectors[vector_indices[i]]			) for i in range(_lowerCamelCase			)
    ]
    ##These nodes will assign the centroid Variables the appropriate
    ##values
    _lowerCAmelCase		: Union[str, Any]						    =  tf.placeholder("""float64"""    ,[dim]			)
    _lowerCAmelCase		: Optional[int]						    =  []
    for centroid in centroids:
      cent_assigns.append(tf.assign(_lowerCamelCase    ,_lowerCamelCase			)			)

    ##Variables for cluster assignments of individual vectors(initialized
    ##to 0 at first)
    _lowerCAmelCase		: Dict						    =  [tf.Variable(0			) for i in range(len(_lowerCamelCase			)			)]
    ##These nodes will assign an assignment Variable the appropriate
    ##value
    _lowerCAmelCase		: List[Any]						    =  tf.placeholder("""int32"""			)
    _lowerCAmelCase		: Any						    =  []
    for assignment in assignments:
      cluster_assigns.append(tf.assign(_lowerCamelCase    ,_lowerCamelCase			)			)

    ##Now lets construct the node that will compute the mean
    # The placeholder for the input
    _lowerCAmelCase		: Union[str, Any]						    =  tf.placeholder("""float"""    ,[None, dim]			)
    # The Node/op takes the input and computes a mean along the 0th
    # dimension, i.e. the list of input vectors
    _lowerCAmelCase		: Optional[int]						    =  tf.reduce_mean(_lowerCamelCase    ,0			)

    ##Node for computing Euclidean distances
    # Placeholders for input
    _lowerCAmelCase		: Dict						    =  tf.placeholder("""float"""    ,[dim]			)
    _lowerCAmelCase		: Any						    =  tf.placeholder("""float"""    ,[dim]			)
    _lowerCAmelCase		: List[Any]						    =  tf.sqrt(tf.reduce_sum(tf.pow(tf.sub(_lowerCamelCase    ,_lowerCamelCase			)    ,2			)			)			)

    ##This node will figure out which cluster to assign a vector to,
    ##based on Euclidean distances of the vector from the centroids.
    # Placeholder for input
    _lowerCAmelCase		: Any						    =  tf.placeholder("""float"""    ,[noofclusters]			)
    _lowerCAmelCase		: str						    =  tf.argmin(_lowerCamelCase    ,0			)

    ##INITIALIZING STATE VARIABLES

    ##This will help initialization of all Variables defined with respect
    ##to the graph. The Variable-initializer should be defined after
    ##all the Variables have been constructed, so that each of them
    ##will be included in the initialization.
    _lowerCAmelCase		: Optional[Any]						    =  tf.initialize_all_variables()

    # Initialize all variables
    sess.run(_lowerCamelCase			)

    ##CLUSTERING ITERATIONS

    # Now perform the Expectation-Maximization steps of K-Means clustering
    # iterations. To keep things simple, we will only do a set number of
    # iterations, instead of using a Stopping Criterion.
    _lowerCAmelCase		: List[str]						    =  100
    for _ in range(_lowerCamelCase			):
      ##EXPECTATION STEP
      ##Based on the centroid locations till last iteration, compute
      ##the _expected_ centroid assignments.
      # Iterate over each vector
      for vector_n in range(len(_lowerCamelCase			)			):
        _lowerCAmelCase		: int						    =  vectors[vector_n]
        # Compute Euclidean distance between this vector and each
        # centroid. Remember that this list cannot be named
        #'centroid_distances', since that is the input to the
        # cluster assignment node.
        _lowerCAmelCase		: Any						    =  [
            sess.run(_lowerCamelCase    ,feed_dict={va: vect, va: sess.run(_lowerCamelCase			)}			)
            for centroid in centroids
        ]
        # Now use the cluster assignment node, with the distances
        # as the input
        _lowerCAmelCase		: Any						    =  sess.run(
            _lowerCamelCase    ,feed_dict={centroid_distances: distances}			)
        # Now assign the value to the appropriate state variable
        sess.run(
            cluster_assigns[vector_n]    ,feed_dict={assignment_value: assignment}			)

      ##MAXIMIZATION STEP
      # Based on the expected state computed from the Expectation Step,
      # compute the locations of the centroids so as to maximize the
      # overall objective of minimizing within-cluster Sum-of-Squares
      for cluster_n in range(_lowerCamelCase			):
        # Collect all the vectors assigned to this cluster
        _lowerCAmelCase		: List[Any]						    =  [
            vectors[i]
            for i in range(len(_lowerCamelCase			)			)
            if sess.run(assignments[i]			) == cluster_n
        ]
        # Compute new centroid location
        _lowerCAmelCase		: Optional[int]						    =  sess.run(
            _lowerCamelCase    ,feed_dict={mean_input: array(_lowerCamelCase			)}			)
        # Assign value to appropriate variable
        sess.run(
            cent_assigns[cluster_n]    ,feed_dict={centroid_value: new_location}			)

        # Return centroids and assignments
    _lowerCAmelCase		: Optional[int]						    =  sess.run(_lowerCamelCase			)
    _lowerCAmelCase		: List[Any]						    =  sess.run(_lowerCamelCase			)
    return centroids, assignments




