


'''simple docstring'''







import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()
class  lowercase_					(   unittest.TestCase       ):
    __UpperCAmelCase             =     'hf-internal-testing/tiny-stable-diffusion-pipe'
    __UpperCAmelCase             =     1
    __UpperCAmelCase             =     77
    __UpperCAmelCase             =     32
    __UpperCAmelCase             =     4
    __UpperCAmelCase             =     __UpperCAmelCase             =     64
    __UpperCAmelCase             =     torch.manual_seed(0       )


    def        __a							(					self  , a=0.9999   ):
          UpperCamelCase__  =    UNetaDConditionModel.from_pretrained(self.model_id  , subfolder="unet"   )
          UpperCamelCase__  =    unet.to(a   )
          UpperCamelCase__  =    EMAModel(unet.parameters()  , decay=a  , model_cls=a  , model_config=unet.config   )
          return unet, ema_unet


    def        __a							(					self   ):
          UpperCamelCase__  =    torch.randn(
              self.batch_size  , self.num_in_channels  , self.latent_height  , self.latent_width  , generator=self.generator   ).to(a   )
          UpperCamelCase__  =    torch.randint(0  , 10_00  , size=(self.batch_size,)  , generator=self.generator   ).to(a   )
          UpperCamelCase__  =    torch.randn(
              self.batch_size  , self.prompt_length  , self.text_encoder_hidden_dim  , generator=self.generator   ).to(a   )
          return noisy_latents, timesteps, encoder_hidden_states


    def        __a							(					self  , a   ):
          UpperCamelCase__  =    {}
          for k, param in unet.state_dict().items():
                UpperCamelCase__  =    torch.randn_like(a   ) + (param * torch.randn_like(a   ))
                updated_state_dict.update({k: updated_param}   )
          unet.load_state_dict(a   )
          return unet


    def        __a							(					self   ):
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models()
          # Take the first (hypothetical) EMA step.
          ema_unet.step(unet.parameters()   )
          assert ema_unet.optimization_step == 1

          # Take two more.
          for _ in range(2   ):
                ema_unet.step(unet.parameters()   )
          assert ema_unet.optimization_step == 3


    def        __a							(					self   ):
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models()
          # Since the `unet` is not being updated (i.e., backprop'd)
          # there won't be any difference between the `params` of `unet`
          # and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
          ema_unet.step(unet.parameters()   )
          UpperCamelCase__  =    list(unet.parameters()   )
          for s_param, param in zip(ema_unet.shadow_params  , a   ):
                assert torch.allclose(a  , a   )

          # The above holds true even if we call `ema.step()` multiple times since
          # `unet` params are still not being updated.
          for _ in range(4   ):
                ema_unet.step(unet.parameters()   )
          for s_param, param in zip(ema_unet.shadow_params  , a   ):
                assert torch.allclose(a  , a   )


    def        __a							(					self   ):
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models()
          # Here we simulate the parameter updates for `unet`. Since there might
          # be some parameters which are initialized to zero we take extra care to
          # initialize their values to something non-zero before the multiplication.
          UpperCamelCase__  =    self.simulate_backprop(a   )

          # Take the EMA step.
          ema_unet.step(unet_pseudo_updated_step_one.parameters()   )

          # Now the EMA'd parameters won't be equal to the original model parameters.
          UpperCamelCase__  =    list(unet_pseudo_updated_step_one.parameters()   )
          for s_param, param in zip(ema_unet.shadow_params  , a   ):
                assert ~torch.allclose(a  , a   )

          # Ensure this is the case when we take multiple EMA steps.
          for _ in range(4   ):
                ema_unet.step(unet.parameters()   )
          for s_param, param in zip(ema_unet.shadow_params  , a   ):
                assert ~torch.allclose(a  , a   )


    def        __a							(					self   ):
          # If we call EMA step after a backpropagation consecutively for two times,
          # the shadow params from those two steps should be different.
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models()

          # First backprop + EMA
          UpperCamelCase__  =    self.simulate_backprop(a   )
          ema_unet.step(unet_step_one.parameters()   )
          UpperCamelCase__  =    ema_unet.shadow_params

          # Second backprop + EMA
          UpperCamelCase__  =    self.simulate_backprop(a   )
          ema_unet.step(unet_step_two.parameters()   )
          UpperCamelCase__  =    ema_unet.shadow_params

          for step_one, step_two in zip(a  , a   ):
                assert ~torch.allclose(a  , a   )


    def        __a							(					self   ):
          # If there's no decay even if there are backprops, EMA steps
          # won't take any effect i.e., the shadow params would remain the
          # same.
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models(decay=0.0   )
          UpperCamelCase__  =    self.simulate_backprop(a   )
          ema_unet.step(unet_step_one.parameters()   )
          UpperCamelCase__  =    ema_unet.shadow_params

          UpperCamelCase__  =    self.simulate_backprop(a   )
          ema_unet.step(unet_step_two.parameters()   )
          UpperCamelCase__  =    ema_unet.shadow_params

          for step_one, step_two in zip(a  , a   ):
                assert torch.allclose(a  , a   )


    @skip_mps
    def        __a							(					self   ):
          UpperCamelCase__ ,     UpperCamelCase__  =    self.get_models()
          UpperCamelCase__ ,     UpperCamelCase__ ,     UpperCamelCase__  =    self.get_dummy_inputs()

          with tempfile.TemporaryDirectory() as tmpdir:
                ema_unet.save_pretrained(a   )
                UpperCamelCase__  =    UNetaDConditionModel.from_pretrained(a  , model_cls=a   )
                UpperCamelCase__  =    loaded_unet.to(unet.device   )

          # Since no EMA step has been performed the outputs should match.
          UpperCamelCase__  =    unet(a  , a  , a   ).sample
          UpperCamelCase__  =    loaded_unet(a  , a  , a   ).sample

          assert torch.allclose(a  , a  , atol=1e-4   )









'''simple docstring'''







import tempfile

import torch

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

from .test_schedulers import SchedulerCommonTest
class  lowercase_					(   a__       ):
    __UpperCAmelCase             =     (DPMSolverMultistepScheduler,)
    __UpperCAmelCase             =     (('num_inference_steps', 25),)


    def        __a							(					self  , **a   ):
          UpperCamelCase__  =    {
              "num_train_timesteps": 10_00,
              "beta_start": 0.0001,
              "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(**a   )
          return config


    def        __a							(					self  , a=0  , **a   ):
          UpperCamelCase__  =    dict(self.forward_default_kwargs   )
          UpperCamelCase__  =    kwargs.pop("num_inference_steps"  , a   )
          UpperCamelCase__  =    self.dummy_sample
          UpperCamelCase__  =    0.1 * sample
          UpperCamelCase__  =    [residual + 0.2, residual + 0.15, residual + 0.10]

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

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

                UpperCamelCase__ ,     UpperCamelCase__  =    sample, sample
                for t in range(a  , time_step + scheduler.config.solver_order + 1   ):
                      UpperCamelCase__  =    scheduler.step(a  , a  , a  , **a   ).prev_sample
                      UpperCamelCase__  =    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        __a							(					self   ):
          pass


    def        __a							(					self  , a=0  , **a   ):
          UpperCamelCase__  =    dict(self.forward_default_kwargs   )
          UpperCamelCase__  =    kwargs.pop("num_inference_steps"  , a   )
          UpperCamelCase__  =    self.dummy_sample
          UpperCamelCase__  =    0.1 * sample
          UpperCamelCase__  =    [residual + 0.2, residual + 0.15, residual + 0.10]

          for scheduler_class in self.scheduler_classes:
                UpperCamelCase__  =    self.get_scheduler_config()
                UpperCamelCase__  =    scheduler_class(**a   )
                scheduler.set_timesteps(a   )

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

                with tempfile.TemporaryDirectory() as tmpdirname:
                      scheduler.save_config(a   )
                      UpperCamelCase__  =    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)
                      UpperCamelCase__  =    dummy_past_residuals[: new_scheduler.config.solver_order]

                UpperCamelCase__  =    scheduler.step(a  , a  , a  , **a   ).prev_sample
                UpperCamelCase__  =    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        __a							(					self  , a=None  , **a   ):
          if scheduler is None:
                UpperCamelCase__  =    self.scheduler_classes[0]
                UpperCamelCase__  =    self.get_scheduler_config(**a   )
                UpperCamelCase__  =    scheduler_class(**a   )

          UpperCamelCase__  =    10
          UpperCamelCase__  =    self.dummy_model()
          UpperCamelCase__  =    self.dummy_sample_deter
          scheduler.set_timesteps(a   )

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

          return sample


    def        __a							(					self   ):
          UpperCamelCase__  =    dict(self.forward_default_kwargs   )

          UpperCamelCase__  =    kwargs.pop("num_inference_steps"  , a   )

          for scheduler_class in self.scheduler_classes:
                UpperCamelCase__  =    self.get_scheduler_config()
                UpperCamelCase__  =    scheduler_class(**a   )

                UpperCamelCase__  =    self.dummy_sample
                UpperCamelCase__  =    0.1 * sample

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

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

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

                UpperCamelCase__  =    scheduler.step(a  , a  , a  , **a   ).prev_sample
                UpperCamelCase__  =    scheduler.step(a  , a  , a  , **a   ).prev_sample

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


    def        __a							(					self   ):
          for timesteps in [25, 50, 1_00, 9_99, 10_00]:
                self.check_over_configs(num_train_timesteps=a   )


    def        __a							(					self   ):
          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        __a							(					self   ):
          for prediction_type in ["epsilon", "v_prediction"]:
                self.check_over_configs(prediction_type=a   )


    def        __a							(					self   ):
          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=a  , solver_type=a  , prediction_type=a  , algorithm_type=a  , )
                                  UpperCamelCase__  =    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        __a							(					self   ):
          self.check_over_configs(lower_order_final=a   )
          self.check_over_configs(lower_order_final=a   )


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


    def        __a							(					self   ):
          self.check_over_configs(variance_type=a   )
          self.check_over_configs(variance_type="learned_range"   )


    def        __a							(					self   ):
          for num_inference_steps in [1, 2, 3, 5, 10, 50, 1_00, 9_99, 10_00]:
                self.check_over_forward(num_inference_steps=a  , time_step=0   )


    def        __a							(					self   ):
          UpperCamelCase__  =    self.full_loop()
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.full_loop(thresholding=a  , dynamic_thresholding_ratio=0.87  , sample_max_value=0.5   )
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.full_loop(prediction_type="v_prediction"   )
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.full_loop(prediction_type="v_prediction"  , use_karras_sigmas=a   )
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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


    def        __a							(					self   ):
          # make sure that iterating over schedulers with same config names gives same results
          # for defaults
          UpperCamelCase__  =    DPMSolverMultistepScheduler(**self.get_scheduler_config()   )
          UpperCamelCase__  =    self.full_loop(scheduler=a   )
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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

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

          UpperCamelCase__  =    self.full_loop(scheduler=a   )
          UpperCamelCase__  =    torch.mean(torch.abs(a   )   )

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


    def        __a							(					self   ):
          UpperCamelCase__  =    self.scheduler_classes[0]
          UpperCamelCase__  =    self.get_scheduler_config(thresholding=a  , dynamic_thresholding_ratio=0   )
          UpperCamelCase__  =    scheduler_class(**a   )

          UpperCamelCase__  =    10
          UpperCamelCase__  =    self.dummy_model()
          UpperCamelCase__  =    self.dummy_sample_deter.half()
          scheduler.set_timesteps(a   )

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

          assert sample.dtype == torch.floataa


    def        __a							(					self  , **a   ):
          for scheduler_class in self.scheduler_classes:
                UpperCamelCase__  =    self.get_scheduler_config(**a   )
                UpperCamelCase__  =    scheduler_class(**a   )

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





