






import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest






class 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      __snake_case       =  (PNDMScheduler,)
      __snake_case       =  (('''num_inference_steps''', 50),)







      def       __lowerCAmelCase					(      self      :						int ,					**__UpperCAmelCase      :						Optional[int]  )    ->int:




       """simple docstring"""


       a		  =      {
           '''num_train_timesteps''': 1_000,
           '''beta_start''': 0.0001,
           '''beta_end''': 0.02,
           '''beta_schedule''': '''linear''',
       }

       config.update(**__UpperCAmelCase  )
       return config







      def       __lowerCAmelCase					(      self      :						Optional[Any] ,					__UpperCAmelCase      :						List[str]=0 ,					**__UpperCAmelCase      :						Tuple  )    ->Tuple:




       """simple docstring"""


       a		  =      dict(self.forward_default_kwargs  )
       a		  =      kwargs.pop('''num_inference_steps''' ,					__UpperCAmelCase  )
       a		  =      self.dummy_sample
       a		  =      0.1 * sample
       a		  =      [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

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

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

        a		  =      scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      new_scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample

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

        a		  =      scheduler.step_plms(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      new_scheduler.step_plms(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample

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







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




       """simple docstring"""


       pass







      def       __lowerCAmelCase					(      self      :						List[Any] ,					__UpperCAmelCase      :						List[str]=0 ,					**__UpperCAmelCase      :						Optional[Any]  )    ->Tuple:




       """simple docstring"""


       a		  =      dict(self.forward_default_kwargs  )
       a		  =      kwargs.pop('''num_inference_steps''' ,					__UpperCAmelCase  )
       a		  =      self.dummy_sample
       a		  =      0.1 * sample
       a		  =      [residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

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

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

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

        a		  =      scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      new_scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample

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

        a		  =      scheduler.step_plms(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      new_scheduler.step_plms(__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 ,					**__UpperCAmelCase      :						Optional[int]  )    ->str:




       """simple docstring"""


       a		  =      self.scheduler_classes[0]
       a		  =      self.get_scheduler_config(**__UpperCAmelCase  )
       a		  =      scheduler_class(**__UpperCAmelCase  )

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

       for i, t in enumerate(scheduler.prk_timesteps  ):
        a		  =      model(__UpperCAmelCase ,					__UpperCAmelCase  )
        a		  =      scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase  ).prev_sample

       for i, t in enumerate(scheduler.plms_timesteps  ):
        a		  =      model(__UpperCAmelCase ,					__UpperCAmelCase  )
        a		  =      scheduler.step_plms(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase  ).prev_sample

       return sample







      def       __lowerCAmelCase					(      self      :						Union[str, Any]  )    ->Optional[int]:




       """simple docstring"""


       a		  =      dict(self.forward_default_kwargs  )

       a		  =      kwargs.pop('''num_inference_steps''' ,					__UpperCAmelCase  )

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

        a		  =      self.dummy_sample
        a		  =      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'''  ):
         a		  =      num_inference_steps

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

        a		  =      scheduler.step_prk(__UpperCAmelCase ,					0 ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      scheduler.step_prk(__UpperCAmelCase ,					1 ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample

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

        a		  =      scheduler.step_plms(__UpperCAmelCase ,					0 ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample
        a		  =      scheduler.step_plms(__UpperCAmelCase ,					1 ,					__UpperCAmelCase ,					**__UpperCAmelCase  ).prev_sample

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







      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->List[Any]:




       """simple docstring"""


       for timesteps in [100, 1_000]:
        self.check_over_configs(num_train_timesteps=__UpperCAmelCase  )







      def       __lowerCAmelCase					(      self      :						Dict  )    ->int:




       """simple docstring"""


       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(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      :						Tuple  )    ->List[Any]:




       """simple docstring"""


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







      def       __lowerCAmelCase					(      self      :						Optional[Any]  )    ->Any:




       """simple docstring"""


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







      def       __lowerCAmelCase					(      self      :						Dict  )    ->Dict:




       """simple docstring"""


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







      def       __lowerCAmelCase					(      self      :						Any  )    ->List[Any]:




       """simple docstring"""


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







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




       """simple docstring"""


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







      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->Optional[Any]:




       """simple docstring"""


       a		  =      27

       for scheduler_class in self.scheduler_classes:
        a		  =      self.dummy_sample
        a		  =      0.1 * sample

        a		  =      self.get_scheduler_config()
        a		  =      scheduler_class(**__UpperCAmelCase  )

        scheduler.set_timesteps(__UpperCAmelCase  )

        # 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]  ):
         a		  =      scheduler.step_prk(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase  ).prev_sample







      def       __lowerCAmelCase					(      self      :						Union[str, Any]  )    ->Optional[int]:




       """simple docstring"""


       with self.assertRaises(__UpperCAmelCase  ):
        a		  =      self.scheduler_classes[0]
        a		  =      self.get_scheduler_config()
        a		  =      scheduler_class(**__UpperCAmelCase  )

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







      def       __lowerCAmelCase					(      self      :						Any  )    ->Optional[int]:




       """simple docstring"""


       a		  =      self.full_loop()
       a		  =      torch.sum(torch.abs(__UpperCAmelCase  )  )
       a		  =      torch.mean(torch.abs(__UpperCAmelCase  )  )

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







      def       __lowerCAmelCase					(      self      :						Dict  )    ->str:




       """simple docstring"""


       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() - 67.3986  ) < 1e-2
       assert abs(result_mean.item() - 0.0878  ) < 1e-3







      def       __lowerCAmelCase					(      self      :						str  )    ->Tuple:




       """simple docstring"""


       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() - 230.0399  ) < 1e-2
       assert abs(result_mean.item() - 0.2995  ) < 1e-3







      def       __lowerCAmelCase					(      self      :						Optional[int]  )    ->Any:




       """simple docstring"""


       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() - 186.9482  ) < 1e-2
       assert abs(result_mean.item() - 0.2434  ) < 1e-3










import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






class 		lowercase_					(	lowercase ,  unittest.TestCase   ):






      '''simple docstring'''




      __snake_case       =  KandinskyVaaPriorPipeline
      __snake_case       =  ['''prompt''']
      __snake_case       =  ['''prompt''', '''negative_prompt''']
      __snake_case       =  [
          '''num_images_per_prompt''',
          '''generator''',
          '''num_inference_steps''',
          '''latents''',
          '''negative_prompt''',
          '''guidance_scale''',
          '''output_type''',
          '''return_dict''',
      ]
      __snake_case       =  False







      @property
      def       __lowerCAmelCase					(      self      :						Optional[Any]  )    ->Union[str, Any]:




       """simple docstring"""


       return 32







      @property
      def       __lowerCAmelCase					(      self      :						Dict  )    ->Any:




       """simple docstring"""


       return 32







      @property
      def       __lowerCAmelCase					(      self      :						int  )    ->List[str]:




       """simple docstring"""


       return self.time_input_dim







      @property
      def       __lowerCAmelCase					(      self      :						Tuple  )    ->Any:




       """simple docstring"""


       return self.time_input_dim * 4







      @property
      def       __lowerCAmelCase					(      self      :						Any  )    ->List[Any]:




       """simple docstring"""


       return 100







      @property
      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->str:




       """simple docstring"""


       a		  =      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''  )
       return tokenizer







      @property
      def       __lowerCAmelCase					(      self      :						Tuple  )    ->str:




       """simple docstring"""


       torch.manual_seed(0  )
       a		  =      CLIPTextConfig(
           bos_token_id=0 ,					eos_token_id=2 ,					hidden_size=self.text_embedder_hidden_size ,					projection_dim=self.text_embedder_hidden_size ,					intermediate_size=37 ,					layer_norm_eps=1e-0_5 ,					num_attention_heads=4 ,					num_hidden_layers=5 ,					pad_token_id=1 ,					vocab_size=1_000 ,					)
       return CLIPTextModelWithProjection(__UpperCAmelCase  )







      @property
      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->Optional[Any]:




       """simple docstring"""


       torch.manual_seed(0  )

       a		  =      {
           '''num_attention_heads''': 2,
           '''attention_head_dim''': 12,
           '''embedding_dim''': self.text_embedder_hidden_size,
           '''num_layers''': 1,
       }

       a		  =      PriorTransformer(**__UpperCAmelCase  )
       # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
       a		  =      nn.Parameter(torch.ones(model.clip_std.shape  )  )
       return model







      @property
      def       __lowerCAmelCase					(      self      :						Optional[int]  )    ->List[Any]:




       """simple docstring"""


       torch.manual_seed(0  )
       a		  =      CLIPVisionConfig(
           hidden_size=self.text_embedder_hidden_size ,					image_size=224 ,					projection_dim=self.text_embedder_hidden_size ,					intermediate_size=37 ,					num_attention_heads=4 ,					num_channels=3 ,					num_hidden_layers=5 ,					patch_size=14 ,					)

       a		  =      CLIPVisionModelWithProjection(__UpperCAmelCase  )
       return model







      @property
      def       __lowerCAmelCase					(      self      :						Tuple  )    ->int:




       """simple docstring"""


       a		  =      CLIPImageProcessor(
           crop_size=224 ,					do_center_crop=__UpperCAmelCase ,					do_normalize=__UpperCAmelCase ,					do_resize=__UpperCAmelCase ,					image_mean=[0.48145466, 0.4578275, 0.40821073] ,					image_std=[0.26862954, 0.26130258, 0.27577711] ,					resample=3 ,					size=224 ,					)

       return image_processor







      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->Optional[Any]:




       """simple docstring"""


       a		  =      self.dummy_prior
       a		  =      self.dummy_image_encoder
       a		  =      self.dummy_text_encoder
       a		  =      self.dummy_tokenizer
       a		  =      self.dummy_image_processor

       a		  =      UnCLIPScheduler(
           variance_type='''fixed_small_log''' ,					prediction_type='''sample''' ,					num_train_timesteps=1_000 ,					clip_sample=__UpperCAmelCase ,					clip_sample_range=10.0 ,					)

       a		  =      {
           '''prior''': prior,
           '''image_encoder''': image_encoder,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''scheduler''': scheduler,
           '''image_processor''': image_processor,
       }

       return components







      def       __lowerCAmelCase					(      self      :						Optional[Any] ,					__UpperCAmelCase      :						List[Any] ,					__UpperCAmelCase      :						str=0  )    ->int:




       """simple docstring"""


       if str(__UpperCAmelCase  ).startswith('''mps'''  ):
        a		  =      torch.manual_seed(__UpperCAmelCase  )
       else:
        a		  =      torch.Generator(device=__UpperCAmelCase  ).manual_seed(__UpperCAmelCase  )
       a		  =      {
           '''prompt''': '''horse''',
           '''generator''': generator,
           '''guidance_scale''': 4.0,
           '''num_inference_steps''': 2,
           '''output_type''': '''np''',
       }
       return inputs







      def       __lowerCAmelCase					(      self      :						str  )    ->Tuple:




       """simple docstring"""


       a		  =      '''cpu'''

       a		  =      self.get_dummy_components()

       a		  =      self.pipeline_class(**__UpperCAmelCase  )
       a		  =      pipe.to(__UpperCAmelCase  )

       pipe.set_progress_bar_config(disable=__UpperCAmelCase  )

       a		  =      pipe(**self.get_dummy_inputs(__UpperCAmelCase  )  )
       a		  =      output.image_embeds

       a		  =      pipe(
           **self.get_dummy_inputs(__UpperCAmelCase  ) ,					return_dict=__UpperCAmelCase ,					)[0]

       a		  =      image[0, -10:]
       a		  =      image_from_tuple[0, -10:]

       assert image.shape == (1, 32)

       a		  =      np.array(
           [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156]  )

       assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice  ).max() < 1e-2







      @skip_mps
      def       __lowerCAmelCase					(      self      :						List[Any]  )    ->Optional[Any]:




       """simple docstring"""


       a		  =      torch_device == '''cpu'''
       a		  =      True
       a		  =      False

       self._test_inference_batch_single_identical(
           test_max_difference=__UpperCAmelCase ,					relax_max_difference=__UpperCAmelCase ,					test_mean_pixel_difference=__UpperCAmelCase ,					)







      @skip_mps
      def       __lowerCAmelCase					(      self      :						List[str]  )    ->Union[str, Any]:




       """simple docstring"""


       a		  =      torch_device == '''cpu'''
       a		  =      False

       self._test_attention_slicing_forward_pass(
           test_max_difference=__UpperCAmelCase ,					test_mean_pixel_difference=__UpperCAmelCase ,					)


