






"""simple docstring"""
# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
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 			__UpperCamelCase	(		snake_case__					):
	A_		:						Any	    =   botoa.client("""iam"""					)

	A_		:						List[str]	    =   {
	    "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=_A   ,   AssumeRolePolicyDocument=json.dumps(_A   ,   indent=2					)					)
		A_		:						Tuple	    =   {
		    "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=_A   ,   PolicyName=F"""{role_name}_policy_permission"""   ,   PolicyDocument=json.dumps(_A   ,   indent=2					)   ,   )
	except iam_client.exceptions.EntityAlreadyExistsException:
		print(F"""role {role_name} already exists. Using existing one"""					)






def 			__UpperCamelCase	(		snake_case__					):
	A_		:						List[str]	    =   botoa.client("""iam"""					)
	return iam_client.get_role(RoleName=_A					)["Role"]["Arn"]






def 			__UpperCamelCase	(		):
	A_		:						int	    =   _ask_options(
	    """How do you want to authorize?"""   ,   ["""AWS Profile""", """Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) """]   ,   _A   ,   )
	A_		:						Tuple	    =   None
	if credentials_configuration == 0:
		A_		:						List[str]	    =   _ask_field("""Enter your AWS Profile name: [default] """   ,   default="""default"""					)
		A_		:						Optional[int]	    =   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`"""					)
		A_		:						Optional[int]	    =   _ask_field("""AWS Access Key ID: """					)
		A_		:						List[str]	    =   aws_access_key_id

		A_		:						List[Any]	    =   _ask_field("""AWS Secret Access Key: """					)
		A_		:						Tuple	    =   aws_secret_access_key

	A_		:						Optional[Any]	    =   _ask_field("""Enter your AWS Region: [us-east-1]"""   ,   default="""us-east-1"""					)
	A_		:						Union[str, Any]	    =   aws_region

	A_		:						Optional[Any]	    =   _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"""]   ,   _A   ,   )
	if role_management == 0:
		A_		:						Optional[int]	    =   _ask_field("""Enter your IAM role name: """					)
	else:
		A_		:						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(_A					)

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

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

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

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

		if use_custom_options:
			A_		:						Any	    =   _ask_options(
			    """Which mode do you want to use?"""   ,   _A   ,   lambda snake_case__					: TORCH_DYNAMO_MODES[int(_A					)]   ,   default="""default"""   ,   )
			A_		:						str	    =   _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=_A   ,   error_message="""Please enter yes or no."""   ,   )
			A_		:						Any	    =   _ask_field(
			    """Do you want to enable dynamic shape tracing? [yes/NO]: """   ,   _convert_yes_no_to_bool   ,   default=_A   ,   error_message="""Please enter yes or no."""   ,   )
	A_		:						int	    =   "Which EC2 instance type you want to use for your training?"
	if distributed_type != SageMakerDistributedType.NO:
		A_		:						List[str]	    =   _ask_options(
		    _A   ,   _A   ,   lambda snake_case__					: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_A					)]					)
	else:
		eca_instance_query += "? [ml.p3.2xlarge]:"
		A_		:						Tuple	    =   _ask_field(_A   ,   lambda snake_case__					: str(_A					).lower()   ,   default="""ml.p3.2xlarge"""					)

	A_		:						List[Any]	    =   1
	if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
		A_		:						List[Any]	    =   _ask_field(
		    """How many machines do you want use? [1]: """   ,   _A   ,   default=1   ,   )

	A_		:						Tuple	    =   _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=_A   ,   compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER   ,   distributed_type=_A   ,   use_cpu=_A   ,   dynamo_config=_A   ,   eca_instance_type=_A   ,   profile=_A   ,   region=_A   ,   iam_role_name=_A   ,   mixed_precision=_A   ,   num_machines=_A   ,   sagemaker_inputs_file=_A   ,   sagemaker_metrics_file=_A   ,   )








"""simple docstring"""
import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format="%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s",
    datefmt="%m/%d/%Y %H:%M:%S",
    level=logging.INFO,
)
_lowerCAmelCase         =							logging.getLogger(__name__)






def 			__UpperCamelCase	(		snake_case__					):
	A_		:						List[str]	    =   git.Repo(search_parent_directories=snake_case__					)
	A_		:						List[str]	    =   {
	    """repo_id""": str(snake_case__					),
	    """repo_sha""": str(repo.head.object.hexsha					),
	    """repo_branch""": str(repo.active_branch					),
	}

	with open(os.path.join(snake_case__   ,   """git_log.json"""					)   ,   """w"""					) as f:
		json.dump(snake_case__   ,   snake_case__   ,   indent=4					)






def 			__UpperCamelCase	(		snake_case__					):
	if params.n_gpu <= 0:
		A_		:						Dict	    =   0
		A_		:						str	    =   -1
		A_		:						int	    =   True
		A_		:						Union[str, Any]	    =   False
		return

	assert torch.cuda.is_available()

	logger.info("""Initializing GPUs"""					)
	if params.n_gpu > 1:
		assert params.local_rank != -1

		A_		:						str	    =   int(os.environ["""WORLD_SIZE"""]					)
		A_		:						int	    =   int(os.environ["""N_GPU_NODE"""]					)
		A_		:						int	    =   int(os.environ["""RANK"""]					)

		# number of nodes / node ID
		A_		:						Optional[int]	    =   params.world_size // params.n_gpu_per_node
		A_		:						Optional[int]	    =   params.global_rank // params.n_gpu_per_node
		A_		:						Tuple	    =   True

		assert params.n_nodes == int(os.environ["""N_NODES"""]					)
		assert params.node_id == int(os.environ["""NODE_RANK"""]					)

	# local job (single GPU)
	else:
		assert params.local_rank == -1

		A_		:						Dict	    =   1
		A_		:						Tuple	    =   0
		A_		:						Dict	    =   0
		A_		:						Optional[int]	    =   0
		A_		:						List[str]	    =   1
		A_		:						List[Any]	    =   1
		A_		:						List[Any]	    =   False

	# sanity checks
	assert params.n_nodes >= 1
	assert 0 <= params.node_id < params.n_nodes
	assert 0 <= params.local_rank <= params.global_rank < params.world_size
	assert params.world_size == params.n_nodes * params.n_gpu_per_node

	# define whether this is the master process / if we are in multi-node distributed mode
	A_		:						Optional[int]	    =   params.node_id == 0 and params.local_rank == 0
	A_		:						Optional[Any]	    =   params.n_nodes > 1

	# summary
	A_		:						str	    =   F"""--- Global rank: {params.global_rank} - """
	logger.info(PREFIX + """Number of nodes: %i""" % params.n_nodes					)
	logger.info(PREFIX + """Node ID        : %i""" % params.node_id					)
	logger.info(PREFIX + """Local rank     : %i""" % params.local_rank					)
	logger.info(PREFIX + """World size     : %i""" % params.world_size					)
	logger.info(PREFIX + """GPUs per node  : %i""" % params.n_gpu_per_node					)
	logger.info(PREFIX + """Master         : %s""" % str(params.is_master					)					)
	logger.info(PREFIX + """Multi-node     : %s""" % str(params.multi_node					)					)
	logger.info(PREFIX + """Multi-GPU      : %s""" % str(params.multi_gpu					)					)
	logger.info(PREFIX + """Hostname       : %s""" % socket.gethostname()					)

	# set GPU device
	torch.cuda.set_device(params.local_rank					)

	# initialize multi-GPU
	if params.multi_gpu:
		logger.info("""Initializing PyTorch distributed"""					)
		torch.distributed.init_process_group(
		    init_method="""env://"""   ,   backend="""nccl"""   ,   )






def 			__UpperCamelCase	(		snake_case__					):
	np.random.seed(args.seed					)
	torch.manual_seed(args.seed					)
	if args.n_gpu > 0:
		torch.cuda.manual_seed_all(args.seed					)

