



"""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 	UpperCAmelCase     (		A	:						List[Any]       ):

		'''simple docstring'''


		_UpperCAmelCase							 =		botoa.client('iam'       )

		_UpperCAmelCase							 =		{
		    '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       )       )
				_UpperCAmelCase							 =		{
				    '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	:						int       ):

		'''simple docstring'''


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






def 	UpperCAmelCase     (		):

		'''simple docstring'''


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

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

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

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

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

		_UpperCAmelCase							 =		_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.'     ,				)
		_UpperCAmelCase							 =		None
		if is_sagemaker_inputs_enabled:
				_UpperCAmelCase							 =		_ask_field(
				    'Enter the path to the SageMaker inputs TSV file with columns (channel_name, data_location): '     ,				lambda A       : str(A       ).lower()     ,				)

		_UpperCAmelCase							 =		_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.'     ,				)
		_UpperCAmelCase							 =		None
		if is_sagemaker_metrics_enabled:
				_UpperCAmelCase							 =		_ask_field(
				    'Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): '     ,				lambda A       : str(A       ).lower()     ,				)

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

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

		_UpperCAmelCase							 =		_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 gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 			lowercase__			(      A,					unittest.TestCase    ):
							'''simple docstring'''







							_UpperCAmelCase      				=					DiTPipeline
							_UpperCAmelCase      				=					CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
							_UpperCAmelCase      				=					PipelineTesterMixin.required_optional_params - {
							    '''latents''',
							    '''num_images_per_prompt''',
							    '''callback''',
							    '''callback_steps''',
							}
							_UpperCAmelCase      				=					CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
							_UpperCAmelCase      				=					False






							def        lowerCamelCase_   (  self						)			->							str:
									torch.manual_seed(0						)
									_UpperCAmelCase							 =		TransformeraDModel(
									    sample_size=16	,       num_layers=2	,       patch_size=4	,       attention_head_dim=8	,       num_attention_heads=2	,       in_channels=4	,       out_channels=8	,       attention_bias=snake_case	,       activation_fn='gelu-approximate'	,       num_embeds_ada_norm=1000	,       norm_type='ada_norm_zero'	,       norm_elementwise_affine=snake_case	,       )
									_UpperCAmelCase							 =		AutoencoderKL()
									_UpperCAmelCase							 =		DDIMScheduler()
									_UpperCAmelCase							 =		{'transformer': transformer.eval(), 'vae': vae.eval(), 'scheduler': scheduler}
									return components






							def        lowerCamelCase_   (  self	,       snake_case	,       snake_case=0						)			->							Optional[Any]:
									if str(snake_case						).startswith('mps'						):
											_UpperCAmelCase							 =		torch.manual_seed(snake_case						)
									else:
											_UpperCAmelCase							 =		torch.Generator(device=snake_case						).manual_seed(snake_case						)
									_UpperCAmelCase							 =		{
									    'class_labels': [1],
									    'generator': generator,
									    'num_inference_steps': 2,
									    'output_type': 'numpy',
									}
									return inputs






							def        lowerCamelCase_   (  self						)			->							List[Any]:
									_UpperCAmelCase							 =		'cpu'

									_UpperCAmelCase							 =		self.get_dummy_components()
									_UpperCAmelCase							 =		self.pipeline_class(**snake_case						)
									pipe.to(snake_case						)
									pipe.set_progress_bar_config(disable=snake_case						)

									_UpperCAmelCase							 =		self.get_dummy_inputs(snake_case						)
									_UpperCAmelCase							 =		pipe(**snake_case						).images
									_UpperCAmelCase							 =		image[0, -3:, -3:, -1]

									self.assertEqual(image.shape	,       (1, 16, 16, 3)						)
									_UpperCAmelCase							 =		np.array([0.2946, 0.6601, 0.4329, 0.3296, 0.4144, 0.5319, 0.7273, 0.5013, 0.4457]						)
									_UpperCAmelCase							 =		np.abs(image_slice.flatten() - expected_slice						).max()
									self.assertLessEqual(snake_case	,       1E-3						)






							def        lowerCamelCase_   (  self						)			->							Any:
									self._test_inference_batch_single_identical(relax_max_difference=snake_case	,       expected_max_diff=1E-3						)






							@unittest.skipIf(
							    torch_device != 'cuda' or not is_xformers_available()	,       reason='XFormers attention is only available with CUDA and `xformers` installed'	,       )
							def        lowerCamelCase_   (  self						)			->							Optional[int]:
									self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3						)







@require_torch_gpu
@slow
class 			lowercase__			(      unittest.TestCase    ):
							'''simple docstring'''







							def        lowerCamelCase_   (  self						)			->							int:
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






							def        lowerCamelCase_   (  self						)			->							int:
									_UpperCAmelCase							 =		torch.manual_seed(0						)

									_UpperCAmelCase							 =		DiTPipeline.from_pretrained('facebook/DiT-XL-2-256'						)
									pipe.to('cuda'						)

									_UpperCAmelCase							 =		['vase', 'umbrella', 'white shark', 'white wolf']
									_UpperCAmelCase							 =		pipe.get_label_ids(snake_case						)

									_UpperCAmelCase							 =		pipe(snake_case	,       generator=snake_case	,       num_inference_steps=40	,       output_type='np'						).images

									for word, image in zip(snake_case	,       snake_case						):
											_UpperCAmelCase							 =		load_numpy(
											    f'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy'						)
											assert np.abs((expected_image - image).max()						) < 1E-2






							def        lowerCamelCase_   (  self						)			->							int:
									_UpperCAmelCase							 =		DiTPipeline.from_pretrained('facebook/DiT-XL-2-512'						)
									_UpperCAmelCase							 =		DPMSolverMultistepScheduler.from_config(pipe.scheduler.config						)
									pipe.to('cuda'						)

									_UpperCAmelCase							 =		['vase', 'umbrella']
									_UpperCAmelCase							 =		pipe.get_label_ids(snake_case						)

									_UpperCAmelCase							 =		torch.manual_seed(0						)
									_UpperCAmelCase							 =		pipe(snake_case	,       generator=snake_case	,       num_inference_steps=25	,       output_type='np'						).images

									for word, image in zip(snake_case	,       snake_case						):
											_UpperCAmelCase							 =		load_numpy(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
											    f'/dit/{word}_512.npy'						)

											assert np.abs((expected_image - image).max()						) < 1E-1


