






'''simple docstring'''

import argparse
import datetime
import io
import itertools
import json
import math
import os
import platform
import re
import shlex
import subprocess
import sys
from pathlib import Path
from statistics import fmean

import pandas as pd
import torch
from tqdm import tqdm

import transformers


UpperCamelCase      					=float("nan")




class      A     :






 """simple docstring"""





 def __init__(				self      ,  __lowerCAmelCase			):
       UpperCamelCase_					: Tuple            =				sys.stdout
       UpperCamelCase_					: str            =				open(_A      ,  """a"""			)
 def __getattr__(				self      ,  __lowerCAmelCase			):
       return getattr(self.stdout      ,  _A			)





 def 	_UpperCAmelCase							(				self      ,  __lowerCAmelCase			):
       self.stdout.write(_A			)
       # strip tqdm codes
       self.file.write(re.sub(R"""^.*\r"""      ,  """"""      ,  _A      ,  0      ,  re.M			)			)






def 				snake_case       (  a_				:			Any=80     , a_				:			List[Any]=False						)							->					Optional[Any]:





      """simple docstring"""

      UpperCamelCase_					: List[str]            =				[]

      # deal with critical env vars
      UpperCamelCase_					: int            =				["""CUDA_VISIBLE_DEVICES"""]
      for key in env_keys:
            UpperCamelCase_					: str            =				os.environ.get(a_     , a_						)
            if val is not None:
                  cmd.append(f"{key}={val}"						)

    # python executable (not always needed if the script is executable)
      UpperCamelCase_					: Union[str, Any]            =				sys.executable if full_python_path else sys.executable.split("""/"""						)[-1]
      cmd.append(a_						)

      # now the normal args
      cmd += list(map(shlex.quote     , sys.argv						)						)

      # split up into up to MAX_WIDTH lines with shell multi-line escapes
      UpperCamelCase_					: List[Any]            =				[]
      UpperCamelCase_					: Dict            =				""""""
      while len(a_						) > 0:
            current_line += f"{cmd.pop(0						)} "
            if len(a_						) == 0 or len(a_						) + len(cmd[0]						) + 1 > max_width - 1:
                  lines.append(a_						)
                  UpperCamelCase_					: List[Any]            =				""""""
      return "\\\n".join(a_						)



def 				snake_case       (  a_				:			int     , a_				:			Optional[int]						)							->					int:





      """simple docstring"""

      UpperCamelCase_					: int            =				re.sub(r"""[\\\n]+"""     , """ """     , args.base_cmd						)

      # remove --output_dir if any and set our own
      UpperCamelCase_					: Union[str, Any]            =				re.sub("""--output_dir\s+[^\s]+"""     , """"""     , args.base_cmd						)
      args.base_cmd += f" --output_dir {output_dir}"

      # ensure we have --overwrite_output_dir
      UpperCamelCase_					: Optional[Any]            =				re.sub("""--overwrite_output_dir\s+"""     , """"""     , args.base_cmd						)
      args.base_cmd += " --overwrite_output_dir"

      return [sys.executable] + shlex.split(args.base_cmd						)



def 				snake_case       (  a_				:			List[str]     , a_				:			Union[str, Any]     , a_				:			Optional[Any]     , a_				:			Any     , a_				:			List[str]     , a_				:			List[Any]     , a_				:			Tuple						)							->					str:





      """simple docstring"""

      if 0:
            import random
            from time import sleep

            sleep(0						)
            return dict(
                {k: random.uniform(0     , 100						) for k in metric_keys}     , **{target_metric_key: random.choice([nan, 10.31, 100.2, 55.6_666, 222.22_222_222]						)}     , )

      UpperCamelCase_					: Tuple            =				subprocess.run(a_     , capture_output=a_     , text=a_						)

      if verbose:
            print("""STDOUT"""     , result.stdout						)
            print("""STDERR"""     , result.stderr						)

      # save the streams
      UpperCamelCase_					: List[Any]            =				variation.replace(""" """     , """-"""						)
      with open(Path(a_						) / f"log.{prefix}.stdout.txt"     , """w"""						) as f:
            f.write(result.stdout						)
      with open(Path(a_						) / f"log.{prefix}.stderr.txt"     , """w"""						) as f:
            f.write(result.stderr						)

      if result.returncode != 0:
            if verbose:
                  print("""failed"""						)
            return {target_metric_key: nan}

      with io.open(f"{output_dir}/all_results.json"     , """r"""     , encoding="""utf-8"""						) as f:
            UpperCamelCase_					: Optional[int]            =				json.load(a_						)

      # filter out just the keys we want
      return {k: v for k, v in metrics.items() if k in metric_keys}



def 				snake_case       (  a_				:			List[Any]     , a_				:			Any     , a_				:			List[str]     , a_				:			Union[str, Any]     , a_				:			Optional[Any]     , a_				:			Any     , a_				:			Optional[int]     , a_				:			Union[str, Any]     , a_				:			List[str]     , a_				:			List[str]     , )							->					Dict:





      """simple docstring"""

      UpperCamelCase_					: Tuple            =				[]
      UpperCamelCase_					: List[Any]            =				[]
      UpperCamelCase_					: Tuple            =				f"{id}: {variation:<{longest_variation_len}}"
      UpperCamelCase_					: List[str]            =				f"{preamble}: "
      UpperCamelCase_					: List[Any]            =				set(report_metric_keys + [target_metric_key]						)
      for i in tqdm(range(a_						)     , desc=a_     , leave=a_						):
            UpperCamelCase_					: List[str]            =				process_run_single(
                a_     , a_     , a_     , a_     , a_     , a_     , a_						)
            UpperCamelCase_					: Optional[int]            =				single_run_metrics[target_metric_key]
            if not math.isnan(a_						):
                  metrics.append(a_						)
                  results.append(a_						)
                  outcome += "✓"
            else:
                  outcome += "✘"
      UpperCamelCase_					: Any            =				f"\33[2K\r{outcome}"
      if len(a_						) > 0:
            UpperCamelCase_					: Optional[int]            =				{k: fmean([x[k] for x in metrics]						) for k in metrics[0].keys()}
            UpperCamelCase_					: Any            =				round(mean_metrics[target_metric_key]     , 2						)
            UpperCamelCase_					: Any            =				f"{outcome} {mean_target}"
            if len(a_						) > 1:
                  results_str += f" {tuple(round(a_     , 2						) for x in results						)}"
            print(a_						)
            UpperCamelCase_					: Tuple            =				variation
            return mean_metrics
      else:
            print(a_						)
            return {variation_key: variation, target_metric_key: nan}



def 				snake_case       (  )							->					Any:





      """simple docstring"""

      UpperCamelCase_					: int            =				torch.cuda.get_device_properties(torch.device("""cuda"""						)						)
      return f"\nDatetime    : {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'						)}\n\nSoftware:\ntransformers: {transformers.__version__}\ntorch       : {torch.__version__}\ncuda        : {torch.version.cuda}\npython      : {platform.python_version()}\n\nHardware:\n{torch.cuda.device_count()} GPUs      : {properties.name}, {properties.total_memory/2**30:0.2f}GB\n"



def 				snake_case       (  a_				:			List[str]     , a_				:			Optional[Any]     , a_				:			Optional[int]     , a_				:			int     , a_				:			int						)							->					int:





      """simple docstring"""

      UpperCamelCase_					: List[str]            =				pd.DataFrame(a_						)
      UpperCamelCase_					: str            =				"""variation"""
      UpperCamelCase_					: int            =				"""diff_%"""

      UpperCamelCase_					: List[str]            =				nan
      if base_variation is not None and len(df[df[variation_key] == base_variation]						):
            # this may still return nan
            UpperCamelCase_					: Union[str, Any]            =				df.loc[df[variation_key] == base_variation][target_metric_key].item()
      if math.isnan(a_						):
            # as a fallback, use the minimal value as the sentinel
            UpperCamelCase_					: List[str]            =				df.loc[df[target_metric_key] != nan][target_metric_key].min()

      # create diff column if possible
      if not math.isnan(a_						):
            UpperCamelCase_					: str            =				df.apply(
                lambda a_						: round(100 * (r[target_metric_key] - sentinel_value) / sentinel_value						)
                if not math.isnan(r[target_metric_key]						)
                else 0     , axis="""columns"""     , )

      # re-order columns
      UpperCamelCase_					: Tuple            =				[variation_key, target_metric_key, diff_key, *report_metric_keys]
      UpperCamelCase_					: Dict            =				df.reindex(a_     , axis="""columns"""						)  # reorder cols

      # capitalize
      UpperCamelCase_					: Any            =				df.rename(str.capitalize     , axis="""columns"""						)

      # make the cols as narrow as possible
      UpperCamelCase_					: int            =				df.rename(lambda a_						: c.replace("""_"""     , """<br>"""						)     , axis="""columns"""						)
      UpperCamelCase_					: Tuple            =				df.rename(lambda a_						: c.replace("""_"""     , """\n"""						)     , axis="""columns"""						)

      UpperCamelCase_					: List[str]            =				["""""", """Copy between the cut-here-lines and paste as is to github or a forum"""]
      report += ["----------8<-----------------8<--------"]
      report += ["*** Results:", df_github.to_markdown(index=a_     , floatfmt=""".2f"""						)]
      report += ["```"]
      report += ["*** Setup:", get_versions()]
      report += ["*** The benchmark command line was:", get_original_command()]
      report += ["```"]
      report += ["----------8<-----------------8<--------"]
      report += ["*** Results (console):", df_console.to_markdown(index=a_     , floatfmt=""".2f"""						)]

      print("""\n\n""".join(a_						)						)



def 				snake_case       (  )							->					str:





      """simple docstring"""

      UpperCamelCase_					: Tuple            =				argparse.ArgumentParser()
      parser.add_argument(
          """--base-cmd"""     , default=a_     , type=a_     , required=a_     , help="""Base cmd"""     , )
      parser.add_argument(
          """--variations"""     , default=a_     , type=a_     , nargs="""+"""     , required=a_     , help="""Multi-dimensional variations, example: '|--fp16|--bf16' '|--tf32'"""     , )
      parser.add_argument(
          """--base-variation"""     , default=a_     , type=a_     , help="""Baseline variation to compare to. if None the minimal target value will be used to compare against"""     , )
      parser.add_argument(
          """--target-metric-key"""     , default=a_     , type=a_     , required=a_     , help="""Target metric key in output_dir/all_results.json, e.g., train_samples_per_second"""     , )
      parser.add_argument(
          """--report-metric-keys"""     , default=""""""     , type=a_     , help="""Report metric keys - other metric keys from output_dir/all_results.json to report, e.g., train_loss. Use a single argument e.g., 'train_loss train_samples"""     , )
      parser.add_argument(
          """--repeat-times"""     , default=1     , type=a_     , help="""How many times to re-run each variation - an average will be reported"""     , )
      parser.add_argument(
          """--output_dir"""     , default="""output_benchmark"""     , type=a_     , help="""The output directory where all the benchmark reports will go to and additionally this directory will be used to override --output_dir in the script that is being benchmarked"""     , )
      parser.add_argument(
          """--verbose"""     , default=a_     , action="""store_true"""     , help="""Whether to show the outputs of each run or just the benchmark progress"""     , )
      UpperCamelCase_					: str            =				parser.parse_args()

      UpperCamelCase_					: Union[str, Any]            =				args.output_dir
      Path(a_						).mkdir(exist_ok=a_						)
      UpperCamelCase_					: Dict            =				get_base_command(a_     , a_						)

      # split each dimension into its --foo variations
      UpperCamelCase_					: Union[str, Any]            =				[list(map(str.strip     , re.split(r"""\|"""     , a_						)						)						) for x in args.variations]
      # build a cartesian product of dimensions and convert those back into cmd-line arg strings,
      # while stripping white space for inputs that were empty
      UpperCamelCase_					: int            =				list(map(str.strip     , map(""" """.join     , itertools.product(*a_						)						)						)						)
      UpperCamelCase_					: int            =				max(len(a_						) for x in variations						)

      # split wanted keys
      UpperCamelCase_					: Union[str, Any]            =				args.report_metric_keys.split()

      # capture prints into a log file for convenience
      UpperCamelCase_					: List[str]            =				f"benchmark-report-{datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S'						)}.txt"
      print(f"\nNote: each run's output is also logged under {output_dir}/log.*.std*.txt"						)
      print(f"and this script's output is also piped into {report_fn}"						)

      UpperCamelCase_					: Union[str, Any]            =				Tee(a_						)

      print(f"\n*** Running {len(a_						)} benchmarks:"						)
      print(f"Base command: {' '.join(a_						)}"						)

      UpperCamelCase_					: Optional[Any]            =				"""variation"""
      UpperCamelCase_					: Any            =				[]
      for id, variation in enumerate(tqdm(a_     , desc="""Total completion: """     , leave=a_						)						):
            UpperCamelCase_					: Optional[int]            =				base_cmd + variation.split()
            results.append(
                process_run(
                    id + 1     , a_     , a_     , a_     , a_     , args.target_metric_key     , a_     , args.repeat_times     , a_     , args.verbose     , )						)

      process_results(a_     , args.target_metric_key     , a_     , args.base_variation     , a_						)


if __name__ == "__main__":
      main()













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

			"""simple docstring"""







			def __init__(      self						,   _A						,   _A=1_3						,   _A=2						,   _A=2_4						,   _A=1_6						,   _A=True						,   _A=True						,   _A=3_2						,   _A=5						,   _A=4						,   _A=3_7						,   _A="gelu"						,   _A=0.1						,   _A=0.1						,   _A=1_0						,   _A=0.02						,   _A=None						,   _A=2						,   _A=2						,   ):






									'''simple docstring'''







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

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







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		Optional[int]				   =			floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]    )

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

									UpperCamelCase				:		Dict				   =			self.get_config()

									return config, input_values, labels







			def   _a      (      self    ):






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







			def   _a      (      self						,   _A						,   _A						,   _A    ):






									'''simple docstring'''







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







			def   _a      (      self    ):






									'''simple docstring'''







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






@require_torch
class 							lowercase__			(   __SCREAMING_SNAKE_CASE					,     __SCREAMING_SNAKE_CASE					,     unittest.TestCase	):

			"""simple docstring"""
			__lowerCAmelCase     :				List[Any]										=  (
			    (
			        ASTModel,
			        ASTForAudioClassification,
			    )
			    if is_torch_available()
			    else ()
			)
			__lowerCAmelCase     :				Any										=  (
			    {"""audio-classification""": ASTForAudioClassification, """feature-extraction""": ASTModel}
			    if is_torch_available()
			    else {}
			)
			__lowerCAmelCase     :				Union[str, Any]										=  False
			__lowerCAmelCase     :				Any										=  False
			__lowerCAmelCase     :				List[Any]										=  False
			__lowerCAmelCase     :				Union[str, Any]										=  False







			def   _a      (      self						,   _A						,   _A						,   _A						,   _A						,   _A    ):






									'''simple docstring'''







									if pipeline_test_casse_name == "AudioClassificationPipelineTests":
															return True

									return False







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		str				   =			ASTModelTester(self    )
									UpperCamelCase				:		int				   =			ConfigTester(self						,   config_class=_A						,   has_text_modality=_A						,   hidden_size=3_7    )







			def   _a      (      self    ):






									'''simple docstring'''







									self.config_tester.run_common_tests()







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






									'''simple docstring'''







									pass







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase   ,						UpperCamelCase				:		int				   =			self.model_tester.prepare_config_and_inputs_for_common()

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







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase   ,						UpperCamelCase				:		Union[str, Any]				   =			self.model_tester.prepare_config_and_inputs_for_common()

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

															UpperCamelCase				:		Any				   =			["""input_values"""]
															self.assertListEqual(arg_names[:1]						,   _A    )







			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		int				   =			self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*_A    )







			@slow
			def   _a      (      self    ):






									'''simple docstring'''







									for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
															UpperCamelCase				:		int				   =			ASTModel.from_pretrained(_A    )
															self.assertIsNotNone(_A    )


def      UpperCamelCase							():
						UpperCamelCase				:		Optional[Any]				   =			hf_hub_download(
						    repo_id="""nielsr/audio-spectogram-transformer-checkpoint"""							,     filename="""sample_audio.flac"""							,     repo_type="""dataset"""     )

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

						return audio, sampling_rate






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

			"""simple docstring"""







			@cached_property
			def   _a      (      self    ):






									'''simple docstring'''







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







			@slow
			def   _a      (      self    ):






									'''simple docstring'''







									UpperCamelCase				:		Union[str, Any]				   =			self.default_feature_extractor
									UpperCamelCase				:		Any				   =			ASTForAudioClassification.from_pretrained("""MIT/ast-finetuned-audioset-10-10-0.4593"""    ).to(_A    )

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

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

									# verify the logits
									UpperCamelCase				:		List[Any]				   =			torch.Size((1, 5_2_7)    )
									self.assertEqual(outputs.logits.shape						,   _A    )

									UpperCamelCase				:		Optional[Any]				   =			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    )    )



