




'''simple docstring'''

import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest





class __magic_name__(						__lowerCAmelCase      ):
						UpperCAmelCase_		:					Optional[int]    =  (DPMSolverMultistepScheduler,)
						UpperCAmelCase_		:					Tuple    =  (("""num_inference_steps""", 25),)






						def       __lowerCAmelCase(   self						:						Optional[Any]					,   **__UpperCamelCase						:						Any       ):



									'''simple docstring'''






									snake_case__          =						{
									    """num_train_timesteps""": 1_0_0_0,
									    """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""",
									    """lower_order_final""": False,
									    """lambda_min_clipped""": -float("""inf"""       ),
									    """variance_type""": None,
									}

									config.update(**__UpperCamelCase       )
									return config






						def       __lowerCAmelCase(   self						:						Optional[Any]					,   __UpperCamelCase						:						Union[str, Any]=0					,   **__UpperCamelCase						:						Tuple       ):



									'''simple docstring'''






									snake_case__          =						dict(self.forward_default_kwargs       )
									snake_case__          =						kwargs.pop("""num_inference_steps"""					,   __UpperCamelCase       )
									snake_case__          =						self.dummy_sample
									snake_case__          =						0.1 * sample
									snake_case__          =						[residual + 0.2, residual + 0.15, residual + 0.10]

									for scheduler_class in self.scheduler_classes:
												snake_case__          =						self.get_scheduler_config(**__UpperCamelCase       )
												snake_case__          =						scheduler_class(**__UpperCamelCase       )
												scheduler.set_timesteps(__UpperCamelCase       )
												# copy over dummy past residuals
												snake_case__          =						dummy_past_residuals[: scheduler.config.solver_order]

												with tempfile.TemporaryDirectory() as tmpdirname:
															scheduler.save_config(__UpperCamelCase       )
															snake_case__          =						scheduler_class.from_pretrained(__UpperCamelCase       )
															new_scheduler.set_timesteps(__UpperCamelCase       )
															# copy over dummy past residuals
															snake_case__          =						dummy_past_residuals[: new_scheduler.config.solver_order]

												snake_case__    ,			snake_case__          =						sample, sample
												for t in range(__UpperCamelCase					,   time_step + scheduler.config.solver_order + 1       ):
															snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample
															snake_case__          =						new_scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample

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






						def       __lowerCAmelCase(   self						:						int       ):



									'''simple docstring'''






									pass






						def       __lowerCAmelCase(   self						:						Dict					,   __UpperCamelCase						:						Tuple=0					,   **__UpperCamelCase						:						Optional[int]       ):



									'''simple docstring'''






									snake_case__          =						dict(self.forward_default_kwargs       )
									snake_case__          =						kwargs.pop("""num_inference_steps"""					,   __UpperCamelCase       )
									snake_case__          =						self.dummy_sample
									snake_case__          =						0.1 * sample
									snake_case__          =						[residual + 0.2, residual + 0.15, residual + 0.10]

									for scheduler_class in self.scheduler_classes:
												snake_case__          =						self.get_scheduler_config()
												snake_case__          =						scheduler_class(**__UpperCamelCase       )
												scheduler.set_timesteps(__UpperCamelCase       )

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

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

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

												snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample
												snake_case__          =						new_scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample

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






						def       __lowerCAmelCase(   self						:						Dict					,   __UpperCamelCase						:						List[str]=None					,   **__UpperCamelCase						:						List[Any]       ):



									'''simple docstring'''






									if scheduler is None:
												snake_case__          =						self.scheduler_classes[0]
												snake_case__          =						self.get_scheduler_config(**__UpperCamelCase       )
												snake_case__          =						scheduler_class(**__UpperCamelCase       )

									snake_case__          =						1_0
									snake_case__          =						self.dummy_model()
									snake_case__          =						self.dummy_sample_deter
									scheduler.set_timesteps(__UpperCamelCase       )

									for i, t in enumerate(scheduler.timesteps       ):
												snake_case__          =						model(__UpperCamelCase					,   __UpperCamelCase       )
												snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase       ).prev_sample

									return sample






						def       __lowerCAmelCase(   self						:						str       ):



									'''simple docstring'''






									snake_case__          =						dict(self.forward_default_kwargs       )

									snake_case__          =						kwargs.pop("""num_inference_steps"""					,   __UpperCamelCase       )

									for scheduler_class in self.scheduler_classes:
												snake_case__          =						self.get_scheduler_config()
												snake_case__          =						scheduler_class(**__UpperCamelCase       )

												snake_case__          =						self.dummy_sample
												snake_case__          =						0.1 * sample

												if num_inference_steps is not None and hasattr(__UpperCamelCase					,   """set_timesteps"""       ):
															scheduler.set_timesteps(__UpperCamelCase       )
												elif num_inference_steps is not None and not hasattr(__UpperCamelCase					,   """set_timesteps"""       ):
															snake_case__          =						num_inference_steps

												# copy over dummy past residuals (must be done after set_timesteps)
												snake_case__          =						[residual + 0.2, residual + 0.15, residual + 0.10]
												snake_case__          =						dummy_past_residuals[: scheduler.config.solver_order]

												snake_case__          =						scheduler.timesteps[5]
												snake_case__          =						scheduler.timesteps[6]

												snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample
												snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase					,   **__UpperCamelCase       ).prev_sample

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






						def       __lowerCAmelCase(   self						:						Optional[int]       ):



									'''simple docstring'''






									for timesteps in [2_5, 5_0, 1_0_0, 9_9_9, 1_0_0_0]:
												self.check_over_configs(num_train_timesteps=__UpperCamelCase       )






						def       __lowerCAmelCase(   self						:						str       ):



									'''simple docstring'''






									self.check_over_configs(thresholding=__UpperCamelCase       )
									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=__UpperCamelCase					,   prediction_type=__UpperCamelCase					,   sample_max_value=__UpperCamelCase					,   algorithm_type="""dpmsolver++"""					,   solver_order=__UpperCamelCase					,   solver_type=__UpperCamelCase					,   )






						def       __lowerCAmelCase(   self						:						Optional[Any]       ):



									'''simple docstring'''






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






						def       __lowerCAmelCase(   self						:						int       ):



									'''simple docstring'''






									for algorithm_type in ["dpmsolver", "dpmsolver++", "sde-dpmsolver", "sde-dpmsolver++"]:
												for solver_type in ["midpoint", "heun"]:
															for order in [1, 2, 3]:
																		for prediction_type in ["epsilon", "sample"]:
																					if algorithm_type in ["sde-dpmsolver", "sde-dpmsolver++"]:
																								if order == 3:
																											continue
																					else:
																								self.check_over_configs(
																								    solver_order=__UpperCamelCase					,   solver_type=__UpperCamelCase					,   prediction_type=__UpperCamelCase					,   algorithm_type=__UpperCamelCase					,   )
																					snake_case__          =						self.full_loop(
																					    solver_order=__UpperCamelCase					,   solver_type=__UpperCamelCase					,   prediction_type=__UpperCamelCase					,   algorithm_type=__UpperCamelCase					,   )
																					assert not torch.isnan(__UpperCamelCase       ).any(), "Samples have nan numbers"






						def       __lowerCAmelCase(   self						:						int       ):



									'''simple docstring'''






									self.check_over_configs(lower_order_final=__UpperCamelCase       )
									self.check_over_configs(lower_order_final=__UpperCamelCase       )






						def       __lowerCAmelCase(   self						:						Union[str, Any]       ):



									'''simple docstring'''






									self.check_over_configs(lambda_min_clipped=-float("""inf"""       )       )
									self.check_over_configs(lambda_min_clipped=-5.1       )






						def       __lowerCAmelCase(   self						:						str       ):



									'''simple docstring'''






									self.check_over_configs(variance_type=__UpperCamelCase       )
									self.check_over_configs(variance_type="""learned_range"""       )






						def       __lowerCAmelCase(   self						:						Any       ):



									'''simple docstring'''






									for num_inference_steps in [1, 2, 3, 5, 1_0, 5_0, 1_0_0, 9_9_9, 1_0_0_0]:
												self.check_over_forward(num_inference_steps=__UpperCamelCase					,   time_step=0       )






						def       __lowerCAmelCase(   self						:						List[Any]       ):



									'''simple docstring'''






									snake_case__          =						self.full_loop()
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 0.33_01       ) < 1E-3






						def       __lowerCAmelCase(   self						:						Dict       ):



									'''simple docstring'''






									snake_case__          =						self.full_loop(thresholding=__UpperCamelCase					,   dynamic_thresholding_ratio=0.87					,   sample_max_value=0.5       )
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 1.13_64       ) < 1E-3






						def       __lowerCAmelCase(   self						:						Any       ):



									'''simple docstring'''






									snake_case__          =						self.full_loop(prediction_type="""v_prediction"""       )
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 0.22_51       ) < 1E-3






						def       __lowerCAmelCase(   self						:						Tuple       ):



									'''simple docstring'''






									snake_case__          =						self.full_loop(prediction_type="""v_prediction"""					,   use_karras_sigmas=__UpperCamelCase       )
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 0.20_96       ) < 1E-3






						def       __lowerCAmelCase(   self						:						Tuple       ):



									'''simple docstring'''






									snake_case__          =						DPMSolverMultistepScheduler(**self.get_scheduler_config()       )
									snake_case__          =						self.full_loop(scheduler=__UpperCamelCase       )
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 0.33_01       ) < 1E-3

									snake_case__          =						DPMSolverSinglestepScheduler.from_config(scheduler.config       )
									snake_case__          =						UniPCMultistepScheduler.from_config(scheduler.config       )
									snake_case__          =						DEISMultistepScheduler.from_config(scheduler.config       )
									snake_case__          =						DPMSolverMultistepScheduler.from_config(scheduler.config       )

									snake_case__          =						self.full_loop(scheduler=__UpperCamelCase       )
									snake_case__          =						torch.mean(torch.abs(__UpperCamelCase       )       )

									assert abs(result_mean.item() - 0.33_01       ) < 1E-3






						def       __lowerCAmelCase(   self						:						Any       ):



									'''simple docstring'''






									snake_case__          =						self.scheduler_classes[0]
									snake_case__          =						self.get_scheduler_config(thresholding=__UpperCamelCase					,   dynamic_thresholding_ratio=0       )
									snake_case__          =						scheduler_class(**__UpperCamelCase       )

									snake_case__          =						1_0
									snake_case__          =						self.dummy_model()
									snake_case__          =						self.dummy_sample_deter.half()
									scheduler.set_timesteps(__UpperCamelCase       )

									for i, t in enumerate(scheduler.timesteps       ):
												snake_case__          =						model(__UpperCamelCase					,   __UpperCamelCase       )
												snake_case__          =						scheduler.step(__UpperCamelCase					,   __UpperCamelCase					,   __UpperCamelCase       ).prev_sample

									assert sample.dtype == torch.floataa







						def       __lowerCAmelCase(   self						:						List[str]					,   **__UpperCamelCase						:						Optional[int]       ):



									'''simple docstring'''






									for scheduler_class in self.scheduler_classes:
												snake_case__          =						self.get_scheduler_config(**__UpperCamelCase       )
												snake_case__          =						scheduler_class(**__UpperCamelCase       )

												scheduler.set_timesteps(scheduler.config.num_train_timesteps       )
												assert len(scheduler.timesteps.unique()       ) == scheduler.num_inference_steps







import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

from tqdm import auto as tqdm_lib


lowerCamelCase					    =	threading.Lock()
lowerCamelCase					    =	None

lowerCamelCase					    =	{
    '''debug''': logging.DEBUG,
    '''info''': logging.INFO,
    '''warning''': logging.WARNING,
    '''error''': logging.ERROR,
    '''critical''': logging.CRITICAL,
}

lowerCamelCase					    =	logging.WARNING

lowerCamelCase					    =	True
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _lowercase     			=						os.getenv('DIFFUSERS_VERBOSITY',   _lowercase   )
    if env_level_str:
        if env_level_str in log_levels:
            return log_levels[env_level_str]
        else:
            logging.getLogger().warning(
                f'''Unknown option DIFFUSERS_VERBOSITY={env_level_str}, '''
                f'''has to be one of: { ", ".join(log_levels.keys()   ) }'''   )
    return _default_log_level
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return __name__.split('.'   )[0]
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return logging.getLogger(_get_library_name()   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    global _default_handler

    with _lock:
        if _default_handler:
            # This library has already configured the library root logger.
            return
        _lowercase     			=						logging.StreamHandler()  # Set sys.stderr as stream.
        _lowercase     			=						sys.stderr.flush

        # Apply our default configuration to the library root logger.
        _lowercase     			=						_get_library_root_logger()
        library_root_logger.addHandler(_default_handler   )
        library_root_logger.setLevel(_get_default_logging_level()   )
        _lowercase     			=						False
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    global _default_handler

    with _lock:
        if not _default_handler:
            return

        _lowercase     			=						_get_library_root_logger()
        library_root_logger.removeHandler(_default_handler   )
        library_root_logger.setLevel(logging.NOTSET   )
        _lowercase     			=						None
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return log_levels
def 							UpperCAmelCase   ( _lowercase = None   ):


    '''simple docstring'''



    if name is None:
        _lowercase     			=						_get_library_name()

    _configure_library_root_logger()
    return logging.getLogger(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _configure_library_root_logger()
    return _get_library_root_logger().getEffectiveLevel()
def 							UpperCAmelCase   ( _lowercase   ):


    '''simple docstring'''



    _configure_library_root_logger()
    _get_library_root_logger().setLevel(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return set_verbosity(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return set_verbosity(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return set_verbosity(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    return set_verbosity(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _configure_library_root_logger()

    assert _default_handler is not None
    _get_library_root_logger().removeHandler(_default_handler   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _configure_library_root_logger()

    assert _default_handler is not None
    _get_library_root_logger().addHandler(_default_handler   )
def 							UpperCAmelCase   ( _lowercase   ):


    '''simple docstring'''



    _configure_library_root_logger()

    assert handler is not None
    _get_library_root_logger().addHandler(_lowercase   )
def 							UpperCAmelCase   ( _lowercase   ):


    '''simple docstring'''



    _configure_library_root_logger()

    assert handler is not None and handler not in _get_library_root_logger().handlers
    _get_library_root_logger().removeHandler(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _configure_library_root_logger()
    _lowercase     			=						False
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _configure_library_root_logger()
    _lowercase     			=						True
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _lowercase     			=						_get_library_root_logger().handlers

    for handler in handlers:
        _lowercase     			=						logging.Formatter('[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'   )
        handler.setFormatter(_lowercase   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    _lowercase     			=						_get_library_root_logger().handlers

    for handler in handlers:
        handler.setFormatter(_lowercase   )
def 							UpperCAmelCase   ( self,   *_lowercase,   **_lowercase   ):


    '''simple docstring'''



    _lowercase     			=						os.getenv('DIFFUSERS_NO_ADVISORY_WARNINGS',   _lowercase   )
    if no_advisory_warnings:
        return
    self.warning(*_lowercase,   **_lowercase   )


lowerCamelCase					    =	warning_advice

class        snake_case_				:







   """simple docstring"""



   def __init__(     self:						int ,							*snake_case__:						str ,							**snake_case__:						Any						)	->		Any:  # pylint: disable=unused-argument


       """simple docstring"""



       _lowercase     			=						args[0] if args else None

   def __iter__(     self:						Union[str, Any]						)	->		str:


       """simple docstring"""



       return iter(self._iterator						)

   def __getattr__(     self:						int ,							snake_case__:						Any						)	->		List[Any]:


       """simple docstring"""



       def empty_fn(*snake_case__:						Optional[int] ,							**snake_case__:						Union[str, Any]						):  # pylint: disable=unused-argument
           return

       return empty_fn

   def __enter__(     self:						List[Any]						)	->		Tuple:


       """simple docstring"""



       return self

   def __exit__(     self:						List[str] ,							snake_case__:						Optional[Any] ,							snake_case__:						Dict ,							snake_case__:						List[str]						)	->		Optional[Any]:


       """simple docstring"""



       return






class        snake_case_				:







   """simple docstring"""



   def __call__(     self:						List[str] ,							*snake_case__:						Tuple ,							**snake_case__:						Union[str, Any]						)	->		List[str]:


       """simple docstring"""



       if _tqdm_active:
           return tqdm_lib.tqdm(*snake_case__ ,							**snake_case__						)
       else:
           return EmptyTqdm(*snake_case__ ,							**snake_case__						)

   def       SCREAMING_SNAKE_CASE__					(     self:						Tuple ,							*snake_case__:						Union[str, Any] ,							**snake_case__:						str						)	->		Optional[int]:


       """simple docstring"""



       _lowercase     			=						None
       if _tqdm_active:
           return tqdm_lib.tqdm.set_lock(*snake_case__ ,							**snake_case__						)

   def       SCREAMING_SNAKE_CASE__					(     self:						int						)	->		int:


       """simple docstring"""



       if _tqdm_active:
           return tqdm_lib.tqdm.get_lock()






lowerCamelCase					    =	_tqdm_cls()
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    global _tqdm_active
    return bool(_tqdm_active   )
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    global _tqdm_active
    _lowercase     			=						True
def 							UpperCAmelCase   ( ):


    '''simple docstring'''



    global _tqdm_active
    _lowercase     			=						False
