




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 						_snake_case					( _snake_case							: Any							)   -> List[str]:






      '''simple docstring'''


      _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=A_    ,  AssumeRolePolicyDocument=json.dumps(A_    ,  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=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 						_snake_case					( _snake_case							: Any							)   -> List[Any]:






      '''simple docstring'''


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







def 						_snake_case					( )   -> Any:






      '''simple docstring'''


      _A        =       _ask_options(
          'How do you want to authorize?'    ,  ['AWS Profile', 'Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) ']    ,  A_    ,  )
      _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']    ,  A_    ,  )
      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(A_							)

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

      _A        =       _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.'    ,  )
      _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 _snake_case							: str(A_							).lower()    ,  )

      _A        =       _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.'    ,  )
      _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 _snake_case							: str(A_							).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=A_    ,  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=A_    ,  error_message='Please enter yes or no.'    ,  )

            if use_custom_options:
                  _A        =       _ask_options(
                      'Which mode do you want to use?'    ,  A_    ,  lambda _snake_case							: TORCH_DYNAMO_MODES[int(A_							)]    ,  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=A_    ,  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=A_    ,  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(
                A_    ,  A_    ,  lambda _snake_case							: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(A_							)]							)
      else:
            eca_instance_query += "? [ml.p3.2xlarge]:"
            _A        =       _ask_field(A_    ,  lambda _snake_case							: str(A_							).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]: '    ,  A_    ,  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=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 dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder





@dataclass
class     lowercase_	(					__lowerCAmelCase							):



 '''simple docstring'''
 UpperCAmelCase   :					"DiagonalGaussianDistribution"





class     lowercase_	(					__lowerCAmelCase      ,     __lowerCAmelCase							):



 '''simple docstring'''
 UpperCAmelCase   :					List[Any]				  =     True

 @register_to_config
 def __init__(			self	:	List[str]   ,							_UpperCAmelCase	:	int = 3   ,							_UpperCAmelCase	:	int = 3   ,							_UpperCAmelCase	:	Tuple[str] = ("DownEncoderBlock2D",)   ,							_UpperCAmelCase	:	Tuple[str] = ("UpDecoderBlock2D",)   ,							_UpperCAmelCase	:	Tuple[int] = (64,)   ,							_UpperCAmelCase	:	int = 1   ,							_UpperCAmelCase	:	str = "silu"   ,							_UpperCAmelCase	:	int = 4   ,							_UpperCAmelCase	:	int = 32   ,							_UpperCAmelCase	:	int = 32   ,							_UpperCAmelCase	:	float = 0.1_8215   ,							):
       super().__init__()

       # pass init params to Encoder
       _A        =       Encoder(
           in_channels=_UpperCAmelCase   ,							out_channels=_UpperCAmelCase   ,							down_block_types=_UpperCAmelCase   ,							block_out_channels=_UpperCAmelCase   ,							layers_per_block=_UpperCAmelCase   ,							act_fn=_UpperCAmelCase   ,							norm_num_groups=_UpperCAmelCase   ,							double_z=_UpperCAmelCase   ,							)

       # pass init params to Decoder
       _A        =       Decoder(
           in_channels=_UpperCAmelCase   ,							out_channels=_UpperCAmelCase   ,							up_block_types=_UpperCAmelCase   ,							block_out_channels=_UpperCAmelCase   ,							layers_per_block=_UpperCAmelCase   ,							norm_num_groups=_UpperCAmelCase   ,							act_fn=_UpperCAmelCase   ,							)

       _A        =       nn.Convad(2 * latent_channels   ,							2 * latent_channels   ,							1		)
       _A        =       nn.Convad(_UpperCAmelCase   ,							_UpperCAmelCase   ,							1		)

       _A        =       False
       _A        =       False

       # only relevant if vae tiling is enabled
       _A        =       self.config.sample_size
       _A        =       (
           self.config.sample_size[0]
           if isinstance(self.config.sample_size   ,							(list, tuple)		)
           else self.config.sample_size
       )
       _A        =       int(sample_size / (2 ** (len(self.config.block_out_channels		) - 1))		)
       _A        =       0.25

 def       lowerCAmelCase_							(			self	:	Union[str, Any]   ,							_UpperCAmelCase	:	int   ,							_UpperCAmelCase	:	Tuple=False		):
       if isinstance(_UpperCAmelCase   ,							(Encoder, Decoder)		):
             _A        =       value

 def       lowerCAmelCase_							(			self	:	List[str]   ,							_UpperCAmelCase	:	bool = True		):
       _A        =       use_tiling

 def       lowerCAmelCase_							(			self	:	Union[str, Any]		):
       self.enable_tiling(_UpperCAmelCase		)

 def       lowerCAmelCase_							(			self	:	Dict		):
       _A        =       True

 def       lowerCAmelCase_							(			self	:	str		):
       _A        =       False

 @property
 # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
 def       lowerCAmelCase_							(			self	:	str		):
       _A        =       {}

       def fn_recursive_add_processors(_UpperCAmelCase	:	str   ,							_UpperCAmelCase	:	torch.nn.Module   ,							_UpperCAmelCase	:	Dict[str, AttentionProcessor]		):
             if hasattr(_UpperCAmelCase   ,							'set_processor'		):
                   _A        =       module.processor

             for sub_name, child in module.named_children():
                   fn_recursive_add_processors(F'''{name}.{sub_name}'''   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)

             return processors

       for name, module in self.named_children():
             fn_recursive_add_processors(_UpperCAmelCase   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)

       return processors

 def       lowerCAmelCase_							(			self	:	str   ,							_UpperCAmelCase	:	Union[AttentionProcessor, Dict[str, AttentionProcessor]]		):
       _A        =       len(self.attn_processors.keys()		)

       if isinstance(_UpperCAmelCase   ,							_UpperCAmelCase		) and len(_UpperCAmelCase		) != count:
             raise ValueError(
                 F'''A dict of processors was passed, but the number of processors {len(_UpperCAmelCase		)} does not match the'''
                 F''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''		)

       def fn_recursive_attn_processor(_UpperCAmelCase	:	str   ,							_UpperCAmelCase	:	torch.nn.Module   ,							_UpperCAmelCase	:	int		):
             if hasattr(_UpperCAmelCase   ,							'set_processor'		):
                   if not isinstance(_UpperCAmelCase   ,							_UpperCAmelCase		):
                         module.set_processor(_UpperCAmelCase		)
                   else:
                         module.set_processor(processor.pop(F'''{name}.processor'''		)		)

             for sub_name, child in module.named_children():
                   fn_recursive_attn_processor(F'''{name}.{sub_name}'''   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)

       for name, module in self.named_children():
             fn_recursive_attn_processor(_UpperCAmelCase   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)

 def       lowerCAmelCase_							(			self	:	List[str]		):
       self.set_attn_processor(AttnProcessor()		)

 @apply_forward_hook
 def       lowerCAmelCase_							(			self	:	int   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = True		):
       if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
             return self.tiled_encode(_UpperCAmelCase   ,							return_dict=_UpperCAmelCase		)

       if self.use_slicing and x.shape[0] > 1:
             _A        =       [self.encoder(_UpperCAmelCase		) for x_slice in x.split(1		)]
             _A        =       torch.cat(_UpperCAmelCase		)
       else:
             _A        =       self.encoder(_UpperCAmelCase		)

       _A        =       self.quant_conv(_UpperCAmelCase		)
       _A        =       DiagonalGaussianDistribution(_UpperCAmelCase		)

       if not return_dict:
             return (posterior,)

       return AutoencoderKLOutput(latent_dist=_UpperCAmelCase		)

 def       lowerCAmelCase_							(			self	:	Any   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = True		):
       if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
             return self.tiled_decode(_UpperCAmelCase   ,							return_dict=_UpperCAmelCase		)

       _A        =       self.post_quant_conv(_UpperCAmelCase		)
       _A        =       self.decoder(_UpperCAmelCase		)

       if not return_dict:
             return (dec,)

       return DecoderOutput(sample=_UpperCAmelCase		)

 @apply_forward_hook
 def       lowerCAmelCase_							(			self	:	str   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = True		):
       if self.use_slicing and z.shape[0] > 1:
             _A        =       [self._decode(_UpperCAmelCase		).sample for z_slice in z.split(1		)]
             _A        =       torch.cat(_UpperCAmelCase		)
       else:
             _A        =       self._decode(_UpperCAmelCase		).sample

       if not return_dict:
             return (decoded,)

       return DecoderOutput(sample=_UpperCAmelCase		)

 def       lowerCAmelCase_							(			self	:	str   ,							_UpperCAmelCase	:	int   ,							_UpperCAmelCase	:	Dict   ,							_UpperCAmelCase	:	Any		):
       _A        =       min(a.shape[2]   ,							b.shape[2]   ,							_UpperCAmelCase		)
       for y in range(_UpperCAmelCase		):
             _A        =       a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
       return b

 def       lowerCAmelCase_							(			self	:	Optional[Any]   ,							_UpperCAmelCase	:	List[str]   ,							_UpperCAmelCase	:	List[Any]   ,							_UpperCAmelCase	:	List[Any]		):
       _A        =       min(a.shape[3]   ,							b.shape[3]   ,							_UpperCAmelCase		)
       for x in range(_UpperCAmelCase		):
             _A        =       a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
       return b

 def       lowerCAmelCase_							(			self	:	str   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = True		):
       _A        =       int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)		)
       _A        =       int(self.tile_latent_min_size * self.tile_overlap_factor		)
       _A        =       self.tile_latent_min_size - blend_extent

       # Split the image into 512x512 tiles and encode them separately.
       _A        =       []
       for i in range(0   ,							x.shape[2]   ,							_UpperCAmelCase		):
             _A        =       []
             for j in range(0   ,							x.shape[3]   ,							_UpperCAmelCase		):
                   _A        =       x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                   _A        =       self.encoder(_UpperCAmelCase		)
                   _A        =       self.quant_conv(_UpperCAmelCase		)
                   row.append(_UpperCAmelCase		)
             rows.append(_UpperCAmelCase		)
       _A        =       []
       for i, row in enumerate(_UpperCAmelCase		):
             _A        =       []
             for j, tile in enumerate(_UpperCAmelCase		):
                   # blend the above tile and the left tile
                   # to the current tile and add the current tile to the result row
                   if i > 0:
                         _A        =       self.blend_v(rows[i - 1][j]   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)
                   if j > 0:
                         _A        =       self.blend_h(row[j - 1]   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)
                   result_row.append(tile[:, :, :row_limit, :row_limit]		)
             result_rows.append(torch.cat(_UpperCAmelCase   ,							dim=3		)		)

       _A        =       torch.cat(_UpperCAmelCase   ,							dim=2		)
       _A        =       DiagonalGaussianDistribution(_UpperCAmelCase		)

       if not return_dict:
             return (posterior,)

       return AutoencoderKLOutput(latent_dist=_UpperCAmelCase		)

 def       lowerCAmelCase_							(			self	:	Tuple   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = True		):
       _A        =       int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)		)
       _A        =       int(self.tile_sample_min_size * self.tile_overlap_factor		)
       _A        =       self.tile_sample_min_size - blend_extent

       # Split z into overlapping 64x64 tiles and decode them separately.
       # The tiles have an overlap to avoid seams between tiles.
       _A        =       []
       for i in range(0   ,							z.shape[2]   ,							_UpperCAmelCase		):
             _A        =       []
             for j in range(0   ,							z.shape[3]   ,							_UpperCAmelCase		):
                   _A        =       z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                   _A        =       self.post_quant_conv(_UpperCAmelCase		)
                   _A        =       self.decoder(_UpperCAmelCase		)
                   row.append(_UpperCAmelCase		)
             rows.append(_UpperCAmelCase		)
       _A        =       []
       for i, row in enumerate(_UpperCAmelCase		):
             _A        =       []
             for j, tile in enumerate(_UpperCAmelCase		):
                   # blend the above tile and the left tile
                   # to the current tile and add the current tile to the result row
                   if i > 0:
                         _A        =       self.blend_v(rows[i - 1][j]   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)
                   if j > 0:
                         _A        =       self.blend_h(row[j - 1]   ,							_UpperCAmelCase   ,							_UpperCAmelCase		)
                   result_row.append(tile[:, :, :row_limit, :row_limit]		)
             result_rows.append(torch.cat(_UpperCAmelCase   ,							dim=3		)		)

       _A        =       torch.cat(_UpperCAmelCase   ,							dim=2		)
       if not return_dict:
             return (dec,)

       return DecoderOutput(sample=_UpperCAmelCase		)



 def       lowerCAmelCase_							(			self	:	List[str]   ,							_UpperCAmelCase	:	torch.FloatTensor   ,							_UpperCAmelCase	:	bool = False   ,							_UpperCAmelCase	:	bool = True   ,							_UpperCAmelCase	:	Optional[torch.Generator] = None   ,							):
       _A        =       sample
       _A        =       self.encode(_UpperCAmelCase		).latent_dist
       if sample_posterior:
             _A        =       posterior.sample(generator=_UpperCAmelCase		)
       else:
             _A        =       posterior.mode()
       _A        =       self.decode(_UpperCAmelCase		).sample

       if not return_dict:
             return (dec,)

       return DecoderOutput(sample=_UpperCAmelCase		)

