



"""simple docstring"""







import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest







class  UpperCamelCase__  (      lowercase_    ):




      """simple docstring"""




      _SCREAMING_SNAKE_CASE    							=	(DPMSolverMultistepScheduler,)
      _SCREAMING_SNAKE_CASE    							=	(("""num_inference_steps""", 25),)



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[int]       ,     **SCREAMING_SNAKE_CASE_							: Union[str, Any]			):
        lowerCAmelCase_  :							Union[str, Any]								=					{
            '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(**SCREAMING_SNAKE_CASE_			)
        return config



      def 	SCREAMING_SNAKE_CASE__					( self							: List[Any]       ,     SCREAMING_SNAKE_CASE_							: str=0       ,     **SCREAMING_SNAKE_CASE_							: int			):
        lowerCAmelCase_  :							List[Any]								=					dict(self.forward_default_kwargs			)
        lowerCAmelCase_  :							Tuple								=					kwargs.pop('num_inference_steps'       ,     SCREAMING_SNAKE_CASE_			)
        lowerCAmelCase_  :							Optional[Any]								=					self.dummy_sample
        lowerCAmelCase_  :							Dict								=					0.1 * sample
        lowerCAmelCase_  :							str								=					[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
          lowerCAmelCase_  :							List[str]								=					self.get_scheduler_config(**SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							List[str]								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)
          scheduler.set_timesteps(SCREAMING_SNAKE_CASE_			)
          # copy over dummy past residuals
          lowerCAmelCase_  :							List[Any]								=					dummy_past_residuals[: scheduler.config.solver_order]

          with tempfile.TemporaryDirectory() as tmpdirname:
            scheduler.save_config(SCREAMING_SNAKE_CASE_			)
            lowerCAmelCase_  :							Dict								=					scheduler_class.from_pretrained(SCREAMING_SNAKE_CASE_			)
            new_scheduler.set_timesteps(SCREAMING_SNAKE_CASE_			)
            # copy over dummy past residuals
            lowerCAmelCase_  :							List[str]								=					dummy_past_residuals[: new_scheduler.config.solver_order]

          lowerCAmelCase_		,lowerCAmelCase_  :							Union[str, Any]								=					sample, sample
          for t in range(SCREAMING_SNAKE_CASE_       ,     time_step + scheduler.config.solver_order + 1			):
            lowerCAmelCase_  :							int								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample
            lowerCAmelCase_  :							Any								=					new_scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Tuple			):
        pass



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[Any]       ,     SCREAMING_SNAKE_CASE_							: List[Any]=0       ,     **SCREAMING_SNAKE_CASE_							: Optional[int]			):
        lowerCAmelCase_  :							Union[str, Any]								=					dict(self.forward_default_kwargs			)
        lowerCAmelCase_  :							Optional[int]								=					kwargs.pop('num_inference_steps'       ,     SCREAMING_SNAKE_CASE_			)
        lowerCAmelCase_  :							Any								=					self.dummy_sample
        lowerCAmelCase_  :							Tuple								=					0.1 * sample
        lowerCAmelCase_  :							Optional[Any]								=					[residual + 0.2, residual + 0.15, residual + 0.10]

        for scheduler_class in self.scheduler_classes:
          lowerCAmelCase_  :							str								=					self.get_scheduler_config()
          lowerCAmelCase_  :							str								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)
          scheduler.set_timesteps(SCREAMING_SNAKE_CASE_			)

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

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

            # copy over dummy past residual (must be after setting timesteps)
            lowerCAmelCase_  :							Union[str, Any]								=					dummy_past_residuals[: new_scheduler.config.solver_order]

          lowerCAmelCase_  :							List[str]								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample
          lowerCAmelCase_  :							Any								=					new_scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Union[str, Any]       ,     SCREAMING_SNAKE_CASE_							: List[Any]=None       ,     **SCREAMING_SNAKE_CASE_							: Optional[Any]			):
        if scheduler is None:
          lowerCAmelCase_  :							List[str]								=					self.scheduler_classes[0]
          lowerCAmelCase_  :							str								=					self.get_scheduler_config(**SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							Any								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)

        lowerCAmelCase_  :							Any								=					1_0
        lowerCAmelCase_  :							Union[str, Any]								=					self.dummy_model()
        lowerCAmelCase_  :							Optional[int]								=					self.dummy_sample_deter
        scheduler.set_timesteps(SCREAMING_SNAKE_CASE_			)

        for i, t in enumerate(scheduler.timesteps			):
          lowerCAmelCase_  :							int								=					model(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							Union[str, Any]								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			).prev_sample

        return sample



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[int]			):
        lowerCAmelCase_  :							Tuple								=					dict(self.forward_default_kwargs			)

        lowerCAmelCase_  :							str								=					kwargs.pop('num_inference_steps'       ,     SCREAMING_SNAKE_CASE_			)

        for scheduler_class in self.scheduler_classes:
          lowerCAmelCase_  :							Dict								=					self.get_scheduler_config()
          lowerCAmelCase_  :							Any								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)

          lowerCAmelCase_  :							Dict								=					self.dummy_sample
          lowerCAmelCase_  :							Union[str, Any]								=					0.1 * sample

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

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

          lowerCAmelCase_  :							Union[str, Any]								=					scheduler.timesteps[5]
          lowerCAmelCase_  :							Optional[Any]								=					scheduler.timesteps[6]

          lowerCAmelCase_  :							Tuple								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample
          lowerCAmelCase_  :							List[Any]								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     **SCREAMING_SNAKE_CASE_			).prev_sample

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[Any]			):
        for timesteps in [2_5, 5_0, 1_0_0, 9_9_9, 1_0_0_0]:
          self.check_over_configs(num_train_timesteps=SCREAMING_SNAKE_CASE_			)



      def 	SCREAMING_SNAKE_CASE__					( self							: List[str]			):
        self.check_over_configs(thresholding=SCREAMING_SNAKE_CASE_			)
        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=SCREAMING_SNAKE_CASE_       ,     prediction_type=SCREAMING_SNAKE_CASE_       ,     sample_max_value=SCREAMING_SNAKE_CASE_       ,     algorithm_type='dpmsolver++'       ,     solver_order=SCREAMING_SNAKE_CASE_       ,     solver_type=SCREAMING_SNAKE_CASE_       ,     )



      def 	SCREAMING_SNAKE_CASE__					( self							: List[str]			):
        for prediction_type in ["epsilon", "v_prediction"]:
          self.check_over_configs(prediction_type=SCREAMING_SNAKE_CASE_			)



      def 	SCREAMING_SNAKE_CASE__					( self							: Dict			):
        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=SCREAMING_SNAKE_CASE_       ,     solver_type=SCREAMING_SNAKE_CASE_       ,     prediction_type=SCREAMING_SNAKE_CASE_       ,     algorithm_type=SCREAMING_SNAKE_CASE_       ,     )
                lowerCAmelCase_  :							int								=					self.full_loop(
                    solver_order=SCREAMING_SNAKE_CASE_       ,     solver_type=SCREAMING_SNAKE_CASE_       ,     prediction_type=SCREAMING_SNAKE_CASE_       ,     algorithm_type=SCREAMING_SNAKE_CASE_       ,     )
                assert not torch.isnan(SCREAMING_SNAKE_CASE_			).any(), "Samples have nan numbers"



      def 	SCREAMING_SNAKE_CASE__					( self							: str			):
        self.check_over_configs(lower_order_final=SCREAMING_SNAKE_CASE_			)
        self.check_over_configs(lower_order_final=SCREAMING_SNAKE_CASE_			)



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[int]			):
        self.check_over_configs(lambda_min_clipped=-float('inf'			)			)
        self.check_over_configs(lambda_min_clipped=-5.1			)



      def 	SCREAMING_SNAKE_CASE__					( self							: List[str]			):
        self.check_over_configs(variance_type=SCREAMING_SNAKE_CASE_			)
        self.check_over_configs(variance_type='learned_range'			)



      def 	SCREAMING_SNAKE_CASE__					( self							: str			):
        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=SCREAMING_SNAKE_CASE_       ,     time_step=0			)



      def 	SCREAMING_SNAKE_CASE__					( self							: Dict			):
        lowerCAmelCase_  :							List[str]								=					self.full_loop()
        lowerCAmelCase_  :							Dict								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[int]			):
        lowerCAmelCase_  :							Any								=					self.full_loop(thresholding=SCREAMING_SNAKE_CASE_       ,     dynamic_thresholding_ratio=0.87       ,     sample_max_value=0.5			)
        lowerCAmelCase_  :							int								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Union[str, Any]			):
        lowerCAmelCase_  :							Optional[Any]								=					self.full_loop(prediction_type='v_prediction'			)
        lowerCAmelCase_  :							Union[str, Any]								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Tuple			):
        lowerCAmelCase_  :							Any								=					self.full_loop(prediction_type='v_prediction'       ,     use_karras_sigmas=SCREAMING_SNAKE_CASE_			)
        lowerCAmelCase_  :							List[str]								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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



      def 	SCREAMING_SNAKE_CASE__					( self							: Optional[int]			):
        # make sure that iterating over schedulers with same config names gives same results
        # for defaults
        lowerCAmelCase_  :							Optional[int]								=					DPMSolverMultistepScheduler(**self.get_scheduler_config()			)
        lowerCAmelCase_  :							List[str]								=					self.full_loop(scheduler=SCREAMING_SNAKE_CASE_			)
        lowerCAmelCase_  :							Tuple								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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

        lowerCAmelCase_  :							Optional[int]								=					DPMSolverSinglestepScheduler.from_config(scheduler.config			)
        lowerCAmelCase_  :							Any								=					UniPCMultistepScheduler.from_config(scheduler.config			)
        lowerCAmelCase_  :							Any								=					DEISMultistepScheduler.from_config(scheduler.config			)
        lowerCAmelCase_  :							Optional[int]								=					DPMSolverMultistepScheduler.from_config(scheduler.config			)

        lowerCAmelCase_  :							Tuple								=					self.full_loop(scheduler=SCREAMING_SNAKE_CASE_			)
        lowerCAmelCase_  :							Optional[int]								=					torch.mean(torch.abs(SCREAMING_SNAKE_CASE_			)			)

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



      def 	SCREAMING_SNAKE_CASE__					( self							: List[str]			):
        lowerCAmelCase_  :							int								=					self.scheduler_classes[0]
        lowerCAmelCase_  :							Tuple								=					self.get_scheduler_config(thresholding=SCREAMING_SNAKE_CASE_       ,     dynamic_thresholding_ratio=0			)
        lowerCAmelCase_  :							Optional[Any]								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)

        lowerCAmelCase_  :							Dict								=					1_0
        lowerCAmelCase_  :							List[str]								=					self.dummy_model()
        lowerCAmelCase_  :							Optional[int]								=					self.dummy_sample_deter.half()
        scheduler.set_timesteps(SCREAMING_SNAKE_CASE_			)

        for i, t in enumerate(scheduler.timesteps			):
          lowerCAmelCase_  :							Tuple								=					model(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							Optional[Any]								=					scheduler.step(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			).prev_sample

        assert sample.dtype == torch.floataa





      def 	SCREAMING_SNAKE_CASE__					( self							: str       ,     **SCREAMING_SNAKE_CASE_							: Any			):
        for scheduler_class in self.scheduler_classes:
          lowerCAmelCase_  :							Tuple								=					self.get_scheduler_config(**SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							Tuple								=					scheduler_class(**SCREAMING_SNAKE_CASE_			)

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










"""simple docstring"""







import unittest
from timeit import timeit

def 							UpperCamelCase_       (       lowerCAmelCase__ :							int       ,  lowerCAmelCase__ :							int		)		->  int:

  """simple docstring"""




  lowerCAmelCase_  :							Any								=					first_num if first_num >= second_num else second_num
  lowerCAmelCase_  :							Union[str, Any]								=					max_num
  while (common_mult % first_num > 0) or (common_mult % second_num > 0):
    common_mult += max_num
  return common_mult

def 							UpperCamelCase_       (       lowerCAmelCase__ :							int       ,  lowerCAmelCase__ :							int		)		->  int:

  """simple docstring"""




  return b if a == 0 else greatest_common_divisor(b % a       ,  lowerCAmelCase__		)

def 							UpperCamelCase_       (       lowerCAmelCase__ :							int       ,  lowerCAmelCase__ :							int		)		->  int:

  """simple docstring"""




  return first_num // greatest_common_divisor(lowerCAmelCase__       ,  lowerCAmelCase__		) * second_num

def 							UpperCamelCase_       (       )		->  Dict:

  """simple docstring"""




  lowerCAmelCase_  :							str								=					(
      'from __main__ import least_common_multiple_slow, least_common_multiple_fast'
  )
  print(
      'least_common_multiple_slow():'       ,  timeit('least_common_multiple_slow(1000, 999)'       ,  setup=lowerCAmelCase__		)       ,  )
  print(
      'least_common_multiple_fast():'       ,  timeit('least_common_multiple_fast(1000, 999)'       ,  setup=lowerCAmelCase__		)       ,  )







class  UpperCamelCase__  (      unittest.TestCase    ):




      """simple docstring"""




      _SCREAMING_SNAKE_CASE    							=	(
          (10, 20),
          (13, 15),
          (4, 31),
          (10, 42),
          (43, 34),
          (5, 12),
          (12, 25),
          (10, 25),
          (6, 9),
      )
      _SCREAMING_SNAKE_CASE    							=	(20, 195, 124, 210, 1462, 60, 300, 50, 18)



      def 	SCREAMING_SNAKE_CASE__					( self							: List[str]			):
        for i, (first_num, second_num) in enumerate(self.test_inputs			):
          lowerCAmelCase_  :							Union[str, Any]								=					least_common_multiple_slow(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			)
          lowerCAmelCase_  :							Any								=					least_common_multiple_fast(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_			)
          with self.subTest(i=SCREAMING_SNAKE_CASE_			):
            self.assertEqual(SCREAMING_SNAKE_CASE_       ,     self.expected_results[i]			)
            self.assertEqual(SCREAMING_SNAKE_CASE_       ,     self.expected_results[i]			)


if __name__ == "__main__":
     benchmark()
     unittest.main()





