





"""simple docstring"""





# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion

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

import flax
import jax.numpy as jnp

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils_flax import (
    CommonSchedulerState,
    FlaxKarrasDiffusionSchedulers,
    FlaxSchedulerMixin,
    FlaxSchedulerOutput,
    add_noise_common,
    get_velocity_common,
)





@flax.struct.dataclass
class 			snake_case		:
      lowerCamelCase__						= 42
      lowerCamelCase__						= 42

      # setable values
      lowerCamelCase__						= 42
      lowerCamelCase__						= 42
      lowerCamelCase__						= None

      @classmethod
      def 						SCREAMING_SNAKE_CASE_			(			cls   :Tuple						,							_lowerCamelCase   :CommonSchedulerState						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							):
       return cls(
           common=_lowerCamelCase						,							final_alpha_cumprod=_lowerCamelCase						,							init_noise_sigma=_lowerCamelCase						,							timesteps=_lowerCamelCase						,							)







@dataclass
class 			snake_case		(   __UpperCAmelCase   ):
      lowerCamelCase__						= 42







class 			snake_case		(   __UpperCAmelCase ,     __UpperCAmelCase   ):
      lowerCamelCase__						= [e.name for e in FlaxKarrasDiffusionSchedulers]

      lowerCamelCase__						= 42

      @property
      def 						SCREAMING_SNAKE_CASE_			(			self   :int						):
       return True

      @register_to_config
      def __init__(			self   :List[Any]						,							_lowerCamelCase   :int = 1_0_0_0						,							_lowerCamelCase   :float = 0.0_0_0_1						,							_lowerCamelCase   :float = 0.0_2						,							_lowerCamelCase   :str = "linear"						,							_lowerCamelCase   :Optional[jnp.ndarray] = None						,							_lowerCamelCase   :bool = True						,							_lowerCamelCase   :int = 0						,							_lowerCamelCase   :str = "epsilon"						,							_lowerCamelCase   :jnp.dtype = jnp.floataa						,							):
       __SCREAMING_SNAKE_CASE :			Dict												=       dtype

      def 						SCREAMING_SNAKE_CASE_			(			self   :List[str]						,							_lowerCamelCase   :Optional[CommonSchedulerState] = None						):
       if common is None:
        __SCREAMING_SNAKE_CASE :			int												=       CommonSchedulerState.create(self						)

       # At every step in ddim, we are looking into the previous alphas_cumprod
       # For the final step, there is no previous alphas_cumprod because we are already at 0
       # `set_alpha_to_one` decides whether we set this parameter simply to one or
       # whether we use the final alpha of the "non-previous" one.
       __SCREAMING_SNAKE_CASE :			int												=       (
           jnp.array(1.0						,							dtype=self.dtype						) if self.config.set_alpha_to_one else common.alphas_cumprod[0]
       )

       # standard deviation of the initial noise distribution
       __SCREAMING_SNAKE_CASE :			int												=       jnp.array(1.0						,							dtype=self.dtype						)

       __SCREAMING_SNAKE_CASE :			Optional[Any]												=       jnp.arange(0						,							self.config.num_train_timesteps						).round()[::-1]

       return DDIMSchedulerState.create(
           common=_lowerCamelCase						,							final_alpha_cumprod=_lowerCamelCase						,							init_noise_sigma=_lowerCamelCase						,							timesteps=_lowerCamelCase						,							)

      def 						SCREAMING_SNAKE_CASE_			(			self   :Union[str, Any]						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :Optional[int] = None						):
       return sample

      def 						SCREAMING_SNAKE_CASE_			(			self   :Optional[int]						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :int						,							_lowerCamelCase   :Tuple = ()						):
       __SCREAMING_SNAKE_CASE :			int												=       self.config.num_train_timesteps // num_inference_steps
       # creates integer timesteps by multiplying by ratio
       # rounding to avoid issues when num_inference_step is power of 3
       __SCREAMING_SNAKE_CASE :			List[Any]												=       (jnp.arange(0						,							_lowerCamelCase						) * step_ratio).round()[::-1] + self.config.steps_offset

       return state.replace(
           num_inference_steps=_lowerCamelCase						,							timesteps=_lowerCamelCase						,							)

      def 						SCREAMING_SNAKE_CASE_			(			self   :Tuple						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :Tuple						,							_lowerCamelCase   :int						):
       __SCREAMING_SNAKE_CASE :			int												=       state.common.alphas_cumprod[timestep]
       __SCREAMING_SNAKE_CASE :			int												=       jnp.where(
           prev_timestep >= 0						,							state.common.alphas_cumprod[prev_timestep]						,							state.final_alpha_cumprod						)
       __SCREAMING_SNAKE_CASE :			Any												=       1 - alpha_prod_t
       __SCREAMING_SNAKE_CASE :			Any												=       1 - alpha_prod_t_prev

       __SCREAMING_SNAKE_CASE :			Union[str, Any]												=       (beta_prod_t_prev / beta_prod_t) * (1 - alpha_prod_t / alpha_prod_t_prev)

       return variance

      def 						SCREAMING_SNAKE_CASE_			(			self   :str						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :int						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :float = 0.0						,							_lowerCamelCase   :bool = True						,							):
       if state.num_inference_steps is None:
        raise ValueError(
            '''Number of inference steps is \'None\', you need to run \'set_timesteps\' after creating the scheduler'''						)

       # See formulas (12) and (16) of DDIM paper https://arxiv.org/pdf/2010.02502.pdf
       # Ideally, read DDIM paper in-detail understanding

       # Notation (<variable name> -> <name in paper>
       # - pred_noise_t -> e_theta(x_t, t)
       # - pred_original_sample -> f_theta(x_t, t) or x_0
       # - std_dev_t -> sigma_t
       # - eta -> η
       # - pred_sample_direction -> "direction pointing to x_t"
       # - pred_prev_sample -> "x_t-1"

       # 1. get previous step value (=t-1)
       __SCREAMING_SNAKE_CASE :			Any												=       timestep - self.config.num_train_timesteps // state.num_inference_steps

       __SCREAMING_SNAKE_CASE :			int												=       state.common.alphas_cumprod
       __SCREAMING_SNAKE_CASE :			List[str]												=       state.final_alpha_cumprod

       # 2. compute alphas, betas
       __SCREAMING_SNAKE_CASE :			Dict												=       alphas_cumprod[timestep]
       __SCREAMING_SNAKE_CASE :			Tuple												=       jnp.where(prev_timestep >= 0						,							alphas_cumprod[prev_timestep]						,							_lowerCamelCase						)

       __SCREAMING_SNAKE_CASE :			int												=       1 - alpha_prod_t

       # 3. compute predicted original sample from predicted noise also called
       # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
       if self.config.prediction_type == "epsilon":
        __SCREAMING_SNAKE_CASE :			int												=       (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
        __SCREAMING_SNAKE_CASE :			Dict												=       model_output
       elif self.config.prediction_type == "sample":
        __SCREAMING_SNAKE_CASE :			List[Any]												=       model_output
        __SCREAMING_SNAKE_CASE :			Union[str, Any]												=       (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
       elif self.config.prediction_type == "v_prediction":
        __SCREAMING_SNAKE_CASE :			Any												=       (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
        __SCREAMING_SNAKE_CASE :			List[Any]												=       (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
       else:
        raise ValueError(
            f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or'''
            ''' `v_prediction`'''						)

       # 4. compute variance: "sigma_t(η)" -> see formula (16)
       # σ_t = sqrt((1 − α_t−1)/(1 − α_t)) * sqrt(1 − α_t/α_t−1)
       __SCREAMING_SNAKE_CASE :			str												=       self._get_variance(_lowerCamelCase						,							_lowerCamelCase						,							_lowerCamelCase						)
       __SCREAMING_SNAKE_CASE :			Optional[Any]												=       eta * variance ** 0.5

       # 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
       __SCREAMING_SNAKE_CASE :			Any												=       (1 - alpha_prod_t_prev - std_dev_t**2) ** 0.5 * pred_epsilon

       # 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
       __SCREAMING_SNAKE_CASE :			int												=       alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

       if not return_dict:
        return (prev_sample, state)

       return FlaxDDIMSchedulerOutput(prev_sample=_lowerCamelCase						,							state=_lowerCamelCase						)

      def 						SCREAMING_SNAKE_CASE_			(			self   :Dict						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							):
       return add_noise_common(state.common						,							_lowerCamelCase						,							_lowerCamelCase						,							_lowerCamelCase						)

      def 						SCREAMING_SNAKE_CASE_			(			self   :List[str]						,							_lowerCamelCase   :DDIMSchedulerState						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							_lowerCamelCase   :jnp.ndarray						,							):
       return get_velocity_common(state.common						,							_lowerCamelCase						,							_lowerCamelCase						,							_lowerCamelCase						)



      def __len__(			self   :Tuple						):
       return self.config.num_train_timesteps









"""simple docstring"""







from typing import Optional

import torch
from PIL import Image
from tqdm.auto import tqdm
from transformers import CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, DiffusionPipeline, UNetaDConditionModel
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import (
    deprecate,
)





class      UpperCamelCase_       (				a_     ):



  def __init__(					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__ = 0.93       ,						snake_case__ = True       ,						)	->	Optional[Any]:



         """simple docstring"""





         UpperCAmelCase					=	mixing_coeff
         UpperCAmelCase					=	leapfrog_steps

         super().__init__()
         self.register_modules(
             vae=snake_case__       ,						text_encoder=snake_case__       ,						tokenizer=snake_case__       ,						unet=snake_case__       ,						scheduler=snake_case__       ,						)

         UpperCAmelCase					=	2 ** (len(self.vae.config.block_out_channels ) - 1)
         UpperCAmelCase					=	VaeImageProcessor(vae_scale_factor=self.vae_scale_factor )



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ = None       ,						snake_case__ = False )	->	Dict:



         """simple docstring"""





         UpperCAmelCase					=	self.tokenizer(
             snake_case__       ,						padding="""max_length"""       ,						max_length=self.tokenizer.model_max_length       ,						truncation=snake_case__       ,						return_tensors="""pt"""       ,						)

         UpperCAmelCase					=	self.text_encoder(text_inputs.input_ids.to(self.device ) ).last_hidden_state

         UpperCAmelCase					=	prompt_embeds.to(dtype=self.text_encoder.dtype       ,						device=self.device )

         if do_classifier_free_guidance:
                UpperCAmelCase					=	"""""" if negative_prompt is None else negative_prompt

                UpperCAmelCase					=	self.tokenizer(
                    snake_case__       ,						padding="""max_length"""       ,						max_length=self.tokenizer.model_max_length       ,						truncation=snake_case__       ,						return_tensors="""pt"""       ,						)

                UpperCAmelCase					=	self.text_encoder(uncond_input.input_ids.to(self.device ) ).last_hidden_state

                UpperCAmelCase					=	torch.cat([negative_prompt_embeds, prompt_embeds] )

         return prompt_embeds



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ )	->	List[Any]:



         """simple docstring"""





         UpperCAmelCase					=	self.mixing_coeff * x + (1 - self.mixing_coeff) * y
         UpperCAmelCase					=	self.mixing_coeff * y + (1 - self.mixing_coeff) * x

         return [x, y]



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ )	->	Dict:



         """simple docstring"""





         UpperCAmelCase					=	(y - (1 - self.mixing_coeff) * x) / self.mixing_coeff
         UpperCAmelCase					=	(x - (1 - self.mixing_coeff) * y) / self.mixing_coeff

         return [x, y]



  def 						UpperCamelCase_ (					self       ,						snake_case__ )	->	Dict:



         """simple docstring"""





         UpperCAmelCase					=	int(snake_case__ )

         UpperCAmelCase					=	self.scheduler.alphas_cumprod[t] if t >= 0 else self.scheduler.final_alpha_cumprod

         return alpha_prod, 1 - alpha_prod



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						)	->	int:



         """simple docstring"""





         UpperCAmelCase					=	timestep - self.scheduler.config.num_train_timesteps / self.scheduler.num_inference_steps

         UpperCAmelCase			,				UpperCAmelCase					=	self._get_alpha_and_beta(snake_case__ )
         UpperCAmelCase			,				UpperCAmelCase					=	self._get_alpha_and_beta(snake_case__ )

         UpperCAmelCase					=	(alpha_prod_t_prev / alpha_prod_t) ** 0.5
         UpperCAmelCase					=	-a_t * (beta_prod_t**0.5) + beta_prod_t_prev**0.5

         UpperCAmelCase					=	(base - b_t * model_output) / a_t

         return model_input, next_model_input.to(base.dtype )



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						)	->	str:



         """simple docstring"""





         UpperCAmelCase					=	timestep - self.scheduler.config.num_train_timesteps / self.scheduler.num_inference_steps

         UpperCAmelCase			,				UpperCAmelCase					=	self._get_alpha_and_beta(snake_case__ )
         UpperCAmelCase			,				UpperCAmelCase					=	self._get_alpha_and_beta(snake_case__ )

         UpperCAmelCase					=	(alpha_prod_t_prev / alpha_prod_t) ** 0.5
         UpperCAmelCase					=	-a_t * (beta_prod_t**0.5) + beta_prod_t_prev**0.5
         UpperCAmelCase					=	a_t * base + b_t * model_output

         return model_input, next_model_input.to(base.dtype )



  @torch.no_grad()
  def 						UpperCamelCase_ (					self       ,						snake_case__ )	->	str:



         """simple docstring"""





         UpperCAmelCase					=	1 / self.vae.config.scaling_factor * latents
         UpperCAmelCase					=	self.vae.decode(snake_case__ ).sample
         UpperCAmelCase					=	(image / 2 + 0.5).clamp(0       ,						1 )
         return image



  @torch.no_grad()
  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__ = None       ,						)	->	List[Any]:



         """simple docstring"""





         UpperCAmelCase					=	guidance_scale > 1.0

         UpperCAmelCase					=	image.to(device=self.device       ,						dtype=text_embeds.dtype )
         UpperCAmelCase					=	self.vae.encode(snake_case__ ).latent_dist.sample(snake_case__ )

         UpperCAmelCase					=	self.vae.config.scaling_factor * latent

         UpperCAmelCase					=	[latent.clone(), latent.clone()]

         for i, t in tqdm(enumerate(snake_case__ )       ,						total=len(snake_case__ ) ):
                UpperCAmelCase					=	self.noise_mixing_layer(x=coupled_latents[0]       ,						y=coupled_latents[1] )

                # j - model_input index, k - base index
                for j in range(2 ):
                       UpperCAmelCase					=	j ^ 1

                       if self.leapfrog_steps:
                              if i % 2 == 0:
                                     UpperCAmelCase			,				UpperCAmelCase					=	j, k

                       UpperCAmelCase					=	coupled_latents[j]
                       UpperCAmelCase					=	coupled_latents[k]

                       UpperCAmelCase					=	torch.cat([model_input] * 2 ) if do_classifier_free_guidance else model_input

                       UpperCAmelCase					=	self.unet(snake_case__       ,						snake_case__       ,						encoder_hidden_states=snake_case__ ).sample

                       if do_classifier_free_guidance:
                              UpperCAmelCase			,				UpperCAmelCase					=	noise_pred.chunk(2 )
                              UpperCAmelCase					=	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                       UpperCAmelCase			,				UpperCAmelCase					=	self.noise_step(
                           base=snake_case__       ,						model_input=snake_case__       ,						model_output=snake_case__       ,						timestep=snake_case__       ,						)

                       UpperCAmelCase					=	model_input

         return coupled_latents



  @torch.no_grad()
  def __call__(					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__ = 3.0       ,						snake_case__ = 50       ,						snake_case__ = 0.8       ,						snake_case__ = None       ,						snake_case__ = None       ,						snake_case__ = "pil"       ,						)	->	str:



         """simple docstring"""





         UpperCAmelCase					=	guidance_scale > 1.0

         UpperCAmelCase					=	self.image_processor.preprocess(snake_case__ )

         UpperCAmelCase					=	self._encode_prompt(snake_case__       ,						snake_case__       ,						snake_case__ )
         UpperCAmelCase					=	self._encode_prompt(snake_case__       ,						snake_case__       ,						snake_case__ )

         self.scheduler.set_timesteps(snake_case__       ,						self.device )

         UpperCAmelCase					=	num_inference_steps - int(num_inference_steps * strength )
         UpperCAmelCase					=	self.scheduler.timesteps[t_limit:]
         UpperCAmelCase					=	fwd_timesteps.flip(0 )

         UpperCAmelCase					=	self.prepare_latents(snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__ )

         for i, t in tqdm(enumerate(snake_case__ )       ,						total=len(snake_case__ ) ):
                # j - model_input index, k - base index
                for k in range(2 ):
                       UpperCAmelCase					=	k ^ 1

                       if self.leapfrog_steps:
                              if i % 2 == 1:
                                     UpperCAmelCase			,				UpperCAmelCase					=	j, k

                       UpperCAmelCase					=	coupled_latents[j]
                       UpperCAmelCase					=	coupled_latents[k]

                       UpperCAmelCase					=	torch.cat([model_input] * 2 ) if do_classifier_free_guidance else model_input

                       UpperCAmelCase					=	self.unet(snake_case__       ,						snake_case__       ,						encoder_hidden_states=snake_case__ ).sample

                       if do_classifier_free_guidance:
                              UpperCAmelCase			,				UpperCAmelCase					=	noise_pred.chunk(2 )
                              UpperCAmelCase					=	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                       UpperCAmelCase			,				UpperCAmelCase					=	self.denoise_step(
                           base=snake_case__       ,						model_input=snake_case__       ,						model_output=snake_case__       ,						timestep=snake_case__       ,						)

                       UpperCAmelCase					=	model_input

                UpperCAmelCase					=	self.denoise_mixing_layer(x=coupled_latents[0]       ,						y=coupled_latents[1] )

         # either one is fine
         UpperCAmelCase					=	coupled_latents[0]

         if output_type not in ["latent", "pt", "np", "pil"]:
                UpperCAmelCase					=	(
                    f'''the output_type {output_type} is outdated. Please make sure to set it to one of these instead: '''
                    """`pil`, `np`, `pt`, `latent`"""
                )
                deprecate("""Unsupported output_type"""       ,						"""1.0.0"""       ,						snake_case__       ,						standard_warn=snake_case__ )
                UpperCAmelCase					=	"""np"""

         if output_type == "latent":
                UpperCAmelCase					=	final_latent
         else:
                UpperCAmelCase					=	self.decode_latents(snake_case__ )
                UpperCAmelCase					=	self.image_processor.postprocess(snake_case__       ,						output_type=snake_case__ )

         return image
