





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": 650, "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": 600, "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": 600, "eval_accuracy": 0.6, "eval_loss": 0.7},
        },
    ]     )
class 			lowercase				(							unittest.TestCase     ):



						"""simple docstring"""

						def 		__UpperCAmelCase       (    self			:		Optional[Any]					):





										'''simple docstring'''







										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=lowerCAmelCase_ ,				)
										assert hasattr(self ,				'env'					)







						def 		__UpperCAmelCase       (    self			:		Tuple ,				lowerCamelCase_			:		Optional[Any]					):





										'''simple docstring'''







										_snake_case :			Optional[Any]  						=   f'''{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}'''
										# distributed data settings
										_snake_case :			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=lowerCAmelCase_ ,				instance_count=lowerCAmelCase_ ,				instance_type=self.instance_type ,				debugger_hook_config=lowerCAmelCase_ ,				hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path} ,				metric_definitions=self.env.metric_definitions ,				distribution=lowerCAmelCase_ ,				py_version='py36' ,				)







						def 		__UpperCAmelCase       (    self			:		List[str] ,				lowerCamelCase_			:		Union[str, Any]					):





										'''simple docstring'''







										TrainingJobAnalytics(lowerCAmelCase_					).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''					)







						@parameterized.expand([(2,)]					)
						def 		__UpperCAmelCase       (    self			:		List[Any] ,				lowerCamelCase_			:		int					):





										'''simple docstring'''







										_snake_case :			str  						=   self.create_estimator(lowerCAmelCase_					)

										# run training
										estimator.fit()

										# result dataframe
										_snake_case :			Tuple  						=   TrainingJobAnalytics(estimator.latest_training_job.name					).dataframe()

										# extract kpis
										_snake_case :			Any  						=   list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']					)
										_snake_case :			str  						=   list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']					)
										# get train time from SageMaker job, this includes starting, preprocessing, stopping
										_snake_case :			List[str]  						=   (
										    Session().describe_training_job(estimator.latest_training_job.name					).get('TrainingTimeInSeconds' ,				99_99_99					)
										)

										# 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} ,				lowerCAmelCase_					)












import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
lowercase_		:				Optional[int]				=						object()

# For specifying empty leaf dict `{}`
lowercase_		:				List[Any]				=						object()

def       A__(		__lowerCAmelCase	,			__lowerCAmelCase ):
				_snake_case :			Optional[Any]  						=   tuple((re.compile(x + '$' ) for x in qs) )
				for i in range(len(__lowerCAmelCase ) - len(__lowerCAmelCase ) + 1 ):
								_snake_case :			Tuple  						=   [x.match(__lowerCAmelCase ) for x, y in zip(__lowerCAmelCase	,			ks[i:] )]
								if matches and all(__lowerCAmelCase ):
												return True
				return False

def       A__(		__lowerCAmelCase ):
				def replace(__lowerCAmelCase	,			__lowerCAmelCase ):
								for rule, replacement in rules:
												if _match(__lowerCAmelCase	,			__lowerCAmelCase ):
																return replacement
								return val

				return replace

def       A__(		):
				return [
				    # embeddings
				    (("transformer", "wpe", "embedding"), P('mp'	,			__lowerCAmelCase )),
				    (("transformer", "wte", "embedding"), P('mp'	,			__lowerCAmelCase )),
				    # atention
				    (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(__lowerCAmelCase	,			'mp' )),
				    (("attention", "out_proj", "kernel"), P('mp'	,			__lowerCAmelCase )),
				    (("attention", "out_proj", "bias"), None),
				    # mlp
				    (("mlp", "c_fc", "kernel"), P(__lowerCAmelCase	,			'mp' )),
				    (("mlp", "c_fc", "bias"), P('mp' )),
				    (("mlp", "c_proj", "kernel"), P('mp'	,			__lowerCAmelCase )),
				    (("mlp", "c_proj", "bias"), None),
				    # layer norms
				    ((r"ln_\d+", "bias"), None),
				    ((r"\d+", r"ln_\d+", "scale"), None),
				    (("ln_f", "bias"), None),
				    (("ln_f", "scale"), None),
				]

def       A__(		__lowerCAmelCase ):
				_snake_case :			Optional[Any]  						=   _get_partition_rules()
				_snake_case :			Optional[Any]  						=   _replacement_rules(__lowerCAmelCase )
				_snake_case :			str  						=   {k: _unmatched for k in flatten_dict(__lowerCAmelCase )}
				_snake_case :			str  						=   {k: replace(__lowerCAmelCase	,			__lowerCAmelCase ) for k, v in initd.items()}
				assert _unmatched not in result.values(), "Incomplete partition spec."
				return freeze(unflatten_dict(__lowerCAmelCase ) )






