
import json
import os

from ...utils.constants import SAGEMAKER_PARALLEL_EC2_INSTANCES, TORCH_DYNAMO_MODES
from ...utils.dataclasses import ComputeEnvironment, SageMakerDistributedType
from ...utils.imports import is_botoa_available
from .config_args import SageMakerConfig
from .config_utils import (
    DYNAMO_BACKENDS,
    _ask_field,
    _ask_options,
    _convert_dynamo_backend,
    _convert_mixed_precision,
    _convert_sagemaker_distributed_mode,
    _convert_yes_no_to_bool,
)


if is_botoa_available():
		import botoa  # noqa: F401




def       _SCREAMING_SNAKE_CASE   (						_lowerCamelCase					:     Any)   ->       str:






							'''simple docstring'''





							__UpperCamelCase      :  int       =   botoa.client("iam")

							__UpperCamelCase      :  Optional[Any]       =   {
							    "Version": "2012-10-17",
							    "Statement": [
							        {"Effect": "Allow", "Principal": {"Service": "sagemaker.amazonaws.com"}, "Action": "sts:AssumeRole"}
							    ],
							}
							try:
														# create the role, associated with the chosen trust policy
														iam_client.create_role(
														    RoleName=_lowerCamelCase		,    AssumeRolePolicyDocument=json.dumps(_lowerCamelCase		,    indent=2))
														__UpperCamelCase      :  str       =   {
														    "Version": "2012-10-17",
														    "Statement": [
														        {
														            "Effect": "Allow",
														            "Action": [
														                "sagemaker:*",
														                "ecr:GetDownloadUrlForLayer",
														                "ecr:BatchGetImage",
														                "ecr:BatchCheckLayerAvailability",
														                "ecr:GetAuthorizationToken",
														                "cloudwatch:PutMetricData",
														                "cloudwatch:GetMetricData",
														                "cloudwatch:GetMetricStatistics",
														                "cloudwatch:ListMetrics",
														                "logs:CreateLogGroup",
														                "logs:CreateLogStream",
														                "logs:DescribeLogStreams",
														                "logs:PutLogEvents",
														                "logs:GetLogEvents",
														                "s3:CreateBucket",
														                "s3:ListBucket",
														                "s3:GetBucketLocation",
														                "s3:GetObject",
														                "s3:PutObject",
														            ],
														            "Resource": "*",
														        }
														    ],
														}
														# attach policy to role
														iam_client.put_role_policy(
														    RoleName=_lowerCamelCase		,    PolicyName=F'{role_name}_policy_permission'		,    PolicyDocument=json.dumps(_lowerCamelCase		,    indent=2)		,    )
							except iam_client.exceptions.EntityAlreadyExistsException:
														print(F'role {role_name} already exists. Using existing one')




def       _SCREAMING_SNAKE_CASE   (						_lowerCamelCase					:     int)   ->       Optional[int]:






							'''simple docstring'''





							__UpperCamelCase      :  str       =   botoa.client("iam")
							return iam_client.get_role(RoleName=_lowerCamelCase)["Role"]["Arn"]




def       _SCREAMING_SNAKE_CASE   (						)   ->       Any:






							'''simple docstring'''





							__UpperCamelCase      :  Tuple       =   _ask_options(
							    "How do you want to authorize?"		,    ["AWS Profile", "Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) "]		,    _lowerCamelCase		,    )
							__UpperCamelCase      :  str       =   None
							if credentials_configuration == 0:
														__UpperCamelCase      :  int       =   _ask_field("Enter your AWS Profile name: [default] "		,    default="default")
														__UpperCamelCase      :  Union[str, Any]       =   aws_profile
							else:
														print(
														    "Note you will need to provide AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY when you launch you training script with,"
														    "`accelerate launch --aws_access_key_id XXX --aws_secret_access_key YYY`")
														__UpperCamelCase      :  int       =   _ask_field("AWS Access Key ID: ")
														__UpperCamelCase      :  Dict       =   aws_access_key_id

														__UpperCamelCase      :  Any       =   _ask_field("AWS Secret Access Key: ")
														__UpperCamelCase      :  int       =   aws_secret_access_key

							__UpperCamelCase      :  List[Any]       =   _ask_field("Enter your AWS Region: [us-east-1]"		,    default="us-east-1")
							__UpperCamelCase      :  Any       =   aws_region

							__UpperCamelCase      :  Tuple       =   _ask_options(
							    "Do you already have an IAM Role for executing Amazon SageMaker Training Jobs?"		,    ["Provide IAM Role name", "Create new IAM role using credentials"]		,    _lowerCamelCase		,    )
							if role_management == 0:
														__UpperCamelCase      :  int       =   _ask_field("Enter your IAM role name: ")
							else:
														__UpperCamelCase      :  Union[str, Any]       =   "accelerate_sagemaker_execution_role"
														print(F'Accelerate will create an iam role \"{iam_role_name}\" using the provided credentials')
														_create_iam_role_for_sagemaker(_lowerCamelCase)

							__UpperCamelCase      :  List[Any]       =   _ask_field(
							    "Do you want to use custom Docker image? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
							__UpperCamelCase      :  Optional[Any]       =   None
							if is_custom_docker_image:
														__UpperCamelCase      :  str       =   _ask_field("Enter your Docker image: "		,    lambda _lowerCamelCase: str(_lowerCamelCase).lower())

							__UpperCamelCase      :  List[Any]       =   _ask_field(
							    "Do you want to provide SageMaker input channels with data locations? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
							__UpperCamelCase      :  Union[str, Any]       =   None
							if is_sagemaker_inputs_enabled:
														__UpperCamelCase      :  str       =   _ask_field(
														    "Enter the path to the SageMaker inputs TSV file with columns (channel_name, data_location): "		,    lambda _lowerCamelCase: str(_lowerCamelCase).lower()		,    )

							__UpperCamelCase      :  Optional[int]       =   _ask_field(
							    "Do you want to enable SageMaker metrics? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
							__UpperCamelCase      :  Dict       =   None
							if is_sagemaker_metrics_enabled:
														__UpperCamelCase      :  Any       =   _ask_field(
														    "Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): "		,    lambda _lowerCamelCase: str(_lowerCamelCase).lower()		,    )

							__UpperCamelCase      :  Optional[Any]       =   _ask_options(
							    "What is the distributed mode?"		,    ["No distributed training", "Data parallelism"]		,    _convert_sagemaker_distributed_mode		,    )
							__UpperCamelCase      :  List[str]       =   {}
							__UpperCamelCase      :  Any       =   _ask_field(
							    "Do you wish to optimize your script with torch dynamo?[yes/NO]:"		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
							if use_dynamo:
														__UpperCamelCase      :  List[Any]       =   "dynamo_"
														__UpperCamelCase      :  List[str]       =   _ask_options(
														    "Which dynamo backend would you like to use?"		,    [x.lower() for x in DYNAMO_BACKENDS]		,    _convert_dynamo_backend		,    default=2		,    )
														__UpperCamelCase      :  int       =   _ask_field(
														    "Do you want to customize the defaults sent to torch.compile? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )

														if use_custom_options:
																					__UpperCamelCase      :  str       =   _ask_options(
																					    "Which mode do you want to use?"		,    _lowerCamelCase		,    lambda _lowerCamelCase: TORCH_DYNAMO_MODES[int(_lowerCamelCase)]		,    default="default"		,    )
																					__UpperCamelCase      :  Tuple       =   _ask_field(
																					    "Do you want the fullgraph mode or it is ok to break model into several subgraphs? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
																					__UpperCamelCase      :  Dict       =   _ask_field(
																					    "Do you want to enable dynamic shape tracing? [yes/NO]: "		,    _convert_yes_no_to_bool		,    default=_lowerCamelCase		,    error_message="Please enter yes or no."		,    )
							__UpperCamelCase      :  Optional[int]       =   "Which EC2 instance type you want to use for your training?"
							if distributed_type != SageMakerDistributedType.NO:
														__UpperCamelCase      :  Optional[Any]       =   _ask_options(
														    _lowerCamelCase		,    _lowerCamelCase		,    lambda _lowerCamelCase: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_lowerCamelCase)])
							else:
														eca_instance_query += "? [ml.p3.2xlarge]:"
														__UpperCamelCase      :  List[Any]       =   _ask_field(_lowerCamelCase		,    lambda _lowerCamelCase: str(_lowerCamelCase).lower()		,    default="ml.p3.2xlarge")

							__UpperCamelCase      :  int       =   1
							if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
														__UpperCamelCase      :  Union[str, Any]       =   _ask_field(
														    "How many machines do you want use? [1]: "		,    _lowerCamelCase		,    default=1		,    )

							__UpperCamelCase      :  Union[str, Any]       =   _ask_options(
							    "Do you wish to use FP16 or BF16 (mixed precision)?"		,    ["no", "fp16", "bf16", "fp8"]		,    _convert_mixed_precision		,    )

							if use_dynamo and mixed_precision == "no":
														print(
														    "Torch dynamo used without mixed precision requires TF32 to be efficient. Accelerate will enable it by default when launching your scripts.")

							return SageMakerConfig(
							    image_uri=_lowerCamelCase		,    compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER		,    distributed_type=_lowerCamelCase		,    use_cpu=_lowerCamelCase		,    dynamo_config=_lowerCamelCase		,    eca_instance_type=_lowerCamelCase		,    profile=_lowerCamelCase		,    region=_lowerCamelCase		,    iam_role_name=_lowerCamelCase		,    mixed_precision=_lowerCamelCase		,    num_machines=_lowerCamelCase		,    sagemaker_inputs_file=_lowerCamelCase		,    sagemaker_metrics_file=_lowerCamelCase		,    )

"""simple docstring"""






import tempfile
import unittest

from transformers import AutoModelForSeqaSeqLM, AutoTokenizer
from transformers.testing_utils import (
    is_torch_available,
    require_optimum,
    require_torch,
    slow,
)


if is_torch_available():
					import torch



@require_torch
@require_optimum
@slow
class 			lowercase					( unittest.TestCase		):






			def 					UpperCAmelCase							(self  :	Tuple			)       ->	Optional[int]:





				"""simple docstring"""


				lowerCAmelCase												=  '''hf-internal-testing/tiny-random-t5'''
				lowerCAmelCase												=  AutoTokenizer.from_pretrained(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  AutoModelForSeqaSeqLM.from_pretrained(SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  tokenizer('''This is me'''      ,return_tensors='''pt'''			)

				lowerCAmelCase												=  model.to_bettertransformer()

				self.assertTrue(any('''BetterTransformer''' in mod.__class__.__name__ for _, mod in model.named_modules()			)			)

				lowerCAmelCase												=  model.generate(**SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  model.reverse_bettertransformer()

				self.assertFalse(any('''BetterTransformer''' in mod.__class__.__name__ for _, mod in model.named_modules()			)			)

				with tempfile.TemporaryDirectory() as tmpdirname:
					model.save_pretrained(SCREAMING_SNAKE_CASE_			)

					lowerCAmelCase												=  AutoModelForSeqaSeqLM.from_pretrained(SCREAMING_SNAKE_CASE_			)

					self.assertFalse(
					    any('''BetterTransformer''' in mod.__class__.__name__ for _, mod in model_reloaded.named_modules()			)			)

					lowerCAmelCase												=  model_reloaded.generate(**SCREAMING_SNAKE_CASE_			)
					self.assertTrue(torch.allclose(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)






			def 					UpperCAmelCase							(self  :	str			)       ->	Tuple:





				"""simple docstring"""


				lowerCAmelCase												=  '''hf-internal-testing/tiny-random-t5'''
				lowerCAmelCase												=  AutoModelForSeqaSeqLM.from_pretrained(SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  model.to_bettertransformer()

				with tempfile.TemporaryDirectory() as tmpdirname:
					with self.assertRaises(SCREAMING_SNAKE_CASE_			):
						model.save_pretrained(SCREAMING_SNAKE_CASE_			)

					lowerCAmelCase												=  model.reverse_bettertransformer()
					model.save_pretrained(SCREAMING_SNAKE_CASE_			)


