
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_(  a__					):



    """simple docstring"""



    SCREAMING_SNAKE_CASE							:							List[Any]								=      botoa.client('''iam'''					)

    SCREAMING_SNAKE_CASE							:							Dict								=      {
        '''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					)					)
        SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      {
            '''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_(  a__					):



    """simple docstring"""



    SCREAMING_SNAKE_CASE							:							Optional[int]								=      botoa.client('''iam'''					)
    return iam_client.get_role(RoleName=_A					)["Role"]["Arn"]







def 							UpperCAmelCase_(  ):



    """simple docstring"""



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

        SCREAMING_SNAKE_CASE							:							Tuple								=      _ask_field('''AWS Secret Access Key: '''					)
        SCREAMING_SNAKE_CASE							:							Dict								=      aws_secret_access_key

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

    SCREAMING_SNAKE_CASE							:							str								=      _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:
        SCREAMING_SNAKE_CASE							:							Dict								=      _ask_field('''Enter your IAM role name: '''					)
    else:
        SCREAMING_SNAKE_CASE							:							Dict								=      '''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					)

    SCREAMING_SNAKE_CASE							:							List[Any]								=      _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.'''      ,		)
    SCREAMING_SNAKE_CASE							:							List[str]								=      None
    if is_custom_docker_image:
        SCREAMING_SNAKE_CASE							:							List[str]								=      _ask_field('''Enter your Docker image: '''      ,		lambda a__					: str(_A					).lower()					)

    SCREAMING_SNAKE_CASE							:							Dict								=      _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.'''      ,		)
    SCREAMING_SNAKE_CASE							:							List[str]								=      None
    if is_sagemaker_inputs_enabled:
        SCREAMING_SNAKE_CASE							:							List[Any]								=      _ask_field(
            '''Enter the path to the SageMaker inputs TSV file with columns (channel_name, data_location): '''      ,		lambda a__					: str(_A					).lower()      ,		)

    SCREAMING_SNAKE_CASE							:							int								=      _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.'''      ,		)
    SCREAMING_SNAKE_CASE							:							List[str]								=      None
    if is_sagemaker_metrics_enabled:
        SCREAMING_SNAKE_CASE							:							Dict								=      _ask_field(
            '''Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): '''      ,		lambda a__					: str(_A					).lower()      ,		)

    SCREAMING_SNAKE_CASE							:							str								=      _ask_options(
        '''What is the distributed mode?'''      ,		['''No distributed training''', '''Data parallelism''']      ,		_convert_sagemaker_distributed_mode      ,		)
    SCREAMING_SNAKE_CASE							:							Union[str, Any]								=      {}
    SCREAMING_SNAKE_CASE							:							List[str]								=      _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:
        SCREAMING_SNAKE_CASE							:							Dict								=      '''dynamo_'''
        SCREAMING_SNAKE_CASE							:							str								=      _ask_options(
            '''Which dynamo backend would you like to use?'''      ,		[x.lower() for x in DYNAMO_BACKENDS]      ,		_convert_dynamo_backend      ,		default=2      ,		)
        SCREAMING_SNAKE_CASE							:							str								=      _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:
            SCREAMING_SNAKE_CASE							:							List[str]								=      _ask_options(
                '''Which mode do you want to use?'''      ,		_A      ,		lambda a__					: TORCH_DYNAMO_MODES[int(_A					)]      ,		default='''default'''      ,		)
            SCREAMING_SNAKE_CASE							:							Dict								=      _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.'''      ,		)
            SCREAMING_SNAKE_CASE							:							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.'''      ,		)
    SCREAMING_SNAKE_CASE							:							Optional[int]								=      '''Which EC2 instance type you want to use for your training?'''
    if distributed_type != SageMakerDistributedType.NO:
        SCREAMING_SNAKE_CASE							:							List[Any]								=      _ask_options(
            _A      ,		_A      ,		lambda a__					: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_A					)]					)
    else:
        eca_instance_query += "? [ml.p3.2xlarge]:"
        SCREAMING_SNAKE_CASE							:							str								=      _ask_field(_A      ,		lambda a__					: str(_A					).lower()      ,		default='''ml.p3.2xlarge'''					)

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

    SCREAMING_SNAKE_CASE							:							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=_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 unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()






class 				a__(					unittest.TestCase	):


			'''simple docstring'''






			@property
			def 							a_       (   self):




					"""simple docstring"""


					torch.manual_seed(0)
					lowerCAmelCase   					=  UNetaDModel(
					    block_out_channels=(32, 64)       ,    layers_per_block=2       ,    sample_size=32       ,    in_channels=3       ,    out_channels=3       ,    down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")       ,    up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")       ,    )
					return model







			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.dummy_uncond_unet
					lowerCAmelCase   					=  PNDMScheduler()

					lowerCAmelCase   					=  PNDMPipeline(unet=__lowerCAmelCase       ,    scheduler=__lowerCAmelCase)
					pndm.to(__lowerCAmelCase)
					pndm.set_progress_bar_config(disable=__lowerCAmelCase)

					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    num_inference_steps=20       ,    output_type="""numpy""").images

					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    num_inference_steps=20       ,    output_type="""numpy"""       ,    return_dict=__lowerCAmelCase)[0]

					lowerCAmelCase   					=  image[0, -3:, -3:, -1]
					lowerCAmelCase   					=  image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 32, 32, 3)
					lowerCAmelCase   					=  np.array([1.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0])

					assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2
					assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1E-2







@slow
@require_torch
class 				a__(					unittest.TestCase	):


			'''simple docstring'''






			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  """google/ddpm-cifar10-32"""

					lowerCAmelCase   					=  UNetaDModel.from_pretrained(__lowerCAmelCase)
					lowerCAmelCase   					=  PNDMScheduler()

					lowerCAmelCase   					=  PNDMPipeline(unet=__lowerCAmelCase       ,    scheduler=__lowerCAmelCase)
					pndm.to(__lowerCAmelCase)
					pndm.set_progress_bar_config(disable=__lowerCAmelCase)
					lowerCAmelCase   					=  torch.manual_seed(0)
					lowerCAmelCase   					=  pndm(generator=__lowerCAmelCase       ,    output_type="""numpy""").images

					lowerCAmelCase   					=  image[0, -3:, -3:, -1]

					assert image.shape == (1, 32, 32, 3)
					lowerCAmelCase   					=  np.array([0.1564, 0.14645, 0.1406, 0.14715, 0.12425, 0.14045, 0.13115, 0.12175, 0.125])

					assert np.abs(image_slice.flatten() - expected_slice).max() < 1E-2




