"""simple docstring"""


from typing import Callable, List, Optional, Union

import torch

from ...models import UNetaDModel
from ...schedulers import CMStochasticIterativeScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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


__A       			=      """
    Examples:
        ```py
        >>> import torch

        >>> from diffusers import ConsistencyModelPipeline

        >>> device = \"cuda\"
        >>> # Load the cd_imagenet64_l2 checkpoint.
        >>> model_id_or_path = \"openai/diffusers-cd_imagenet64_l2\"
        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
        >>> pipe.to(device)

        >>> # Onestep Sampling
        >>> image = pipe(num_inference_steps=1).images[0]
        >>> image.save(\"cd_imagenet64_l2_onestep_sample.png\")

        >>> # Onestep sampling, class-conditional image generation
        >>> # ImageNet-64 class label 145 corresponds to king penguins
        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
        >>> image.save(\"cd_imagenet64_l2_onestep_sample_penguin.png\")

        >>> # Multistep sampling, class-conditional image generation
        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
        >>> image.save(\"cd_imagenet64_l2_multistep_sample_penguin.png\")
        ```
"""






class 					a					(				A_							):



  def __init__(		self   :    Dict						,  lowerCamelCase_   :    UNetaDModel						,  lowerCamelCase_   :    CMStochasticIterativeScheduler     )      -> None:
      super().__init__()

      self.register_modules(
          unet=lowerCamelCase_						,  scheduler=lowerCamelCase_						,  )

      __a			      =						None



  def        lowerCAmelCase_				(		self   :    List[Any]						,  lowerCamelCase_   :    List[str]=0     )      -> Optional[int]:
      if is_accelerate_available() and is_accelerate_version(""">="""						,  """0.14.0"""     ):
          from accelerate import cpu_offload
      else:
          raise ImportError("""`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher"""     )

      __a			      =						torch.device(F"""cuda:{gpu_id}"""     )

      if self.device.type != "cpu":
          self.to("""cpu"""						,  silence_dtype_warnings=lowerCamelCase_     )
          torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

      for cpu_offloaded_model in [self.unet]:
          cpu_offload(lowerCamelCase_						,  lowerCamelCase_     )

      if self.safety_checker is not None:
          cpu_offload(self.safety_checker						,  execution_device=lowerCamelCase_						,  offload_buffers=lowerCamelCase_     )



  def        lowerCAmelCase_				(		self   :    Dict						,  lowerCamelCase_   :    Optional[int]=0     )      -> Dict:
      if is_accelerate_available() and is_accelerate_version(""">="""						,  """0.17.0.dev0"""     ):
          from accelerate import cpu_offload_with_hook
      else:
          raise ImportError("""`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."""     )

      __a			      =						torch.device(F"""cuda:{gpu_id}"""     )

      if self.device.type != "cpu":
          self.to("""cpu"""						,  silence_dtype_warnings=lowerCamelCase_     )
          torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

      __a			      =						None
      for cpu_offloaded_model in [self.unet]:
          __a      ,  __a			      =						cpu_offload_with_hook(lowerCamelCase_						,  lowerCamelCase_						,  prev_module_hook=lowerCamelCase_     )

      if self.safety_checker is not None:
          __a      ,  __a			      =						cpu_offload_with_hook(self.safety_checker						,  lowerCamelCase_						,  prev_module_hook=lowerCamelCase_     )

      # We'll offload the last model manually.
      __a			      =						hook



  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def        lowerCAmelCase_				(		self   :    List[str]     )      -> Dict:
      if not hasattr(self.unet						,  """_hf_hook"""     ):
          return self.device
      for module in self.unet.modules():
          if (
              hasattr(lowerCamelCase_						,  """_hf_hook"""     )
              and hasattr(module._hf_hook						,  """execution_device"""     )
              and module._hf_hook.execution_device is not None
          ):
              return torch.device(module._hf_hook.execution_device     )
      return self.device



  def        lowerCAmelCase_				(		self   :    Tuple						,  lowerCamelCase_   :    Optional[Any]						,  lowerCamelCase_   :    Dict						,  lowerCamelCase_   :    Dict						,  lowerCamelCase_   :    List[str]						,  lowerCamelCase_   :    Dict						,  lowerCamelCase_   :    Dict						,  lowerCamelCase_   :    int						,  lowerCamelCase_   :    Tuple=None     )      -> Union[str, Any]:
      __a			      =						(batch_size, num_channels, height, width)
      if isinstance(lowerCamelCase_						,  lowerCamelCase_     ) and len(lowerCamelCase_     ) != batch_size:
          raise ValueError(
              F"""You have passed a list of generators of length {len(lowerCamelCase_     )}, but requested an effective batch"""
              F""" size of {batch_size}. Make sure the batch size matches the length of the generators."""     )

      if latents is None:
          __a			      =						randn_tensor(lowerCamelCase_						,  generator=lowerCamelCase_						,  device=lowerCamelCase_						,  dtype=lowerCamelCase_     )
      else:
          __a			      =						latents.to(device=lowerCamelCase_						,  dtype=lowerCamelCase_     )

      # scale the initial noise by the standard deviation required by the scheduler
      __a			      =						latents * self.scheduler.init_noise_sigma
      return latents



  def        lowerCAmelCase_				(		self   :    Any						,  lowerCamelCase_   :    torch.FloatTensor						,  lowerCamelCase_   :    str = "pil"     )      -> Any:
      if output_type not in ["pt", "np", "pil"]:
          raise ValueError(
              F"""output_type={output_type} is not supported. Make sure to choose one of ['pt', 'np', or 'pil']"""     )

      # Equivalent to diffusers.VaeImageProcessor.denormalize
      __a			      =						(sample / 2 + 0.5).clamp(0						,  1     )
      if output_type == "pt":
          return sample

      # Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
      __a			      =						sample.cpu().permute(0						,  2						,  3						,  1     ).numpy()
      if output_type == "np":
          return sample

      # Output_type must be 'pil'
      __a			      =						self.numpy_to_pil(lowerCamelCase_     )
      return sample



  def        lowerCAmelCase_				(		self   :    List[Any]						,  lowerCamelCase_   :    Optional[int]						,  lowerCamelCase_   :    Any						,  lowerCamelCase_   :    List[str]=None     )      -> Dict:
      if self.unet.config.num_class_embeds is not None:
          if isinstance(lowerCamelCase_						,  lowerCamelCase_     ):
              __a			      =						torch.tensor(lowerCamelCase_						,  dtype=torch.int     )
          elif isinstance(lowerCamelCase_						,  lowerCamelCase_     ):
              assert batch_size == 1, "Batch size must be 1 if classes is an int"
              __a			      =						torch.tensor([class_labels]						,  dtype=torch.int     )
          elif class_labels is None:
              # Randomly generate batch_size class labels
              # TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
              __a			      =						torch.randint(0						,  self.unet.config.num_class_embeds						,  size=(batch_size,)     )
          __a			      =						class_labels.to(lowerCamelCase_     )
      else:
          __a			      =						None
      return class_labels



  def        lowerCAmelCase_				(		self   :    int						,  lowerCamelCase_   :    Optional[int]						,  lowerCamelCase_   :    Optional[Any]						,  lowerCamelCase_   :    Optional[Any]						,  lowerCamelCase_   :    str						,  lowerCamelCase_   :    List[Any]						,  lowerCamelCase_   :    Optional[Any]     )      -> Optional[Any]:
      if num_inference_steps is None and timesteps is None:
          raise ValueError("""Exactly one of `num_inference_steps` or `timesteps` must be supplied."""     )

      if num_inference_steps is not None and timesteps is not None:
          logger.warning(
              F"""Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;"""
              """ `timesteps` will be used over `num_inference_steps`."""     )

      if latents is not None:
          __a			      =						(batch_size, 3, img_size, img_size)
          if latents.shape != expected_shape:
              raise ValueError(F"""The shape of latents is {latents.shape} but is expected to be {expected_shape}."""     )

      if (callback_steps is None) or (
          callback_steps is not None and (not isinstance(lowerCamelCase_						,  lowerCamelCase_     ) or callback_steps <= 0)
      ):
          raise ValueError(
              F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              F""" {type(lowerCamelCase_     )}."""     )



  @torch.no_grad()
  @replace_example_docstring(lowerCamelCase_     )
  def __call__(		self   :    Optional[int]						,  lowerCamelCase_   :    int = 1						,  lowerCamelCase_   :    Optional[Union[torch.Tensor, List[int], int]] = None						,  lowerCamelCase_   :    int = 1						,  lowerCamelCase_   :    List[int] = None						,  lowerCamelCase_   :    Optional[Union[torch.Generator, List[torch.Generator]]] = None						,  lowerCamelCase_   :    Optional[torch.FloatTensor] = None						,  lowerCamelCase_   :    Optional[str] = "pil"						,  lowerCamelCase_   :    bool = True						,  lowerCamelCase_   :    Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,  lowerCamelCase_   :    int = 1						,  )      -> Dict:
      __a			      =						self.unet.config.sample_size
      __a			      =						self._execution_device

      # 1. Check inputs
      self.check_inputs(lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_     )

      # 2. Prepare image latents
      # Sample image latents x_0 ~ N(0, sigma_0^2 * I)
      __a			      =						self.prepare_latents(
          batch_size=lowerCamelCase_						,  num_channels=self.unet.config.in_channels						,  height=lowerCamelCase_						,  width=lowerCamelCase_						,  dtype=self.unet.dtype						,  device=lowerCamelCase_						,  generator=lowerCamelCase_						,  latents=lowerCamelCase_						,  )

      # 3. Handle class_labels for class-conditional models
      __a			      =						self.prepare_class_labels(lowerCamelCase_						,  lowerCamelCase_						,  class_labels=lowerCamelCase_     )

      # 4. Prepare timesteps
      if timesteps is not None:
          self.scheduler.set_timesteps(timesteps=lowerCamelCase_						,  device=lowerCamelCase_     )
          __a			      =						self.scheduler.timesteps
          __a			      =						len(lowerCamelCase_     )
      else:
          self.scheduler.set_timesteps(lowerCamelCase_     )
          __a			      =						self.scheduler.timesteps

      # 5. Denoising loop
      # Multistep sampling: implements Algorithm 1 in the paper
      with self.progress_bar(total=lowerCamelCase_     ) as progress_bar:
          for i, t in enumerate(lowerCamelCase_     ):
              __a			      =						self.scheduler.scale_model_input(lowerCamelCase_						,  lowerCamelCase_     )
              __a			      =						self.unet(lowerCamelCase_						,  lowerCamelCase_						,  class_labels=lowerCamelCase_						,  return_dict=lowerCamelCase_     )[0]

              __a			      =						self.scheduler.step(lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_						,  generator=lowerCamelCase_     )[0]

              # call the callback, if provided
              progress_bar.update()
              if callback is not None and i % callback_steps == 0:
                  callback(lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_     )

        # 6. Post-process image sample
      __a			      =						self.postprocess_image(lowerCamelCase_						,  output_type=lowerCamelCase_     )

      # Offload last model to CPU
      if hasattr(self						,  """final_offload_hook"""     ) and self.final_offload_hook is not None:
          self.final_offload_hook.offload()

      if not return_dict:
          return (image,)

      return ImagePipelineOutput(images=lowerCamelCase_     )

"""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 					a					:
  A_ :				CommonSchedulerState
  A_ :				jnp.ndarray

  # setable values
  A_ :				jnp.ndarray
  A_ :				jnp.ndarray
  A_ :				Optional[int] 			=							None



  @classmethod
  def        lowerCAmelCase_				(		cls   :    Optional[int]						,  lowerCamelCase_   :    CommonSchedulerState						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  )      -> Optional[Any]:
      return cls(
          common=lowerCamelCase_						,  final_alpha_cumprod=lowerCamelCase_						,  init_noise_sigma=lowerCamelCase_						,  timesteps=lowerCamelCase_						,  )






@dataclass
class 					a					(				A_							):
  A_ :				DDIMSchedulerState






class 					a					(				A_   ,		A_							):
  A_ :				Optional[int] 			=							[e.name for e in FlaxKarrasDiffusionSchedulers]

  A_ :				jnp.dtype



  @property
  def        lowerCAmelCase_				(		self   :    Optional[int]     )      -> Optional[Any]:
      return True



  @register_to_config
  def __init__(		self   :    Optional[int]						,  lowerCamelCase_   :    int = 10_00						,  lowerCamelCase_   :    float = 0.00_01						,  lowerCamelCase_   :    float = 0.02						,  lowerCamelCase_   :    str = "linear"						,  lowerCamelCase_   :    Optional[jnp.ndarray] = None						,  lowerCamelCase_   :    bool = True						,  lowerCamelCase_   :    int = 0						,  lowerCamelCase_   :    str = "epsilon"						,  lowerCamelCase_   :    jnp.dtype = jnp.floataa						,  )      -> Any:
      __a			      =						dtype



  def        lowerCAmelCase_				(		self   :    Dict						,  lowerCamelCase_   :    Optional[CommonSchedulerState] = None     )      -> DDIMSchedulerState:
      if common is None:
          __a			      =						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.
      __a			      =						(
          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
      __a			      =						jnp.array(1.0						,  dtype=self.dtype     )

      __a			      =						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        lowerCAmelCase_				(		self   :    str						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    Optional[int] = None     )      -> jnp.ndarray:
      return sample



  def        lowerCAmelCase_				(		self   :    List[str]						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    int						,  lowerCamelCase_   :    Tuple = ()     )      -> DDIMSchedulerState:
      __a			      =						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
      __a			      =						(jnp.arange(0						,  lowerCamelCase_     ) * step_ratio).round()[::-1] + self.config.steps_offset

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



  def        lowerCAmelCase_				(		self   :    int						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    int						,  lowerCamelCase_   :    Any     )      -> str:
      __a			      =						state.common.alphas_cumprod[timestep]
      __a			      =						jnp.where(
          prev_timestep >= 0						,  state.common.alphas_cumprod[prev_timestep]						,  state.final_alpha_cumprod     )
      __a			      =						1 - alpha_prod_t
      __a			      =						1 - alpha_prod_t_prev

      __a			      =						(beta_prod_t_prev / beta_prod_t) * (1 - alpha_prod_t / alpha_prod_t_prev)

      return variance



  def        lowerCAmelCase_				(		self   :    List[str]						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    int						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    float = 0.0						,  lowerCamelCase_   :    bool = True						,  )      -> Union[FlaxDDIMSchedulerOutput, Tuple]:
      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)
      __a			      =						timestep - self.config.num_train_timesteps // state.num_inference_steps

      __a			      =						state.common.alphas_cumprod
      __a			      =						state.final_alpha_cumprod

      # 2. compute alphas, betas
      __a			      =						alphas_cumprod[timestep]
      __a			      =						jnp.where(prev_timestep >= 0						,  alphas_cumprod[prev_timestep]						,  lowerCamelCase_     )

      __a			      =						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":
          __a			      =						(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
          __a			      =						model_output
      elif self.config.prediction_type == "sample":
          __a			      =						model_output
          __a			      =						(sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
      elif self.config.prediction_type == "v_prediction":
          __a			      =						(alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
          __a			      =						(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)
      __a			      =						self._get_variance(lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_     )
      __a			      =						eta * variance ** 0.5

      # 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
      __a			      =						(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
      __a			      =						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        lowerCAmelCase_				(		self   :    str						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  )      -> jnp.ndarray:
      return add_noise_common(state.common						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_     )



  def        lowerCAmelCase_				(		self   :    Any						,  lowerCamelCase_   :    DDIMSchedulerState						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  lowerCamelCase_   :    jnp.ndarray						,  )      -> jnp.ndarray:
      return get_velocity_common(state.common						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_     )



  def __len__(		self   :    Tuple     )      -> Union[str, Any]:
      return self.config.num_train_timesteps
