'''simple docstring'''




import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest


class       UpperCAmelCase__  (       _snake_case			):






   """simple docstring"""
   A							:       int     =    (DPMSolverSinglestepScheduler,)
   A							:       List[str]     =    (('''num_inference_steps''', 25),)
   def 		_lowerCamelCase (self       ,				**_a				)    ->     Optional[Any]:
        lowercase_					:       Optional[Any]     				=					{
            'num_train_timesteps': 1_000,
            'beta_start': 0.00_01,
            'beta_end': 0.02,
            'beta_schedule': 'linear',
            'solver_order': 2,
            'prediction_type': 'epsilon',
            'thresholding': False,
            'sample_max_value': 1.0,
            'algorithm_type': 'dpmsolver++',
            'solver_type': 'midpoint',
            'lambda_min_clipped': -float('inf'				),
            'variance_type': None,
        }

        config.update(**_a				)
        return config
   def 		_lowerCamelCase (self       ,				_a=0       ,				**_a				)    ->     Tuple:
        lowercase_					:       List[str]     				=					dict(self.forward_default_kwargs				)
        lowercase_					:       List[Any]     				=					kwargs.pop('num_inference_steps'       ,				_a				)
        lowercase_					:       Optional[int]     				=					self.dummy_sample
        lowercase_					:       Optional[Any]     				=					0.1 * sample
        lowercase_					:       List[Any]     				=					[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
             lowercase_					:       Dict     				=					self.get_scheduler_config(**_a				)
             lowercase_					:       Tuple     				=					scheduler_class(**_a				)
             scheduler.set_timesteps(_a				)
             # copy over dummy past residuals
             lowercase_					:       Optional[Any]     				=					dummy_past_residuals[: scheduler.config.solver_order]

             with tempfile.TemporaryDirectory() as tmpdirname:
                  scheduler.save_config(_a				)
                  lowercase_					:       List[Any]     				=					scheduler_class.from_pretrained(_a				)
                  new_scheduler.set_timesteps(_a				)
                  # copy over dummy past residuals
                  lowercase_					:       Optional[int]     				=					dummy_past_residuals[: new_scheduler.config.solver_order]

             lowercase_				,lowercase_					:       str     				=					sample, sample
             for t in range(_a       ,				time_step + scheduler.config.solver_order + 1				):
                  lowercase_					:       Dict     				=					scheduler.step(_a       ,				_a       ,				_a       ,				**_a				).prev_sample
                  lowercase_					:       Dict     				=					new_scheduler.step(_a       ,				_a       ,				_a       ,				**_a				).prev_sample

                  assert torch.sum(torch.abs(output - new_output				)				) < 1e-5, "Scheduler outputs are not identical"
   def 		_lowerCamelCase (self				)    ->     Tuple:
        pass
   def 		_lowerCamelCase (self       ,				_a=0       ,				**_a				)    ->     Tuple:
        lowercase_					:       Optional[int]     				=					dict(self.forward_default_kwargs				)
        lowercase_					:       List[Any]     				=					kwargs.pop('num_inference_steps'       ,				_a				)
        lowercase_					:       int     				=					self.dummy_sample
        lowercase_					:       int     				=					0.1 * sample
        lowercase_					:       Union[str, Any]     				=					[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
             lowercase_					:       Optional[int]     				=					self.get_scheduler_config()
             lowercase_					:       Dict     				=					scheduler_class(**_a				)
             scheduler.set_timesteps(_a				)

             # copy over dummy past residuals (must be after setting timesteps)
             lowercase_					:       str     				=					dummy_past_residuals[: scheduler.config.solver_order]

             with tempfile.TemporaryDirectory() as tmpdirname:
                  scheduler.save_config(_a				)
                  lowercase_					:       Tuple     				=					scheduler_class.from_pretrained(_a				)
                  # copy over dummy past residuals
                  new_scheduler.set_timesteps(_a				)

                  # copy over dummy past residual (must be after setting timesteps)
                  lowercase_					:       Dict     				=					dummy_past_residuals[: new_scheduler.config.solver_order]

             lowercase_					:       Tuple     				=					scheduler.step(_a       ,				_a       ,				_a       ,				**_a				).prev_sample
             lowercase_					:       Optional[int]     				=					new_scheduler.step(_a       ,				_a       ,				_a       ,				**_a				).prev_sample

             assert torch.sum(torch.abs(output - new_output				)				) < 1e-5, "Scheduler outputs are not identical"
   def 		_lowerCamelCase (self       ,				_a=None       ,				**_a				)    ->     str:
        if scheduler is None:
             lowercase_					:       List[Any]     				=					self.scheduler_classes[0]
             lowercase_					:       Optional[Any]     				=					self.get_scheduler_config(**_a				)
             lowercase_					:       Any     				=					scheduler_class(**_a				)

        lowercase_					:       Optional[Any]     				=					self.scheduler_classes[0]
        lowercase_					:       List[Any]     				=					self.get_scheduler_config(**_a				)
        lowercase_					:       str     				=					scheduler_class(**_a				)

        lowercase_					:       Tuple     				=					10
        lowercase_					:       List[Any]     				=					self.dummy_model()
        lowercase_					:       Any     				=					self.dummy_sample_deter
        scheduler.set_timesteps(_a				)

        for i, t in enumerate(scheduler.timesteps				):
             lowercase_					:       str     				=					model(_a       ,				_a				)
             lowercase_					:       Any     				=					scheduler.step(_a       ,				_a       ,				_a				).prev_sample

        return sample
   def 		_lowerCamelCase (self				)    ->     str:
        lowercase_					:       Optional[Any]     				=					DPMSolverSinglestepScheduler(**self.get_scheduler_config()				)
        lowercase_					:       Tuple     				=					50
        lowercase_					:       int     				=					self.dummy_model()
        lowercase_					:       Optional[int]     				=					self.dummy_sample_deter
        scheduler.set_timesteps(_a				)

        # make sure that the first t is uneven
        for i, t in enumerate(scheduler.timesteps[3:]				):
             lowercase_					:       Any     				=					model(_a       ,				_a				)
             lowercase_					:       Any     				=					scheduler.step(_a       ,				_a       ,				_a				).prev_sample

        lowercase_					:       Tuple     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.25_74				) < 1e-3
   def 		_lowerCamelCase (self				)    ->     str:
        for timesteps in [25, 50, 100, 999, 1_000]:
             self.check_over_configs(num_train_timesteps=_a				)
   def 		_lowerCamelCase (self				)    ->     str:
        # make sure that iterating over schedulers with same config names gives same results
        # for defaults
        lowercase_					:       Optional[Any]     				=					DPMSolverSinglestepScheduler(**self.get_scheduler_config()				)
        lowercase_					:       List[str]     				=					self.full_loop(scheduler=_a				)
        lowercase_					:       Union[str, Any]     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.27_91				) < 1e-3

        lowercase_					:       List[str]     				=					DEISMultistepScheduler.from_config(scheduler.config				)
        lowercase_					:       int     				=					DPMSolverMultistepScheduler.from_config(scheduler.config				)
        lowercase_					:       Dict     				=					UniPCMultistepScheduler.from_config(scheduler.config				)
        lowercase_					:       List[str]     				=					DPMSolverSinglestepScheduler.from_config(scheduler.config				)

        lowercase_					:       List[str]     				=					self.full_loop(scheduler=_a				)
        lowercase_					:       List[str]     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.27_91				) < 1e-3
   def 		_lowerCamelCase (self				)    ->     Union[str, Any]:
        self.check_over_configs(thresholding=_a				)
        for order in [1, 2, 3]:
             for solver_type in ["midpoint", "heun"]:
                  for threshold in [0.5, 1.0, 2.0]:
                       for prediction_type in ["epsilon", "sample"]:
                            self.check_over_configs(
                                thresholding=_a       ,				prediction_type=_a       ,				sample_max_value=_a       ,				algorithm_type='dpmsolver++'       ,				solver_order=_a       ,				solver_type=_a       ,				)
   def 		_lowerCamelCase (self				)    ->     List[str]:
        for prediction_type in ["epsilon", "v_prediction"]:
             self.check_over_configs(prediction_type=_a				)
   def 		_lowerCamelCase (self				)    ->     Union[str, Any]:
        for algorithm_type in ["dpmsolver", "dpmsolver++"]:
             for solver_type in ["midpoint", "heun"]:
                  for order in [1, 2, 3]:
                       for prediction_type in ["epsilon", "sample"]:
                            self.check_over_configs(
                                solver_order=_a       ,				solver_type=_a       ,				prediction_type=_a       ,				algorithm_type=_a       ,				)
                            lowercase_					:       Tuple     				=					self.full_loop(
                                solver_order=_a       ,				solver_type=_a       ,				prediction_type=_a       ,				algorithm_type=_a       ,				)
                            assert not torch.isnan(_a				).any(), "Samples have nan numbers"
   def 		_lowerCamelCase (self				)    ->     Union[str, Any]:
        self.check_over_configs(lower_order_final=_a				)
        self.check_over_configs(lower_order_final=_a				)
   def 		_lowerCamelCase (self				)    ->     Tuple:
        self.check_over_configs(lambda_min_clipped=-float('inf'				)				)
        self.check_over_configs(lambda_min_clipped=-5.1				)
   def 		_lowerCamelCase (self				)    ->     Tuple:
        self.check_over_configs(variance_type=_a				)
        self.check_over_configs(variance_type='learned_range'				)
   def 		_lowerCamelCase (self				)    ->     List[Any]:
        for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
             self.check_over_forward(num_inference_steps=_a       ,				time_step=0				)
   def 		_lowerCamelCase (self				)    ->     List[str]:
        lowercase_					:       Optional[Any]     				=					self.full_loop()
        lowercase_					:       List[str]     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.27_91				) < 1e-3
   def 		_lowerCamelCase (self				)    ->     Optional[Any]:
        lowercase_					:       Any     				=					self.full_loop(use_karras_sigmas=_a				)
        lowercase_					:       Dict     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.22_48				) < 1e-3
   def 		_lowerCamelCase (self				)    ->     str:
        lowercase_					:       List[Any]     				=					self.full_loop(prediction_type='v_prediction'				)
        lowercase_					:       Optional[Any]     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.14_53				) < 1e-3
   def 		_lowerCamelCase (self				)    ->     List[str]:
        lowercase_					:       Optional[int]     				=					self.full_loop(prediction_type='v_prediction'       ,				use_karras_sigmas=_a				)
        lowercase_					:       Optional[Any]     				=					torch.mean(torch.abs(_a				)				)

        assert abs(result_mean.item() - 0.06_49				) < 1e-3





   def 		_lowerCamelCase (self				)    ->     int:
        lowercase_					:       List[str]     				=					self.scheduler_classes[0]
        lowercase_					:       List[Any]     				=					self.get_scheduler_config(thresholding=_a       ,				dynamic_thresholding_ratio=0				)
        lowercase_					:       Optional[int]     				=					scheduler_class(**_a				)

        lowercase_					:       Tuple     				=					10
        lowercase_					:       List[str]     				=					self.dummy_model()
        lowercase_					:       int     				=					self.dummy_sample_deter.half()
        scheduler.set_timesteps(_a				)

        for i, t in enumerate(scheduler.timesteps				):
             lowercase_					:       List[str]     				=					model(_a       ,				_a				)
             lowercase_					:       Any     				=					scheduler.step(_a       ,				_a       ,				_a				).prev_sample

        assert sample.dtype == torch.floataa





'''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   (       SCREAMING_SNAKE_CASE_							):
     lowercase_					:       str     				=					botoa.client('iam'							)

     lowercase_					:       Optional[int]     				=					{
         '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=SCREAMING_SNAKE_CASE_  ,       AssumeRolePolicyDocument=json.dumps(SCREAMING_SNAKE_CASE_  ,       indent=2							)							)
          lowercase_					:       Optional[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=SCREAMING_SNAKE_CASE_  ,       PolicyName=f'''{role_name}_policy_permission'''  ,       PolicyDocument=json.dumps(SCREAMING_SNAKE_CASE_  ,       indent=2							)  ,       )
     except iam_client.exceptions.EntityAlreadyExistsException:
          print(f'''role {role_name} already exists. Using existing one'''							)

def _UpperCamelCase   (       SCREAMING_SNAKE_CASE_							):
     lowercase_					:       List[Any]     				=					botoa.client('iam'							)
     return iam_client.get_role(RoleName=SCREAMING_SNAKE_CASE_							)["Role"]["Arn"]






def _UpperCamelCase   (       ):
     lowercase_					:       Dict     				=					_ask_options(
         'How do you want to authorize?'  ,       ['AWS Profile', 'Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) ']  ,       SCREAMING_SNAKE_CASE_  ,       )
     lowercase_					:       List[Any]     				=					None
     if credentials_configuration == 0:
          lowercase_					:       List[Any]     				=					_ask_field('Enter your AWS Profile name: [default] '  ,       default='default'							)
          lowercase_					:       Optional[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`'							)
          lowercase_					:       Optional[Any]     				=					_ask_field('AWS Access Key ID: '							)
          lowercase_					:       Optional[int]     				=					aws_access_key_id

          lowercase_					:       Optional[Any]     				=					_ask_field('AWS Secret Access Key: '							)
          lowercase_					:       Optional[int]     				=					aws_secret_access_key

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

     lowercase_					:       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']  ,       SCREAMING_SNAKE_CASE_  ,       )
     if role_management == 0:
          lowercase_					:       int     				=					_ask_field('Enter your IAM role name: '							)
     else:
          lowercase_					:       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(SCREAMING_SNAKE_CASE_							)

     lowercase_					:       List[Any]     				=					_ask_field(
         'Do you want to use custom Docker image? [yes/NO]: '  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
     lowercase_					:       Optional[Any]     				=					None
     if is_custom_docker_image:
          lowercase_					:       List[Any]     				=					_ask_field('Enter your Docker image: '  ,       lambda SCREAMING_SNAKE_CASE_							: str(SCREAMING_SNAKE_CASE_							).lower()							)

     lowercase_					:       Any     				=					_ask_field(
         'Do you want to provide SageMaker input channels with data locations? [yes/NO]: '  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
     lowercase_					:       Optional[Any]     				=					None
     if is_sagemaker_inputs_enabled:
          lowercase_					:       Dict     				=					_ask_field(
              'Enter the path to the SageMaker inputs TSV file with columns (channel_name, data_location): '  ,       lambda SCREAMING_SNAKE_CASE_							: str(SCREAMING_SNAKE_CASE_							).lower()  ,       )

     lowercase_					:       Dict     				=					_ask_field(
         'Do you want to enable SageMaker metrics? [yes/NO]: '  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
     lowercase_					:       Optional[Any]     				=					None
     if is_sagemaker_metrics_enabled:
          lowercase_					:       Dict     				=					_ask_field(
              'Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): '  ,       lambda SCREAMING_SNAKE_CASE_							: str(SCREAMING_SNAKE_CASE_							).lower()  ,       )

     lowercase_					:       int     				=					_ask_options(
         'What is the distributed mode?'  ,       ['No distributed training', 'Data parallelism']  ,       _convert_sagemaker_distributed_mode  ,       )
     lowercase_					:       List[str]     				=					{}
     lowercase_					:       Optional[Any]     				=					_ask_field(
         'Do you wish to optimize your script with torch dynamo?[yes/NO]:'  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
     if use_dynamo:
          lowercase_					:       List[Any]     				=					'dynamo_'
          lowercase_					:       List[Any]     				=					_ask_options(
              'Which dynamo backend would you like to use?'  ,       [x.lower() for x in DYNAMO_BACKENDS]  ,       _convert_dynamo_backend  ,       default=2  ,       )
          lowercase_					:       int     				=					_ask_field(
              'Do you want to customize the defaults sent to torch.compile? [yes/NO]: '  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )

          if use_custom_options:
               lowercase_					:       Union[str, Any]     				=					_ask_options(
                   'Which mode do you want to use?'  ,       SCREAMING_SNAKE_CASE_  ,       lambda SCREAMING_SNAKE_CASE_							: TORCH_DYNAMO_MODES[int(SCREAMING_SNAKE_CASE_							)]  ,       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=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
               lowercase_					:       Tuple     				=					_ask_field(
                   'Do you want to enable dynamic shape tracing? [yes/NO]: '  ,       _convert_yes_no_to_bool  ,       default=SCREAMING_SNAKE_CASE_  ,       error_message='Please enter yes or no.'  ,       )
     lowercase_					:       List[str]     				=					'Which EC2 instance type you want to use for your training?'
     if distributed_type != SageMakerDistributedType.NO:
          lowercase_					:       Optional[Any]     				=					_ask_options(
              SCREAMING_SNAKE_CASE_  ,       SCREAMING_SNAKE_CASE_  ,       lambda SCREAMING_SNAKE_CASE_							: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(SCREAMING_SNAKE_CASE_							)]							)
     else:
          eca_instance_query += "? [ml.p3.2xlarge]:"
          lowercase_					:       Tuple     				=					_ask_field(SCREAMING_SNAKE_CASE_  ,       lambda SCREAMING_SNAKE_CASE_							: str(SCREAMING_SNAKE_CASE_							).lower()  ,       default='ml.p3.2xlarge'							)

     lowercase_					:       str     				=					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]: '  ,       SCREAMING_SNAKE_CASE_  ,       default=1  ,       )

     lowercase_					:       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=SCREAMING_SNAKE_CASE_  ,       compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER  ,       distributed_type=SCREAMING_SNAKE_CASE_  ,       use_cpu=SCREAMING_SNAKE_CASE_  ,       dynamo_config=SCREAMING_SNAKE_CASE_  ,       eca_instance_type=SCREAMING_SNAKE_CASE_  ,       profile=SCREAMING_SNAKE_CASE_  ,       region=SCREAMING_SNAKE_CASE_  ,       iam_role_name=SCREAMING_SNAKE_CASE_  ,       mixed_precision=SCREAMING_SNAKE_CASE_  ,       num_machines=SCREAMING_SNAKE_CASE_  ,       sagemaker_inputs_file=SCREAMING_SNAKE_CASE_  ,       sagemaker_metrics_file=SCREAMING_SNAKE_CASE_  ,       )




