

'''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


__lowerCamelCase									=     float('''nan''')


class        A__   :



    def __init__(       self		,					UpperCamelCase__							)    ->   Dict:





           '''simple docstring'''
           A_						       =   sys.stdout
           A_						       =   open(UpperCamelCase_		,					"""a"""							)



    def __getattr__(       self		,					UpperCamelCase__							)    ->   Any:





           '''simple docstring'''
           return getattr(self.stdout		,					UpperCamelCase_							)







    def        snake_case_      (       self		,					UpperCamelCase__							)    ->   Dict:





           '''simple docstring'''
           self.stdout.write(UpperCamelCase_							)
           # strip tqdm codes
           self.file.write(re.sub(R"""^.*\r"""		,					""""""		,					UpperCamelCase_		,					0		,					re.M							)							)

def    UpperCAmelCase__     (   UpperCAmelCase__=80,							UpperCAmelCase__=False					)      ->      List[str]:
       A_						       =   []

       # deal with critical env vars
       A_						       =   ['''CUDA_VISIBLE_DEVICES''']
       for key in env_keys:
              A_						       =   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)
       A_						       =   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
       A_						       =   []
       A_						       =   ''''''
       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					)
                     A_						       =   ''''''
       return "\\\n".join(__a					)
def    UpperCAmelCase__     (   UpperCAmelCase__,							UpperCAmelCase__					)      ->      Union[str, Any]:
       A_						       =   re.sub(r"""[\\\n]+""",							""" """,							args.base_cmd					)

       # remove --output_dir if any and set our own
       A_						       =   re.sub("""--output_dir\s+[^\s]+""",							"""""",							args.base_cmd					)
       args.base_cmd += F''' --output_dir {output_dir}'''

       # ensure we have --overwrite_output_dir
       A_						       =   re.sub("""--overwrite_output_dir\s+""",							"""""",							args.base_cmd					)
       args.base_cmd += " --overwrite_output_dir"

       return [sys.executable] + shlex.split(args.base_cmd					)
def    UpperCAmelCase__     (   UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__					)      ->      Tuple:






       if 0:
              import random
              from time import sleep

              sleep(0					)
              return dict(
                  {k: random.uniform(0,							1_00					) for k in metric_keys},							**{target_metric_key: random.choice([nan, 10.31, 1_00.2, 55.66_66, 2_22.22_22_22_22]					)},							)

       A_						       =   subprocess.run(__a,							capture_output=__a,							text=__a					)

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

       # save the streams
       A_						       =   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:
              A_						       =   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    UpperCAmelCase__     (   UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							)      ->      List[str]:
       A_						       =   []
       A_						       =   []
       A_						       =   F'''{id}: {variation:<{longest_variation_len}}'''
       A_						       =   F'''{preamble}: '''
       A_						       =   set(report_metric_keys + [target_metric_key]					)
       for i in tqdm(range(__a					),							desc=__a,							leave=__a					):
              A_						       =   process_run_single(
                  __a,							__a,							__a,							__a,							__a,							__a,							__a					)
              A_						       =   single_run_metrics[target_metric_key]
              if not math.isnan(__a					):
                     metrics.append(__a					)
                     results.append(__a					)
                     outcome += "✓"
              else:
                     outcome += "✘"
       A_						       =   F'''\33[2K\r{outcome}'''
       if len(__a					) > 0:
              A_						       =   {k: fmean([x[k] for x in metrics]					) for k in metrics[0].keys()}
              A_						       =   round(mean_metrics[target_metric_key],							2					)
              A_						       =   F'''{outcome} {mean_target}'''
              if len(__a					) > 1:
                     results_str += F''' {tuple(round(__a,							2					) for x in results					)}'''
              print(__a					)
              A_						       =   variation
              return mean_metrics
       else:
              print(__a					)
              return {variation_key: variation, target_metric_key: nan}
def    UpperCAmelCase__     (   )      ->      Optional[Any]:
       A_						       =   torch.cuda.get_device_properties(torch.device("""cuda"""					)					)
       return F'''
Datetime    : {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"					)}

Software:
transformers: {transformers.__version__}
torch       : {torch.__version__}
cuda        : {torch.version.cuda}
python      : {platform.python_version()}

Hardware:
{torch.cuda.device_count()} GPUs      : {properties.name}, {properties.total_memory/2**30:0.2f}GB
'''
def    UpperCAmelCase__     (   UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__,							UpperCAmelCase__					)      ->      Optional[Any]:
       A_						       =   pd.DataFrame(__a					)
       A_						       =   '''variation'''
       A_						       =   '''diff_%'''

       A_						       =   nan
       if base_variation is not None and len(df[df[variation_key] == base_variation]					):
              # this may still return nan
              A_						       =   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
              A_						       =   df.loc[df[target_metric_key] != nan][target_metric_key].min()

       # create diff column if possible
       if not math.isnan(__a					):
              A_						       =   df.apply(
                  lambda UpperCAmelCase__					: round(1_00 * (r[target_metric_key] - sentinel_value) / sentinel_value					)
                  if not math.isnan(r[target_metric_key]					)
                  else 0,							axis="""columns""",							)

       # re-order columns
       A_						       =   [variation_key, target_metric_key, diff_key, *report_metric_keys]
       A_						       =   df.reindex(__a,							axis="""columns"""					)  # reorder cols

       # capitalize
       A_						       =   df.rename(str.capitalize,							axis="""columns"""					)

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

       A_						       =   ['''''', '''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    UpperCAmelCase__     (   )      ->      Tuple:
       A_						       =   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""",							)
       A_						       =   parser.parse_args()

       A_						       =   args.output_dir
       Path(__a					).mkdir(exist_ok=__a					)
       A_						       =   get_base_command(__a,							__a					)

       # split each dimension into its --foo variations
       A_						       =   [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
       A_						       =   list(map(str.strip,							map(""" """.join,							itertools.product(*__a					)					)					)					)
       A_						       =   max(len(__a					) for x in variations					)

       # split wanted keys
       A_						       =   args.report_metric_keys.split()

       # capture prints into a log file for convenience
       A_						       =   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}'''					)

       A_						       =   Tee(__a					)

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

       A_						       =   '''variation'''
       A_						       =   []
       for id, variation in enumerate(tqdm(__a,							desc="""Total completion: """,							leave=__a					)					):
              A_						       =   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 importlib
import os
import sys


# This is required to make the module import works (when the python process is running from the root of the repo)
sys.path.append('''.''')


def      a (     __a  )    ->      List[str]:



					'''simple docstring'''






					UpperCamelCase__  :int				   =					test_file.split(os.path.sep  )
					if components[0:2] != ["tests", "models"]:
										raise ValueError(
										    '''`test_file` should start with `tests/models/` (with `/` being the OS specific path separator). Got '''
										    f'''{test_file} instead.'''  )
					UpperCamelCase__  :List[Any]				   =					components[-1]
					if not test_fn.endswith('''py'''  ):
										raise ValueError(f'''`test_file` should be a python file. Got {test_fn} instead.'''  )
					if not test_fn.startswith('''test_modeling_'''  ):
										raise ValueError(
										    f'''`test_file` should point to a file name of the form `test_modeling_*.py`. Got {test_fn} instead.'''  )

					UpperCamelCase__  :Optional[Any]				   =					components[:-1] + [test_fn.replace('''.py'''					,   ''''''  )]
					UpperCamelCase__  :str				   =					'''.'''.join(__a  )

					return test_module_path


def      a (     __a  )    ->      Tuple:



					'''simple docstring'''






					UpperCamelCase__  :Dict				   =					get_module_path(__a  )
					UpperCamelCase__  :List[str]				   =					importlib.import_module(__a  )

					return test_module


def      a (     __a  )    ->      int:



					'''simple docstring'''






					UpperCamelCase__  :Dict				   =					[]
					UpperCamelCase__  :Optional[Any]				   =					get_test_module(__a  )
					for attr in dir(__a  ):
										if attr.endswith('''ModelTester'''  ):
															tester_classes.append(getattr(__a					,   __a  )  )

    # sort with class names
					return sorted(__a					,   key=lambda __a  : x.__name__  )


def      a (     __a  )    ->      List[Any]:



					'''simple docstring'''






					UpperCamelCase__  :List[Any]				   =					[]
					UpperCamelCase__  :Dict				   =					get_test_module(__a  )
					for attr in dir(__a  ):
										UpperCamelCase__  :Dict				   =					getattr(__a					,   __a  )
										# (TF/Flax)ModelTesterMixin is also an attribute in specific model test module. Let's exclude them by checking
										# `all_model_classes` is not empty (which also excludes other special classes).
										UpperCamelCase__  :Optional[int]				   =					getattr(__a					,   '''all_model_classes'''					,   []  )
										if len(__a  ) > 0:
															test_classes.append(__a  )

    # sort with class names
					return sorted(__a					,   key=lambda __a  : x.__name__  )


def      a (     __a  )    ->      Union[str, Any]:



					'''simple docstring'''






					UpperCamelCase__  :int				   =					get_test_classes(__a  )
					UpperCamelCase__  :Any				   =					set()
					for test_class in test_classes:
										model_classes.update(test_class.all_model_classes  )

					# sort with class names
					return sorted(__a					,   key=lambda __a  : x.__name__  )


def      a (     __a  )    ->      Dict:



					'''simple docstring'''






					UpperCamelCase__  :Optional[Any]				   =					test_class()
					if hasattr(__a					,   '''setUp'''  ):
										test.setUp()

					UpperCamelCase__  :Union[str, Any]				   =					None
					if hasattr(__a					,   '''model_tester'''  ):
										# `(TF/Flax)ModelTesterMixin` has this attribute default to `None`. Let's skip this case.
										if test.model_tester is not None:
															UpperCamelCase__  :List[str]				   =					test.model_tester.__class__

					return model_tester


def      a (     __a					,   __a  )    ->      Any:



					'''simple docstring'''






					UpperCamelCase__  :str				   =					get_test_classes(__a  )

					UpperCamelCase__  :List[str]				   =					[]
					for test_class in test_classes:
										if model_class in test_class.all_model_classes:
															target_test_classes.append(__a  )

    # sort with class names
					return sorted(__a					,   key=lambda __a  : x.__name__  )


def      a (     __a					,   __a  )    ->      Union[str, Any]:



					'''simple docstring'''






					UpperCamelCase__  :Optional[int]				   =					get_test_classes_for_model(__a					,   __a  )

					UpperCamelCase__  :List[str]				   =					[]
					for test_class in test_classes:
										UpperCamelCase__  :int				   =					get_model_tester_from_test_class(__a  )
										if tester_class is not None:
															tester_classes.append(__a  )

    # sort with class names
					return sorted(__a					,   key=lambda __a  : x.__name__  )


def      a (     __a  )    ->      int:



					'''simple docstring'''






					UpperCamelCase__  :Any				   =					get_test_classes(__a  )
					UpperCamelCase__  :Dict				   =					{test_class: get_model_tester_from_test_class(__a  ) for test_class in test_classes}
					return test_tester_mapping


def      a (     __a  )    ->      List[str]:



					'''simple docstring'''






					UpperCamelCase__  :Any				   =					get_model_classes(__a  )
					UpperCamelCase__  :Tuple				   =					{
					    model_class: get_test_classes_for_model(__a					,   __a  ) for model_class in model_classes
					}
					return model_test_mapping


def      a (     __a  )    ->      Tuple:



					'''simple docstring'''






					UpperCamelCase__  :List[str]				   =					get_model_classes(__a  )
					UpperCamelCase__  :Union[str, Any]				   =					{
					    model_class: get_tester_classes_for_model(__a					,   __a  ) for model_class in model_classes
					}
					return model_to_tester_mapping







def      a (     __a  )    ->      List[Any]:



					'''simple docstring'''






					if isinstance(__a					,   __a  ):
										return o
					elif isinstance(__a					,   __a  ):
										return o.__name__
					elif isinstance(__a					,   (list, tuple)  ):
										return [to_json(__a  ) for x in o]
					elif isinstance(__a					,   __a  ):
										return {to_json(__a  ): to_json(__a  ) for k, v in o.items()}
					else:
										return o