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     :						Dict       ,  __UpperCAmelCase     :						int       ,  __UpperCAmelCase     :						Any=1_3       ,  __UpperCAmelCase     :						Dict=2       ,  __UpperCAmelCase     :						Tuple=2_4       ,  __UpperCAmelCase     :						List[str]=1_6       ,  __UpperCAmelCase     :						List[Any]=True       ,  __UpperCAmelCase     :						Union[str, Any]=True       ,  __UpperCAmelCase     :						Dict=3_2       ,  __UpperCAmelCase     :						str=5       ,  __UpperCAmelCase     :						Dict=4       ,  __UpperCAmelCase     :						Tuple=3_7       ,  __UpperCAmelCase     :						int="gelu"       ,  __UpperCAmelCase     :						Dict=0.1       ,  __UpperCAmelCase     :						Dict=0.1       ,  __UpperCAmelCase     :						Optional[int]=1_0       ,  __UpperCAmelCase     :						Optional[Any]=0.02       ,  __UpperCAmelCase     :						List[Any]=None       ,  __UpperCAmelCase     :						Dict=2       ,  __UpperCAmelCase     :						Tuple=2       ,  )						->				Tuple:
         """simple docstring"""



         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     :						str		)						->				str:
         """simple docstring"""



         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[Any]		)						->				Optional[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 			lowercase_      (self     :						Union[str, Any]       ,  __UpperCAmelCase     :						List[str]       ,  __UpperCAmelCase     :						Optional[int]       ,  __UpperCAmelCase     :						Optional[int]		)						->				Optional[Any]:
         """simple docstring"""



         UpperCAmelCase__								=  ASTModel(config=__UpperCAmelCase		)
         model.to(__UpperCAmelCase		)
         model.eval()
         UpperCAmelCase__								=  model(__UpperCAmelCase		)
         self.parent.assertEqual(result.last_hidden_state.shape       ,  (self.batch_size, self.seq_length, self.hidden_size)		)





   def 			lowercase_      (self     :						Union[str, Any]		)						->				Union[str, Any]:
         """simple docstring"""



         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     ):
   __UpperCAmelCase : List[Any]      =			(
       (
           ASTModel,
           ASTForAudioClassification,
       )
       if is_torch_available()
       else ()
   )
   __UpperCAmelCase : List[str]      =			(
       {'audio-classification': ASTForAudioClassification, 'feature-extraction': ASTModel}
       if is_torch_available()
       else {}
   )
   __UpperCAmelCase : Optional[Any]      =			False
   __UpperCAmelCase : Tuple      =			False
   __UpperCAmelCase : Any      =			False
   __UpperCAmelCase : Any      =			False


   def 			lowercase_      (self     :						Union[str, Any]       ,  __UpperCAmelCase     :						Any       ,  __UpperCAmelCase     :						Optional[int]       ,  __UpperCAmelCase     :						Union[str, Any]       ,  __UpperCAmelCase     :						Dict       ,  __UpperCAmelCase     :						Tuple		)						->				Any:
         """simple docstring"""



         if pipeline_test_casse_name == "AudioClassificationPipelineTests":
               return True

         return False


   def 			lowercase_      (self     :						Optional[Any]		)						->				Tuple:
         """simple docstring"""



         UpperCAmelCase__								=  ASTModelTester(self		)
         UpperCAmelCase__								=  ConfigTester(self       ,  config_class=__UpperCAmelCase       ,  has_text_modality=__UpperCAmelCase       ,  hidden_size=3_7		)


   def 			lowercase_      (self     :						Union[str, Any]		)						->				Tuple:
         """simple docstring"""



         self.config_tester.run_common_tests()


   @unittest.skip(reason="AST does not use inputs_embeds"		)
   def 			lowercase_      (self     :						Dict		)						->				List[str]:
         """simple docstring"""



         pass


   def 			lowercase_      (self     :						int		)						->				Any:
         """simple docstring"""



         UpperCAmelCase__		,      UpperCAmelCase__								=  self.model_tester.prepare_config_and_inputs_for_common()

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


   def 			lowercase_      (self     :						str		)						->				str:
         """simple docstring"""



         UpperCAmelCase__		,      UpperCAmelCase__								=  self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
               UpperCAmelCase__								=  model_class(__UpperCAmelCase		)
               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]       ,  __UpperCAmelCase		)


   def 			lowercase_      (self     :						Union[str, Any]		)						->				Optional[int]:
         """simple docstring"""



         UpperCAmelCase__								=  self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*__UpperCAmelCase		)





   @slow
   def 			lowercase_      (self     :						Optional[Any]		)						->				Union[str, Any]:
         """simple docstring"""



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




def    lowerCAmelCase_ (     )     ->				Dict:







      '''simple docstring'''


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

      UpperCAmelCase__		,      UpperCAmelCase__								=  torchaudio.load(__A							)

      return audio, sampling_rate







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


   @cached_property
   def 			lowercase_      (self     :						List[Any]		)						->				List[str]:
         """simple docstring"""



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





   @slow
   def 			lowercase_      (self     :						Optional[int]		)						->				Any:
         """simple docstring"""



         UpperCAmelCase__								=  self.default_feature_extractor
         UpperCAmelCase__								=  ASTForAudioClassification.from_pretrained("MIT/ast-finetuned-audioset-10-10-0.4593"		).to(__UpperCAmelCase		)

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

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

         # verify the logits
         UpperCAmelCase__								=  torch.Size((1, 5_2_7)		)
         self.assertEqual(outputs.logits.shape       ,  __UpperCAmelCase		)

         UpperCAmelCase__								=  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 gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


A_						=				re.compile(r'''\s+''')


def 					UpperCAmelCase__       (snake_case__							:					Optional[int]    ):
  """simple docstring"""

  return {"hash": hashlib.mda(re.sub(snake_case__		,   """"""		,   example["""content"""]    ).encode("""utf-8"""    )    ).hexdigest()}


def 					UpperCAmelCase__       (snake_case__							:					Dict    ):
  """simple docstring"""

  _snake_case							:		Any								=  [len(snake_case__    ) for line in example["""content"""].splitlines()]
  return {"line_mean": np.mean(snake_case__    ), "line_max": max(snake_case__    )}


def 					UpperCAmelCase__       (snake_case__							:					List[Any]    ):
  """simple docstring"""

  _snake_case							:		Tuple								=  np.mean([c.isalnum() for c in example["""content"""]]    )
  return {"alpha_frac": alpha_frac}


def 					UpperCAmelCase__       (snake_case__							:					List[Any]		,   snake_case__							:					List[Any]    ):
  """simple docstring"""

  if example["hash"] in uniques:
    uniques.remove(example["""hash"""]    )
    return True
  else:
    return False


def 					UpperCAmelCase__       (snake_case__							:					Optional[Any]		,   snake_case__							:					List[str]=5    ):
  """simple docstring"""

  _snake_case							:		Any								=  ["""auto-generated""", """autogenerated""", """automatically generated"""]
  _snake_case							:		Tuple								=  example["""content"""].splitlines()
  for _, line in zip(range(snake_case__    )		,   snake_case__    ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"autogenerated": True}
  else:
    return {"autogenerated": False}


def 					UpperCAmelCase__       (snake_case__							:					Any		,   snake_case__							:					Union[str, Any]=5		,   snake_case__							:					Any=0.05    ):
  """simple docstring"""

  _snake_case							:		Optional[Any]								=  ["""unit tests""", """test file""", """configuration file"""]
  _snake_case							:		List[Any]								=  example["""content"""].splitlines()
  _snake_case							:		Dict								=  0
  _snake_case							:		str								=  0
  # first test
  for _, line in zip(range(snake_case__    )		,   snake_case__    ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"config_or_test": True}
    # second test
  _snake_case							:		Optional[int]								=  example["""content"""].count("""\n"""    )
  _snake_case							:		Tuple								=  int(coeff * nlines    )
  for line in lines:
    count_config += line.lower().count("""config"""    )
    count_test += line.lower().count("""test"""    )
    if count_config > threshold or count_test > threshold:
      return {"config_or_test": True}
  return {"config_or_test": False}


def 					UpperCAmelCase__       (snake_case__							:					str    ):
  """simple docstring"""

  _snake_case							:		Optional[int]								=  ["""def """, """class """, """for """, """while """]
  _snake_case							:		str								=  example["""content"""].splitlines()
  for line in lines:
    for keyword in keywords:
      if keyword in line.lower():
        return {"has_no_keywords": False}
  return {"has_no_keywords": True}


def 					UpperCAmelCase__       (snake_case__							:					List[str]		,   snake_case__							:					List[str]=4    ):
  """simple docstring"""

  _snake_case							:		List[Any]								=  example["""content"""].splitlines()
  _snake_case							:		str								=  0
  for line in lines:
    counter += line.lower().count("""="""    )
    if counter > minimum:
      return {"has_few_assignments": False}
  return {"has_few_assignments": True}


def 					UpperCAmelCase__       (snake_case__							:					List[str]    ):
  """simple docstring"""

  _snake_case							:		Optional[Any]								=  tokenizer(example["""content"""]		,   truncation=snake_case__    )["""input_ids"""]
  _snake_case							:		Optional[Any]								=  len(example["""content"""]    ) / len(snake_case__    )
  return {"ratio": ratio}


def 					UpperCAmelCase__       (snake_case__							:					Optional[int]    ):
  """simple docstring"""

  _snake_case							:		Optional[int]								=  {}
  results.update(get_hash(snake_case__    )    )
  results.update(line_stats(snake_case__    )    )
  results.update(alpha_stats(snake_case__    )    )
  results.update(char_token_ratio(snake_case__    )    )
  results.update(is_autogenerated(snake_case__    )    )
  results.update(is_config_or_test(snake_case__    )    )
  results.update(has_no_keywords(snake_case__    )    )
  results.update(has_few_assignments(snake_case__    )    )
  return results


def 					UpperCAmelCase__       (snake_case__							:					Tuple		,   snake_case__							:					List[Any]		,   snake_case__							:					List[str]    ):
  """simple docstring"""

  if not check_uniques(snake_case__		,   snake_case__    ):
    return False
  elif example["autogenerated"]:
    return False
  elif example["line_max"] > args.line_max:
    return False
  elif example["line_mean"] > args.line_mean:
    return False
  elif example["alpha_frac"] < args.alpha_frac:
    return False
  elif example["ratio"] < args.min_token_ratio:
    return False
  elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_few_assignments"]:
    return False
  else:
    return True


def 					UpperCAmelCase__       (snake_case__							:					Optional[Any]    ):
  """simple docstring"""

  with open(snake_case__		,   """rb"""    ) as f_in:
    with gzip.open(str(snake_case__    ) + """.gz"""		,   """wb"""		,   compresslevel=6    ) as f_out:
      shutil.copyfileobj(snake_case__		,   snake_case__    )
  os.unlink(snake_case__    )


# Settings
A_						=				HfArgumentParser(PreprocessingArguments)
A_						=				parser.parse_args()
if args.num_workers is None:
     A_						=				multiprocessing.cpu_count()
A_						=				AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
A_						=				time.time()
A_						=				load_dataset(args.dataset_name, split='''train''')
print(F'''Time to load dataset: {time.time()-t_start:.2f}''')

# Run preprocessing
A_						=				time.time()
A_						=				ds.map(preprocess, num_proc=args.num_workers)
print(F'''Time to preprocess dataset: {time.time()-t_start:.2f}''')

# Deduplicate hashes
A_						=				set(ds.unique('''hash'''))
A_						=				len(uniques) / len(ds)
print(F'''Fraction of duplicates: {1-frac:.2%}''')

# Deduplicate data and apply heuristics
A_						=				time.time()
A_						=				ds.filter(filter, fn_kwargs={'''uniques''': uniques, '''args''': args})
print(F'''Time to filter dataset: {time.time()-t_start:.2f}''')
print(F'''Size of filtered dataset: {len(ds_filter)}''')

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
     A_						=				time.time()
     A_	,    A_						=				deduplicate_dataset(ds_filter, args.jaccard_threshold)
     print(F'''Time to deduplicate dataset: {time.time()-t_start:.2f}''')
     print(F'''Size of deduplicate dataset: {len(ds_filter)}''')

# Save data in batches of samples_per_file
A_						=				Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
     with open(output_dir / '''duplicate_clusters.json''', '''w''') as f:
          json.dump(duplicate_clusters, f)

A_						=				output_dir / '''data'''
data_dir.mkdir(exist_ok=True)

A_						=				time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
     A_						=				str(data_dir / F'''file-{file_number+1:012}.json''')
     A_						=				min(len(ds_filter), index + args.samples_per_file)
     ds_filter.select(list(range(index, end_index))).to_json(file_path)
     compress_file(file_path)






print(F'''Time to save dataset: {time.time()-t_start:.2f}''')


