


"""simple docstring"""






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


__snake_case												=							threading.Lock()
__snake_case												=							None

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

__snake_case												=							logging.WARNING

__snake_case												=							True
def    A_ ( ):






     """simple docstring"""

     _a     						=					os.getenv('''DIFFUSERS_VERBOSITY''',  _lowerCAmelCase							)
     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    A_ ( ):






     """simple docstring"""

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






     """simple docstring"""

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






     """simple docstring"""

     global _default_handler

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

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






     """simple docstring"""

     global _default_handler

     with _lock:
          if not _default_handler:
               return

          _a     						=					_get_library_root_logger()
          library_root_logger.removeHandler(_default_handler							)
          library_root_logger.setLevel(logging.NOTSET							)
          _a     						=					None
def    A_ ( ):






     """simple docstring"""

     return log_levels
def    A_ ( _lowerCAmelCase  :       Optional[str] = None							):






     """simple docstring"""

     if name is None:
          _a     						=					_get_library_name()

     _configure_library_root_logger()
     return logging.getLogger(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     _configure_library_root_logger()
     return _get_library_root_logger().getEffectiveLevel()
def    A_ ( _lowerCAmelCase  :       int							):






     """simple docstring"""

     _configure_library_root_logger()
     _get_library_root_logger().setLevel(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     return set_verbosity(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     return set_verbosity(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     return set_verbosity(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     return set_verbosity(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     _configure_library_root_logger()

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






     """simple docstring"""

     _configure_library_root_logger()

     assert _default_handler is not None
     _get_library_root_logger().addHandler(_default_handler							)
def    A_ ( _lowerCAmelCase  :       logging.Handler							):






     """simple docstring"""

     _configure_library_root_logger()

     assert handler is not None
     _get_library_root_logger().addHandler(_lowerCAmelCase							)
def    A_ ( _lowerCAmelCase  :       logging.Handler							):






     """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(_lowerCAmelCase							)
def    A_ ( ):






     """simple docstring"""

     _configure_library_root_logger()
     _a     						=					False
def    A_ ( ):






     """simple docstring"""

     _configure_library_root_logger()
     _a     						=					True
def    A_ ( ):






     """simple docstring"""

     _a     						=					_get_library_root_logger().handlers

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






     """simple docstring"""

     _a     						=					_get_library_root_logger().handlers

     for handler in handlers:
          handler.setFormatter(_lowerCAmelCase							)
def    A_ ( self  :       Union[str, Any],  *_lowerCAmelCase  :       Any,  **_lowerCAmelCase  :       str							):






     """simple docstring"""

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


__snake_case												=							warning_advice


class      __lowerCamelCase	:

     '''simple docstring'''


     def __init__(   self     ,					*__UpperCAmelCase     ,					**__UpperCAmelCase				)							->				Dict:  # pylint: disable=unused-argument
          _a     						=					args[0] if args else None


     def __iter__(   self				)							->				int:
          return iter(self._iterator				)


     def __getattr__(   self     ,					__UpperCAmelCase				)							->				Tuple:

          def empty_fn(*__UpperCAmelCase     ,					**__UpperCAmelCase				):  # pylint: disable=unused-argument
               return

          return empty_fn


     def __enter__(   self				)							->				Tuple:
          return self





     def __exit__(   self     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				)							->				int:
          return







class      __lowerCamelCase	:

     '''simple docstring'''


     def __call__(   self     ,					*__UpperCAmelCase     ,					**__UpperCAmelCase				)							->				List[str]:
          if _tqdm_active:
               return tqdm_lib.tqdm(*__UpperCAmelCase     ,					**__UpperCAmelCase				)
          else:
               return EmptyTqdm(*__UpperCAmelCase     ,					**__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self     ,					*__UpperCAmelCase     ,					**__UpperCAmelCase				)							->				Any:
          _a     						=					None
          if _tqdm_active:
               return tqdm_lib.tqdm.set_lock(*__UpperCAmelCase     ,					**__UpperCAmelCase				)





     def 				_UpperCAmelCase   (   self				)							->				Tuple:
          if _tqdm_active:
               return tqdm_lib.tqdm.get_lock()







__snake_case												=							_tqdm_cls()
def    A_ ( ):






     """simple docstring"""

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






     """simple docstring"""

     global _tqdm_active
     _a     						=					True

def    A_ ( ):






     """simple docstring"""

     global _tqdm_active
     _a     						=					False



"""simple docstring"""






import torch

from diffusers import DDIMScheduler

from .test_schedulers import SchedulerCommonTest


class      __lowerCamelCase	(       a__					):

     '''simple docstring'''
     A_							:						Tuple     =		(DDIMScheduler,)
     A_							:						Tuple     =		(('eta', 0.0), ('num_inference_steps', 50))


     def 				_UpperCAmelCase   (   self     ,					**__UpperCAmelCase				)							->				str:
          _a     						=					{
              '''num_train_timesteps''': 1000,
              '''beta_start''': 0.0001,
              '''beta_end''': 0.02,
              '''beta_schedule''': '''linear''',
              '''clip_sample''': True,
          }

          config.update(**__UpperCAmelCase				)
          return config


     def 				_UpperCAmelCase   (   self     ,					**__UpperCAmelCase				)							->				List[Any]:
          _a     						=					self.scheduler_classes[0]
          _a     						=					self.get_scheduler_config(**__UpperCAmelCase				)
          _a     						=					scheduler_class(**__UpperCAmelCase				)

          _a						,   _a     						=					10, 0.0

          _a     						=					self.dummy_model()
          _a     						=					self.dummy_sample_deter

          scheduler.set_timesteps(__UpperCAmelCase				)

          for t in scheduler.timesteps:
               _a     						=					model(__UpperCAmelCase     ,					__UpperCAmelCase				)
               _a     						=					scheduler.step(__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase				).prev_sample

          return sample


     def 				_UpperCAmelCase   (   self				)							->				str:
          for timesteps in [100, 500, 1000]:
               self.check_over_configs(num_train_timesteps=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Dict:
          for steps_offset in [0, 1]:
               self.check_over_configs(steps_offset=__UpperCAmelCase				)

          _a     						=					self.scheduler_classes[0]
          _a     						=					self.get_scheduler_config(steps_offset=1				)
          _a     						=					scheduler_class(**__UpperCAmelCase				)
          scheduler.set_timesteps(5				)
          assert torch.equal(scheduler.timesteps     ,					torch.LongTensor([801, 601, 401, 201, 1]				)				)


     def 				_UpperCAmelCase   (   self				)							->				int:
          for beta_start, beta_end in zip([0.0001, 0.001, 0.01, 0.1]     ,					[0.002, 0.02, 0.2, 2]				):
               self.check_over_configs(beta_start=__UpperCAmelCase     ,					beta_end=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Union[str, Any]:
          for schedule in ["linear", "squaredcos_cap_v2"]:
               self.check_over_configs(beta_schedule=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          for prediction_type in ["epsilon", "v_prediction"]:
               self.check_over_configs(prediction_type=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				int:
          for clip_sample in [True, False]:
               self.check_over_configs(clip_sample=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Union[str, Any]:
          for timestep_spacing in ["trailing", "leading"]:
               self.check_over_configs(timestep_spacing=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          for rescale_betas_zero_snr in [True, False]:
               self.check_over_configs(rescale_betas_zero_snr=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          self.check_over_configs(thresholding=__UpperCAmelCase				)
          for threshold in [0.5, 1.0, 2.0]:
               for prediction_type in ["epsilon", "v_prediction"]:
                    self.check_over_configs(
                        thresholding=__UpperCAmelCase     ,					prediction_type=__UpperCAmelCase     ,					sample_max_value=__UpperCAmelCase     ,					)


     def 				_UpperCAmelCase   (   self				)							->				int:
          for t in [1, 10, 49]:
               self.check_over_forward(time_step=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Tuple:
          for t, num_inference_steps in zip([1, 10, 50]     ,					[10, 50, 500]				):
               self.check_over_forward(time_step=__UpperCAmelCase     ,					num_inference_steps=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				Optional[Any]:
          for t, eta in zip([1, 10, 49]     ,					[0.0, 0.5, 1.0]				):
               self.check_over_forward(time_step=__UpperCAmelCase     ,					eta=__UpperCAmelCase				)


     def 				_UpperCAmelCase   (   self				)							->				int:
          _a     						=					self.scheduler_classes[0]
          _a     						=					self.get_scheduler_config()
          _a     						=					scheduler_class(**__UpperCAmelCase				)

          assert torch.sum(torch.abs(scheduler._get_variance(0     ,					0				) - 0.0				)				) < 1e-5
          assert torch.sum(torch.abs(scheduler._get_variance(420     ,					400				) - 0.14771				)				) < 1e-5
          assert torch.sum(torch.abs(scheduler._get_variance(980     ,					960				) - 0.32460				)				) < 1e-5
          assert torch.sum(torch.abs(scheduler._get_variance(0     ,					0				) - 0.0				)				) < 1e-5
          assert torch.sum(torch.abs(scheduler._get_variance(487     ,					486				) - 0.00979				)				) < 1e-5
          assert torch.sum(torch.abs(scheduler._get_variance(999     ,					998				) - 0.02				)				) < 1e-5


     def 				_UpperCAmelCase   (   self				)							->				Optional[int]:
          _a     						=					self.full_loop()

          _a     						=					torch.sum(torch.abs(__UpperCAmelCase				)				)
          _a     						=					torch.mean(torch.abs(__UpperCAmelCase				)				)

          assert abs(result_sum.item() - 172.0067				) < 1e-2
          assert abs(result_mean.item() - 0.223967				) < 1e-3


     def 				_UpperCAmelCase   (   self				)							->				Any:
          _a     						=					self.full_loop(prediction_type='''v_prediction'''				)

          _a     						=					torch.sum(torch.abs(__UpperCAmelCase				)				)
          _a     						=					torch.mean(torch.abs(__UpperCAmelCase				)				)

          assert abs(result_sum.item() - 52.5302				) < 1e-2
          assert abs(result_mean.item() - 0.0684				) < 1e-3


     def 				_UpperCAmelCase   (   self				)							->				List[str]:
          # We specify different beta, so that the first alpha is 0.99
          _a     						=					self.full_loop(set_alpha_to_one=__UpperCAmelCase     ,					beta_start=0.01				)
          _a     						=					torch.sum(torch.abs(__UpperCAmelCase				)				)
          _a     						=					torch.mean(torch.abs(__UpperCAmelCase				)				)

          assert abs(result_sum.item() - 149.8295				) < 1e-2
          assert abs(result_mean.item() - 0.1951				) < 1e-3





     def 				_UpperCAmelCase   (   self				)							->				int:
          # We specify different beta, so that the first alpha is 0.99
          _a     						=					self.full_loop(set_alpha_to_one=__UpperCAmelCase     ,					beta_start=0.01				)
          _a     						=					torch.sum(torch.abs(__UpperCAmelCase				)				)
          _a     						=					torch.mean(torch.abs(__UpperCAmelCase				)				)

          assert abs(result_sum.item() - 149.0784				) < 1e-2
          assert abs(result_mean.item() - 0.1941				) < 1e-3