




'''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


a_   :	str							   = logging.get_logger(__name__)  # pylint: disable=invalid-name


@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerDiscrete
class 						__UpperCamelCase (				lowerCamelCase__       ):
       lowercase			:					torch.FloatTensor
       lowercase			:					Optional[torch.FloatTensor]   		=None





def   a_							(   __snake_case							:    int     , __snake_case							:    Any=0.9_9_9     , __snake_case							:    List[Any]="cosine"     , )							->		List[str]:



       """simple docstring"""
       if alpha_transform_type == "cosine":

              def alpha_bar_fn(__snake_case							:    Any     ):
                     return math.cos((t + 0.0_0_8) / 1.0_0_8 * math.pi / 2     ) ** 2

       elif alpha_transform_type == "exp":

              def alpha_bar_fn(__snake_case							:    int     ):
                     return math.exp(t * -12.0     )

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

       lowerCamelCase_							=[]
       for i in range(__snake_case     ):
              lowerCamelCase_							=i / num_diffusion_timesteps
              lowerCamelCase_							=(i + 1) / num_diffusion_timesteps
              betas.append(min(1 - alpha_bar_fn(__snake_case     ) / alpha_bar_fn(__snake_case     )     , __snake_case     )     )
       return torch.tensor(__snake_case     , dtype=torch.floataa     )


class 						__UpperCamelCase (				lowerCamelCase__    ,      lowerCamelCase__       ):
       lowercase			:					Union[str, Any]   		=[e.name for e in KarrasDiffusionSchedulers]
       lowercase			:					List[str]   		=1






       @register_to_config
       def __init__(							self,	lowerCAmelCase = 1_000,	lowerCAmelCase = 0.0_0_0_1,	lowerCAmelCase = 0.0_2,	lowerCAmelCase = "linear",	lowerCAmelCase = None,	lowerCAmelCase = "epsilon",	lowerCAmelCase = "linear",	lowerCAmelCase = False,	lowerCAmelCase = "linspace",	lowerCAmelCase = 0,	):



              """simple docstring"""







              if trained_betas is not None:
                     lowerCamelCase_							=torch.tensor(lowerCAmelCase,	dtype=torch.floataa						)
              elif beta_schedule == "linear":
                     lowerCamelCase_							=torch.linspace(lowerCAmelCase,	lowerCAmelCase,	lowerCAmelCase,	dtype=torch.floataa						)
              elif beta_schedule == "scaled_linear":
                     # this schedule is very specific to the latent diffusion model.
                     lowerCamelCase_							=(
                         torch.linspace(beta_start**0.5,	beta_end**0.5,	lowerCAmelCase,	dtype=torch.floataa						) ** 2
                     )
              elif beta_schedule == "squaredcos_cap_v2":
                     # Glide cosine schedule
                     lowerCamelCase_							=betas_for_alpha_bar(lowerCAmelCase						)
              else:
                     raise NotImplementedError(f'''{beta_schedule} does is not implemented for {self.__class__}'''						)

              lowerCamelCase_							=1.0 - self.betas
              lowerCamelCase_							=torch.cumprod(self.alphas,	dim=0						)

              lowerCamelCase_							=np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5						)
              lowerCamelCase_							=np.concatenate([sigmas[::-1], [0.0]]						).astype(np.floataa						)
              lowerCamelCase_							=torch.from_numpy(lowerCAmelCase						)

              # setable values
              lowerCamelCase_							=None
              lowerCamelCase_							=np.linspace(0,	num_train_timesteps - 1,	lowerCAmelCase,	dtype=lowerCAmelCase						)[::-1].copy()
              lowerCamelCase_							=torch.from_numpy(lowerCAmelCase						)
              lowerCamelCase_							=False
              lowerCamelCase_							=use_karras_sigmas






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







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

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






       def   lowercase__  (							self,	lowerCAmelCase,	lowerCAmelCase						):



              """simple docstring"""







              if isinstance(lowerCAmelCase,	torch.Tensor						):
                     lowerCamelCase_							=timestep.to(self.timesteps.device						)
              lowerCamelCase_							=(self.timesteps == timestep).nonzero().item()
              lowerCamelCase_							=self.sigmas[step_index]

              lowerCamelCase_							=sample / ((sigma**2 + 1) ** 0.5)

              lowerCamelCase_							=True
              return sample






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



              """simple docstring"""







              lowerCamelCase_							=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":
                     lowerCamelCase_							=np.linspace(0,	self.config.num_train_timesteps - 1,	lowerCAmelCase,	dtype=lowerCAmelCase						)[
                         ::-1
                     ].copy()
              elif self.config.timestep_spacing == "leading":
                     lowerCamelCase_							=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
                     lowerCamelCase_							=(np.arange(0,	lowerCAmelCase						) * step_ratio).round()[::-1].copy().astype(lowerCAmelCase						)
                     timesteps += self.config.steps_offset
              elif self.config.timestep_spacing == "trailing":
                     lowerCamelCase_							=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
                     lowerCamelCase_							=(np.arange(self.config.num_train_timesteps,	0,	-step_ratio						)).round().copy().astype(lowerCAmelCase						)
                     timesteps -= 1
              else:
                     raise ValueError(
                         f'''{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'.'''						)

              lowerCamelCase_							=np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5						)
              lowerCamelCase_							=np.log(lowerCAmelCase						)

              if self.config.interpolation_type == "linear":
                     lowerCamelCase_							=np.interp(lowerCAmelCase,	np.arange(0,	len(lowerCAmelCase						)						),	lowerCAmelCase						)
              elif self.config.interpolation_type == "log_linear":
                     lowerCamelCase_							=torch.linspace(np.log(sigmas[-1]						),	np.log(sigmas[0]						),	num_inference_steps + 1						).exp()
              else:
                     raise ValueError(
                         f'''{self.config.interpolation_type} is not implemented. Please specify interpolation_type to either'''
                         ''' \'linear\' or \'log_linear\''''						)

              if self.use_karras_sigmas:
                     lowerCamelCase_							=self._convert_to_karras(in_sigmas=lowerCAmelCase,	num_inference_steps=self.num_inference_steps						)
                     lowerCamelCase_							=np.array([self._sigma_to_t(lowerCAmelCase,	lowerCAmelCase						) for sigma in sigmas]						)

              lowerCamelCase_							=np.concatenate([sigmas, [0.0]]						).astype(np.floataa						)
              lowerCamelCase_							=torch.from_numpy(lowerCAmelCase						).to(device=lowerCAmelCase						)
              if str(lowerCAmelCase						).startswith('''mps'''						):
                     # mps does not support float64
                     lowerCamelCase_							=torch.from_numpy(lowerCAmelCase						).to(lowerCAmelCase,	dtype=torch.floataa						)
              else:
                     lowerCamelCase_							=torch.from_numpy(lowerCAmelCase						).to(device=lowerCAmelCase						)






       def   lowercase__  (							self,	lowerCAmelCase,	lowerCAmelCase						):



              """simple docstring"""







              lowerCamelCase_							=np.log(lowerCAmelCase						)

              # get distribution
              lowerCamelCase_							=log_sigma - log_sigmas[:, np.newaxis]

              # get sigmas range
              lowerCamelCase_							=np.cumsum((dists >= 0),	axis=0						).argmax(axis=0						).clip(max=log_sigmas.shape[0] - 2						)
              lowerCamelCase_							=low_idx + 1

              lowerCamelCase_							=log_sigmas[low_idx]
              lowerCamelCase_							=log_sigmas[high_idx]

              # interpolate sigmas
              lowerCamelCase_							=(low - log_sigma) / (low - high)
              lowerCamelCase_							=np.clip(lowerCAmelCase,	0,	1						)

              # transform interpolation to time range
              lowerCamelCase_							=(1 - w) * low_idx + w * high_idx
              lowerCamelCase_							=t.reshape(sigma.shape						)
              return t






       def   lowercase__  (							self,	lowerCAmelCase,	lowerCAmelCase						):



              """simple docstring"""







              lowerCamelCase_							=in_sigmas[-1].item()
              lowerCamelCase_							=in_sigmas[0].item()

              lowerCamelCase_							=7.0  # 7.0 is the value used in the paper
              lowerCamelCase_							=np.linspace(0,	1,	lowerCAmelCase						)
              lowerCamelCase_							=sigma_min ** (1 / rho)
              lowerCamelCase_							=sigma_max ** (1 / rho)
              lowerCamelCase_							=(max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
              return sigmas






       def   lowercase__  (							self,	lowerCAmelCase,	lowerCAmelCase,	lowerCAmelCase,	lowerCAmelCase = 0.0,	lowerCAmelCase = 0.0,	lowerCAmelCase = float('''inf'''						),	lowerCAmelCase = 1.0,	lowerCAmelCase = None,	lowerCAmelCase = True,	):



              """simple docstring"""







              if (
                  isinstance(lowerCAmelCase,	lowerCAmelCase						)
                  or isinstance(lowerCAmelCase,	torch.IntTensor						)
                  or isinstance(lowerCAmelCase,	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(lowerCAmelCase,	torch.Tensor						):
                     lowerCamelCase_							=timestep.to(self.timesteps.device						)

              lowerCamelCase_							=(self.timesteps == timestep).nonzero().item()
              lowerCamelCase_							=self.sigmas[step_index]

              lowerCamelCase_							=min(s_churn / (len(self.sigmas						) - 1),	2**0.5 - 1						) if s_tmin <= sigma <= s_tmax else 0.0

              lowerCamelCase_							=randn_tensor(
                  model_output.shape,	dtype=model_output.dtype,	device=model_output.device,	generator=lowerCAmelCase						)

              lowerCamelCase_							=noise * s_noise
              lowerCamelCase_							=sigma * (gamma + 1)

              if gamma > 0:
                     lowerCamelCase_							=sample + eps * (sigma_hat**2 - sigma**2) ** 0.5

              # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
              # NOTE: "original_sample" should not be an expected prediction_type but is left in for
              # backwards compatibility
              if self.config.prediction_type == "original_sample" or self.config.prediction_type == "sample":
                     lowerCamelCase_							=model_output
              elif self.config.prediction_type == "epsilon":
                     lowerCamelCase_							=sample - sigma_hat * model_output
              elif self.config.prediction_type == "v_prediction":
                     # * c_out + input * c_skip
                     lowerCamelCase_							=model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
              else:
                     raise ValueError(
                         f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`'''						)

              # 2. Convert to an ODE derivative
              lowerCamelCase_							=(sample - pred_original_sample) / sigma_hat

              lowerCamelCase_							=self.sigmas[step_index + 1] - sigma_hat

              lowerCamelCase_							=sample + derivative * dt

              if not return_dict:
                     return (prev_sample,)

              return EulerDiscreteSchedulerOutput(prev_sample=lowerCAmelCase,	pred_original_sample=lowerCAmelCase						)






       def   lowercase__  (							self,	lowerCAmelCase,	lowerCAmelCase,	lowerCAmelCase,	):



              """simple docstring"""







              lowerCamelCase_							=self.sigmas.to(device=original_samples.device,	dtype=original_samples.dtype						)
              if original_samples.device.type == "mps" and torch.is_floating_point(lowerCAmelCase						):
                     # mps does not support float64
                     lowerCamelCase_							=self.timesteps.to(original_samples.device,	dtype=torch.floataa						)
                     lowerCamelCase_							=timesteps.to(original_samples.device,	dtype=torch.floataa						)
              else:
                     lowerCamelCase_							=self.timesteps.to(original_samples.device						)
                     lowerCamelCase_							=timesteps.to(original_samples.device						)

              lowerCamelCase_							=[(schedule_timesteps == t).nonzero().item() for t in timesteps]

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

              lowerCamelCase_							=original_samples + noise * sigma
              return noisy_samples






       def __len__(							self						):



              """simple docstring"""







              return self.config.num_train_timesteps










'''simple docstring'''




import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()


class 						__UpperCamelCase (				lowerCamelCase__    ,      unittest.TestCase       ):
       lowercase			:					Tuple   		=KandinskyPipeline
       lowercase			:					Optional[int]   		=[
           'prompt',
           'image_embeds',
           'negative_image_embeds',
       ]
       lowercase			:					Union[str, Any]   		=['prompt', 'negative_prompt', 'image_embeds', 'negative_image_embeds']
       lowercase			:					Dict   		=[
           'generator',
           'height',
           'width',
           'latents',
           'guidance_scale',
           'negative_prompt',
           'num_inference_steps',
           'return_dict',
           'guidance_scale',
           'num_images_per_prompt',
           'output_type',
           'return_dict',
       ]
       lowercase			:					Dict   		=False






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return 32






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return 32






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return self.time_input_dim






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return self.time_input_dim * 4






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return 100






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              lowerCamelCase_							=XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''						)
              return tokenizer






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              torch.manual_seed(0						)
              lowerCamelCase_							=MCLIPConfig(
                  numDims=self.cross_attention_dim,	transformerDimensions=self.text_embedder_hidden_size,	hidden_size=self.text_embedder_hidden_size,	intermediate_size=37,	num_attention_heads=4,	num_hidden_layers=5,	vocab_size=1_005,	)

              lowerCamelCase_							=MultilingualCLIP(lowerCAmelCase						)
              lowerCamelCase_							=text_encoder.eval()

              return text_encoder






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              torch.manual_seed(0						)

              lowerCamelCase_							={
                  '''in_channels''': 4,
                  # Out channels is double in channels because predicts mean and variance
                  '''out_channels''': 8,
                  '''addition_embed_type''': '''text_image''',
                  '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
                  '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
                  '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
                  '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
                  '''layers_per_block''': 1,
                  '''encoder_hid_dim''': self.text_embedder_hidden_size,
                  '''encoder_hid_dim_type''': '''text_image_proj''',
                  '''cross_attention_dim''': self.cross_attention_dim,
                  '''attention_head_dim''': 4,
                  '''resnet_time_scale_shift''': '''scale_shift''',
                  '''class_embed_type''': None,
              }

              lowerCamelCase_							=UNetaDConditionModel(**lowerCAmelCase						)
              return model






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              return {
                  "block_out_channels": [32, 64],
                  "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
                  "in_channels": 3,
                  "latent_channels": 4,
                  "layers_per_block": 1,
                  "norm_num_groups": 8,
                  "norm_type": "spatial",
                  "num_vq_embeddings": 12,
                  "out_channels": 3,
                  "up_block_types": [
                      "AttnUpDecoderBlock2D",
                      "UpDecoderBlock2D",
                  ],
                  "vq_embed_dim": 4,
              }






       @property
       def   lowercase__  (							self						):



              """simple docstring"""







              torch.manual_seed(0						)
              lowerCamelCase_							=VQModel(**self.dummy_movq_kwargs						)
              return model






       def   lowercase__  (							self						):



              """simple docstring"""







              lowerCamelCase_							=self.dummy_text_encoder
              lowerCamelCase_							=self.dummy_tokenizer
              lowerCamelCase_							=self.dummy_unet
              lowerCamelCase_							=self.dummy_movq

              lowerCamelCase_							=DDIMScheduler(
                  num_train_timesteps=1_000,	beta_schedule='''linear''',	beta_start=0.0_0_0_8_5,	beta_end=0.0_1_2,	clip_sample=lowerCAmelCase,	set_alpha_to_one=lowerCAmelCase,	steps_offset=1,	prediction_type='''epsilon''',	thresholding=lowerCAmelCase,	)

              lowerCamelCase_							={
                  '''text_encoder''': text_encoder,
                  '''tokenizer''': tokenizer,
                  '''unet''': unet,
                  '''scheduler''': scheduler,
                  '''movq''': movq,
              }
              return components






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



              """simple docstring"""







              lowerCamelCase_							=floats_tensor((1, self.cross_attention_dim),	rng=random.Random(lowerCAmelCase						)						).to(lowerCAmelCase						)
              lowerCamelCase_							=floats_tensor((1, self.cross_attention_dim),	rng=random.Random(seed + 1						)						).to(lowerCAmelCase						)
              if str(lowerCAmelCase						).startswith('''mps'''						):
                     lowerCamelCase_							=torch.manual_seed(lowerCAmelCase						)
              else:
                     lowerCamelCase_							=torch.Generator(device=lowerCAmelCase						).manual_seed(lowerCAmelCase						)
              lowerCamelCase_							={
                  '''prompt''': '''horse''',
                  '''image_embeds''': image_embeds,
                  '''negative_image_embeds''': negative_image_embeds,
                  '''generator''': generator,
                  '''height''': 64,
                  '''width''': 64,
                  '''guidance_scale''': 4.0,
                  '''num_inference_steps''': 2,
                  '''output_type''': '''np''',
              }
              return inputs






       def   lowercase__  (							self						):



              """simple docstring"""







              lowerCamelCase_							='''cpu'''

              lowerCamelCase_							=self.get_dummy_components()

              lowerCamelCase_							=self.pipeline_class(**lowerCAmelCase						)
              lowerCamelCase_							=pipe.to(lowerCAmelCase						)

              pipe.set_progress_bar_config(disable=lowerCAmelCase						)

              lowerCamelCase_							=pipe(**self.get_dummy_inputs(lowerCAmelCase						)						)
              lowerCamelCase_							=output.images

              lowerCamelCase_							=pipe(
                  **self.get_dummy_inputs(lowerCAmelCase						),	return_dict=lowerCAmelCase,	)[0]

              lowerCamelCase_							=image[0, -3:, -3:, -1]
              lowerCamelCase_							=image_from_tuple[0, -3:, -3:, -1]

              assert image.shape == (1, 64, 64, 3)

              lowerCamelCase_							=np.array(
                  [0.3_2_8_6_6_3, 1.0, 0.2_3_2_1_6_8_7_3, 1.0, 0.9_2_7_1_7_5_6_4, 0.4_6_3_9_0_4_6, 0.9_6_8_9_4_7_7_7, 0.3_1_7_1_3_3_7_8, 0.6_2_9_3_9_5_3]						)

              assert (
                  np.abs(image_slice.flatten() - expected_slice						).max() < 1e-2
              ), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''
              assert (
                  np.abs(image_from_tuple_slice.flatten() - expected_slice						).max() < 1e-2
              ), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''





@slow
@require_torch_gpu
class 						__UpperCamelCase (				unittest.TestCase       ):






       def   lowercase__  (							self						):



              """simple docstring"""







              super().tearDown()
              gc.collect()
              torch.cuda.empty_cache()






       def   lowercase__  (							self						):



              """simple docstring"""







              lowerCamelCase_							=load_numpy(
                  '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
                  '''/kandinsky/kandinsky_text2img_cat_fp16.npy'''						)

              lowerCamelCase_							=KandinskyPriorPipeline.from_pretrained(
                  '''kandinsky-community/kandinsky-2-1-prior''',	torch_dtype=torch.floataa						)
              pipe_prior.to(lowerCAmelCase						)

              lowerCamelCase_							=KandinskyPipeline.from_pretrained('''kandinsky-community/kandinsky-2-1''',	torch_dtype=torch.floataa						)
              lowerCamelCase_							=pipeline.to(lowerCAmelCase						)
              pipeline.set_progress_bar_config(disable=lowerCAmelCase						)

              lowerCamelCase_							='''red cat, 4k photo'''

              lowerCamelCase_							=torch.Generator(device='''cuda'''						).manual_seed(0						)
              lowerCamelCase_,   lowerCamelCase_							=pipe_prior(
                  lowerCAmelCase,	generator=lowerCAmelCase,	num_inference_steps=5,	negative_prompt='''''',	).to_tuple()

              lowerCamelCase_							=torch.Generator(device='''cuda'''						).manual_seed(0						)
              lowerCamelCase_							=pipeline(
                  lowerCAmelCase,	image_embeds=lowerCAmelCase,	negative_image_embeds=lowerCAmelCase,	generator=lowerCAmelCase,	num_inference_steps=100,	output_type='''np''',	)

              lowerCamelCase_							=output.images[0]

              assert image.shape == (512, 512, 3)

              assert_mean_pixel_difference(lowerCAmelCase,	lowerCAmelCase						)




