
import collections
import importlib.util
import os
import re
from pathlib import Path


lowercase  :					List[str]													=       'src/transformers'


# Matches is_xxx_available()
lowercase  :					Union[str, Any]													=       re.compile(R'is\_([a-z_]*)_available()')
# Catches a one-line _import_struct = {xxx}
lowercase  :					List[Any]													=       re.compile(R'^_import_structure\s+=\s+\{([^\}]+)\}')
# Catches a line with a key-values pattern: "bla": ["foo", "bar"]
lowercase  :					Optional[Any]													=       re.compile(R'\s+"\S*":\s+\[([^\]]*)\]')
# Catches a line if not is_foo_available
lowercase  :					Tuple													=       re.compile(R'^\s*if\s+not\s+is\_[a-z_]*\_available\(\)')
# Catches a line _import_struct["bla"].append("foo")
lowercase  :					Optional[Any]													=       re.compile(R'^\s*_import_structure\["\S*"\]\.append\("(\S*)"\)')
# Catches a line _import_struct["bla"].extend(["foo", "bar"]) or _import_struct["bla"] = ["foo", "bar"]
lowercase  :					Optional[int]													=       re.compile(R'^\s*_import_structure\[\S*\](?:\.extend\(|\s*=\s+)\[([^\]]*)\]')
# Catches a line with an object between quotes and a comma:     "MyModel",
lowercase  :					int													=       re.compile('^\s+"([^"]+)",')
# Catches a line with objects between brackets only:    ["foo", "bar"],
lowercase  :					Any													=       re.compile('^\s+\[([^\]]+)\]')
# Catches a line with from foo import bar, bla, boo
lowercase  :					Optional[Any]													=       re.compile(R'\s+from\s+\S*\s+import\s+([^\(\s].*)\n')
# Catches a line with try:
lowercase  :					Any													=       re.compile(R'^\s*try:')
# Catches a line with else:
lowercase  :					Optional[Any]													=       re.compile(R'^\s*else:')




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			Union[str, Any])					->    Optional[Any]:






							'''simple docstring'''





							if _re_test_backend.search(_lowerCamelCase) is None:
														return None
							__UpperCamelCase	:   Optional[Any]			   =					[b[0] for b in _re_backend.findall(_lowerCamelCase)]
							backends.sort()
							return "_and_".join(_lowerCamelCase)




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			List[Any])					->    str:






							'''simple docstring'''





							with open(_lowerCamelCase						,  "r"						,  encoding="utf-8"						,  newline="\n") as f:
														__UpperCamelCase	:   Union[str, Any]			   =					f.readlines()

							__UpperCamelCase	:   List[Any]			   =					0
							while line_index < len(_lowerCamelCase) and not lines[line_index].startswith("_import_structure = {"):
														line_index += 1

							# If this is a traditional init, just return.
							if line_index >= len(_lowerCamelCase):
														return None

							# First grab the objects without a specific backend in _import_structure
							__UpperCamelCase	:   Union[str, Any]			   =					[]
							while not lines[line_index].startswith("if TYPE_CHECKING") and find_backend(lines[line_index]) is None:
														__UpperCamelCase	:   int			   =					lines[line_index]
														# If we have everything on a single line, let's deal with it.
														if _re_one_line_import_struct.search(_lowerCamelCase):
																					__UpperCamelCase	:   List[Any]			   =					_re_one_line_import_struct.search(_lowerCamelCase).groups()[0]
																					__UpperCamelCase	:   int			   =					re.findall("\[([^\]]+)\]"						,  _lowerCamelCase)
																					for imp in imports:
																												objects.extend([obj[1:-1] for obj in imp.split(", ")])
																					line_index += 1
																					continue
														__UpperCamelCase	:   Union[str, Any]			   =					_re_import_struct_key_value.search(_lowerCamelCase)
														if single_line_import_search is not None:
																					__UpperCamelCase	:   Optional[Any]			   =					[obj[1:-1] for obj in single_line_import_search.groups()[0].split(", ") if len(_lowerCamelCase) > 0]
																					objects.extend(_lowerCamelCase)
														elif line.startswith(" " * 8 + "\""):
																					objects.append(line[9:-3])
														line_index += 1

							__UpperCamelCase	:   List[str]			   =					{"none": objects}
							# Let's continue with backend-specific objects in _import_structure
							while not lines[line_index].startswith("if TYPE_CHECKING"):
														# If the line is an if not is_backend_available, we grab all objects associated.
														__UpperCamelCase	:   Dict			   =					find_backend(lines[line_index])
														# Check if the backend declaration is inside a try block:
														if _re_try.search(lines[line_index - 1]) is None:
																					__UpperCamelCase	:   Optional[Any]			   =					None

														if backend is not None:
																					line_index += 1

																					# Scroll until we hit the else block of try-except-else
																					while _re_else.search(lines[line_index]) is None:
																												line_index += 1

																					line_index += 1

																					__UpperCamelCase	:   List[Any]			   =					[]
																					# Until we unindent, add backend objects to the list
																					while len(lines[line_index]) <= 1 or lines[line_index].startswith(" " * 4):
																												__UpperCamelCase	:   Dict			   =					lines[line_index]
																												if _re_import_struct_add_one.search(_lowerCamelCase) is not None:
																																			objects.append(_re_import_struct_add_one.search(_lowerCamelCase).groups()[0])
																												elif _re_import_struct_add_many.search(_lowerCamelCase) is not None:
																																			__UpperCamelCase	:   str			   =					_re_import_struct_add_many.search(_lowerCamelCase).groups()[0].split(", ")
																																			__UpperCamelCase	:   List[str]			   =					[obj[1:-1] for obj in imports if len(_lowerCamelCase) > 0]
																																			objects.extend(_lowerCamelCase)
																												elif _re_between_brackets.search(_lowerCamelCase) is not None:
																																			__UpperCamelCase	:   Dict			   =					_re_between_brackets.search(_lowerCamelCase).groups()[0].split(", ")
																																			__UpperCamelCase	:   List[str]			   =					[obj[1:-1] for obj in imports if len(_lowerCamelCase) > 0]
																																			objects.extend(_lowerCamelCase)
																												elif _re_quote_object.search(_lowerCamelCase) is not None:
																																			objects.append(_re_quote_object.search(_lowerCamelCase).groups()[0])
																												elif line.startswith(" " * 8 + "\""):
																																			objects.append(line[9:-3])
																												elif line.startswith(" " * 12 + "\""):
																																			objects.append(line[13:-3])
																												line_index += 1

																					__UpperCamelCase	:   List[str]			   =					objects
														else:
																					line_index += 1

    # At this stage we are in the TYPE_CHECKING part, first grab the objects without a specific backend
							__UpperCamelCase	:   Tuple			   =					[]
							while (
							    line_index < len(_lowerCamelCase)
							    and find_backend(lines[line_index]) is None
							    and not lines[line_index].startswith("else")
							):
														__UpperCamelCase	:   Tuple			   =					lines[line_index]
														__UpperCamelCase	:   Optional[Any]			   =					_re_import.search(_lowerCamelCase)
														if single_line_import_search is not None:
																					objects.extend(single_line_import_search.groups()[0].split(", "))
														elif line.startswith(" " * 8):
																					objects.append(line[8:-2])
														line_index += 1

							__UpperCamelCase	:   Dict			   =					{"none": objects}
							# Let's continue with backend-specific objects
							while line_index < len(_lowerCamelCase):
														# If the line is an if is_backend_available, we grab all objects associated.
														__UpperCamelCase	:   Union[str, Any]			   =					find_backend(lines[line_index])
														# Check if the backend declaration is inside a try block:
														if _re_try.search(lines[line_index - 1]) is None:
																					__UpperCamelCase	:   int			   =					None

														if backend is not None:
																					line_index += 1

																					# Scroll until we hit the else block of try-except-else
																					while _re_else.search(lines[line_index]) is None:
																												line_index += 1

																					line_index += 1

																					__UpperCamelCase	:   List[str]			   =					[]
																					# Until we unindent, add backend objects to the list
																					while len(lines[line_index]) <= 1 or lines[line_index].startswith(" " * 8):
																												__UpperCamelCase	:   List[Any]			   =					lines[line_index]
																												__UpperCamelCase	:   Optional[Any]			   =					_re_import.search(_lowerCamelCase)
																												if single_line_import_search is not None:
																																			objects.extend(single_line_import_search.groups()[0].split(", "))
																												elif line.startswith(" " * 12):
																																			objects.append(line[12:-2])
																												line_index += 1

																					__UpperCamelCase	:   int			   =					objects
														else:
																					line_index += 1

							return import_dict_objects, type_hint_objects




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			Union[str, Any]						,  _lowerCamelCase :			Optional[Any])					->    Union[str, Any]:






							'''simple docstring'''





							def find_duplicates(_lowerCamelCase :			List[Any]):
														return [k for k, v in collections.Counter(_lowerCamelCase).items() if v > 1]

							if list(import_dict_objects.keys()) != list(type_hint_objects.keys()):
														return ["Both sides of the init do not have the same backends!"]

							__UpperCamelCase	:   int			   =					[]
							for key in import_dict_objects.keys():
														__UpperCamelCase	:   List[str]			   =					find_duplicates(import_dict_objects[key])
														if duplicate_imports:
																					errors.append(F'Duplicate _import_structure definitions for: {duplicate_imports}')
														__UpperCamelCase	:   Union[str, Any]			   =					find_duplicates(type_hint_objects[key])
														if duplicate_type_hints:
																					errors.append(F'Duplicate TYPE_CHECKING objects for: {duplicate_type_hints}')

														if sorted(set(import_dict_objects[key])) != sorted(set(type_hint_objects[key])):
																					__UpperCamelCase	:   Dict			   =					"base imports" if key == "none" else F'{key} backend'
																					errors.append(F'Differences for {name}:')
																					for a in type_hint_objects[key]:
																												if a not in import_dict_objects[key]:
																																			errors.append(F'  {a} in TYPE_HINT but not in _import_structure.')
																					for a in import_dict_objects[key]:
																												if a not in type_hint_objects[key]:
																																			errors.append(F'  {a} in _import_structure but not in TYPE_HINT.')
							return errors




def 	_SCREAMING_SNAKE_CASE					(		)					->    Any:






							'''simple docstring'''





							__UpperCamelCase	:   Optional[Any]			   =					[]
							for root, _, files in os.walk(_lowerCamelCase):
														if "__init__.py" in files:
																					__UpperCamelCase	:   Union[str, Any]			   =					os.path.join(_lowerCamelCase						,  "__init__.py")
																					__UpperCamelCase	:   List[str]			   =					parse_init(_lowerCamelCase)
																					if objects is not None:
																												__UpperCamelCase	:   List[Any]			   =					analyze_results(*_lowerCamelCase)
																												if len(_lowerCamelCase) > 0:
																																			__UpperCamelCase	:   int			   =					F'Problem in {fname}, both halves do not define the same objects.\n{errors[0]}'
																																			failures.append("\n".join(_lowerCamelCase))
							if len(_lowerCamelCase) > 0:
														raise ValueError("\n\n".join(_lowerCamelCase))




def 	_SCREAMING_SNAKE_CASE					(		)					->    Optional[Any]:






							'''simple docstring'''





							__UpperCamelCase	:   Tuple			   =					[]
							for path, directories, files in os.walk(_lowerCamelCase):
														for folder in directories:
																					# Ignore private modules
																					if folder.startswith("_"):
																												directories.remove(_lowerCamelCase)
																												continue
																					# Ignore leftovers from branches (empty folders apart from pycache)
																					if len(list((Path(_lowerCamelCase) / folder).glob("*.py"))) == 0:
																												continue
																					__UpperCamelCase	:   int			   =					str((Path(_lowerCamelCase) / folder).relative_to(_lowerCamelCase))
																					__UpperCamelCase	:   int			   =					short_path.replace(os.path.sep						,  ".")
																					submodules.append(_lowerCamelCase)
														for fname in files:
																					if fname == "__init__.py":
																												continue
																					__UpperCamelCase	:   Optional[int]			   =					str((Path(_lowerCamelCase) / fname).relative_to(_lowerCamelCase))
																					__UpperCamelCase	:   List[str]			   =					short_path.replace(".py"						,  "").replace(os.path.sep						,  ".")
																					if len(submodule.split(".")) == 1:
																												submodules.append(_lowerCamelCase)
							return submodules


lowercase  :					List[Any]													=       [
    'convert_pytorch_checkpoint_to_tf2',
    'modeling_flax_pytorch_utils',
]




def 	_SCREAMING_SNAKE_CASE					(		)					->    Optional[Any]:






							'''simple docstring'''





							__UpperCamelCase	:   int			   =					importlib.util.spec_from_file_location(
							    "transformers"						,  os.path.join(_lowerCamelCase						,  "__init__.py")						,  submodule_search_locations=[PATH_TO_TRANSFORMERS]						,  )
							__UpperCamelCase	:   Optional[Any]			   =					spec.loader.load_module()

							__UpperCamelCase	:   Tuple			   =					[
							    module
							    for module in get_transformers_submodules()
							    if module not in IGNORE_SUBMODULES and module not in transformers._import_structure.keys()
							]
							if len(_lowerCamelCase) > 0:
														__UpperCamelCase	:   List[str]			   =					"\n".join(F'- {module}' for module in module_not_registered)
														raise ValueError(
														    "The following submodules are not properly registered in the main init of Transformers:\n"
														    F'{list_of_modules}\n'
														    "Make sure they appear somewhere in the keys of `_import_structure` with an empty list as value.")


if __name__ == "__main__":
		check_all_inits()
		check_submodules()

import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
		from sagemaker import Session, TrainingJobAnalytics
		from sagemaker.huggingface import HuggingFace




@pytest.mark.skipif(
    literal_eval(os.getenv('TEST_SAGEMAKER'     ,						'False')) is not True     ,						reason='Skipping test because should only be run when releasing minor transformers version'     ,						)
@pytest.mark.usefixtures('sm_env')
@parameterized_class(
    [
        {
            'framework': 'pytorch',
            'script': 'run_glue.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.p3.16xlarge',
            'results': {'train_runtime': 6_5_0, 'eval_accuracy': 0.7, 'eval_loss': 0.6},
        },
        {
            'framework': 'pytorch',
            'script': 'run_ddp.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.p3.16xlarge',
            'results': {'train_runtime': 6_0_0, 'eval_accuracy': 0.7, 'eval_loss': 0.6},
        },
        {
            'framework': 'tensorflow',
            'script': 'run_tf_dist.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.p3.16xlarge',
            'results': {'train_runtime': 6_0_0, 'eval_accuracy': 0.6, 'eval_loss': 0.7},
        },
    ])
class  lowerCamelCase__							(				unittest.TestCase):




	'''simple docstring'''




	def    _lowerCamelCase				(					self		:List[Any]						)		-> Any:
								if self.framework == "pytorch":
															subprocess.run(
															    f'cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py'.split()     ,   encoding="utf-8"     ,   check=a     ,   )
								assert hasattr(self     ,   "env"						)







	def    _lowerCamelCase				(					self		:Any     ,   a		:Optional[Any]						)		-> Dict:
								__UpperCamelCase	:   str			   =					f'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
								# distributed data settings
								__UpperCamelCase	:   Optional[int]			   =					{"smdistributed": {"dataparallel": {"enabled": True}}} if self.script != "run_ddp.py" else None

								# creates estimator
								return HuggingFace(
								    entry_point=self.script     ,   source_dir=self.env.test_path     ,   role=self.env.role     ,   image_uri=self.env.image_uri     ,   base_job_name=a     ,   instance_count=a     ,   instance_type=self.instance_type     ,   debugger_hook_config=a     ,   hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}     ,   metric_definitions=self.env.metric_definitions     ,   distribution=a     ,   py_version="py36"     ,   )







	def    _lowerCamelCase				(					self		:Dict     ,   a		:Dict						)		-> Optional[int]:
								TrainingJobAnalytics(a						).export_csv(f'{self.env.test_path}/{job_name}_metrics.csv'						)







	@parameterized.expand([(2,)]						)
	def    _lowerCamelCase				(					self		:Dict     ,   a		:Tuple						)		-> List[Any]:
								# create estimator
								__UpperCamelCase	:   int			   =					self.create_estimator(a						)

								# run training
								estimator.fit()

								# result dataframe
								__UpperCamelCase	:   Optional[int]			   =					TrainingJobAnalytics(estimator.latest_training_job.name						).dataframe()

								# extract kpis
								__UpperCamelCase	:   Optional[Any]			   =					list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]						)
								__UpperCamelCase	:   Tuple			   =					list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]						)
								# get train time from SageMaker job, this includes starting, preprocessing, stopping
								__UpperCamelCase	:   int			   =					(
								    Session().describe_training_job(estimator.latest_training_job.name						).get("TrainingTimeInSeconds"     ,   9_9_9_9_9_9						)
								)

								# assert kpis
								assert train_runtime <= self.results["train_runtime"]
								assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy						)
								assert all(t <= self.results["eval_loss"] for t in eval_loss						)

								# dump tests result into json file to share in PR
								with open(f'{estimator.latest_training_job.name}.json'     ,   "w"						) as outfile:
															json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}     ,   a						)