




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




	"""simple docstring"""






	A_  :				List[str]	   =		botoa.client('''iam'''	)

	A_  :				List[Any]	   =		{
	    '''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_  :				List[str]	   =		{
		    '''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_      (		_UpperCAmelCase	):




	"""simple docstring"""






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

def 		lowercase_      (		):




	"""simple docstring"""






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

		A_  :				Optional[Any]	   =		_ask_field('''AWS Secret Access Key: '''	)
		A_  :				str	   =		aws_secret_access_key

	A_  :				Dict	   =		_ask_field('''Enter your AWS Region: [us-east-1]'''  ,      default='''us-east-1'''	)
	A_  :				List[str]	   =		aws_region

	A_  :				int	   =		_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_  :				List[Any]	   =		_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(UpperCAmelCase__	)

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

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

	A_  :				Optional[int]	   =		_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_  :				Tuple	   =		None
	if is_sagemaker_metrics_enabled:
		A_  :				Union[str, Any]	   =		_ask_field(
		    '''Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): '''  ,      lambda _UpperCAmelCase	: str(UpperCAmelCase__	).lower()  ,      )

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

	A_  :				List[str]	   =		1
	if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
		A_  :				Dict	   =		_ask_field(
		    '''How many machines do you want use? [1]: '''  ,      UpperCAmelCase__  ,      default=1  ,      )

	A_  :				int	   =		_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__  ,      )











"""simple docstring"""

import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import TransformeraDModel, VQDiffusionPipeline, VQDiffusionScheduler, VQModel
from diffusers.pipelines.vq_diffusion.pipeline_vq_diffusion import LearnedClassifierFreeSamplingEmbeddings
from diffusers.utils import load_numpy, slow, torch_device
from diffusers.utils.testing_utils import require_torch_gpu


_lowerCamelCase  :			Any        =       False

class        lowercase      (    unittest.TestCase):







				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				@property
				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					return 12







				@property
				def 					a_  (							self   :					List[str]       ):



					"""simple docstring"""

					return 12







				@property
				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					return 32







				@property
				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					torch.manual_seed(0       )
					A_  :				Union[str, Any]	   =		VQModel(
					    block_out_channels=[32, 64]      ,     in_channels=3      ,     out_channels=3      ,     down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']      ,     up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']      ,     latent_channels=3      ,     num_vq_embeddings=self.num_embed      ,     vq_embed_dim=3      ,     )
					return model







				@property
				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				int	   =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''       )
					return tokenizer







				@property
				def 					a_  (							self   :					int       ):



					"""simple docstring"""

					torch.manual_seed(0       )
					A_  :				Optional[int]	   =		CLIPTextConfig(
					    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=self.text_embedder_hidden_size      ,     intermediate_size=37      ,     layer_norm_eps=1E-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=10_00      ,     )
					return CLIPTextModel(_lowerCamelCase       )







				@property
				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					torch.manual_seed(0       )

					A_  :				Optional[Any]	   =		12
					A_  :				Optional[int]	   =		12

					A_  :				int	   =		{
					    '''attention_bias''': True,
					    '''cross_attention_dim''': 32,
					    '''attention_head_dim''': height * width,
					    '''num_attention_heads''': 1,
					    '''num_vector_embeds''': self.num_embed,
					    '''num_embeds_ada_norm''': self.num_embeds_ada_norm,
					    '''norm_num_groups''': 32,
					    '''sample_size''': width,
					    '''activation_fn''': '''geglu-approximate''',
					}

					A_  :				Tuple	   =		TransformeraDModel(**_lowerCamelCase       )
					return model







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		'''cpu'''

					A_  :				Union[str, Any]	   =		self.dummy_vqvae
					A_  :				str	   =		self.dummy_text_encoder
					A_  :				List[Any]	   =		self.dummy_tokenizer
					A_  :				int	   =		self.dummy_transformer
					A_  :				Any	   =		VQDiffusionScheduler(self.num_embed       )
					A_  :				Optional[Any]	   =		LearnedClassifierFreeSamplingEmbeddings(learnable=_lowerCamelCase       )

					A_  :				Dict	   =		VQDiffusionPipeline(
					    vqvae=_lowerCamelCase      ,     text_encoder=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     transformer=_lowerCamelCase      ,     scheduler=_lowerCamelCase      ,     learned_classifier_free_sampling_embeddings=_lowerCamelCase      ,     )
					A_  :				List[Any]	   =		pipe.to(_lowerCamelCase       )
					pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				List[Any]	   =		'''teddy bear playing in the pool'''

					A_  :				List[Any]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[Any]	   =		pipe([prompt]      ,     generator=_lowerCamelCase      ,     num_inference_steps=2      ,     output_type='''np'''       )
					A_  :				Any	   =		output.images

					A_  :				List[str]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[Any]	   =		pipe(
					    [prompt]      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     return_dict=_lowerCamelCase      ,     num_inference_steps=2       )[0]

					A_  :				Optional[int]	   =		image[0, -3:, -3:, -1]
					A_  :				Optional[int]	   =		image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 24, 24, 3)

					A_  :				Optional[int]	   =		np.array([0.6551, 0.6168, 0.5008, 0.5676, 0.5659, 0.4295, 0.6073, 0.5599, 0.4992]       )

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







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		'''cpu'''

					A_  :				int	   =		self.dummy_vqvae
					A_  :				List[str]	   =		self.dummy_text_encoder
					A_  :				Optional[Any]	   =		self.dummy_tokenizer
					A_  :				Any	   =		self.dummy_transformer
					A_  :				Any	   =		VQDiffusionScheduler(self.num_embed       )
					A_  :				Optional[int]	   =		LearnedClassifierFreeSamplingEmbeddings(
					    learnable=_lowerCamelCase      ,     hidden_size=self.text_embedder_hidden_size      ,     length=tokenizer.model_max_length       )

					A_  :				int	   =		VQDiffusionPipeline(
					    vqvae=_lowerCamelCase      ,     text_encoder=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     transformer=_lowerCamelCase      ,     scheduler=_lowerCamelCase      ,     learned_classifier_free_sampling_embeddings=_lowerCamelCase      ,     )
					A_  :				List[Any]	   =		pipe.to(_lowerCamelCase       )
					pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				Any	   =		'''teddy bear playing in the pool'''

					A_  :				str	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				Optional[Any]	   =		pipe([prompt]      ,     generator=_lowerCamelCase      ,     num_inference_steps=2      ,     output_type='''np'''       )
					A_  :				Tuple	   =		output.images

					A_  :				Optional[int]	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				List[str]	   =		pipe(
					    [prompt]      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     return_dict=_lowerCamelCase      ,     num_inference_steps=2       )[0]

					A_  :				Optional[int]	   =		image[0, -3:, -3:, -1]
					A_  :				Optional[Any]	   =		image_from_tuple[0, -3:, -3:, -1]

					assert image.shape == (1, 24, 24, 3)

					A_  :				str	   =		np.array([0.6693, 0.6075, 0.4959, 0.5701, 0.5583, 0.4333, 0.6171, 0.5684, 0.4988]       )

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

@slow
@require_torch_gpu
class        lowercase      (    unittest.TestCase):







				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					A_  :				str	   =		load_numpy(
					    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
					    '''/vq_diffusion/teddy_bear_pool_classifier_free_sampling.npy'''       )

					A_  :				int	   =		VQDiffusionPipeline.from_pretrained('''microsoft/vq-diffusion-ithq'''       )
					A_  :				Tuple	   =		pipeline.to(_lowerCamelCase       )
					pipeline.set_progress_bar_config(disable=_lowerCamelCase       )

					# requires GPU generator for gumbel softmax
					# don't use GPU generator in tests though
					A_  :				Dict	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(0       )
					A_  :				Union[str, Any]	   =		pipeline(
					    '''teddy bear playing in the pool'''      ,     num_images_per_prompt=1      ,     generator=_lowerCamelCase      ,     output_type='''np'''      ,     )

					A_  :				Optional[int]	   =		output.images[0]

					assert image.shape == (2_56, 2_56, 3)
					assert np.abs(expected_image - image       ).max() < 2.0





