
"""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 					_snake_case (						snake_case__      :   List[Any]   ):
	A        =      botoa.client('iam'   )

	A        =      {
	    '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=snake_case__		,					AssumeRolePolicyDocument=json.dumps(snake_case__		,					indent=2   )   )
		A        =      {
		    '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=snake_case__		,					PolicyName=F'{role_name}_policy_permission'		,					PolicyDocument=json.dumps(snake_case__		,					indent=2   )		,					)
	except iam_client.exceptions.EntityAlreadyExistsException:
		print(F'role {role_name} already exists. Using existing one'   )






def 					_snake_case (						snake_case__      :   int   ):
	A        =      botoa.client('iam'   )
	return iam_client.get_role(RoleName=snake_case__   )["Role"]["Arn"]






def 					_snake_case (						):
	A        =      _ask_options(
	    'How do you want to authorize?'		,					['AWS Profile', 'Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) ']		,					snake_case__		,					)
	A        =      None
	if credentials_configuration == 0:
		A        =      _ask_field('Enter your AWS Profile name: [default] '		,					default='default'   )
		A        =      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        =      _ask_field('AWS Access Key ID: '   )
		A        =      aws_access_key_id

		A        =      _ask_field('AWS Secret Access Key: '   )
		A        =      aws_secret_access_key

	A        =      _ask_field('Enter your AWS Region: [us-east-1]'		,					default='us-east-1'   )
	A        =      aws_region

	A        =      _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']		,					snake_case__		,					)
	if role_management == 0:
		A        =      _ask_field('Enter your IAM role name: '   )
	else:
		A        =      'accelerate_sagemaker_execution_role'
		print(F'Accelerate will create an iam role "{iam_role_name}" using the provided credentials'   )
		_create_iam_role_for_sagemaker(snake_case__   )

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

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

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

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

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

	A        =      1
	if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
		A        =      _ask_field(
		    'How many machines do you want use? [1]: '		,					snake_case__		,					default=1		,					)

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







from pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url
class       _UpperCAmelCase					(       A__	):

				"""simple docstring"""




				lowercase__											=       """"""
				lowercase__											=       """hf-legacy"""  # "hf://"" is reserved for hffs




				def __init__(     self						:    List[str],  lowerCamelCase						:    Optional[DatasetInfo] = None,  lowerCamelCase						:    Optional[str] = None,  **lowerCamelCase						:    str,  ):

								'''simple docstring'''
								super().__init__(self,  **lowerCamelCase  )
								lowercase__    =			repo_info
								lowercase__    =			token
								lowercase__    =			None




				def 				lowercase__   (     self						:    Optional[Any]  ):

								'''simple docstring'''
								if self.dir_cache is None:
												lowercase__    =			{}
												for hf_file in self.repo_info.siblings:
																# TODO(QL): add sizes
																lowercase__    =			{
																    '''name''': hf_file.rfilename,
																    '''size''': None,
																    '''type''': '''file''',
																}
																self.dir_cache.update(
																    {
																        str(lowerCamelCase  ): {'''name''': str(lowerCamelCase  ), '''size''': None, '''type''': '''directory'''}
																        for d in list(PurePosixPath(hf_file.rfilename  ).parents  )[:-1]
																    }  )




				def 				lowercase__   (     self						:    Any,  lowerCamelCase						:    str,  lowerCamelCase						:    str = "rb",  **lowerCamelCase						:    Any,  ):

								'''simple docstring'''
								if not isinstance(self.repo_info,  lowerCamelCase  ):
												raise NotImplementedError(F"""Open is only implemented for dataset repositories, but got {self.repo_info}"""  )
								lowercase__    =			hf_hub_url(self.repo_info.id,  lowerCamelCase,  revision=self.repo_info.sha  )
								return fsspec.open(
								    lowerCamelCase,  mode=lowerCamelCase,  headers=get_authentication_headers_for_url(lowerCamelCase,  use_auth_token=self.token  ),  client_kwargs={'''trust_env''': True},  ).open()




				def 				lowercase__   (     self						:    Dict,  lowerCamelCase						:    Any,  **lowerCamelCase						:    int  ):

								'''simple docstring'''
								self._get_dirs()
								lowercase__    =			self._strip_protocol(lowerCamelCase  )
								if path in self.dir_cache:
												return self.dir_cache[path]
								else:
												raise FileNotFoundError(lowerCamelCase  )




				def 				lowercase__   (     self						:    Optional[Any],  lowerCamelCase						:    List[str],  lowerCamelCase						:    Optional[int]=False,  **lowerCamelCase						:    str  ):

								'''simple docstring'''
								self._get_dirs()
								lowercase__    =			PurePosixPath(path.strip('''/'''  )  )
								lowercase__    =			{}
								for p, f in self.dir_cache.items():
												lowercase__    =			PurePosixPath(p.strip('''/'''  )  )
												lowercase__    =			p.parent
												if root == path:
																lowercase__    =			f
								lowercase__    =			list(paths.values()  )
								if detail:
												return out
								else:
												return sorted(f['''name'''] for f in out  )




