






"""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 )     ->       int:
				lowercase    :      Tuple       = botoa.client("""iam""" )

				lowercase    :      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=_A  ,				AssumeRolePolicyDocument=json.dumps(_A  ,				indent=2 ) )
								lowercase    :      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=_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:
				lowercase    :      int       = botoa.client("""iam""" )
				return iam_client.get_role(RoleName=_A )["Role"]["Arn"]


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

								lowercase    :      List[str]       = _ask_field("""AWS Secret Access Key: """ )
								lowercase    :      List[Any]       = aws_secret_access_key

				lowercase    :      Any       = _ask_field("""Enter your AWS Region: [us-east-1]"""  ,				default="""us-east-1""" )
				lowercase    :      List[Any]       = aws_region

				lowercase    :      Optional[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"""]  ,				_A  ,				)
				if role_management == 0:
								lowercase    :      int       = _ask_field("""Enter your IAM role name: """ )
				else:
								lowercase    :      List[str]       = """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 )

				lowercase    :      str       = _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."""  ,				)
				lowercase    :      int       = None
				if is_custom_docker_image:
								lowercase    :      Tuple       = _ask_field("""Enter your Docker image: """  ,				lambda _A : str(_A ).lower() )

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

				lowercase    :      Optional[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."""  ,				)
				lowercase    :      Optional[Any]       = None
				if is_sagemaker_metrics_enabled:
								lowercase    :      Tuple       = _ask_field(
								    """Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): """  ,				lambda _A : str(_A ).lower()  ,				)

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

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

				lowercase    :      List[str]       = _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"""







from typing import List, Optional, Tuple, Union

import torch

from ...schedulers import DDIMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


class      UpperCamelCase		(__snake_case     ):






						def __init__(		self  :Any			, __magic_name__  :Union[str, Any]			, __magic_name__  :Optional[Any]				)      ->Dict:
										super().__init__()

										# make sure scheduler can always be converted to DDIM
										lowercase    :      Optional[Any]       = DDIMScheduler.from_config(scheduler.config				)

										self.register_modules(unet=__magic_name__			, scheduler=__magic_name__				)






						@torch.no_grad()
						def __call__(		self  :Optional[int]			, __magic_name__  :int = 1			, __magic_name__  :Optional[Union[torch.Generator, List[torch.Generator]]] = None			, __magic_name__  :float = 0.0			, __magic_name__  :int = 50			, __magic_name__  :Optional[bool] = None			, __magic_name__  :Optional[str] = "pil"			, __magic_name__  :bool = True			, )      ->Union[ImagePipelineOutput, Tuple]:

										# Sample gaussian noise to begin loop
										if isinstance(self.unet.config.sample_size			, __magic_name__				):
														lowercase    :      int       = (
														    batch_size,
														    self.unet.config.in_channels,
														    self.unet.config.sample_size,
														    self.unet.config.sample_size,
														)
										else:
														lowercase    :      Tuple       = (batch_size, self.unet.config.in_channels, *self.unet.config.sample_size)

										if isinstance(__magic_name__			, __magic_name__				) and len(__magic_name__				) != batch_size:
														raise ValueError(
														    f"""You have passed a list of generators of length {len(__magic_name__				)}, but requested an effective batch"""
														    f""" size of {batch_size}. Make sure the batch size matches the length of the generators."""				)

										lowercase    :      Dict       = randn_tensor(__magic_name__			, generator=__magic_name__			, device=self.device			, dtype=self.unet.dtype				)

										# set step values
										self.scheduler.set_timesteps(__magic_name__				)

										for t in self.progress_bar(self.scheduler.timesteps				):
														# 1. predict noise model_output
														lowercase    :      Union[str, Any]       = self.unet(__magic_name__			, __magic_name__				).sample

														# 2. predict previous mean of image x_t-1 and add variance depending on eta
														# eta corresponds to η in paper and should be between [0, 1]
														# do x_t -> x_t-1
														lowercase    :      int       = self.scheduler.step(
														    __magic_name__			, __magic_name__			, __magic_name__			, eta=__magic_name__			, use_clipped_model_output=__magic_name__			, generator=__magic_name__				).prev_sample

										lowercase    :      Optional[Any]       = (image / 2 + 0.5).clamp(0			, 1				)
										lowercase    :      Tuple       = image.cpu().permute(0			, 2			, 3			, 1				).numpy()
										if output_type == "pil":
														lowercase    :      Any       = self.numpy_to_pil(__magic_name__				)

										if not return_dict:
														return (image,)

										return ImagePipelineOutput(images=__magic_name__				)
