


# 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   _lowercase(					__a				:						List[str]     ):
							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   _lowercase(					__a				:						Dict     ):
							a__      =botoa.client('iam'     )
							return iam_client.get_role(RoleName=_UpperCAmelCase     )["Role"]["Arn"]







def   _lowercase(					):
							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 __a     : 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 __a     : 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 __a     : 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 __a     : 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 __a     : SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_UpperCAmelCase     )]     )
							else:
														eca_instance_query += "? [ml.p3.2xlarge]:"
														a__      =_ask_field(_UpperCAmelCase    ,     lambda __a     : 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    ,     )








import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      a__      :     Optional[Any]         =		None

try:
      import msvcrt
except ImportError:
      a__      :     List[str]         =		None

try:
      import fcntl
except ImportError:
      a__      :     Optional[int]         =		None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      a__      :     List[str]         =		OSError


# Data
# ------------------------------------------------
a__      :     Union[str, Any]         =		[
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

a__      :     str         =		'3.0.12'


a__      :     Optional[int]         =		None

def      UpperCAmelCase_    (    )	->				Tuple:
     '''simple docstring'''


     global _logger
     A_          =       _logger or logging.getLogger(__name__				)
     return _logger





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock_file
       return None




  def __str__(							self ):



       """simple docstring"""





       A_          =       f'The file lock \'{self.lock_file}\' could not be acquired.'
       return temp





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock
       return None


  def __enter__(							self ):



       """simple docstring"""





       return self.lock




  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.lock.release()
       return None





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       max_filename_length if max_filename_length is not None else 2_5_5
       # Hash the filename if it's too long
       A_          =       self.hash_filename_if_too_long(__snake_case ,__snake_case )
       # The path to the lock file.
       A_          =       lock_file

       # The file descriptor for the *_lock_file* as it is returned by the
       # os.open() function.
       # This file lock is only NOT None, if the object currently holds the
       # lock.
       A_          =       None

       # The default timeout value.
       A_          =       timeout

       # We use this lock primarily for the lock counter.
       A_          =       threading.Lock()

       # The lock counter is used for implementing the nested locking
       # mechanism. Whenever the lock is acquired, the counter is increased and
       # the lock is only released, when this value is 0 again.
       A_          =       0
       return None


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._timeout


  @timeout.setter
  def 						__UpperCAmelCase     (							self ,__snake_case ):



       """simple docstring"""





       A_          =       float(__snake_case )
       return None


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file_fd is not None


  def 						__UpperCAmelCase     (							self ,__snake_case=None ,__snake_case=0.05 ):



       """simple docstring"""





       if timeout is None:
            A_          =       self.timeout

       # Increment the number right at the beginning.
       # We can still undo it, if something fails.
       with self._thread_lock:
            self._lock_counter += 1

       A_          =       id(self )
       A_          =       self._lock_file
       A_          =       time.time()
       try:
            while True:
                 with self._thread_lock:
                      if not self.is_locked:
                           logger().debug(f'Attempting to acquire lock {lock_id} on {lock_filename}' )
                           self._acquire()

                 if self.is_locked:
                      logger().debug(f'Lock {lock_id} acquired on {lock_filename}' )
                      break
                 elif timeout >= 0 and time.time() - start_time > timeout:
                      logger().debug(f'Timeout on acquiring lock {lock_id} on {lock_filename}' )
                      raise Timeout(self._lock_file )
                 else:
                      logger().debug(
                          f'Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ...' )
                      time.sleep(__snake_case )
       except:  # noqa
            # Something did go wrong, so decrement the counter.
            with self._thread_lock:
                 A_          =       max(0 ,self._lock_counter - 1 )

            raise
       return _Acquire_ReturnProxy(lock=self )


  def 						__UpperCAmelCase     (							self ,__snake_case=False ):



       """simple docstring"""





       with self._thread_lock:
            if self.is_locked:
                 self._lock_counter -= 1

                 if self._lock_counter == 0 or force:
                      A_          =       id(self )
                      A_          =       self._lock_file

                      logger().debug(f'Attempting to release lock {lock_id} on {lock_filename}' )
                      self._release()
                      A_          =       0
                      logger().debug(f'Lock {lock_id} released on {lock_filename}' )

       return None


  def __enter__(							self ):



       """simple docstring"""





       self.acquire()
       return self


  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.release()
       return None


  def __del__(							self ):



       """simple docstring"""





       self.release(force=__snake_case )
       return None




  def 						__UpperCAmelCase     (							self ,__snake_case ,__snake_case ):



       """simple docstring"""





       A_          =       os.path.basename(__snake_case )
       if len(__snake_case ) > max_length and max_length > 0:
            A_          =       os.path.dirname(__snake_case )
            A_          =       str(hash(__snake_case ) )
            A_          =       filename[: max_length - len(__snake_case ) - 8] + '''...''' + hashed_filename + '''.lock'''
            return os.path.join(__snake_case ,__snake_case )
       else:
            return path





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       from .file_utils import relative_to_absolute_path

       super().__init__(__snake_case ,timeout=__snake_case ,max_filename_length=__snake_case )
       A_          =       '''\\\\?\\''' + relative_to_absolute_path(self.lock_file )


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC

       try:
            A_          =       os.open(self._lock_file ,__snake_case )
       except OSError:
            pass
       else:
            try:
                 msvcrt.locking(__snake_case ,msvcrt.LK_NBLCK ,1 )
            except OSError:
                 os.close(__snake_case )
            else:
                 A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       None
       msvcrt.locking(__snake_case ,msvcrt.LK_UNLCK ,1 )
       os.close(__snake_case )

       try:
            os.remove(self._lock_file )
       # Probably another instance of the application
       # that acquired the file lock.
       except OSError:
            pass
       return None





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       os.statvfs(os.path.dirname(__snake_case ) ).f_namemax
       super().__init__(__snake_case ,timeout=__snake_case ,max_filename_length=__snake_case )


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC
       A_          =       os.open(self._lock_file ,__snake_case )

       try:
            fcntl.flock(__snake_case ,fcntl.LOCK_EX | fcntl.LOCK_NB )
       except OSError:
            os.close(__snake_case )
       else:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       None
       fcntl.flock(__snake_case ,fcntl.LOCK_UN )
       os.close(__snake_case )
       return None





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
            A_          =       os.open(self._lock_file ,__snake_case )
       except OSError:
            pass
       else:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       os.close(self._lock_file_fd )
       A_          =       None

       try:
            os.remove(self._lock_file )
       # The file is already deleted and that's what we want.
       except OSError:
            pass
       return None




a__      :     Union[str, Any]         =		None

if msvcrt:
      a__      :     Dict         =		WindowsFileLock
elif fcntl:
      a__      :     Dict         =		UnixFileLock
else:
      a__      :     Optional[int]         =		SoftFileLock

      if warnings is not None:
            warnings.warn('only soft file lock is available')



