






"""simple docstring"""


from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


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

_UpperCamelCase    :   List[Any]   			=						'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")\n        >>> pipe_prior.to("cuda")\n        >>> prompt = "red cat, 4k photo"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")\n        >>> pipe.to("cuda")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save("cat.png")\n        ```\n'




def       snake_case  (A_				:Dict     ,      A_				:Optional[Any]     ,      A_				:Optional[Any]=8				):







  '''simple docstring'''

  a					:      List[str]    =       height // scale_factor**2
  if height % scale_factor**2 != 0:
    new_height += 1
  a					:      Optional[Any]    =       width // scale_factor**2
  if width % scale_factor**2 != 0:
    new_width += 1
  return new_height * scale_factor, new_width * scale_factor






class 						snake_case  (  snake_case_	):





    def __init__(							self     : Optional[int]    ,		A     : UNetaDConditionModel    ,		A     : DDPMScheduler    ,		A     : VQModel    ,		):


      '''simple docstring'''




      super().__init__()

      self.register_modules(
          unet=A    ,		scheduler=A    ,		movq=A    ,		)
      a					:      Optional[Any]    =       2 ** (len(self.movq.config.block_out_channels      ) - 1)





    def     lowerCamelCase__				(							self     : Optional[Any]    ,		A     : Tuple    ,		A     : int    ,		A     : Any    ,		A     : Optional[int]    ,		A     : Union[str, Any]    ,		A     : Optional[int]      ):


      '''simple docstring'''




      if latents is None:
        a					:      Optional[Any]    =       randn_tensor(A    ,		generator=A    ,		device=A    ,		dtype=A      )
      else:
        if latents.shape != shape:
          raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''      )
        a					:      int    =       latents.to(A      )

      a					:      Optional[Any]    =       latents * scheduler.init_noise_sigma
      return latents





    def     lowerCamelCase__				(							self     : List[Any]    ,		A     : List[str]=0      ):


      '''simple docstring'''




      if is_accelerate_available():
        from accelerate import cpu_offload
      else:
        raise ImportError('Please install accelerate via `pip install accelerate`'      )

      a					:      str    =       torch.device(F'''cuda:{gpu_id}'''      )

      a					:      Dict    =       [
          self.unet,
          self.movq,
      ]
      for cpu_offloaded_model in models:
        if cpu_offloaded_model is not None:
          cpu_offload(A    ,		A      )





    def     lowerCamelCase__				(							self     : Union[str, Any]    ,		A     : Any=0      ):


      '''simple docstring'''




      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					:      Optional[int]    =       torch.device(F'''cuda:{gpu_id}'''      )

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

      a					:      List[Any]    =       None
      for cpu_offloaded_model in [self.unet, self.movq]:
        a					:      Optional[int]    =       cpu_offload_with_hook(A    ,		A    ,		prev_module_hook=A      )

      # We'll offload the last model manually.
      a					:      Tuple    =       hook





    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def     lowerCamelCase__				(							self     : str      ):


      '''simple docstring'''




      if not hasattr(self.unet    ,		'_hf_hook'      ):
        return self.device
      for module in self.unet.modules():
        if (
            hasattr(A    ,		'_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





    @torch.no_grad()
    @replace_example_docstring(A      )
    def __call__(							self     : int    ,		A     : Union[torch.FloatTensor, List[torch.FloatTensor]]    ,		A     : Union[torch.FloatTensor, List[torch.FloatTensor]]    ,		A     : int = 5_1_2    ,		A     : int = 5_1_2    ,		A     : int = 1_0_0    ,		A     : float = 4.0    ,		A     : int = 1    ,		A     : Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,		A     : Optional[torch.FloatTensor] = None    ,		A     : Optional[str] = "pil"    ,		A     : bool = True    ,		):


      '''simple docstring'''




      a					:      Optional[Any]    =       self._execution_device

      a					:      str    =       guidance_scale > 1.0

      if isinstance(A    ,		A      ):
        a					:      Tuple    =       torch.cat(A    ,		dim=0      )
      a					:      str    =       image_embeds.shape[0] * num_images_per_prompt
      if isinstance(A    ,		A      ):
        a					:      Any    =       torch.cat(A    ,		dim=0      )

      if do_classifier_free_guidance:
        a					:      int    =       image_embeds.repeat_interleave(A    ,		dim=0      )
        a					:      str    =       negative_image_embeds.repeat_interleave(A    ,		dim=0      )

      a					:      str    =       torch.cat([negative_image_embeds, image_embeds]    ,		dim=0      ).to(dtype=self.unet.dtype    ,		device=A      )

      self.scheduler.set_timesteps(A    ,		device=A      )
      a					:      int    =       self.scheduler.timesteps

      a					:      Union[str, Any]    =       self.unet.config.in_channels

      a					:      Union[str, Any]    =       downscale_height_and_width(A    ,		A    ,		self.movq_scale_factor      )

      # create initial latent
      a					:      Tuple    =       self.prepare_latents(
          (batch_size, num_channels_latents, height, width)    ,		image_embeds.dtype    ,		A    ,		A    ,		A    ,		self.scheduler    ,		)

      for i, t in enumerate(self.progress_bar(A      )      ):
        # expand the latents if we are doing classifier free guidance
        a					:      List[str]    =       torch.cat([latents] * 2      ) if do_classifier_free_guidance else latents

        a					:      int    =       {'image_embeds': image_embeds}
        a					:      Dict    =       self.unet(
            sample=A    ,		timestep=A    ,		encoder_hidden_states=A    ,		added_cond_kwargs=A    ,		return_dict=A    ,		)[0]

        if do_classifier_free_guidance:
          a					:      Tuple    =       noise_pred.split(latents.shape[1]    ,		dim=1      )
          a					:      List[Any]    =       noise_pred.chunk(2      )
          a					:      Union[str, Any]    =       variance_pred.chunk(2      )
          a					:      Optional[int]    =       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
          a					:      Any    =       torch.cat([noise_pred, variance_pred_text]    ,		dim=1      )

        if not (
            hasattr(self.scheduler.config    ,		'variance_type'      )
            and self.scheduler.config.variance_type in ["learned", "learned_range"]
        ):
          a					:      List[str]    =       noise_pred.split(latents.shape[1]    ,		dim=1      )

        # compute the previous noisy sample x_t -> x_t-1
        a					:      Dict    =       self.scheduler.step(
            A    ,		A    ,		A    ,		generator=A    ,		)[0]
      # post-processing
      a					:      int    =       self.movq.decode(A    ,		force_not_quantize=A      )['sample']

      if output_type not in ["pt", "np", "pil"]:
        raise ValueError(F'''Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'''      )

      if output_type in ["np", "pil"]:
        a					:      Tuple    =       image * 0.5 + 0.5
        a					:      Union[str, Any]    =       image.clamp(0    ,		1      )
        a					:      List[Any]    =       image.cpu().permute(0    ,		2    ,		3    ,		1      ).float().numpy()

      if output_type == "pil":
        a					:      List[Any]    =       self.numpy_to_pil(A      )

      if not return_dict:
        return (image,)

      return ImagePipelineOutput(images=A      )












"""simple docstring"""


from __future__ import annotations

import math

import numpy as np
from numpy.linalg import norm




def       snake_case  (A_				:np.ndarray     ,      A_				:np.ndarray				):







  '''simple docstring'''

  return math.sqrt(sum(pow(a - b     ,      2				) for a, b in zip(A_     ,      A_				)				)				)




def       snake_case  (A_				:np.ndarray     ,      A_				:np.ndarray				):







  '''simple docstring'''

  if dataset.ndim != value_array.ndim:
    a					:      Optional[Any]    =       (
        'Wrong input data\'s dimensions... '
        f'''dataset : {dataset.ndim}, value_array : {value_array.ndim}'''
    )
    raise ValueError(A_				)

  try:
    if dataset.shape[1] != value_array.shape[1]:
      a					:      Optional[int]    =       (
          'Wrong input data\'s shape... '
          f'''dataset : {dataset.shape[1]}, value_array : {value_array.shape[1]}'''
      )
      raise ValueError(A_				)
  except IndexError:
    if dataset.ndim != value_array.ndim:
      raise TypeError('Wrong shape'				)

  if dataset.dtype != value_array.dtype:
    a					:      Optional[Any]    =       (
        'Input data have different datatype... '
        f'''dataset : {dataset.dtype}, value_array : {value_array.dtype}'''
    )
    raise TypeError(A_				)

  a					:      Tuple    =       []

  for value in value_array:
    a					:      List[Any]    =       euclidean(A_     ,      dataset[0]				)
    a					:      int    =       dataset[0].tolist()

    for dataset_value in dataset[1:]:
      a					:      Optional[int]    =       euclidean(A_     ,      A_				)

      if dist > temp_dist:
        a					:      List[str]    =       temp_dist
        a					:      int    =       dataset_value.tolist()

    answer.append([vector, dist]				)

  return answer




def       snake_case  (A_				:np.ndarray     ,      A_				:np.ndarray				):







  '''simple docstring'''

  return np.dot(A_     ,      A_				) / (norm(A_				) * norm(A_				))


if __name__ == "__main__":
   import doctest

   doctest.testmod()




