


# 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 argparse

from ...utils.dataclasses import (
    ComputeEnvironment,
    DistributedType,
    DynamoBackend,
    PrecisionType,
    SageMakerDistributedType,
)
from ..menu import BulletMenu


lowerCamelCase		   =			[
    'EAGER',
    'AOT_EAGER',
    'INDUCTOR',
    'NVFUSER',
    'AOT_NVFUSER',
    'AOT_CUDAGRAPHS',
    'OFI',
    'FX2TRT',
    'ONNXRT',
    'IPEX',
]




def       a_ (       SCREAMING_SNAKE_CASE__   : Optional[Any]						,     SCREAMING_SNAKE_CASE__   : List[str]=None						,     SCREAMING_SNAKE_CASE__   : Optional[Any]=None						,     SCREAMING_SNAKE_CASE__   : Any=None	):
       '''simple docstring'''





       _lowerCamelCase			:						Union[str, Any]            =True
       while ask_again:
              _lowerCamelCase			:						List[Any]            =input(SCREAMING_SNAKE_CASE__	)
              try:
                     if default is not None and len(SCREAMING_SNAKE_CASE__	) == 0:
                            return default
                     return convert_value(SCREAMING_SNAKE_CASE__	) if convert_value is not None else result
              except Exception:
                     if error_message is not None:
                            print(SCREAMING_SNAKE_CASE__	)




def       a_ (       SCREAMING_SNAKE_CASE__   : str						,     SCREAMING_SNAKE_CASE__   : List[str]=[]						,     SCREAMING_SNAKE_CASE__   : Any=None						,     SCREAMING_SNAKE_CASE__   : Union[str, Any]=0	):
       '''simple docstring'''





       _lowerCamelCase			:						Optional[Any]            =BulletMenu(SCREAMING_SNAKE_CASE__						,     SCREAMING_SNAKE_CASE__	)
       _lowerCamelCase			:						List[str]            =menu.run(default_choice=SCREAMING_SNAKE_CASE__	)
       return convert_value(SCREAMING_SNAKE_CASE__	) if convert_value is not None else result




def       a_ (       SCREAMING_SNAKE_CASE__   : Any	):
       '''simple docstring'''





       _lowerCamelCase			:						List[str]            =int(SCREAMING_SNAKE_CASE__	)
       return ComputeEnvironment(['LOCAL_MACHINE', 'AMAZON_SAGEMAKER'][value]	)




def       a_ (       SCREAMING_SNAKE_CASE__   : Union[str, Any]	):
       '''simple docstring'''





       _lowerCamelCase			:						int            =int(SCREAMING_SNAKE_CASE__	)
       return DistributedType(['NO', 'MULTI_CPU', 'MULTI_XPU', 'MULTI_GPU', 'MULTI_NPU', 'TPU'][value]	)




def       a_ (       SCREAMING_SNAKE_CASE__   : Tuple	):
       '''simple docstring'''





       _lowerCamelCase			:						List[str]            =int(SCREAMING_SNAKE_CASE__	)
       return DynamoBackend(DYNAMO_BACKENDS[value]	).value




def       a_ (       SCREAMING_SNAKE_CASE__   : Tuple	):
       '''simple docstring'''





       _lowerCamelCase			:						Dict            =int(SCREAMING_SNAKE_CASE__	)
       return PrecisionType(['no', 'fp16', 'bf16', 'fp8'][value]	)




def       a_ (       SCREAMING_SNAKE_CASE__   : List[str]	):
       '''simple docstring'''





       _lowerCamelCase			:						Union[str, Any]            =int(SCREAMING_SNAKE_CASE__	)
       return SageMakerDistributedType(['NO', 'DATA_PARALLEL', 'MODEL_PARALLEL'][value]	)




def       a_ (       SCREAMING_SNAKE_CASE__   : Tuple	):
       '''simple docstring'''





       return {"yes": True, "no": False}[value.lower()]




class A    (  argparse.RawDescriptionHelpFormatter	):






      def 			lowerCamelCase			(   self		:	str     ,  lowercase_		:	List[Any]     ,  lowercase_		:	int     ,  lowercase_		:	Optional[Any]     ,  lowercase_		:	Optional[int]					)   ->		Dict:
             """simple docstring"""




             _lowerCamelCase			:						str            =super()._format_usage(lowercase_     ,  lowercase_     ,  lowercase_     ,  lowercase_					)
             _lowerCamelCase			:						Dict            =usage.replace('<command> [<args>] '     ,  ''					)
             return usage






import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest




class A    (  UpperCamelCase_	):
      UpperCamelCase__				:    List[str] =(PNDMScheduler,)
      UpperCamelCase__				:    Dict =(('num_inference_steps', 50),)






      def 			lowerCamelCase			(   self		:	Dict     ,  **lowercase_		:	Dict					)   ->		str:
             """simple docstring"""




             _lowerCamelCase			:						List[Any]            ={
                 'num_train_timesteps': 1000,
                 'beta_start': 0.0001,
                 'beta_end': 0.02,
                 'beta_schedule': 'linear',
             }

             config.update(**lowercase_					)
             return config






      def 			lowerCamelCase			(   self		:	Any     ,  lowercase_		:	str=0     ,  **lowercase_		:	str					)   ->		Optional[int]:
             """simple docstring"""




             _lowerCamelCase			:						Optional[Any]            =dict(self.forward_default_kwargs					)
             _lowerCamelCase			:						Optional[Any]            =kwargs.pop('num_inference_steps'     ,  lowercase_					)
             _lowerCamelCase			:						Tuple            =self.dummy_sample
             _lowerCamelCase			:						int            =0.1 * sample
             _lowerCamelCase			:						int            =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

             for scheduler_class in self.scheduler_classes:
                    _lowerCamelCase			:						str            =self.get_scheduler_config(**lowercase_					)
                    _lowerCamelCase			:						str            =scheduler_class(**lowercase_					)
                    scheduler.set_timesteps(lowercase_					)
                    # copy over dummy past residuals
                    _lowerCamelCase			:						Any            =dummy_past_residuals[:]

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

                    _lowerCamelCase			:						Union[str, Any]            =scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						List[str]            =new_scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample

                    assert torch.sum(torch.abs(output - new_output					)					) < 1E-5, "Scheduler outputs are not identical"

                    _lowerCamelCase			:						Optional[int]            =scheduler.step_plms(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						Optional[Any]            =new_scheduler.step_plms(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample

                    assert torch.sum(torch.abs(output - new_output					)					) < 1E-5, "Scheduler outputs are not identical"






      def 			lowerCamelCase			(   self		:	Any					)   ->		Any:
             """simple docstring"""




             pass






      def 			lowerCamelCase			(   self		:	Union[str, Any]     ,  lowercase_		:	List[str]=0     ,  **lowercase_		:	int					)   ->		int:
             """simple docstring"""




             _lowerCamelCase			:						Any            =dict(self.forward_default_kwargs					)
             _lowerCamelCase			:						Dict            =kwargs.pop('num_inference_steps'     ,  lowercase_					)
             _lowerCamelCase			:						Union[str, Any]            =self.dummy_sample
             _lowerCamelCase			:						Optional[int]            =0.1 * sample
             _lowerCamelCase			:						List[str]            =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

             for scheduler_class in self.scheduler_classes:
                    _lowerCamelCase			:						Any            =self.get_scheduler_config()
                    _lowerCamelCase			:						int            =scheduler_class(**lowercase_					)
                    scheduler.set_timesteps(lowercase_					)

                    # copy over dummy past residuals (must be after setting timesteps)
                    _lowerCamelCase			:						Any            =dummy_past_residuals[:]

                    with tempfile.TemporaryDirectory() as tmpdirname:
                           scheduler.save_config(lowercase_					)
                           _lowerCamelCase			:						Optional[Any]            =scheduler_class.from_pretrained(lowercase_					)
                           # copy over dummy past residuals
                           new_scheduler.set_timesteps(lowercase_					)

                           # copy over dummy past residual (must be after setting timesteps)
                           _lowerCamelCase			:						Optional[int]            =dummy_past_residuals[:]

                    _lowerCamelCase			:						List[str]            =scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						str            =new_scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample

                    assert torch.sum(torch.abs(output - new_output					)					) < 1E-5, "Scheduler outputs are not identical"

                    _lowerCamelCase			:						Optional[Any]            =scheduler.step_plms(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						List[str]            =new_scheduler.step_plms(lowercase_     ,  lowercase_     ,  lowercase_     ,  **lowercase_					).prev_sample

                    assert torch.sum(torch.abs(output - new_output					)					) < 1E-5, "Scheduler outputs are not identical"






      def 			lowerCamelCase			(   self		:	Optional[Any]     ,  **lowercase_		:	Any					)   ->		Tuple:
             """simple docstring"""




             _lowerCamelCase			:						Union[str, Any]            =self.scheduler_classes[0]
             _lowerCamelCase			:						Optional[Any]            =self.get_scheduler_config(**lowercase_					)
             _lowerCamelCase			:						Any            =scheduler_class(**lowercase_					)

             _lowerCamelCase			:						Union[str, Any]            =10
             _lowerCamelCase			:						str            =self.dummy_model()
             _lowerCamelCase			:						Union[str, Any]            =self.dummy_sample_deter
             scheduler.set_timesteps(lowercase_					)

             for i, t in enumerate(scheduler.prk_timesteps					):
                    _lowerCamelCase			:						Union[str, Any]            =model(lowercase_     ,  lowercase_					)
                    _lowerCamelCase			:						Any            =scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_					).prev_sample

             for i, t in enumerate(scheduler.plms_timesteps					):
                    _lowerCamelCase			:						List[Any]            =model(lowercase_     ,  lowercase_					)
                    _lowerCamelCase			:						int            =scheduler.step_plms(lowercase_     ,  lowercase_     ,  lowercase_					).prev_sample

             return sample






      def 			lowerCamelCase			(   self		:	Dict					)   ->		List[Any]:
             """simple docstring"""




             _lowerCamelCase			:						Any            =dict(self.forward_default_kwargs					)

             _lowerCamelCase			:						Optional[int]            =kwargs.pop('num_inference_steps'     ,  lowercase_					)

             for scheduler_class in self.scheduler_classes:
                    _lowerCamelCase			:						List[str]            =self.get_scheduler_config()
                    _lowerCamelCase			:						List[Any]            =scheduler_class(**lowercase_					)

                    _lowerCamelCase			:						Union[str, Any]            =self.dummy_sample
                    _lowerCamelCase			:						Optional[int]            =0.1 * sample

                    if num_inference_steps is not None and hasattr(lowercase_     ,  'set_timesteps'					):
                           scheduler.set_timesteps(lowercase_					)
                    elif num_inference_steps is not None and not hasattr(lowercase_     ,  'set_timesteps'					):
                           _lowerCamelCase			:						Tuple            =num_inference_steps

                    # copy over dummy past residuals (must be done after set_timesteps)
                    _lowerCamelCase			:						List[Any]            =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
                    _lowerCamelCase			:						Union[str, Any]            =dummy_past_residuals[:]

                    _lowerCamelCase			:						List[Any]            =scheduler.step_prk(lowercase_     ,  0     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						str            =scheduler.step_prk(lowercase_     ,  1     ,  lowercase_     ,  **lowercase_					).prev_sample

                    self.assertEqual(output_a.shape     ,  sample.shape					)
                    self.assertEqual(output_a.shape     ,  output_a.shape					)

                    _lowerCamelCase			:						Tuple            =scheduler.step_plms(lowercase_     ,  0     ,  lowercase_     ,  **lowercase_					).prev_sample
                    _lowerCamelCase			:						Optional[Any]            =scheduler.step_plms(lowercase_     ,  1     ,  lowercase_     ,  **lowercase_					).prev_sample

                    self.assertEqual(output_a.shape     ,  sample.shape					)
                    self.assertEqual(output_a.shape     ,  output_a.shape					)






      def 			lowerCamelCase			(   self		:	Dict					)   ->		Any:
             """simple docstring"""




             for timesteps in [100, 1000]:
                    self.check_over_configs(num_train_timesteps=lowercase_					)






      def 			lowerCamelCase			(   self		:	Optional[int]					)   ->		Optional[Any]:
             """simple docstring"""




             for steps_offset in [0, 1]:
                    self.check_over_configs(steps_offset=lowercase_					)

             _lowerCamelCase			:						Optional[int]            =self.scheduler_classes[0]
             _lowerCamelCase			:						Dict            =self.get_scheduler_config(steps_offset=1					)
             _lowerCamelCase			:						List[Any]            =scheduler_class(**lowercase_					)
             scheduler.set_timesteps(10					)
             assert torch.equal(
                 scheduler.timesteps     ,  torch.LongTensor(
                     [901, 851, 851, 801, 801, 751, 751, 701, 701, 651, 651, 601, 601, 501, 401, 301, 201, 101, 1]					)     ,  )






      def 			lowerCamelCase			(   self		:	List[str]					)   ->		str:
             """simple docstring"""




             for beta_start, beta_end in zip([0.0001, 0.001]     ,  [0.002, 0.02]					):
                    self.check_over_configs(beta_start=lowercase_     ,  beta_end=lowercase_					)






      def 			lowerCamelCase			(   self		:	Tuple					)   ->		Union[str, Any]:
             """simple docstring"""




             for schedule in ["linear", "squaredcos_cap_v2"]:
                    self.check_over_configs(beta_schedule=lowercase_					)






      def 			lowerCamelCase			(   self		:	Tuple					)   ->		Any:
             """simple docstring"""




             for prediction_type in ["epsilon", "v_prediction"]:
                    self.check_over_configs(prediction_type=lowercase_					)






      def 			lowerCamelCase			(   self		:	Optional[Any]					)   ->		Union[str, Any]:
             """simple docstring"""




             for t in [1, 5, 10]:
                    self.check_over_forward(time_step=lowercase_					)






      def 			lowerCamelCase			(   self		:	List[str]					)   ->		List[Any]:
             """simple docstring"""




             for t, num_inference_steps in zip([1, 5, 10]     ,  [10, 50, 100]					):
                    self.check_over_forward(num_inference_steps=lowercase_					)






      def 			lowerCamelCase			(   self		:	int					)   ->		Optional[Any]:
             """simple docstring"""




             _lowerCamelCase			:						Optional[int]            =27

             for scheduler_class in self.scheduler_classes:
                    _lowerCamelCase			:						Dict            =self.dummy_sample
                    _lowerCamelCase			:						List[Any]            =0.1 * sample

                    _lowerCamelCase			:						List[Any]            =self.get_scheduler_config()
                    _lowerCamelCase			:						List[str]            =scheduler_class(**lowercase_					)

                    scheduler.set_timesteps(lowercase_					)

                    # before power of 3 fix, would error on first step, so we only need to do two
                    for i, t in enumerate(scheduler.prk_timesteps[:2]					):
                           _lowerCamelCase			:						Tuple            =scheduler.step_prk(lowercase_     ,  lowercase_     ,  lowercase_					).prev_sample






      def 			lowerCamelCase			(   self		:	List[str]					)   ->		Optional[int]:
             """simple docstring"""




             with self.assertRaises(lowercase_					):
                    _lowerCamelCase			:						Dict            =self.scheduler_classes[0]
                    _lowerCamelCase			:						Optional[Any]            =self.get_scheduler_config()
                    _lowerCamelCase			:						List[Any]            =scheduler_class(**lowercase_					)

                    scheduler.step_plms(self.dummy_sample     ,  1     ,  self.dummy_sample					).prev_sample






      def 			lowerCamelCase			(   self		:	List[str]					)   ->		List[str]:
             """simple docstring"""




             _lowerCamelCase			:						Tuple            =self.full_loop()
             _lowerCamelCase			:						Optional[Any]            =torch.sum(torch.abs(lowercase_					)					)
             _lowerCamelCase			:						Union[str, Any]            =torch.mean(torch.abs(lowercase_					)					)

             assert abs(result_sum.item() - 198.1318					) < 1E-2
             assert abs(result_mean.item() - 0.2580					) < 1E-3






      def 			lowerCamelCase			(   self		:	Any					)   ->		Optional[int]:
             """simple docstring"""




             _lowerCamelCase			:						int            =self.full_loop(prediction_type='v_prediction'					)
             _lowerCamelCase			:						Dict            =torch.sum(torch.abs(lowercase_					)					)
             _lowerCamelCase			:						str            =torch.mean(torch.abs(lowercase_					)					)

             assert abs(result_sum.item() - 67.3986					) < 1E-2
             assert abs(result_mean.item() - 0.0878					) < 1E-3






      def 			lowerCamelCase			(   self		:	Tuple					)   ->		Optional[Any]:
             """simple docstring"""




             _lowerCamelCase			:						Tuple            =self.full_loop(set_alpha_to_one=lowercase_     ,  beta_start=0.01					)
             _lowerCamelCase			:						Optional[Any]            =torch.sum(torch.abs(lowercase_					)					)
             _lowerCamelCase			:						Union[str, Any]            =torch.mean(torch.abs(lowercase_					)					)

             assert abs(result_sum.item() - 230.0399					) < 1E-2
             assert abs(result_mean.item() - 0.2995					) < 1E-3






      def 			lowerCamelCase			(   self		:	Optional[Any]					)   ->		Any:
             """simple docstring"""




             _lowerCamelCase			:						str            =self.full_loop(set_alpha_to_one=lowercase_     ,  beta_start=0.01					)
             _lowerCamelCase			:						Union[str, Any]            =torch.sum(torch.abs(lowercase_					)					)
             _lowerCamelCase			:						str            =torch.mean(torch.abs(lowercase_					)					)

             assert abs(result_sum.item() - 186.9482					) < 1E-2
             assert abs(result_mean.item() - 0.2434					) < 1E-3


