





# 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_    (    _UpperCAmelCase							:List[str]				)	->				List[str]:
     '''simple docstring'''


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

def      UpperCAmelCase_    (    _UpperCAmelCase							:Dict				)	->				List[str]:
     '''simple docstring'''


     A_          =       botoa.client('''iam'''				)
     return iam_client.get_role(RoleName=_UpperCAmelCase				)["Role"]["Arn"]


def      UpperCAmelCase_    (    )	->				List[str]:
     '''simple docstring'''


     A_          =       _ask_options(
         '''How do you want to authorize?'''  ,						['''AWS Profile''', '''Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) ''']  ,						_UpperCAmelCase  ,						)
     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''']  ,						_UpperCAmelCase  ,						)
     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(_UpperCAmelCase				)

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

     A_          =       _ask_field(
         '''Do you want to provide SageMaker input channels with data locations? [yes/NO]: '''  ,						_convert_yes_no_to_bool  ,						default=_UpperCAmelCase  ,						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 _UpperCAmelCase				: str(_UpperCAmelCase				).lower()  ,						)

     A_          =       _ask_field(
         '''Do you want to enable SageMaker metrics? [yes/NO]: '''  ,						_convert_yes_no_to_bool  ,						default=_UpperCAmelCase  ,						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 _UpperCAmelCase				: str(_UpperCAmelCase				).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=_UpperCAmelCase  ,						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=_UpperCAmelCase  ,						error_message='''Please enter yes or no.'''  ,						)

          if use_custom_options:
               A_          =       _ask_options(
                   '''Which mode do you want to use?'''  ,						_UpperCAmelCase  ,						lambda _UpperCAmelCase				: TORCH_DYNAMO_MODES[int(_UpperCAmelCase				)]  ,						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=_UpperCAmelCase  ,						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=_UpperCAmelCase  ,						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(
              _UpperCAmelCase  ,						_UpperCAmelCase  ,						lambda _UpperCAmelCase				: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_UpperCAmelCase				)]				)
     else:
          eca_instance_query += "? [ml.p3.2xlarge]:"
          A_          =       _ask_field(_UpperCAmelCase  ,						lambda _UpperCAmelCase				: str(_UpperCAmelCase				).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]: '''  ,						_UpperCAmelCase  ,						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=_UpperCAmelCase  ,						compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER  ,						distributed_type=_UpperCAmelCase  ,						use_cpu=_UpperCAmelCase  ,						dynamo_config=_UpperCAmelCase  ,						eca_instance_type=_UpperCAmelCase  ,						profile=_UpperCAmelCase  ,						region=_UpperCAmelCase  ,						iam_role_name=_UpperCAmelCase  ,						mixed_precision=_UpperCAmelCase  ,						num_machines=_UpperCAmelCase  ,						sagemaker_inputs_file=_UpperCAmelCase  ,						sagemaker_metrics_file=_UpperCAmelCase  ,						)










# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

a__      :     Optional[int]         =		TypeVar('T')





class UpperCAmelCase_				( Generic[T] ):


  def __init__(							self ,__snake_case = True ):



       """simple docstring"""





       A_          =       {}  # dictionary of lists
       A_          =       directed


  def 						__UpperCAmelCase     (							self ,__snake_case ,__snake_case ):



       """simple docstring"""





       if not self.directed:  # For undirected graphs
            # if both source vertex and destination vertex are both present in the
            # adjacency list, add destination vertex to source vertex list of adjacent
            # vertices and add source vertex to destination vertex list of adjacent
            # vertices.
            if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                 self.adj_list[source_vertex].append(__snake_case )
                 self.adj_list[destination_vertex].append(__snake_case )
            # if only source vertex is present in adjacency list, add destination vertex
            # to source vertex list of adjacent vertices, then create a new vertex with
            # destination vertex as key and assign a list containing the source vertex
            # as it's first adjacent vertex.
            elif source_vertex in self.adj_list:
                 self.adj_list[source_vertex].append(__snake_case )
                 A_          =       [source_vertex]
            # if only destination vertex is present in adjacency list, add source vertex
            # to destination vertex list of adjacent vertices, then create a new vertex
            # with source vertex as key and assign a list containing the source vertex
            # as it's first adjacent vertex.
            elif destination_vertex in self.adj_list:
                 self.adj_list[destination_vertex].append(__snake_case )
                 A_          =       [destination_vertex]
            # if both source vertex and destination vertex are not present in adjacency
            # list, create a new vertex with source vertex as key and assign a list
            # containing the destination vertex as it's first adjacent vertex also
            # create a new vertex with destination vertex as key and assign a list
            # containing the source vertex as it's first adjacent vertex.
            else:
                 A_          =       [destination_vertex]
                 A_          =       [source_vertex]
       else:  # For directed graphs
            # if both source vertex and destination vertex are present in adjacency
            # list, add destination vertex to source vertex list of adjacent vertices.
            if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                 self.adj_list[source_vertex].append(__snake_case )
            # if only source vertex is present in adjacency list, add destination
            # vertex to source vertex list of adjacent vertices and create a new vertex
            # with destination vertex as key, which has no adjacent vertex
            elif source_vertex in self.adj_list:
                 self.adj_list[source_vertex].append(__snake_case )
                 A_          =       []
            # if only destination vertex is present in adjacency list, create a new
            # vertex with source vertex as key and assign a list containing destination
            # vertex as first adjacent vertex
            elif destination_vertex in self.adj_list:
                 A_          =       [destination_vertex]
            # if both source vertex and destination vertex are not present in adjacency
            # list, create a new vertex with source vertex as key and a list containing
            # destination vertex as it's first adjacent vertex. Then create a new vertex
            # with destination vertex as key, which has no adjacent vertex
            else:
                 A_          =       [destination_vertex]
                 A_          =       []

       return self




  def __repr__(							self ):



       """simple docstring"""





       return pformat(self.adj_list )



