

'''simple docstring'''






import random
import unittest

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

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

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase,unittest.TestCase      ):
    _A			    =							KandinskyVaaPriorEmbaEmbPipeline
    _A			    =							["prompt", "image"]
    _A			    =							["prompt", "image"]
    _A			    =							[
        "num_images_per_prompt",
        "strength",
        "generator",
        "num_inference_steps",
        "latents",
        "negative_prompt",
        "guidance_scale",
        "output_type",
        "return_dict",
    ]
    _A			    =							False




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          return 32




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          return 32




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          return self.time_input_dim




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          return self.time_input_dim * 4




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          return 100




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			int     =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"			)
          return tokenizer




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       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-05				,     num_attention_heads=4				,     num_hidden_layers=5				,     pad_token_id=1				,     vocab_size=1000				,     )
          return CLIPTextModelWithProjection(lowercase__			)




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          torch.manual_seed(0			)

          SCREAMING_SNAKE_CASE_      :			Any     =       {
              "num_attention_heads": 2,
              "attention_head_dim": 12,
              "embedding_dim": self.text_embedder_hidden_size,
              "num_layers": 1,
          }

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       PriorTransformer(**lowercase__			)
          # 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
          SCREAMING_SNAKE_CASE_      :			Dict     =       nn.Parameter(torch.ones(model.clip_std.shape			)			)
          return model




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          torch.manual_seed(0			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       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				,     )

          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       CLIPVisionModelWithProjection(lowercase__			)
          return model




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       CLIPImageProcessor(
              crop_size=224				,     do_center_crop=lowercase__				,     do_normalize=lowercase__				,     do_resize=lowercase__				,     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			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Dict     =       self.dummy_prior
          SCREAMING_SNAKE_CASE_      :			List[str]     =       self.dummy_image_encoder
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       self.dummy_text_encoder
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       self.dummy_tokenizer
          SCREAMING_SNAKE_CASE_      :			int     =       self.dummy_image_processor

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       UnCLIPScheduler(
              variance_type="fixed_small_log"				,     prediction_type="sample"				,     num_train_timesteps=1000				,     clip_sample=lowercase__				,     clip_sample_range=10.0				,     )

          SCREAMING_SNAKE_CASE_      :			Tuple     =       {
              "prior": prior,
              "image_encoder": image_encoder,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "scheduler": scheduler,
              "image_processor": image_processor,
          }

          return components




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__=0			):






          """simple docstring"""






          if str(lowercase__			).startswith("mps"			):
                SCREAMING_SNAKE_CASE_      :			Tuple     =       torch.manual_seed(lowercase__			)
          else:
                SCREAMING_SNAKE_CASE_      :			Tuple     =       torch.Generator(device=lowercase__			).manual_seed(lowercase__			)

          SCREAMING_SNAKE_CASE_      :			Tuple     =       floats_tensor((1, 3, 64, 64)				,     rng=random.Random(lowercase__			)			).to(lowercase__			)
          SCREAMING_SNAKE_CASE_      :			str     =       image.cpu().permute(0				,     2				,     3				,     1			)[0]
          SCREAMING_SNAKE_CASE_      :			Tuple     =       Image.fromarray(np.uinta(lowercase__			)			).convert("RGB"			).resize((256, 256)			)

          SCREAMING_SNAKE_CASE_      :			Tuple     =       {
              "prompt": "horse",
              "image": init_image,
              "strength": 0.5,
              "generator": generator,
              "guidance_scale": 4.0,
              "num_inference_steps": 2,
              "output_type": "np",
          }
          return inputs




    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       "cpu"

          SCREAMING_SNAKE_CASE_      :			Tuple     =       self.get_dummy_components()

          SCREAMING_SNAKE_CASE_      :			str     =       self.pipeline_class(**lowercase__			)
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       pipe.to(lowercase__			)

          pipe.set_progress_bar_config(disable=lowercase__			)

          SCREAMING_SNAKE_CASE_      :			Dict     =       pipe(**self.get_dummy_inputs(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			str     =       output.image_embeds

          SCREAMING_SNAKE_CASE_      :			Dict     =       pipe(
              **self.get_dummy_inputs(lowercase__			)				,     return_dict=lowercase__				,     )[0]

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       image[0, -10:]
          SCREAMING_SNAKE_CASE_      :			Any     =       image_from_tuple[0, -10:]

          assert image.shape == (1, 32)

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       np.array(
              [
                  0.1071284,
                  1.3330271,
                  0.61260223,
                  -0.6691065,
                  -0.3846852,
                  -1.0303661,
                  0.22716111,
                  0.03348901,
                  0.30040675,
                  -0.24805029,
              ]			)

          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			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Any     =       torch_device == "cpu"
          SCREAMING_SNAKE_CASE_      :			str     =       True
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       False

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







    @skip_mps
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			Tuple     =       torch_device == "cpu"
          SCREAMING_SNAKE_CASE_      :			List[str]     =       False

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




'''simple docstring'''






import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin


snake_case_       =       logging.get_logger(__name__)  # pylint: disable=invalid-name
@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerAncestralDiscrete
class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase      ):
    _A			    =							42
    _A			    =							None






def 					__lowerCamelCase			(      SCREAMING_SNAKE_CASE_     :	Any  ,  SCREAMING_SNAKE_CASE_     :	int=0.999  ,  SCREAMING_SNAKE_CASE_     :	Optional[Any]="cosine"  ,  )							->   Optional[int]:
      """simple docstring"""

      if alpha_transform_type == "cosine":

            def alpha_bar_fn(SCREAMING_SNAKE_CASE_     :	Tuple    ):
                  return math.cos((t + 0.008) / 1.008 * math.pi / 2    ) ** 2

      elif alpha_transform_type == "exp":

            def alpha_bar_fn(SCREAMING_SNAKE_CASE_     :	Tuple    ):
                  return math.exp(t * -12.0    )

      else:
            raise ValueError(F"Unsupported alpha_tranform_type: {alpha_transform_type}"    )

      SCREAMING_SNAKE_CASE_      :			List[str]     =       []
      for i in range(SCREAMING_SNAKE_CASE_    ):
            SCREAMING_SNAKE_CASE_      :			List[Any]     =       i / num_diffusion_timesteps
            SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       (i + 1) / num_diffusion_timesteps
            betas.append(min(1 - alpha_bar_fn(SCREAMING_SNAKE_CASE_    ) / alpha_bar_fn(SCREAMING_SNAKE_CASE_    )  ,  SCREAMING_SNAKE_CASE_    )    )
      return torch.tensor(SCREAMING_SNAKE_CASE_  ,  dtype=torch.floataa    )





class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase,_UpperCAmelCase      ):
    _A			    =							[e.name for e in KarrasDiffusionSchedulers]
    _A			    =							1




    @register_to_config
    def __init__(					self				,     lowercase__ = 1000				,     lowercase__ = 0.0001				,     lowercase__ = 0.02				,     lowercase__ = "linear"				,     lowercase__ = None				,     lowercase__ = "epsilon"				,     lowercase__ = "linspace"				,     lowercase__ = 0				,     ):






          """simple docstring"""






          if trained_betas is not None:
                SCREAMING_SNAKE_CASE_      :			Any     =       torch.tensor(lowercase__				,     dtype=torch.floataa			)
          elif beta_schedule == "linear":
                SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       torch.linspace(lowercase__				,     lowercase__				,     lowercase__				,     dtype=torch.floataa			)
          elif beta_schedule == "scaled_linear":
                # this schedule is very specific to the latent diffusion model.
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       (
                    torch.linspace(beta_start**0.5				,     beta_end**0.5				,     lowercase__				,     dtype=torch.floataa			) ** 2
                )
          elif beta_schedule == "squaredcos_cap_v2":
                # Glide cosine schedule
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       betas_for_alpha_bar(lowercase__			)
          else:
                raise NotImplementedError(F"{beta_schedule} does is not implemented for {self.__class__}"			)

          SCREAMING_SNAKE_CASE_      :			Any     =       1.0 - self.betas
          SCREAMING_SNAKE_CASE_      :			Tuple     =       torch.cumprod(self.alphas				,     dim=0			)

          SCREAMING_SNAKE_CASE_      :			Dict     =       np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       np.concatenate([sigmas[::-1], [0.0]]			).astype(np.floataa			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       torch.from_numpy(lowercase__			)

          # setable values
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       None
          SCREAMING_SNAKE_CASE_      :			str     =       np.linspace(0				,     num_train_timesteps - 1				,     lowercase__				,     dtype=lowercase__			)[::-1].copy()
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       torch.from_numpy(lowercase__			)
          SCREAMING_SNAKE_CASE_      :			int     =       False




    @property
    def   __lowerCamelCase      (					self			):






          """simple docstring"""






          if self.config.timestep_spacing in ["linspace", "trailing"]:
                return self.sigmas.max()

          return (self.sigmas.max() ** 2 + 1) ** 0.5




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__			):






          """simple docstring"""






          if isinstance(lowercase__				,     torch.Tensor			):
                SCREAMING_SNAKE_CASE_      :			Any     =       timestep.to(self.timesteps.device			)
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       (self.timesteps == timestep).nonzero().item()
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.sigmas[step_index]
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       sample / ((sigma**2 + 1) ** 0.5)
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       True
          return sample




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__ = None			):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			str     =       num_inference_steps

          # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
          if self.config.timestep_spacing == "linspace":
                SCREAMING_SNAKE_CASE_      :			Dict     =       np.linspace(0				,     self.config.num_train_timesteps - 1				,     lowercase__				,     dtype=lowercase__			)[
                    ::-1
                ].copy()
          elif self.config.timestep_spacing == "leading":
                SCREAMING_SNAKE_CASE_      :			Dict     =       self.config.num_train_timesteps // self.num_inference_steps
                # creates integer timesteps by multiplying by ratio
                # casting to int to avoid issues when num_inference_step is power of 3
                SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       (np.arange(0				,     lowercase__			) * step_ratio).round()[::-1].copy().astype(lowercase__			)
                timesteps += self.config.steps_offset
          elif self.config.timestep_spacing == "trailing":
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.config.num_train_timesteps / self.num_inference_steps
                # creates integer timesteps by multiplying by ratio
                # casting to int to avoid issues when num_inference_step is power of 3
                SCREAMING_SNAKE_CASE_      :			Any     =       (np.arange(self.config.num_train_timesteps				,     0				,     -step_ratio			)).round().copy().astype(lowercase__			)
                timesteps -= 1
          else:
                raise ValueError(
                    F"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."			)

          SCREAMING_SNAKE_CASE_      :			int     =       np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
          SCREAMING_SNAKE_CASE_      :			Any     =       np.interp(lowercase__				,     np.arange(0				,     len(lowercase__			)			)				,     lowercase__			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       np.concatenate([sigmas, [0.0]]			).astype(np.floataa			)
          SCREAMING_SNAKE_CASE_      :			Any     =       torch.from_numpy(lowercase__			).to(device=lowercase__			)
          if str(lowercase__			).startswith("mps"			):
                # mps does not support float64
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       torch.from_numpy(lowercase__			).to(lowercase__				,     dtype=torch.floataa			)
          else:
                SCREAMING_SNAKE_CASE_      :			Tuple     =       torch.from_numpy(lowercase__			).to(device=lowercase__			)




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__				,     lowercase__				,     lowercase__ = None				,     lowercase__ = True				,     ):






          """simple docstring"""






          if (
              isinstance(lowercase__				,     lowercase__			)
              or isinstance(lowercase__				,     torch.IntTensor			)
              or isinstance(lowercase__				,     torch.LongTensor			)
          ):
                raise ValueError(
                    (
                        "Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to"
                        " `EulerDiscreteScheduler.step()` is not supported. Make sure to pass"
                        " one of the `scheduler.timesteps` as a timestep."
                    )				,     )

          if not self.is_scale_input_called:
                logger.warning(
                    "The `scale_model_input` function should be called before `step` to ensure correct denoising. "
                    "See `StableDiffusionPipeline` for a usage example."			)

          if isinstance(lowercase__				,     torch.Tensor			):
                SCREAMING_SNAKE_CASE_      :			Tuple     =       timestep.to(self.timesteps.device			)

          SCREAMING_SNAKE_CASE_      :			Tuple     =       (self.timesteps == timestep).nonzero().item()
          SCREAMING_SNAKE_CASE_      :			str     =       self.sigmas[step_index]

          # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
          if self.config.prediction_type == "epsilon":
                SCREAMING_SNAKE_CASE_      :			Any     =       sample - sigma * model_output
          elif self.config.prediction_type == "v_prediction":
                # * c_out + input * c_skip
                SCREAMING_SNAKE_CASE_      :			Any     =       model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
          elif self.config.prediction_type == "sample":
                raise NotImplementedError("prediction_type not implemented yet: sample"			)
          else:
                raise ValueError(
                    F"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"			)

          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.sigmas[step_index]
          SCREAMING_SNAKE_CASE_      :			Any     =       self.sigmas[step_index + 1]
          SCREAMING_SNAKE_CASE_      :			Any     =       (sigma_to**2 * (sigma_from**2 - sigma_to**2) / sigma_from**2) ** 0.5
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       (sigma_to**2 - sigma_up**2) ** 0.5

          # 2. Convert to an ODE derivative
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       (sample - pred_original_sample) / sigma

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       sigma_down - sigma

          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       sample + derivative * dt

          SCREAMING_SNAKE_CASE_      :			Dict     =       model_output.device
          SCREAMING_SNAKE_CASE_      :			int     =       randn_tensor(model_output.shape				,     dtype=model_output.dtype				,     device=lowercase__				,     generator=lowercase__			)

          SCREAMING_SNAKE_CASE_      :			List[Any]     =       prev_sample + noise * sigma_up

          if not return_dict:
                return (prev_sample,)

          return EulerAncestralDiscreteSchedulerOutput(
              prev_sample=lowercase__				,     pred_original_sample=lowercase__			)




    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__				,     lowercase__				,     ):






          """simple docstring"""






          SCREAMING_SNAKE_CASE_      :			str     =       self.sigmas.to(device=original_samples.device				,     dtype=original_samples.dtype			)
          if original_samples.device.type == "mps" and torch.is_floating_point(lowercase__			):
                # mps does not support float64
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.timesteps.to(original_samples.device				,     dtype=torch.floataa			)
                SCREAMING_SNAKE_CASE_      :			List[Any]     =       timesteps.to(original_samples.device				,     dtype=torch.floataa			)
          else:
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       self.timesteps.to(original_samples.device			)
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       timesteps.to(original_samples.device			)

          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       [(schedule_timesteps == t).nonzero().item() for t in timesteps]

          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       sigmas[step_indices].flatten()
          while len(sigma.shape			) < len(original_samples.shape			):
                SCREAMING_SNAKE_CASE_      :			int     =       sigma.unsqueeze(-1			)

          SCREAMING_SNAKE_CASE_      :			Dict     =       original_samples + noise * sigma
          return noisy_samples







    def __len__(					self			):






          """simple docstring"""






          return self.config.num_train_timesteps

