


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

import pytest
from parameterized import 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.g4dn.xlarge""",
            """results""": {"""train_runtime""": 650, """eval_accuracy""": 0.6, """eval_loss""": 0.9},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.g4dn.xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.3, """eval_loss""": 0.9},
        },
    ]							)
class 				snake_case__(							unittest.TestCase							):







						'''simple docstring'''




						def 						lowercase_ (				self							)					->    List[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=__lowercase   ,						)
										assert hasattr(self   ,						'''env'''							)




						def 						lowercase_ (				self   ,						__lowercase=1							)					->    Dict:
										# 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=f"""{self.env.base_job_name}-single"""   ,						instance_count=__lowercase   ,						instance_type=self.instance_type   ,						debugger_hook_config=__lowercase   ,						hyperparameters={**self.env.hyperparameters, '''model_name_or_path''': self.model_name_or_path}   ,						metric_definitions=self.env.metric_definitions   ,						py_version='''py36'''   ,						)




						def 						lowercase_ (				self   ,						__lowercase							)					->    List[str]:
										TrainingJobAnalytics(__lowercase							).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""							)




						def 						lowercase_ (				self							)					->    int:
										# create estimator
										lowerCAmelCase_ :      Optional[Any]    				=				self.create_estimator()

										# run training
										estimator.fit()

										# result dataframe
										lowerCAmelCase_ :      Any    				=				TrainingJobAnalytics(estimator.latest_training_job.name							).dataframe()

										# extract kpis
										lowerCAmelCase_ :      List[str]    				=				list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']							)
										lowerCAmelCase_ :      Optional[int]    				=				list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']							)
										# get train time from SageMaker job, this includes starting, preprocessing, stopping
										lowerCAmelCase_ :      str    				=				(
										    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}   ,						__lowercase							)



import inspect
import re

from transformers.utils import direct_transformers_import


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_config_docstrings.py
_UpperCAmelCase			:  Any     ="""src/transformers"""


# This is to make sure the transformers module imported is the one in the repo.
_UpperCAmelCase			:  Optional[Any]     =direct_transformers_import(PATH_TO_TRANSFORMERS)

_UpperCAmelCase			:  List[str]     =transformers.models.auto.configuration_auto.CONFIG_MAPPING

# Regex pattern used to find the checkpoint mentioned in the docstring of `config_class`.
# For example, `[bert-base-uncased](https://huggingface.co/bert-base-uncased)`
_UpperCAmelCase			:  Dict     =re.compile(R"""\[(.+?)\]\((https://huggingface\.co/.+?)\)""")


_UpperCAmelCase			:  Any     ={
    """DecisionTransformerConfig""",
    """EncoderDecoderConfig""",
    """MusicgenConfig""",
    """RagConfig""",
    """SpeechEncoderDecoderConfig""",
    """TimmBackboneConfig""",
    """VisionEncoderDecoderConfig""",
    """VisionTextDualEncoderConfig""",
    """LlamaConfig""",
}



def 			lowerCAmelCase (		lowerCAmelCase_ )-> str:
				lowerCAmelCase_ :      Any    				=				None

				# source code of `config_class`
				lowerCAmelCase_ :      Optional[int]    				=				inspect.getsource(lowerCAmelCase_ )
				lowerCAmelCase_ :      str    				=				_re_checkpoint.findall(lowerCAmelCase_ )

				# Each `checkpoint` is a tuple of a checkpoint name and a checkpoint link.
				# For example, `('bert-base-uncased', 'https://huggingface.co/bert-base-uncased')`
				for ckpt_name, ckpt_link in checkpoints:
								# allow the link to end with `/`
								if ckpt_link.endswith('''/''' ):
												lowerCAmelCase_ :      Optional[Any]    				=				ckpt_link[:-1]

								# verify the checkpoint name corresponds to the checkpoint link
								lowerCAmelCase_ :      Tuple    				=				f"""https://huggingface.co/{ckpt_name}"""
								if ckpt_link == ckpt_link_from_name:
												lowerCAmelCase_ :      List[str]    				=				ckpt_name
												break

				return checkpoint



def 			lowerCAmelCase (		)-> Optional[Any]:
				lowerCAmelCase_ :      Tuple    				=				[]

				for config_class in list(CONFIG_MAPPING.values() ):
								# Skip deprecated models
								if "models.deprecated" in config_class.__module__:
												continue
								lowerCAmelCase_ :      int    				=				get_checkpoint_from_config_class(lowerCAmelCase_ )

								lowerCAmelCase_ :      Tuple    				=				config_class.__name__
								if checkpoint is None and name not in CONFIG_CLASSES_TO_IGNORE_FOR_DOCSTRING_CHECKPOINT_CHECK:
												configs_without_checkpoint.append(lowerCAmelCase_ )

				if len(lowerCAmelCase_ ) > 0:
								lowerCAmelCase_ :      List[Any]    				=				'''\n'''.join(sorted(lowerCAmelCase_ ) )
								raise ValueError(f"""The following configurations don't contain any valid checkpoint:\n{message}""" )


if __name__ == "__main__":
						check_config_docstrings_have_checkpoints()